from __future__ import annotations import unittest from pathlib import Path from types import SimpleNamespace from typing import Any, cast from unittest.mock import patch from fastapi import HTTPException from app.config import config from app.core.gemini_client import GeminiModelAdapter from app.core.heavy_resources import HeavyResourceRegistry from app.core.mongo_runtime import MongoRuntime from app.core.services import build_services from app.llm import ( DefaultModelRouter, DefaultModelSelectionPolicy, GeminiAdapter, GroqAdapter, LLMErrorKind, LLMOutputKind, LLMProviderAvailability, LLMProviderCapabilities, LLMProviderError, LLMStructuredRequest, LLMTask, LLMTextRequest, OpenAIAdapterStub, ) from app.llm.models import LLMStructuredResult, LLMTextResult from app.llm.schemas import HistoriaClinicaStructured, LaboratorioStructured from app.services.llm_phase8_benchmark import build_phase8_benchmark_report, render_phase8_benchmark_markdown from modules.processing.cie10.RANGES_CUPS import CUPSRetriever from modules.processing.cie10.RANGES_HTML import CIE10Retriever from modules.processing.resumen_google import HistoriaClinicaRequest from modules.processing.soat.soat_retriever import SOATRetriever class DefaultModelSelectionPolicyTest(unittest.TestCase): def test_resolves_gemini_extract_for_clinical_document_task(self) -> None: route = DefaultModelSelectionPolicy().resolve(LLMTask.CLINICAL_DOCUMENT_EXTRACT) self.assertEqual(route.provider, "gemini") self.assertEqual(route.model, "gemini-2.5-flash-lite") self.assertEqual(route.output_kind, LLMOutputKind.STRUCTURED_OBJECT) self.assertIsNone(route.fallback_rule) def test_resolves_reasoning_route_without_legacy_fallback_for_soat_codes(self) -> None: route = DefaultModelSelectionPolicy().resolve(LLMTask.SOAT_CODE_GENERATION) self.assertEqual(route.provider, "gemini") self.assertEqual(route.model, "gemini-2.5-flash") self.assertEqual(route.output_kind, LLMOutputKind.STRUCTURED_COLLECTION) self.assertIsNone(route.fallback_rule) def test_resolves_gemini_primary_for_cie10_and_cups(self) -> None: route = DefaultModelSelectionPolicy().resolve(LLMTask.CIE10_RESOLUTION) cups_route = DefaultModelSelectionPolicy().resolve(LLMTask.CUPS_RESOLUTION) self.assertEqual(route.provider, "gemini") self.assertEqual(route.model, "gemini-2.5-flash-lite") self.assertEqual(route.output_kind, LLMOutputKind.CONTROLLED_PLAIN_TEXT) self.assertIsNone(route.fallback_rule) self.assertEqual(cups_route.provider, "gemini") self.assertEqual(cups_route.model, "gemini-2.5-flash-lite") def test_resolves_routes_for_soat_chat_and_manual_lookup(self) -> None: chat_route = DefaultModelSelectionPolicy().resolve(LLMTask.SOAT_CHAT) manual_route = DefaultModelSelectionPolicy().resolve(LLMTask.SOAT_MANUAL_CODE_LOOKUP) self.assertEqual(chat_route.provider, "gemini") self.assertEqual(chat_route.model, "gemini-2.5-flash") self.assertTrue(chat_route.metadata["long_context"]) self.assertEqual(manual_route.provider, "gemini") self.assertEqual(manual_route.model, "gemini-2.5-flash") self.assertEqual(manual_route.output_kind, LLMOutputKind.CONTROLLED_PLAIN_TEXT) def test_resolves_prefactura_page_classification_route_with_gemini_escalation(self) -> None: route = DefaultModelSelectionPolicy().resolve(LLMTask.PREFACTURA_PAGE_CLASSIFICATION) self.assertEqual(route.provider, "gemini") self.assertEqual(route.model, "gemini-2.5-flash-lite") self.assertEqual(route.output_kind, LLMOutputKind.STRUCTURED_OBJECT) self.assertIsNone(route.fallback_rule) def test_all_task_routes_expose_minimum_metadata(self) -> None: policy = DefaultModelSelectionPolicy() for task in LLMTask: route = policy.resolve(task) self.assertIn("required_capability", route.metadata) self.assertTrue(route.provider) self.assertTrue(route.model) self.assertIsNotNone(route.output_kind) class FakeProvider: def __init__( self, name: str, *, text_result: str = "ok", structured_result: object | None = None, availability: LLMProviderAvailability | None = None, text_error: Exception | None = None, structured_error: Exception | None = None, ) -> None: self._name = name self._text_result = text_result self._structured_result = structured_result or {"ok": True} self._availability = availability or LLMProviderAvailability(configured=True) self._text_error = text_error self._structured_error = structured_error self.calls: list[tuple[str, str]] = [] def provider_name(self) -> str: return self._name def availability(self) -> LLMProviderAvailability: return self._availability def capabilities(self) -> LLMProviderCapabilities: return LLMProviderCapabilities(text_generation=True, structured_generation=True) def generate_text(self, request: LLMTextRequest) -> LLMTextResult: self.calls.append(("text", request.model or "")) if self._text_error: raise self._text_error return LLMTextResult( content=self._text_result, provider=self._name, model=request.model or "unset", ) def generate_structured(self, request: LLMStructuredRequest) -> LLMStructuredResult: self.calls.append(("structured", request.model or "")) if self._structured_error: raise self._structured_error return LLMStructuredResult( content=self._structured_result, provider=self._name, model=request.model or "unset", output_kind=request.output_kind, ) def close(self) -> None: return None class DefaultModelRouterTest(unittest.TestCase): def test_uses_primary_provider_when_available(self) -> None: gemini = FakeProvider("gemini", text_result="gemini ok") groq = FakeProvider("groq", text_result="groq ok") router = DefaultModelRouter( providers={"gemini": gemini, "groq": groq, "openai": FakeProvider("openai")}, policy=DefaultModelSelectionPolicy(), ) result = router.generate_text( LLMTextRequest( task=LLMTask.HISTORIA_FINAL_SUMMARY, prompt="resume esto", ) ) self.assertEqual(result.content, "gemini ok") self.assertEqual(gemini.calls, [("text", "gemini-2.5-flash")]) self.assertEqual(groq.calls, []) def test_raises_when_primary_provider_fails_without_legacy_fallback(self) -> None: gemini = FakeProvider( "gemini", text_error=LLMProviderError( "temporarily unavailable", kind=LLMErrorKind.TRANSIENT, provider="gemini", model="gemini-2.5-flash-lite", retryable=True, ), ) groq = FakeProvider("groq", text_result="groq fallback") router = DefaultModelRouter( providers={"gemini": gemini, "groq": groq, "openai": FakeProvider("openai")}, policy=DefaultModelSelectionPolicy(), ) with self.assertRaises(LLMProviderError) as ctx: router.generate_text( LLMTextRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae los datos", ) ) self.assertEqual(ctx.exception.provider, "gemini") self.assertEqual(groq.calls, []) def test_raises_for_structured_soat_code_generation_without_legacy_fallback(self) -> None: gemini = FakeProvider( "gemini", structured_error=LLMProviderError( "rate limit", kind=LLMErrorKind.RATE_LIMITED, provider="gemini", model="gemini-2.5-flash", retryable=True, ), ) router = DefaultModelRouter( providers={"gemini": gemini, "groq": FakeProvider("groq"), "openai": FakeProvider("openai")}, policy=DefaultModelSelectionPolicy(), ) with self.assertRaises(LLMProviderError) as ctx: router.generate_structured( LLMStructuredRequest( task=LLMTask.SOAT_CODE_GENERATION, prompt="propón códigos", output_model=SimpleNamespace, output_kind=LLMOutputKind.STRUCTURED_COLLECTION, ) ) self.assertEqual(ctx.exception.provider, "gemini") def test_raises_on_unsupported_capability_without_legacy_fallback(self) -> None: gemini = FakeProvider( "gemini", structured_error=LLMProviderError( "schema mismatch", kind=LLMErrorKind.UNSUPPORTED_CAPABILITY, provider="gemini", model="gemini-2.5-flash", ), ) groq = FakeProvider("groq", structured_result={"ok": True}) router = DefaultModelRouter( providers={"gemini": gemini, "groq": groq, "openai": FakeProvider("openai")}, policy=DefaultModelSelectionPolicy(), ) with self.assertRaises(LLMProviderError) as ctx: router.generate_structured( LLMStructuredRequest( task=LLMTask.SOAT_REASONING, prompt="valida", output_model=SimpleNamespace, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(ctx.exception.provider, "gemini") self.assertEqual(groq.calls, []) class RetrieverTaskRoutingTest(unittest.TestCase): def _router_with_text(self, text_result: str) -> tuple[DefaultModelRouter, FakeProvider, FakeProvider]: gemini = FakeProvider("gemini", text_result=text_result) groq = FakeProvider("groq", text_result=text_result) router = DefaultModelRouter( providers={"gemini": gemini, "groq": groq, "openai": FakeProvider("openai")}, policy=DefaultModelSelectionPolicy(), ) return router, gemini, groq def test_cie10_retriever_uses_task_policy(self) -> None: router, gemini, groq = self._router_with_text("Código: A00 - Descripción: Cólera") retriever = CIE10Retriever.__new__(CIE10Retriever) retriever.llm_router = router retriever.groq_client = None retriever.groq_model = "unused" retriever.buscar = lambda *args, **kwargs: {"detalles": [{"codigo": "A00", "descripcion": "Cólera"}]} result = retriever.asignar_codigo_cie10("diag") self.assertEqual(result, "Código: A00 - Descripción: Cólera") self.assertEqual(gemini.calls, [("text", "gemini-2.5-flash-lite")]) self.assertEqual(groq.calls, []) def test_cups_retriever_uses_task_policy(self) -> None: router, gemini, groq = self._router_with_text("123456 - Lavado quirúrgico") retriever = CUPSRetriever.__new__(CUPSRetriever) retriever.llm_router = router retriever.groq = None retriever.groq_model = "unused" retriever.buscar = lambda *args, **kwargs: [SimpleNamespace(metadata={"codigo": "123456", "nombre": "Lavado quirúrgico"})] result = retriever.asignar_codigos(["Lavado quirúrgico"]) self.assertEqual(result.to_dict(orient="records")[0]["codigo_cups"], "123456 - Lavado quirúrgico") self.assertEqual(gemini.calls, [("text", "gemini-2.5-flash-lite")]) self.assertEqual(groq.calls, []) def test_soat_retriever_manual_lookup_uses_task_policy(self) -> None: router, gemini, groq = self._router_with_text("S123 - Procedimiento SOAT (p.12)") retriever = SOATRetriever.__new__(SOATRetriever) retriever.llm_router = router retriever.groq = None retriever.groq_model = "unused" result = retriever.proponer_codigos_soat( "Lavado quirúrgico", [SimpleNamespace(metadata={"page": 12}, page_content="Procedimiento SOAT relevante")], ) self.assertEqual(result, "S123 - Procedimiento SOAT (p.12)") self.assertEqual(gemini.calls, [("text", "gemini-2.5-flash")]) self.assertEqual(groq.calls, []) class HistoriaRouterIntegrationTest(unittest.TestCase): def test_historia_request_reports_router_provider_instead_of_global_llm_provider(self) -> None: fake_router = SimpleNamespace( _resolve_route=lambda task: SimpleNamespace(provider="gemini"), ) with patch("modules.processing.resumen_google.ClinicalStructuredExtractionService") as extraction_service, patch( "modules.processing.resumen_google.config.LLM_PROVIDER", "groq", ): extraction_service.return_value.extract.side_effect = RuntimeError("boom") with self.assertRaises(HTTPException) as ctx: HistoriaClinicaRequest.extraer_historia_estructurada( "Historia corta", llm_router=fake_router, ) self.assertEqual(ctx.exception.status_code, 500) self.assertIn("gemini", ctx.exception.detail) self.assertNotIn("groq", ctx.exception.detail) class GeminiAdapterTest(unittest.TestCase): def test_maps_503_to_transient_error(self) -> None: def _raise_unavailable(**kwargs): raise Exception("503 UNAVAILABLE") failing_models = SimpleNamespace( generate_content=_raise_unavailable ) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash", client=SimpleNamespace(models=failing_models), ) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_text(LLMTextRequest(task=None, prompt="hola")) self.assertEqual(ctx.exception.kind, LLMErrorKind.TRANSIENT) def test_maps_429_to_rate_limited_error(self) -> None: def _raise_rate_limited(**kwargs): raise Exception("429 RESOURCE_EXHAUSTED") failing_models = SimpleNamespace( generate_content=_raise_rate_limited ) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash", client=SimpleNamespace(models=failing_models), ) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_text(LLMTextRequest(task=None, prompt="hola")) self.assertEqual(ctx.exception.kind, LLMErrorKind.RATE_LIMITED) def test_generates_text_with_mocked_client(self) -> None: models = SimpleNamespace(generate_content=lambda **kwargs: SimpleNamespace(text="respuesta")) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash", client=SimpleNamespace(models=models), ) result = adapter.generate_text(LLMTextRequest(task=None, prompt="hola")) self.assertEqual(result.content, "respuesta") self.assertEqual(result.provider, "gemini") def test_legacy_generate_content_contract_still_works(self) -> None: with patch( "app.core.gemini_client.GeminiAdapter.generate_content_legacy", return_value=SimpleNamespace(text="salida legacy"), ): adapter = GeminiModelAdapter(api_key="secret", model_name="gemini-2.5-flash") result = adapter.generate_content("contenido") self.assertEqual(result.text, "salida legacy") def test_retries_invalid_structured_json_and_recovers(self) -> None: valid_payload = HistoriaClinicaStructured( patient_name="PACIENTE JSON", resumen_clinico="Resumen", diagnosticos=[], procedimientos=[], medicamentos=[], ).model_dump_json(by_alias=True, exclude_none=True, exclude_defaults=True) responses = iter( [ SimpleNamespace(text="{invalid json"), SimpleNamespace(text=valid_payload), ] ) models = SimpleNamespace(generate_content=lambda **kwargs: next(responses)) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash-lite", client=SimpleNamespace(models=models), ) result = adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=HistoriaClinicaStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(result.content["pn"], "PACIENTE JSON") def test_raises_unsupported_capability_after_invalid_structured_retry(self) -> None: responses = iter([SimpleNamespace(text="{invalid"), SimpleNamespace(text="{still invalid")]) models = SimpleNamespace(generate_content=lambda **kwargs: next(responses)) adapter = GeminiAdapter( api_key="secret", default_model="gemini-2.5-flash-lite", client=SimpleNamespace(models=models), ) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=HistoriaClinicaStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(ctx.exception.kind, LLMErrorKind.UNSUPPORTED_CAPABILITY) class GroqAdapterTest(unittest.TestCase): def test_generates_text_with_mocked_client(self) -> None: fake_response = SimpleNamespace( choices=[SimpleNamespace(message=SimpleNamespace(content="respuesta groq"))] ) fake_client = SimpleNamespace( chat=SimpleNamespace( completions=SimpleNamespace(create=lambda **kwargs: fake_response) ) ) adapter = GroqAdapter( api_key="secret", default_model="openai/gpt-oss-120b", client=fake_client, ) result = adapter.generate_text(LLMTextRequest(task=None, prompt="hola")) self.assertEqual(result.content, "respuesta groq") self.assertEqual(result.provider, "groq") def test_exposes_capabilities(self) -> None: adapter = GroqAdapter(api_key="secret", default_model="openai/gpt-oss-120b", client=SimpleNamespace()) capabilities = adapter.capabilities() self.assertTrue(capabilities.text_generation) self.assertTrue(capabilities.structured_generation) def test_raises_controlled_error_for_invalid_structured_model(self) -> None: adapter = GroqAdapter(api_key="secret", default_model="openai/gpt-oss-120b", client=SimpleNamespace()) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=dict, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(ctx.exception.kind, LLMErrorKind.INVALID_CONFIGURATION) def test_retries_invalid_structured_json_and_recovers(self) -> None: valid_payload = LaboratorioStructured( patient_name="PACIENTE JSON", tipo_examen="Hemograma", interpretacion="Normal", resultados=[], valores_alterados=[], ).model_dump_json(by_alias=True, exclude_none=True, exclude_defaults=True) responses = iter( [ SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="{invalid json"))]), SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content=valid_payload))]), ] ) fake_client = SimpleNamespace( chat=SimpleNamespace(completions=SimpleNamespace(create=lambda **kwargs: next(responses))) ) adapter = GroqAdapter(api_key="secret", default_model="openai/gpt-oss-120b", client=fake_client) result = adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=LaboratorioStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(result.content["pn"], "PACIENTE JSON") def test_raises_unsupported_capability_after_invalid_structured_retry(self) -> None: responses = iter( [ SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="{invalid"))]), SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content="{still invalid"))]), ] ) fake_client = SimpleNamespace( chat=SimpleNamespace(completions=SimpleNamespace(create=lambda **kwargs: next(responses))) ) adapter = GroqAdapter(api_key="secret", default_model="openai/gpt-oss-120b", client=fake_client) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=LaboratorioStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(ctx.exception.kind, LLMErrorKind.UNSUPPORTED_CAPABILITY) class OpenAIAdapterStubTest(unittest.TestCase): def test_reports_stub_availability_and_fails_in_controlled_way(self) -> None: adapter = OpenAIAdapterStub(enabled=False) availability = adapter.availability() self.assertFalse(availability.configured) self.assertTrue(availability.is_stub) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_text(LLMTextRequest(task=None, prompt="hola")) self.assertEqual(ctx.exception.kind, LLMErrorKind.INVALID_CONFIGURATION) def test_fails_for_structured_generation_in_controlled_way(self) -> None: adapter = OpenAIAdapterStub(enabled=False) with self.assertRaises(LLMProviderError) as ctx: adapter.generate_structured( LLMStructuredRequest( task=LLMTask.CLINICAL_DOCUMENT_EXTRACT, prompt="extrae", output_model=HistoriaClinicaStructured, output_kind=LLMOutputKind.STRUCTURED_OBJECT, ) ) self.assertEqual(ctx.exception.kind, LLMErrorKind.INVALID_CONFIGURATION) class Phase8BenchmarkReportTest(unittest.TestCase): def test_builds_reproducible_report_for_active_tasks(self) -> None: report = build_phase8_benchmark_report() self.assertEqual(report["schema_version"], "v1") self.assertGreaterEqual(report["total_tasks"], 5) tasks = {item["task"] for item in report["snapshots"]} self.assertIn(LLMTask.HISTORIA_CHUNK_SUMMARY.value, tasks) self.assertIn(LLMTask.CLINICAL_DOCUMENT_EXTRACT.value, tasks) self.assertIn(LLMTask.SOAT_REASONING.value, tasks) self.assertIn(LLMTask.CIE10_RESOLUTION.value, tasks) self.assertIn(LLMTask.CUPS_RESOLUTION.value, tasks) self.assertTrue(any(item["reduced_chars"] > 0 for item in report["snapshots"])) def test_renders_markdown_table(self) -> None: markdown = render_phase8_benchmark_markdown(build_phase8_benchmark_report()) self.assertIn("# Phase 8 LLM Benchmark", markdown) self.assertIn("| Task | Provider | Model |", markdown) self.assertIn("soat_reasoning", markdown) class BuildServicesPhase2IntegrationTest(unittest.TestCase): class FakeResources: def legacy_llm_providers_enabled(self) -> bool: return False def groq_available(self) -> bool: return False def gemini_available(self) -> bool: return True def openai_stub_enabled(self) -> bool: return False def soat_available(self) -> bool: return False def get_groq_client(self): return SimpleNamespace( chat=SimpleNamespace( completions=SimpleNamespace( create=lambda **kwargs: SimpleNamespace( choices=[SimpleNamespace(message=SimpleNamespace(content="groq ok"))] ) ) ) ) def get_gemini_client(self): return SimpleNamespace(generate_content=lambda contents, generation_config=None: SimpleNamespace(text="gemini ok")) def get_groq_provider(self): return SimpleNamespace(provider_name=lambda: "groq") def get_gemini_provider(self): return SimpleNamespace(provider_name=lambda: "gemini") def get_openai_provider(self): return SimpleNamespace(provider_name=lambda: "openai") def get_llm_router(self): return SimpleNamespace(generate_text=lambda request: "router ok") def get_model_selection_policy(self): return SimpleNamespace(resolve=lambda task: task) def get_cie10_retriever(self): return "cie10" def get_cups_retriever(self): return "cups" def get_soat_retriever(self): return "soat" def set_soat_index_dir(self, index_dir: str) -> None: self.index_dir = index_dir def warmup(self, *resource_names: str): return {name: "ready" for name in resource_names} def status(self): return {"llm_router_loaded": True} def test_build_services_exposes_new_llm_fields_without_breaking_legacy_clients(self) -> None: fake_runtime = cast( MongoRuntime, SimpleNamespace( sync_client=object(), async_client=object(), sync_database={"audit_events": object()}, async_database=object(), ), ) fake_resources = cast(HeavyResourceRegistry, self.FakeResources()) with patch("app.core.services.MongoDBStorage", side_effect=lambda *args, **kwargs: SimpleNamespace()), patch( "app.core.services.Jinja2Templates", return_value=SimpleNamespace() ), patch( "app.core.services.ClinicalDocumentService", return_value=SimpleNamespace(batch_case_repository="repo") ), patch( "app.core.services.CaseEpicrisisService", return_value=SimpleNamespace() ), patch( "app.core.services.RdaService", return_value=SimpleNamespace() ), patch( "app.core.services.RipsService", return_value=SimpleNamespace() ), patch( "app.core.services.CaseEpicrisisRuntimeService", return_value=SimpleNamespace() ), patch( "app.core.services.CaseDeletionService", return_value=SimpleNamespace() ), patch( "app.core.services.build_batch_runtime", return_value=SimpleNamespace(recompute_batch_bulk_epicrisis=SimpleNamespace(execute=lambda *args, **kwargs: None)), ), patch("app.core.services.set_audit_repository"): services = build_services(Path("."), mongo_runtime=fake_runtime, heavy_resources=fake_resources) self.assertIn("gemini", services.llm_providers) self.assertNotIn("groq", services.llm_providers) self.assertNotIn("openai", services.llm_providers) self.assertIsNotNone(services.client_gemini) client_gemini = services.client_gemini self.assertIsNotNone(client_gemini) gemini_client = cast(Any, client_gemini) self.assertEqual(gemini_client.generate_content("hola").text, "gemini ok") self.assertIsNone(services.client_groq) self.assertEqual(services.llm_router.generate_text(None), "router ok") def test_build_services_wires_llm_task_cache_repository_to_dedicated_collection(self) -> None: fake_runtime = cast( MongoRuntime, SimpleNamespace( sync_client=object(), async_client=object(), sync_database={"audit_events": object()}, async_database=object(), ), ) fake_resources = cast(HeavyResourceRegistry, self.FakeResources()) storage_calls: list[tuple[tuple, dict]] = [] repo_instances: list[SimpleNamespace] = [] def fake_storage(*args, **kwargs): storage_calls.append((args, kwargs)) collection_name = args[2] return SimpleNamespace(collection=f"collection:{collection_name}") def fake_cache_repository(mongo_cache, **kwargs): repo = SimpleNamespace(collection=getattr(mongo_cache, "collection", None), ensure_indexes=lambda: None) repo_instances.append(repo) return repo with patch("app.core.services.MongoDBStorage", side_effect=fake_storage), patch( "app.core.services.MongoLLMTaskCacheRepository", side_effect=fake_cache_repository, ), patch( "app.core.services.Jinja2Templates", return_value=SimpleNamespace() ), patch( "app.core.services.ClinicalDocumentService", return_value=SimpleNamespace(batch_case_repository="repo") ), patch( "app.core.services.CaseEpicrisisService", return_value=SimpleNamespace() ), patch( "app.core.services.RdaService", return_value=SimpleNamespace() ), patch( "app.core.services.RipsService", return_value=SimpleNamespace() ), patch( "app.core.services.CaseEpicrisisRuntimeService", return_value=SimpleNamespace() ), patch( "app.core.services.CaseDeletionService", return_value=SimpleNamespace() ), patch( "app.core.services.build_batch_runtime", return_value=SimpleNamespace(recompute_batch_bulk_epicrisis=SimpleNamespace(execute=lambda *args, **kwargs: None)), ), patch( "app.core.services.set_audit_repository" ): services = build_services(Path("."), mongo_runtime=fake_runtime, heavy_resources=fake_resources) self.assertGreaterEqual(len(storage_calls), 3) self.assertEqual(storage_calls[1][0][2], "historias_analizadas") self.assertEqual(storage_calls[2][0][2], config.LLM_TASK_CACHE_COLLECTION) self.assertEqual(repo_instances[0].collection, f"collection:{config.LLM_TASK_CACHE_COLLECTION}") self.assertIs(services.llm_task_cache_repository, repo_instances[0]) class HeavyResourceRegistryStatusTest(unittest.TestCase): def test_status_exposes_phase2_provider_flags(self) -> None: registry = HeavyResourceRegistry(Path(".")) status = registry.status() self.assertIn("legacy_llm_providers_enabled", status) self.assertIn("openai_stub_enabled", status) self.assertIn("llm_groq_provider_loaded", status) self.assertIn("llm_gemini_provider_loaded", status) self.assertIn("llm_openai_provider_loaded", status) self.assertIn("llm_router_loaded", status) if __name__ == "__main__": unittest.main()