76 lines
3.5 KiB
Python
76 lines
3.5 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path, PureWindowsPath
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from routes.streams import _media_path, _normalize_chapters, _status_payload
|
|
|
|
|
|
class ReplayWorkflowTest(unittest.TestCase):
|
|
def test_media_path_accepts_nested_file_in_media_root(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
replay = Path(directory, "events", "session.mp4")
|
|
replay.parent.mkdir()
|
|
replay.write_bytes(b"video")
|
|
with patch("routes.streams.settings.replay_media_dir", directory):
|
|
self.assertEqual(_media_path("events/session.mp4"), replay.resolve())
|
|
|
|
def test_media_path_rejects_parent_traversal_and_absolute_paths(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with patch("routes.streams.settings.replay_media_dir", directory):
|
|
for unsafe in ("../secret.mp4", "/tmp/secret.mp4"):
|
|
with self.subTest(unsafe=unsafe), self.assertRaises(HTTPException):
|
|
_media_path(unsafe)
|
|
|
|
def test_media_path_rejects_windows_absolute_path_on_any_platform(self):
|
|
value = str(PureWindowsPath("C:/media/replay.mp4"))
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
with patch("routes.streams.settings.replay_media_dir", directory):
|
|
with self.assertRaises(HTTPException):
|
|
_media_path(value)
|
|
|
|
def test_chapters_are_trimmed_validated_and_sorted(self):
|
|
chapters = _normalize_chapters([
|
|
{"title": " Closing ", "time": "62.5"},
|
|
{"title": "Opening", "time": 0},
|
|
])
|
|
self.assertEqual(chapters, [
|
|
{"title": "Opening", "time": 0.0},
|
|
{"title": "Closing", "time": 62.5},
|
|
])
|
|
|
|
def test_chapters_reject_invalid_title_and_time(self):
|
|
for chapters in ([{"title": "", "time": 1}], [{"title": "Bad", "time": -1}]):
|
|
with self.subTest(chapters=chapters), self.assertRaises(HTTPException):
|
|
_normalize_chapters(chapters)
|
|
|
|
def test_status_payload_exposes_pending_live_and_ended_updates(self):
|
|
stream = SimpleNamespace(id=7, replay_file=None, replay_url=None, chapters_json="[]")
|
|
for status in ("pending", "live", "ended"):
|
|
stream.status = SimpleNamespace(value=status)
|
|
self.assertEqual(_status_payload(stream)["status"], status)
|
|
|
|
stream.replay_file = "session.mp4"
|
|
stream.chapters_json = json.dumps([{"title": "Intro", "time": 0}])
|
|
payload = _status_payload(stream)
|
|
self.assertEqual(payload["replay_local_url"], "/api/streams/7/replay")
|
|
self.assertIsNone(payload["replay_url"])
|
|
self.assertEqual(payload["chapters"], [{"title": "Intro", "time": 0}])
|
|
|
|
stream.replay_file = None
|
|
stream.replay_url = "https://example.com/vod/master.m3u8"
|
|
payload = _status_payload(stream)
|
|
self.assertIsNone(payload["replay_local_url"])
|
|
self.assertEqual(payload["replay_url"], "https://example.com/vod/master.m3u8")
|
|
|
|
def test_replay_url_rejects_non_http(self):
|
|
from routes.streams import _valid_replay_url
|
|
for unsafe in ("ftp://example.com/a.mp4", "file:///etc/passwd", "javascript:alert(1)"):
|
|
with self.subTest(unsafe=unsafe), self.assertRaises(HTTPException):
|
|
_valid_replay_url(unsafe)
|
|
self.assertEqual(_valid_replay_url(" https://cdn.example.com/a.mp4 "), "https://cdn.example.com/a.mp4")
|