完善观众端目录与回放状态降级

backend/tests/test_replay_workflow.py

@@ -0,0 +1,61 @@
+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, 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_url"], "/api/streams/7/replay")
+        self.assertEqual(payload["chapters"], [{"title": "Intro", "time": 0}])