import ast import unittest from pathlib import Path from app.stream_lifecycle import ( RoleRunHandle, StreamLifecycle, classify_pipeline_failure, lifecycle_command_payload, make_stream_failure, stop_role_handles, ) class StreamLifecycleTest(unittest.TestCase): def test_required_roles_gate_running_until_every_first_frame(self): lifecycle = StreamLifecycle() generation = lifecycle.begin() self.assertTrue(lifecycle.configure_roles(generation, ("det", "pose"))) lifecycle.signal_initialized(generation, "det") lifecycle.signal_frame(generation, "det", 10.0) self.assertEqual("starting", lifecycle.snapshot()["observed_state"]) lifecycle.signal_frame(generation, "pose", 11.0) status = lifecycle.snapshot() self.assertEqual("running", status["observed_state"]) self.assertEqual("first_frame", status["roles"]["det"]["state"]) def test_optional_role_does_not_block_running(self): lifecycle = StreamLifecycle() generation = lifecycle.begin() lifecycle.configure_roles(generation, ("det",), ("seg",)) lifecycle.signal_error( generation, make_stream_failure("ROLE_RUNTIME_FAILED", "pipeline_runtime", "seg"), ) lifecycle.signal_frame(generation, "det") status = lifecycle.snapshot() self.assertEqual("running", status["observed_state"]) self.assertEqual("error", status["roles"]["seg"]["state"]) def test_init_failure_and_timeout_become_structured_error(self): lifecycle = StreamLifecycle() generation = lifecycle.begin() lifecycle.configure_roles(generation, ("det",)) failure = make_stream_failure( "PIPELINE_INIT_FAILED", "custom_processor_load", "det" ) lifecycle.signal_error(generation, failure) status = lifecycle.snapshot() self.assertEqual("error", status["observed_state"]) self.assertEqual("PIPELINE_INIT_FAILED", status["last_error"]["error_code"]) generation = lifecycle.begin() lifecycle.configure_roles(generation, ("det",)) timeout = make_stream_failure("FIRST_FRAME_TIMEOUT", "first_frame_wait", "det") lifecycle.signal_error(generation, timeout) self.assertEqual("FIRST_FRAME_TIMEOUT", lifecycle.snapshot()["last_error"]["error_code"]) def test_zero_detection_frame_is_a_valid_first_frame(self): lifecycle = StreamLifecycle() generation = lifecycle.begin() lifecycle.configure_roles(generation, ("det",)) def update(gen, role, event, payload): if event == "frame": return lifecycle.signal_frame(gen, role, payload) return True handle = RoleRunHandle("det", generation, update) handle.mark_frame(frame_time=12.5, detection_count=0) status = lifecycle.snapshot() self.assertEqual("running", status["observed_state"]) self.assertEqual(12.5, status["roles"]["det"]["last_frame_at"]) def test_source_stage_requires_current_generation(self): lifecycle = StreamLifecycle() old_generation = lifecycle.begin() lifecycle.configure_roles(old_generation, ("det",)) current_generation = lifecycle.begin() lifecycle.configure_roles(current_generation, ("det",)) self.assertFalse( lifecycle.signal_source_stage( old_generation, "det", "decoded_frame_received", 10.0 ) ) self.assertTrue( lifecycle.signal_source_stage( current_generation, "det", "decoded_frame_received", 11.0 ) ) status = lifecycle.snapshot()["roles"]["det"] self.assertEqual("decoded_frame_received", status["source_stage"]) self.assertEqual(11.0, status["source_stage_at"]) self.assertEqual({"decoded_frame_received": 11.0}, status["source_stages"]) self.assertTrue( lifecycle.signal_source_stage( current_generation, "det", "inference_frame_completed", 12.0 ) ) status = lifecycle.snapshot()["roles"]["det"] self.assertEqual("inference_frame_completed", status["source_stage"]) self.assertEqual( { "decoded_frame_received": 11.0, "inference_frame_completed": 12.0, }, status["source_stages"], ) self.assertFalse( lifecycle.signal_source_stage(current_generation, "unregistered", "rtp_packet_received") ) def test_stale_generation_signal_is_ignored(self): lifecycle = StreamLifecycle() old_generation = lifecycle.begin() lifecycle.configure_roles(old_generation, ("det",)) current_generation = lifecycle.begin() lifecycle.configure_roles(current_generation, ("det",)) self.assertFalse(lifecycle.signal_frame(old_generation, "det")) self.assertEqual("starting", lifecycle.snapshot()["observed_state"]) def test_stop_is_idempotent_and_preserves_previous_error(self): lifecycle = StreamLifecycle() generation = lifecycle.begin() lifecycle.configure_roles(generation, ("det",)) failure = make_stream_failure("ROLE_RUNTIME_FAILED", "tensor_parse", "det") lifecycle.signal_error(generation, failure) lifecycle.request_stop() lifecycle.finish_stop(generation) lifecycle.request_stop() lifecycle.finish_stop(generation) status = lifecycle.snapshot() self.assertEqual("stopped", status["observed_state"]) self.assertEqual("ROLE_RUNTIME_FAILED", status["last_error"]["error_code"]) def test_process_default_is_stopped(self): status = StreamLifecycle().snapshot() self.assertEqual("stopped", status["desired_state"]) self.assertEqual("stopped", status["observed_state"]) self.assertEqual(0, status["run_generation"]) class _Pipeline: def __init__(self, error=None): self.error = error self.stop_calls = 0 def stop(self): self.stop_calls += 1 if self.error: raise self.error class RoleRunHandleTest(unittest.TestCase): def test_sibling_cleanup_stops_every_role_once(self): det_pipeline = _Pipeline() pose_pipeline = _Pipeline() handles = { "det": RoleRunHandle("det", 1), "pose": RoleRunHandle("pose", 1), } handles["det"].bind_pipeline(det_pipeline) handles["pose"].bind_pipeline(pose_pipeline) self.assertIsNone(stop_role_handles(handles)) self.assertIsNone(stop_role_handles(handles)) self.assertEqual(1, det_pipeline.stop_calls) self.assertEqual(1, pose_pipeline.stop_calls) def test_cleanup_failure_is_sanitized(self): pipeline = _Pipeline(RuntimeError("sensitive-detail-should-not-escape")) handle = RoleRunHandle("det", 1) handle.bind_pipeline(pipeline) failure = stop_role_handles({"det": handle}) self.assertEqual("STOP_CLEANUP_FAILED", failure.error_code) self.assertNotIn("sensitive-detail", str(failure.as_dict())) def test_custom_processor_error_has_stable_stage(self): failure = classify_pipeline_failure( RuntimeError("failed to load libcontextra_nvinferserver_rfdetr.so"), "det", initialized=False, ) self.assertEqual("PIPELINE_INIT_FAILED", failure.error_code) self.assertEqual("custom_processor_load", failure.failure_stage) self.assertEqual("det", failure.role) class ApiCompatibilityTest(unittest.TestCase): def test_start_payload_keeps_state_alias(self): lifecycle = StreamLifecycle() lifecycle.begin() payload = lifecycle_command_payload("stream", lifecycle.snapshot()) self.assertEqual("starting", payload["state"]) self.assertEqual(payload["state"], payload["observed_state"]) self.assertEqual("running", payload["desired_state"]) self.assertEqual(1, payload["run_generation"]) def test_start_route_declares_http_202(self): tree = ast.parse(Path("app/api/streams.py").read_text(encoding="utf-8")) route = next( node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "start_stream" ) status_codes = [ keyword.value.value for decorator in route.decorator_list if isinstance(decorator, ast.Call) for keyword in decorator.keywords if keyword.arg == "status_code" and isinstance(keyword.value, ast.Constant) ] self.assertEqual([202], status_codes) if __name__ == "__main__": unittest.main()