import unittest import threading import time from types import SimpleNamespace from unittest.mock import Mock, patch from fastapi import FastAPI from fastapi.testclient import TestClient from app.api.errors import install_stream_error_handlers from app.api.web_cameras import StartRequest, build_web_router from app.application.streams import StreamApplicationService from app.application.camera_commands import CameraCommands from app.stream_config_store import ConfigPersistenceError from app.stream_lifecycle import StreamLifecycle class WebFacadeTest(unittest.TestCase): def setUp(self): self.manager = Mock() lifecycle = StreamLifecycle() lifecycle.begin() self.manager.apply_and_start.return_value = SimpleNamespace( stream_id="stream", lifecycle=lifecycle, ) self.service = StreamApplicationService(lambda: self.manager, "server") application = FastAPI() install_stream_error_handlers(application) self.commands = CameraCommands(self.service) self.addCleanup(self.commands.close) application.include_router(build_web_router(self.service, self.commands, response_budget=.02)) self.client = TestClient(application) self.uri = "rtsp://" + "private-user:private-pass@camera.invalid/live" def test_accepted_start_and_default_inactive(self): response = self.client.post("/api/v1/cameras/start", json={ "cameraId": "001", "rtspUrl": self.uri, }) self.assertEqual(response.status_code, 202) self.assertEqual(response.json()["cameraId"], "001") self.assertEqual(response.json()["status"], "STARTING") self.assertIsNone(response.json()["streamUrl"]) self.assertNotIn("private-pass", response.text) def test_validation_error_does_not_echo_input(self): response = self.client.post("/api/v1/cameras/start", json={ "cameraId": {"bad": self.uri}, "rtspUrl": self.uri, "unexpected": self.uri, }) self.assertEqual(response.status_code, 422) self.assertNotIn("private-pass", response.text) self.manager.apply_and_start.assert_not_called() def test_save_failure_is_503_and_sanitized(self): self.manager.apply_and_start.side_effect = ConfigPersistenceError() response = self.client.post("/api/v1/cameras/start", json={ "cameraId": 1, "rtspUrl": self.uri, }) self.assertEqual(response.status_code, 503) self.assertEqual(response.json()["detail"]["error_code"], "CONFIG_PERSISTENCE_FAILED") def test_unsupported_scenario_is_detection_only(self): response = self.client.post("/api/v1/cameras/start", json={ "cameraId": 1, "rtspUrl": self.uri, "scenarios": [{"scenarioCode": "FALL", "enabled": True}], }) self.assertEqual(response.status_code, 202) self.assertEqual(response.json()["effectiveScenarios"], []) self.assertEqual(response.json()["skippedScenarios"][0]["code"], "FALL") self.assertEqual(self.manager.apply_and_start.call_args.args[0]["scenarios"], "off") def test_anchor_alias_and_null(self): response = self.client.post("/api/v1/cameras/start", json={ "cameraId": "001", "rtspUrl": self.uri, "rec_camera_id": "ref", "startWallTimeMs": None, "rtpClockRate": None, }) self.assertEqual(response.status_code, 202) spec = self.manager.apply_and_start.call_args.args[0] self.assertEqual(spec["rec_camera_id"], "ref") self.assertIsNone(spec["time_anchor"]["start_wall_time_ms"]) def test_partial_supported_selection(self): self.service.supported_scenarios = frozenset({"INTRUSION"}) response = self.client.post("/api/v1/cameras/start", json={ "cameraId": "001", "rtspUrl": self.uri, "scenarios": [{"scenarioCode": "INTRUSION", "enabled": True}, {"scenarioCode": "FALL", "enabled": True}], }) self.assertEqual(response.status_code, 202) self.assertEqual(self.manager.apply_and_start.call_args.args[0]["scenarios"], "intrusion") def test_pending_http_dedupe_late_result_and_busy(self): gate = threading.Event() group = self.manager.apply_and_start.return_value self.manager.apply_and_start.side_effect = lambda spec: (gate.wait(1), group)[1] body = {"cameraId": "001", "rtspUrl": self.uri} try: before = time.monotonic() first = self.client.post("/api/v1/cameras/start", json=body) self.assertLess(time.monotonic() - before, .5) second = self.client.post("/api/v1/cameras/start", json=body) self.assertEqual(first.json()["operationId"], second.json()["operationId"]) self.assertIsNone(first.json()["observedState"]) busy = self.client.post("/api/v1/cameras/stop", json={"cameraId": "001"}) self.assertEqual(busy.status_code, 503) finally: gate.set() job = self.commands.get(first.json()["operationId"]) self.assertIs(job.future.result(1), group) self.assertEqual(self.manager.apply_and_start.call_count, 1) def test_running_response_and_actual_failure(self): group = self.manager.apply_and_start.return_value group.lifecycle.snapshot = Mock(return_value={ "observed_state": "running", "desired_state": "running", "run_id": "run", "run_generation": 1, "last_error": None, }) result = self.client.post("/api/v1/cameras/start", json={"cameraId": 1, "rtspUrl": self.uri}) self.assertEqual(result.status_code, 200) self.assertEqual(result.json()["result"], "SUCCESS") def test_worker_timeout_is_not_accepted(self): self.manager.apply_and_start.side_effect = TimeoutError("private detail") result = self.client.post("/api/v1/cameras/start", json={"cameraId": 1, "rtspUrl": self.uri}) self.assertEqual(result.status_code, 503) self.assertNotIn("private detail", result.text) def test_missing_stop_is_404(self): self.manager.stop.side_effect = KeyError("private detail") result = self.client.post("/api/v1/cameras/stop", json={"cameraId": 1}) self.assertEqual(result.status_code, 404) def test_secret_repr_and_bool_identity_rejection(self): request = StartRequest.model_validate({"cameraId": 1, "rtspUrl": self.uri}) self.assertNotIn("private-pass", repr(request)) response = self.client.post("/api/v1/cameras/start", json={ "cameraId": True, "rtspUrl": self.uri, }) self.assertEqual(response.status_code, 422) class LegacyResourceHttpTest(unittest.TestCase): def setUp(self): from app.main import create_app # No TestClient context: application startup must not run in HTTP unit tests. self.client = TestClient(create_app()) def test_start_keeps_202_state_alias_and_run_identity(self): lifecycle = StreamLifecycle() lifecycle.begin() with patch("app.api.streams.stream_service") as service: service.start.return_value = SimpleNamespace(lifecycle=lifecycle) response = self.client.post("/api/v1/streams/unit-stream/start") service.start.assert_called_once_with("unit-stream") self.assertEqual(response.status_code, 202) self.assertEqual(response.json()["state"], "starting") self.assertEqual(response.json()["state"], response.json()["observed_state"]) self.assertEqual(response.json()["run_id"], lifecycle.snapshot()["run_id"]) def test_storage_failure_returns_503_not_success(self): with patch("app.api.streams.stream_service") as service: service.add.side_effect = ConfigPersistenceError() response = self.client.post("/api/v1/streams", json={"stream_id": "unit-stream"}) self.assertEqual(response.status_code, 503) self.assertEqual(response.json()["detail"]["error_code"], "CONFIG_PERSISTENCE_FAILED") def test_draft_camera_routes_remain_unpublished(self): response = self.client.post("/api/v1/cameras/start", json={}) self.assertEqual(response.status_code, 404)