import ast import hashlib import importlib import json import os import subprocess import sys import tempfile import unittest from unittest.mock import Mock, patch from pathlib import Path os.environ.setdefault("DATABASE_URL", "postgresql://u:p" + chr(64) + "localhost:1/db") os.environ.setdefault("REDIS_STREAMS_ENABLED", "false") from app import main as app_main from app.scenario_runtime.config_loader import load_scenario_config_api REPOSITORY_ROOT = Path(__file__).resolve().parents[1] # Two diagnostic POSTs replaced the retired debug JPEG snapshot GET (+1). EXPECTED_OPENAPI_OPERATION_COUNT = 27 EXPECTED_OPENAPI_OPERATION_SHA256 = "b56e01d302d00c00233cc4f1637ce326e4e226dcbd2c84614568f508d7cf6cb2" APPLICATION_MODULE_ROOTS = { path.stem for path in (REPOSITORY_ROOT / "app").glob("*.py") } | {"scenario_runtime", "scenarios", "source_adapters"} METADATA_ADDITIONS = {'/api/v1/streams/{stream_id}/settings', '/api/v1/metadata/status'} def _openapi_operation_fingerprint(application): operations = [] for path, path_item in sorted(application.openapi()["paths"].items()): if path in METADATA_ADDITIONS: continue for method, operation in sorted(path_item.items()): if method.lower() not in { "get", "post", "put", "patch", "delete", "options", "head", "trace", }: continue operations.append({ "path": path, "method": method.upper(), "operation_id": operation.get("operationId"), "responses": sorted(operation.get("responses", {})), }) payload = json.dumps(operations, sort_keys=True, separators=(",", ":")) return len(operations), hashlib.sha256(payload.encode()).hexdigest() class CompositionRootTest(unittest.TestCase): def test_diagnostic_operations_and_retired_snapshot_are_explicit(self): paths = app_main.create_app().openapi()['paths'] for path in ('/api/v1/streams/{stream_id}/metrics', '/api/v1/metadata/metrics'): self.assertEqual(set(paths[path]), {'post'}) self.assertNotIn('/api/v1/streams/{stream_id}/snapshot', paths) def test_metadata_additions_are_explicit_without_mounting_web_facade(self): paths = app_main.create_app().openapi()['paths'] self.assertEqual(set(paths['/api/v1/streams/{stream_id}/settings']), {'get', 'patch'}) self.assertEqual(set(paths['/api/v1/metadata/status']), {'get'}) self.assertFalse(any('/cameras' in path for path in paths)) def test_factory_builds_independent_complete_applications(self): first = app_main.create_app() second = app_main.create_app() self.assertIsNot(first, second) for application in (app_main.app, first, second): self.assertEqual(application.title, "Contextra Analytics API") self.assertEqual(application.version, "1.1.0") self.assertEqual( _openapi_operation_fingerprint(application), (EXPECTED_OPENAPI_OPERATION_COUNT, EXPECTED_OPENAPI_OPERATION_SHA256), ) self.assertEqual( [handler.__name__ for handler in application.router.on_startup], [], ) self.assertEqual(application.router.on_shutdown, []) def test_process_globals_have_one_canonical_module_owner(self): inference_engine = importlib.import_module("app.inference_engine") metadata_storage = importlib.import_module("app.metadata_storage") self.assertIs(app_main.engine, inference_engine.engine) self.assertIs(app_main.storage, metadata_storage.storage) self.assertIs(app_main.redis_publisher, metadata_storage.redis_publisher) for bare_name in ( "inference_engine", "inference_config", "metadata_storage", "scenario_runtime", "stream_manager", ): self.assertNotIn(bare_name, sys.modules) def test_factory_creation_does_not_import_or_start_stream_manager(self): command = """ import os import sys os.environ.setdefault('DATABASE_URL', 'postgresql://u:p' + chr(64) + 'localhost:1/db') os.environ.setdefault('REDIS_STREAMS_ENABLED', 'false') import app.main as main assert 'app.stream_manager' not in sys.modules main.create_app() assert 'app.stream_manager' not in sys.modules """ completed = subprocess.run( [sys.executable, "-c", command], cwd=REPOSITORY_ROOT, env={**os.environ, "PYTHONPATH": str(REPOSITORY_ROOT)}, capture_output=True, text=True, timeout=30, ) self.assertEqual(completed.returncode, 0, completed.stderr) class LifespanTest(unittest.IsolatedAsyncioTestCase): async def test_actual_app_lifespan_starts_and_closes_once(self): from app.stream_manager import manager publisher = Mock() publisher.close.return_value = True with patch('app.metadata_runtime.publisher_from_environment', return_value=publisher), \ patch.object(manager, 'configure_metadata') as configure, \ patch.object(manager, 'load_and_autostart') as load: application = app_main.create_app() async with application.router.lifespan_context(application): configure.assert_called_once_with(publisher) load.assert_called_once_with() stopped, worker = application.state.retention self.assertTrue(worker.is_alive()) publisher.close.assert_not_called() self.assertTrue(stopped.is_set()) self.assertFalse(worker.is_alive()) publisher.close.assert_called_once_with(1) self.assertEqual(application.state.cleanup_errors, ()) async def test_partial_startup_failure_closes_acquired_publisher(self): from app.stream_manager import manager publisher = Mock() publisher.close.return_value = True with patch('app.metadata_runtime.publisher_from_environment', return_value=publisher), \ patch.object(manager, 'configure_metadata'), \ patch.object(manager, 'load_and_autostart'), \ patch.object(app_main, '_boot_retention', side_effect=ValueError('invalid retention')): application = app_main.create_app() with self.assertRaisesRegex(ValueError, 'invalid retention'): async with application.router.lifespan_context(application): self.fail('startup must fail') publisher.close.assert_called_once_with(1) async def test_duplicate_guard_failure_keeps_original_and_cleans_only_candidate(self): from app.stream_manager import manager publisher = Mock() publisher.close.side_effect = RuntimeError('secret cleanup detail') with patch('app.metadata_runtime.publisher_from_environment', return_value=publisher), \ patch.object(manager, 'configure_metadata', side_effect=RuntimeError('secret config detail')), \ patch.object(manager, 'close_metadata') as old_close, \ patch.object(app_main, '_boot_retention') as retention: application = app_main.create_app() with self.assertRaisesRegex(RuntimeError, '^Metadata composition failed.$'): async with application.router.lifespan_context(application): self.fail('startup must fail') retention.assert_not_called() old_close.assert_not_called() publisher.close.assert_called_once_with(1) self.assertEqual(application.state.cleanup_errors, ('METADATA_CLEANUP_FAILED',)) self.assertIs(application.state.metadata_publisher, publisher) async def test_cleanup_survivor_is_reported_without_skipping_publisher(self): from app.stream_manager import manager publisher, stopped, worker = Mock(), Mock(), Mock() publisher.close.return_value = True worker.is_alive.return_value = True with patch('app.metadata_runtime.publisher_from_environment', return_value=publisher), \ patch.object(manager, 'configure_metadata'), \ patch.object(manager, 'load_and_autostart'), \ patch.object(app_main, '_boot_retention', return_value=(stopped, worker)): application = app_main.create_app() with self.assertRaisesRegex(RuntimeError, 'Application cleanup incomplete'): async with application.router.lifespan_context(application): pass stopped.set.assert_called_once_with() publisher.close.assert_called_once_with(1) self.assertEqual(application.state.cleanup_errors, ('RETENTION_CLEANUP_INCOMPLETE',)) class CanonicalImportTest(unittest.TestCase): def test_application_imports_are_canonical_and_do_not_mutate_sys_path(self): violations = [] for path in (REPOSITORY_ROOT / "app").rglob("*.py"): tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) for node in ast.walk(tree): if isinstance(node, ast.ImportFrom) and node.module: root = node.module.split(".", 1)[0] if root in APPLICATION_MODULE_ROOTS and root != "app": violations.append(f"{path.relative_to(REPOSITORY_ROOT)}:{node.lineno}:from {node.module}") elif isinstance(node, ast.Import): for alias in node.names: root = alias.name.split(".", 1)[0] if root in APPLICATION_MODULE_ROOTS and root != "app": violations.append(f"{path.relative_to(REPOSITORY_ROOT)}:{node.lineno}:import {alias.name}") elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): owner = node.func.value if ( isinstance(owner, ast.Attribute) and isinstance(owner.value, ast.Name) and owner.value.id == "sys" and owner.attr == "path" and node.func.attr in {"insert", "append"} ): violations.append(f"{path.relative_to(REPOSITORY_ROOT)}:{node.lineno}:sys.path") self.assertEqual(violations, []) def test_scenario_package_import_does_not_change_sys_path(self): before = list(sys.path) scenarios = importlib.import_module("app.scenarios") self.assertIsNotNone(scenarios) self.assertEqual(sys.path, before) self.assertNotIn("scenario_runtime", sys.modules) def test_container_pythonpath_points_to_package_parent(self): dockerfile = (REPOSITORY_ROOT / "docker/Dockerfile").read_text(encoding="utf-8") entrypoint = (REPOSITORY_ROOT / "docker/entrypoint.sh").read_text(encoding="utf-8") self.assertIn("PYTHONPATH=/opt/contextra:${PYTHONPATH}", dockerfile) self.assertIn("PYTHONPATH=/opt/contextra:${PYTHONPATH}", entrypoint) self.assertNotIn("PYTHONPATH=/opt/contextra/app", dockerfile) self.assertNotIn("PYTHONPATH=/opt/contextra/app", entrypoint) def test_module_entrypoint_passes_app_object_to_uvicorn(self): source = (REPOSITORY_ROOT / "app/main.py").read_text(encoding="utf-8") tree = ast.parse(source) calls = [ node for node in ast.walk(tree) if ( isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) and node.func.value.id == "uvicorn" and node.func.attr == "run" ) ] self.assertEqual(len(calls), 1) self.assertIsInstance(calls[0].args[0], ast.Name) self.assertEqual(calls[0].args[0].id, "app") class ScenarioConfigLoaderTest(unittest.TestCase): def test_explicit_loader_preserves_sys_path_and_contract(self): with tempfile.TemporaryDirectory() as directory: config = Path(directory) / "scenario_config.py" config.write_text( "def get_active_scenarios():\n" " return ['intrusion']\n\n" "def get_scenario_runtime_cfg(name):\n" " return {'name': name}\n", encoding="utf-8", ) before = list(sys.path) get_active, get_runtime = load_scenario_config_api([Path(directory)]) self.assertEqual(get_active(), ["intrusion"]) self.assertEqual(get_runtime("intrusion"), {"name": "intrusion"}) self.assertEqual(sys.path, before) self.assertNotIn("scenario_config", sys.modules) def test_missing_or_invalid_config_fails_explicitly(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) with self.assertRaises(ModuleNotFoundError): load_scenario_config_api([root]) (root / "scenario_config.py").write_text("VALUE = 1\n", encoding="utf-8") with self.assertRaisesRegex(ImportError, "contract is invalid"): load_scenario_config_api([root]) if __name__ == "__main__": unittest.main()