from __future__ import annotations import json import unittest from types import SimpleNamespace from unittest.mock import patch from httpx import ASGITransport, AsyncClient from app.config import config from app.core.heavy_resources import LazyResourceProxy from app.main import create_app from app.migrations.definitions import MIGRATIONS from app.migrations.runner import Migration, MigrationRunner class FakeCollection: def __init__(self) -> None: self.documents: list[dict] = [] self.indexes: list[tuple] = [] def create_index(self, keys, **kwargs): self.indexes.append((tuple(keys), tuple(sorted(kwargs.items())))) def find(self, *_args, **_kwargs): return list(self.documents) def insert_one(self, payload): self.documents.append(dict(payload)) class FakeDatabase(dict): def __getitem__(self, key): if key not in self: self[key] = FakeCollection() return dict.__getitem__(self, key) class FakeHeavyResources: def __init__(self, status_overrides: dict[str, dict[str, bool]] | None = None) -> None: self.status_overrides = status_overrides or {} self.describe_calls = 0 def describe_status(self) -> dict[str, dict[str, bool]]: self.describe_calls += 1 base = { "groq_provider": {"enabled": False, "configured": False, "loaded": False}, "gemini_provider": {"enabled": True, "configured": True, "loaded": False}, "openai_stub": {"enabled": False, "loaded": False}, "llm_router": {"loaded": False}, "cie10_resource": {"available": True, "loaded": False}, "cups_resource": {"available": True, "loaded": False}, "soat_resource": {"available": True, "loaded": False}, } for key, values in self.status_overrides.items(): base.setdefault(key, {}).update(values) return base class AppFactoryTest(unittest.IsolatedAsyncioTestCase): async def test_create_app_does_not_build_services_before_startup(self) -> None: with patch("app.main.build_services") as build_services_mock, patch("app.main.init_mongo_runtime") as init_mock: app = create_app() self.assertIsNotNone(app) build_services_mock.assert_not_called() init_mock.assert_not_called() async def test_health_and_ready_routes_have_distinct_semantics(self) -> None: app = create_app() app.state.startup_ready = True app.state.mongo_runtime = SimpleNamespace( ping=self._async_noop, sync_database=self._build_ready_database(), ) proxy_calls = {"count": 0} def _llm_router_loader(): proxy_calls["count"] += 1 return object() app.state.services = self._build_services( heavy_resources=FakeHeavyResources(), llm_router=LazyResourceProxy(_llm_router_loader, name="llm_router"), ) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as client: health = await client.get("/health") ready = await client.get("/ready") self.assertEqual(health.status_code, 200) health_payload = health.json() self.assertEqual(health_payload["status"], "ok") self.assertEqual(health_payload["summary"]["error"], 0) self.assertIn("services", health_payload) self.assertEqual(health_payload["services"]["mongo"]["status"], "ok") self.assertEqual(health_payload["services"]["batch_dispatcher"]["details"]["dispatcher_name"], "inprocess") self.assertEqual(health_payload["services"]["groq_provider"]["status"], "ok") self.assertEqual( health_payload["services"]["groq_provider"]["details"]["reason_code"], "compatibility_disabled", ) self.assertEqual(proxy_calls["count"], 0) self.assertEqual(ready.status_code, 200) self.assertEqual(ready.json()["status"], "ready") serialized = json.dumps(health_payload) self.assertNotIn(config.DB_NAME, serialized) self.assertNotIn(config.MONGO_URI, serialized) self.assertNotIn(config.REDIS_URL, serialized) async def test_health_reports_degraded_when_passive_dependencies_are_unavailable(self) -> None: app = create_app() app.state.startup_ready = True app.state.mongo_runtime = SimpleNamespace( ping=self._async_noop, sync_database=self._build_ready_database(), ) app.state.services = self._build_services( heavy_resources=FakeHeavyResources( { "gemini_provider": {"configured": False}, "soat_resource": {"available": False}, } ) ) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as client: response = await client.get("/health") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["status"], "degraded") self.assertEqual(payload["services"]["gemini_provider"]["status"], "degraded") self.assertEqual(payload["services"]["gemini_provider"]["details"]["reason_code"], "not_configured") self.assertEqual(payload["services"]["soat_resource"]["status"], "degraded") self.assertEqual(payload["services"]["soat_resource"]["details"]["reason_code"], "missing_index") async def test_health_reports_error_when_mongo_ping_fails_without_leaking_error_details(self) -> None: app = create_app() app.state.startup_ready = True app.state.mongo_runtime = SimpleNamespace( ping=self._async_ping_fail, sync_database=self._build_ready_database(), ) app.state.services = self._build_services(heavy_resources=FakeHeavyResources()) async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as client: response = await client.get("/health") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["status"], "error") self.assertEqual(payload["services"]["mongo"]["status"], "error") self.assertEqual(payload["services"]["mongo"]["details"]["reason_code"], "unreachable") serialized = json.dumps(payload) self.assertNotIn(config.MONGO_URI, serialized) self.assertNotIn("secret-db-name", serialized) async def test_health_reports_error_when_celery_dispatcher_cannot_reach_redis(self) -> None: app = create_app() app.state.startup_ready = True app.state.mongo_runtime = SimpleNamespace( ping=self._async_noop, sync_database=self._build_ready_database(), ) app.state.services = self._build_services( heavy_resources=FakeHeavyResources(), batch_runtime=SimpleNamespace(dispatcher_name="celery"), ) with patch( "app.core.health.ping_redis", side_effect=RuntimeError("redis://sensitive-broker:6379/0"), ): async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as client: response = await client.get("/health") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["status"], "error") self.assertEqual(payload["services"]["batch_dispatcher"]["status"], "error") self.assertEqual(payload["services"]["batch_dispatcher"]["details"]["reason_code"], "unreachable") self.assertNotIn(config.REDIS_URL, json.dumps(payload)) async def _async_noop(self) -> None: return None async def _async_ping_fail(self) -> None: raise RuntimeError(f"{config.MONGO_URI} secret-db-name") def _build_ready_database(self) -> FakeDatabase: database = FakeDatabase() database["schema_migrations"].documents = [ {"version": migration.version} for migration in MIGRATIONS ] return database def _build_services(self, **overrides): defaults = { "resources_status": lambda: {"cie10_loaded": False}, "heavy_resources": FakeHeavyResources(), "llm_router": object(), "batch_runtime": SimpleNamespace(dispatcher_name="inprocess"), "individual_ingestion_runtime": SimpleNamespace(dispatcher_name="inprocess"), "clinical_document_service": object(), "case_epicrisis_service": object(), "rda_service": object(), "rips_service": object(), } defaults.update(overrides) return SimpleNamespace(**defaults) class LazyProxyTest(unittest.TestCase): def test_lazy_proxy_loads_resource_once(self) -> None: calls = {"count": 0} class Resource: value = "ready" def loader(): calls["count"] += 1 return Resource() proxy = LazyResourceProxy(loader, name="test") self.assertEqual(proxy.value, "ready") self.assertEqual(proxy.value, "ready") self.assertEqual(calls["count"], 1) class MigrationRunnerTest(unittest.TestCase): def test_runner_is_idempotent(self) -> None: database = FakeDatabase() applied: list[str] = [] def _apply(_database): applied.append("20260507_test") runner = MigrationRunner( database, [Migration(version="20260507_test", description="test", apply=_apply)], ) first = runner.run() second = runner.run() self.assertEqual(first, ["20260507_test"]) self.assertEqual(second, []) self.assertEqual(applied, ["20260507_test"]) if __name__ == "__main__": unittest.main()