import time import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from app.stream_lifecycle import RoleRunHandle, make_stream_failure from app.stream_manager import StreamGroup, StreamManager def _wait_until(predicate, timeout=1.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if predicate(): return True time.sleep(0.01) return bool(predicate()) def _profile(roles=("det",)): return SimpleNamespace( roles=roles, resolved_models={role: f"model-{role}" for role in roles}, score_threshold=0.5, models={role: f"model-{role}" for role in roles}, degraded_roles=(), class_min_confidence={}, scenarios={}, ) class StreamGroupLifecycleTest(unittest.TestCase): def _group(self, roles=("det",)): group = StreamGroup({ "stream_id": "test-stream", "source_type": "rtsp", "uri": "rtsp://camera.invalid/live", "store": False, "scenarios": "off", "active_learning": False, "inference_profile": {"roles": list(roles)}, }) return group def _patch_setup(self, roles): return ( patch("app.stream_manager.resolve_source", return_value={ "uri": "rtsp://camera.invalid/live", "source_type": "rtsp", "resolved_from": None, }), patch("app.stream_manager.resolve_from_spec", return_value=_profile(roles)), patch("app.stream_manager.render_cfg", side_effect=lambda _sid, role, _model: f"/{role}.txt"), ) def test_group_reaches_running_only_after_required_first_frames(self): roles = ("det", "pose") group = self._group(roles) handles = {} def fake_run_pipeline(role, *_args, run_generation=0, on_role_update=None, **_kwargs): handle = RoleRunHandle(role, run_generation, on_role_update) handles[role] = handle handle.mark_initialized() return handle patches = self._patch_setup(roles) with patches[0], patches[1], patches[2], patch( "app.stream_manager.run_pipeline", side_effect=fake_run_pipeline ): generation = group.start() self.assertTrue(_wait_until(lambda: len(handles) == 2)) self.assertEqual("starting", group.observed_state) handles["det"].mark_frame(detection_count=0) self.assertEqual("starting", group.observed_state) handles["pose"].mark_frame(detection_count=0) self.assertTrue(_wait_until(lambda: group.observed_state == "running")) self.assertEqual(generation, group.run_generation) group.stop() self.assertTrue(group._cleanup_done.wait(2)) self.assertEqual("stopped", group.observed_state) self.assertIsNone(group.collector) self.assertIsNone(group.tracker) def test_worker_init_failure_propagates_to_group_error(self): group = self._group() def failing_run_pipeline(role, *_args, run_generation=0, on_role_update=None, **_kwargs): handle = RoleRunHandle(role, run_generation, on_role_update) handle.mark_failure(make_stream_failure( "PIPELINE_INIT_FAILED", "custom_processor_load", role )) return handle patches = self._patch_setup(("det",)) with patches[0], patches[1], patches[2], patch( "app.stream_manager.run_pipeline", side_effect=failing_run_pipeline ): group.start() self.assertTrue(_wait_until(lambda: group.observed_state == "error")) self.assertEqual( "PIPELINE_INIT_FAILED", group.last_error["error_code"] ) self.assertEqual("custom_processor_load", group.last_error["failure_stage"]) def test_first_frame_timeout_propagates_to_group_error(self): group = self._group() def idle_run_pipeline(role, *_args, run_generation=0, on_role_update=None, **_kwargs): handle = RoleRunHandle(role, run_generation, on_role_update) handle.mark_initialized() return handle patches = self._patch_setup(("det",)) with patches[0], patches[1], patches[2], patch( "app.stream_manager.RtspSourceOptions.model_validate", return_value=SimpleNamespace(first_frame_timeout_s=0.05), ), patch("app.stream_manager.run_pipeline", side_effect=idle_run_pipeline): group.start() self.assertTrue(_wait_until(lambda: group.observed_state == "error")) self.assertEqual("FIRST_FRAME_TIMEOUT", group.last_error["error_code"]) def test_rtsp_options_are_passed_to_each_role_adapter(self): group = self._group() captured = {} class _Adapter: def __init__(self, uri, options, gpu_id): captured["uri"] = uri captured["options"] = options captured["gpu_id"] = gpu_id def fake_run_pipeline(role, *_args, run_generation=0, on_role_update=None, **kwargs): captured["adapter"] = kwargs["source_adapter"] handle = RoleRunHandle(role, run_generation, on_role_update) handle.mark_initialized() handle.mark_frame(detection_count=0) return handle patches = self._patch_setup(("det",)) with patches[0], patches[1], patches[2], patch( "app.stream_manager.RtspSourceAdapter", _Adapter ), patch("app.stream_manager.run_pipeline", side_effect=fake_run_pipeline): group.start() self.assertTrue(_wait_until(lambda: group.observed_state == "running")) self.assertEqual("rtsp://camera.invalid/live", captured["uri"]) self.assertEqual(60, captured["options"].first_frame_timeout_s) self.assertEqual(0, captured["gpu_id"]) self.assertIsInstance(captured["adapter"], _Adapter) group.stop() def test_process_load_keeps_even_legacy_autostart_stream_stopped(self): manager = StreamManager(base_path="/missing", overlay_path="/missing") spec = { "stream_id": "legacy-autostart", "source_type": "rtsp", "autostart": True, } with patch.object(manager, "effective_specs", return_value=[spec]), patch.object( StreamGroup, "start" ) as start: manager.load_and_autostart() start.assert_not_called() self.assertEqual("stopped", manager.get("legacy-autostart").state) def test_source_options_survive_runtime_overlay_round_trip(self): with tempfile.TemporaryDirectory() as root: base_path = Path(root) / "streams.yaml" overlay_path = Path(root) / "streams_runtime.yaml" base_path.write_text("streams: []\n", encoding="utf-8") manager = StreamManager( base_path=str(base_path), overlay_path=str(overlay_path) ) manager.add( { "stream_id": "overlay-test", "source_type": "rtsp", "uri": "rtsp://camera.invalid/live", "source_options": { "transport": "auto", "latency_ms": 1500, "disable_audio": True, "reconnect_policy": "none", "first_frame_timeout_s": 75, }, }, start=False, ) loaded = StreamManager( base_path=str(base_path), overlay_path=str(overlay_path) ).effective_specs() self.assertEqual(1, len(loaded)) self.assertEqual("auto", loaded[0]["source_options"]["transport"]) self.assertEqual(75, loaded[0]["source_options"]["first_frame_timeout_s"]) if __name__ == "__main__": unittest.main()