from __future__ import annotations import unittest from datetime import datetime from pathlib import Path from types import SimpleNamespace from typing import cast from unittest.mock import patch from bson import ObjectId from fastapi import FastAPI from fastapi.templating import Jinja2Templates from httpx import ASGITransport, AsyncClient from app.auth import get_current_user from app.core.services import AppServices from app.llm.schemas import DocumentoQuirurgicoStructured, HistoriaClinicaStructured from app.routes import cases, documents, history, pages from app.services.case_deletion_service import CaseDeletionError class SortableDocs: def __init__(self, items): self.items = list(items) def sort(self, fields): docs = list(self.items) for field_name, direction in reversed(fields): reverse = int(direction) < 0 docs.sort(key=lambda item, sort_field=field_name: item.get(sort_field), reverse=reverse) return docs class FakeMongoCollection: def __init__(self, items): self.items = list(items) def find(self, query): requested_case_keys = set(((query.get("case_key") or {}).get("$in")) or []) filtered = [ item for item in self.items if item.get("usuario") == query.get("usuario") and item.get("tipo_documento") == query.get("tipo_documento") and item.get("case_key") in requested_case_keys ] return SortableDocs(filtered) class FakeOwnedDocumentCollection: def __init__(self, items): self.items = {item["_id"]: dict(item) for item in items} def find_one(self, query): item = self.items.get(query.get("_id")) if item is None: return None return dict(item) class FakeSnapshotMongoCollection: def __init__(self, items): self.items = list(items) def find(self, query): excluded = set(((query.get("tipo_documento") or {}).get("$nin")) or []) return [ dict(item) for item in self.items if item.get("usuario") == query.get("usuario") and item.get("tipo_documento") not in excluded ] class FakeSnapshotCaseRepository: def __init__(self, case_doc=None): self.case_doc = dict(case_doc or {}) def get_user_case(self, username, case_key): if not case_key: return None return dict(self.case_doc) class CapturingClinicalDocumentService: def __init__(self): self.last_request = None self.response = { "mensaje": "ok", "tipo_documento": "generico", "case_key": "CASE-AUTO-001", "case_number": "HC-AUTO-001", "patient_id": "123456", "nombre_paciente": "Paciente Auto", "case_resolution_status": "confirmed", "review_required": False, "review_messages": [], } def process_and_persist(self, request): self.last_request = request payload = dict(self.response) payload["tipo_documento"] = request.detected_type payload["case_key"] = payload.get("case_key") or request.case_key payload["case_number"] = payload.get("case_number") or request.case_number payload["patient_id"] = payload.get("patient_id") or request.patient_id payload["nombre_paciente"] = payload.get("nombre_paciente") or request.provided_patient_name return payload class FakeMongoStorageCollection: def __init__(self, latest_doc=None): self.latest_doc = latest_doc or {} def find_one(self, query, sort=None): return dict(self.latest_doc) if self.latest_doc else None class FakeRecordsDataFrame: def __init__(self, records): self._records = list(records) def to_dict(self, orient="records"): if orient != "records": raise AssertionError(f"orient inesperado: {orient}") return list(self._records) class CapturingCie10Retriever: def __init__(self): self.last_html = None def asignar_codigos_batch(self, analisis_html): self.last_html = analisis_html return FakeRecordsDataFrame([{"codigo": "A00", "descripcion": "Diag estructurado"}]) class FakeCaseDeletionService: def __init__(self) -> None: self.last_document_request = None self.last_case_request = None def delete_document(self, *, username: str, document_id: str): self.last_document_request = {"username": username, "document_id": document_id} if document_id == "derived-doc": raise CaseDeletionError("Documento derivado no permitido.", status_code=400) return { "message": "Documento eliminado correctamente.", "deleted_document_id": document_id, "case_key": "CASE-001", "remaining_documents": 1, "case_deleted": False, } def delete_case(self, *, username: str, case_key: str): self.last_case_request = {"username": username, "case_key": case_key} if case_key == "CASE-404": raise CaseDeletionError("Caso no encontrado.", status_code=404) return { "message": "Caso eliminado correctamente.", "case_key": case_key, "deleted_documents": 2, "deleted_derived_artifacts": 3, } class FakeDemoIdentityService: enabled = True def resolve_case_key(self, *, username: str, visible_case_key: str) -> str: if visible_case_key == "00001-paciente-01-caso-00001": return "CASE-001" return visible_case_key def resolve_patient_name(self, *, username: str, visible_patient_name: str) -> str: return visible_patient_name def resolve_patient_id(self, *, username: str, visible_patient_id: str) -> str: return visible_patient_id def project_case_identity( self, *, username: str, case_key: str, case_number: str = "", patient_id: str = "", patient_name: str = "", ) -> dict[str, str]: if case_key == "CASE-001": return { "case_key": "00001-paciente-01-caso-00001", "case_number": case_number, "patient_id": "00001", "patient_name": "paciente-01", "nombre_paciente": "paciente-01", } return { "case_key": case_key, "case_number": case_number, "patient_id": patient_id, "patient_name": patient_name, "nombre_paciente": patient_name, } def project_document(self, *, username: str, document: dict | None): if not document: return document projected = dict(document) if projected.get("case_key") == "CASE-001": projected["case_key"] = "00001-paciente-01-caso-00001" projected["patient_id"] = "00001" projected["nombre_paciente"] = "paciente-01" projected["analisis_html"] = str(projected.get("analisis_html") or "").replace( "Blanca Lija", "paciente-01" ).replace("12345678", "00001") return projected def project_snapshot(self, *, username: str, snapshot: dict) -> dict: return snapshot def project_epicrisis_context(self, *, username: str, context: dict, case_key: str, case_number: str = "", patient_id: str = "", patient_name: str = "") -> dict: return dict(context) class HistorySnapshotGroupingTest(unittest.TestCase): def test_prefers_non_empty_patient_name_from_case_documents(self) -> None: services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=FakeSnapshotMongoCollection( [ { "_id": ObjectId(), "usuario": "auditor", "nombre_archivo": "prefactura.pdf", "tipo_documento": "prefactura", "case_key": "203922", "case_number": "203922", "patient_id": "", "nombre_paciente": "", "fecha_analisis": "2026-06-09T13:57:00-05:00", }, { "_id": ObjectId(), "usuario": "auditor", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "case_key": "203922", "case_number": "203922", "patient_id": "1088025550", "nombre_paciente": "LUISA FERNANDA IBARRA CANO", "fecha_analisis": "2026-06-09T13:56:00-05:00", }, ] ) ), clinical_document_service=SimpleNamespace(repair_existing_documents_for_user=lambda username: {}), ) with patch( "app.routes.history.MongoBatchCaseRepository", return_value=FakeSnapshotCaseRepository({"case_key": "203922", "patient_name": ""}), ): snapshot = history.build_user_histories_snapshot( username="auditor", services=cast(AppServices, services), ) case_group = snapshot["historias_por_caso"]["203922"] self.assertEqual(case_group["patient_name"], "LUISA FERNANDA IBARRA CANO") self.assertEqual(case_group["patient_id"], "1088025550") def test_excludes_llm_task_cache_from_case_snapshot(self) -> None: services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=FakeSnapshotMongoCollection( [ { "_id": ObjectId(), "usuario": "auditor", "nombre_archivo": "cache-1", "tipo_documento": "llm_task_cache", "case_key": "CASE-CACHE", "case_number": "", "patient_id": "", "nombre_paciente": "desconocido", "fecha_analisis": "2026-06-22T10:00:00-05:00", }, { "_id": ObjectId(), "usuario": "auditor", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "case_key": "CASE-001", "case_number": "178474", "patient_id": "93387170", "nombre_paciente": "JULIO CESAR ACOSTA GUEVARA", "fecha_analisis": "2026-06-22T09:58:00-05:00", }, ] ) ), clinical_document_service=SimpleNamespace(repair_existing_documents_for_user=lambda username: {}), ) with patch( "app.routes.history.MongoBatchCaseRepository", return_value=FakeSnapshotCaseRepository({"case_key": "CASE-001", "patient_name": ""}), ): snapshot = history.build_user_histories_snapshot( username="auditor", services=cast(AppServices, services), ) self.assertNotIn("CASE-CACHE", snapshot["historias_por_caso"]) self.assertEqual(list(snapshot["historias_por_caso"]), ["CASE-001"]) class CasesPageAndApiTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(cases.router) self.app.include_router(pages.router) async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") self.app.dependency_overrides[get_current_user] = override_current_user self.app.state.services = SimpleNamespace( templates=Jinja2Templates(directory=str(Path(__file__).resolve().parents[1] / "web")), mongo_analyses=SimpleNamespace(collection=FakeMongoCollection([])), case_deletion_service=FakeCaseDeletionService(), individual_ingestion_runtime=SimpleNamespace( list_sessions=SimpleNamespace(execute=lambda username, include_terminal=False: []) ), batch_runtime=SimpleNamespace( list_user_batches=SimpleNamespace(execute=lambda username, limit=20: []) ), case_epicrisis_runtime_service=SimpleNamespace( case_repository=SimpleNamespace(list_user_cases=lambda username, limit=50: []) ), ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_cases_page_renders_assets(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/casos") self.assertEqual(response.status_code, 200) self.assertIn('/static/css/cases.css', response.text) self.assertEqual(response.text.count('/static/js/auth-guard.js'), 1) self.assertIn('/static/js/case-upload-dialog.js', response.text) self.assertIn('/static/js/app-shell.js', response.text) self.assertIn('/static/js/user-menu.js', response.text) self.assertIn('/static/js/cases.js', response.text) self.assertIn('/static/icons/folders.svg', response.text) self.assertIn('/static/icons/menu-2.svg', response.text) self.assertIn("Casos indexados por", response.text) self.assertNotIn("Documentos del caso", response.text) self.assertIn('data-app-shell-sidebar', response.text) self.assertIn('data-app-shell-toggle', response.text) self.assertIn('data-app-shell-backdrop', response.text) self.assertIn('class="ui-header__leading"', response.text) self.assertIn('data-case-upload-open="support"', response.text) self.assertIn('data-cases-list', response.text) self.assertIn('data-cases-confirm-dialog', response.text) self.assertIn('data-cases-queue-open', response.text) self.assertIn('data-cases-queue-list', response.text) self.assertEqual(response.text.count('data-case-upload-dialog'), 1) self.assertIn('data-case-upload-case-picker', response.text) self.assertIn('data-case-upload-case-search', response.text) self.assertIn('data-case-upload-case-select', response.text) self.assertNotIn("Volver al dashboard", response.text) self.assertNotIn("Volver a casos", response.text) self.assertIn('data-user-menu-root', response.text) self.assertIn("Mi perfil", response.text) self.assertIn("Cerrar sesión", response.text) self.assertNotIn("/ver_historia?", response.text) self.assertNotIn("/subir_documento", response.text) self.assertNotIn("/epicrisis?id=", response.text) async def test_pacientes_redirects_to_casos(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/pacientes", follow_redirects=False) self.assertEqual(response.status_code, 307) self.assertEqual(response.headers.get("location"), "/casos") async def test_subir_documento_redirects_to_casos(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/subir_documento", follow_redirects=False) self.assertEqual(response.status_code, 307) self.assertEqual(response.headers.get("location"), "/casos") async def test_ver_historia_redirects_to_casos_when_case_key_is_present(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get( "/ver_historia?case_key=CASE-001&documento_id=doc-1", follow_redirects=False, ) self.assertEqual(response.status_code, 307) self.assertEqual(response.headers.get("location"), "/casos?case_key=CASE-001&documento_id=doc-1") async def test_cases_api_excludes_legacy_and_filters_by_patient_id(self) -> None: snapshot = { "historias_por_paciente": { "Blanca Lija": [ { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-04-28T08:15:00", "nombre_paciente": "Blanca Lija", } ], "Legacy": [ { "_id": "doc-legacy", "nombre_archivo": "legacy.pdf", "tipo_documento": "factura", "fecha_analisis": "2026-04-20T10:00:00", "nombre_paciente": "Legacy", } ], }, "historias_por_caso": { "CASE-001": { "group_key": "CASE-001", "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Blanca Lija", "patient_id": "12345678", "documentos": [ { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-04-28T08:15:00", } ], "tipos_documento": ["historia_clinica"], "ready_for_epicrisis": True, "epicrisis_status": "completado", "epicrisis_url": "/epicrisis?case_key=CASE-001", }, "legacy::legacy": { "group_key": "legacy::legacy", "case_key": "", "case_number": "", "patient_name": "Legacy", "patient_id": "", "documentos": [ { "_id": "doc-legacy", "nombre_archivo": "legacy.pdf", "tipo_documento": "factura", "fecha_analisis": "2026-04-20T10:00:00", } ], "tipos_documento": ["factura"], "ready_for_epicrisis": False, "epicrisis_status": "pendiente", "epicrisis_url": "", }, }, } self.app.state.services.mongo_analyses = SimpleNamespace( collection=FakeMongoCollection( [ { "usuario": "auditor", "tipo_documento": "rips_case_payload", "case_key": "CASE-001", "version": 2, "fecha_analisis": datetime.fromisoformat("2026-04-28T09:00:00"), "ready_for_cuv": True, "reject_count": 0, "notify_count": 1, } ] ) ) with patch("app.routes.cases.build_user_histories_snapshot", return_value=snapshot): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/casos?q=12345678") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["summary"]["legacy_excluded_count"], 1) self.assertEqual(payload["summary"]["filtered_cases"], 1) self.assertEqual(payload["cases"][0]["case_key"], "CASE-001") self.assertEqual(payload["cases"][0]["rips"]["status"], "rips_listo") async def test_case_detail_groups_documents_by_type(self) -> None: snapshot = { "historias_por_paciente": {}, "historias_por_caso": { "CASE-001": { "group_key": "CASE-001", "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Blanca Lija", "patient_id": "12345678", "documentos": [ { "_id": "doc-2", "nombre_archivo": "factura.pdf", "tipo_documento": "factura", "document_title": "PreFactura de Servicios", "fecha_analisis": "2026-04-28T09:15:00", }, { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "document_title": "Historia Clínica de Urgencias", "fecha_analisis": "2026-04-28T08:15:00", }, ], "tipos_documento": ["factura", "historia_clinica"], "ready_for_epicrisis": True, "epicrisis_status": "completado", "epicrisis_url": "/epicrisis?case_key=CASE-001", } }, } with patch("app.routes.cases.build_user_histories_snapshot", return_value=snapshot): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/casos/CASE-001") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["case"]["case_key"], "CASE-001") self.assertEqual(len(payload["document_groups"]), 2) self.assertEqual(payload["document_groups"][0]["documents"][0]["document_id"], "doc-2") self.assertEqual(payload["document_groups"][0]["documents"][0]["document_title"], "PreFactura de Servicios") async def test_cases_api_projects_demo_aliases_when_enabled(self) -> None: snapshot = { "historias_por_paciente": {}, "historias_por_caso": { "CASE-001": { "group_key": "CASE-001", "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Blanca Lija", "patient_id": "12345678", "documentos": [ { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-04-28T08:15:00", "nombre_paciente": "Blanca Lija", "patient_id": "12345678", } ], "tipos_documento": ["historia_clinica"], "ready_for_epicrisis": True, "epicrisis_status": "completado", "epicrisis_url": "/epicrisis?case_key=CASE-001", } }, } self.app.state.services.demo_identity_service = FakeDemoIdentityService() with patch("app.routes.cases.build_user_histories_snapshot", return_value=snapshot): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/casos/00001-paciente-01-caso-00001") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["case"]["case_key"], "00001-paciente-01-caso-00001") self.assertEqual(payload["case"]["patient_name"], "paciente-01") self.assertEqual(payload["case"]["patient_id"], "00001") async def test_operational_queue_api_returns_mixed_items(self) -> None: self.app.state.services.individual_ingestion_runtime = SimpleNamespace( list_sessions=SimpleNamespace( execute=lambda username, include_terminal=False: [ { "session_id": "manual-1", "active_case_key": "CASE-001", "session_status": "active", "updated_at": "2026-06-26T09:00:00-05:00", "base_upload_id": "upload-1", "uploads": [ { "upload_id": "upload-1", "status": "procesando", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", "updated_at": "2026-06-26T09:00:00-05:00", } ], } ] ) ) self.app.state.services.batch_runtime = SimpleNamespace( list_user_batches=SimpleNamespace( execute=lambda username, limit=20: [ { "batch_id": "batch-1", "nombre_archivo": "prefactura.pdf", "ingestion_mode": "prefactura", "status": "procesando", "updated_at": "2026-06-26T08:00:00-05:00", } ] ) ) self.app.state.services.case_epicrisis_runtime_service = SimpleNamespace( case_repository=SimpleNamespace( list_user_cases=lambda username, limit=50: [ { "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Blanca Lija", "patient_id": "12345678", "epicrisis_status": "fallido", "updated_at": "2099-06-26T12:30:00-05:00", "epicrisis_url": "/epicrisis?case_key=CASE-001", } ] ) ) async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/cola-operativa?include_terminal=true") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["summary"]["groups"], 1) self.assertEqual(payload["summary"]["leaves"], 2) self.assertEqual(payload["items"][0]["item_type"], "manual_session") async def test_cases_api_sorts_mixed_naive_and_aware_datetimes(self) -> None: snapshot = { "historias_por_paciente": {}, "historias_por_caso": { "CASE-001": { "group_key": "CASE-001", "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Paciente Uno", "patient_id": "123", "documentos": [ { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-04-28T08:15:00", } ], "tipos_documento": ["historia_clinica"], "ready_for_epicrisis": True, "epicrisis_status": "completado", "epicrisis_url": "/epicrisis?case_key=CASE-001", }, "CASE-002": { "group_key": "CASE-002", "case_key": "CASE-002", "case_number": "HC-002", "patient_name": "Paciente Dos", "patient_id": "456", "documentos": [ { "_id": "doc-2", "nombre_archivo": "factura.pdf", "tipo_documento": "factura", "fecha_analisis": "2026-04-28T09:20:00-05:00", } ], "tipos_documento": ["factura"], "ready_for_epicrisis": False, "epicrisis_status": "pendiente", "epicrisis_url": "/epicrisis?case_key=CASE-002", }, }, } with patch("app.routes.cases.build_user_histories_snapshot", return_value=snapshot): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/casos") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["cases"][0]["case_key"], "CASE-002") async def test_cases_api_labels_epicrisis_action_by_generation_status(self) -> None: snapshot = { "historias_por_paciente": {}, "historias_por_caso": { "CASE-001": { "group_key": "CASE-001", "case_key": "CASE-001", "case_number": "HC-001", "patient_name": "Paciente Uno", "patient_id": "123", "documentos": [ { "_id": "doc-1", "nombre_archivo": "historia.pdf", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-04-28T08:15:00", } ], "tipos_documento": ["historia_clinica"], "ready_for_epicrisis": True, "epicrisis_status": "completado", "epicrisis_url": "/epicrisis?case_key=CASE-001", }, "CASE-002": { "group_key": "CASE-002", "case_key": "CASE-002", "case_number": "HC-002", "patient_name": "Paciente Dos", "patient_id": "456", "documentos": [ { "_id": "doc-2", "nombre_archivo": "factura.pdf", "tipo_documento": "factura", "fecha_analisis": "2026-04-28T09:20:00", } ], "tipos_documento": ["factura"], "ready_for_epicrisis": False, "epicrisis_status": "pendiente", "epicrisis_url": "/epicrisis?case_key=CASE-002", }, }, } with patch("app.routes.cases.build_user_histories_snapshot", return_value=snapshot): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/casos") self.assertEqual(response.status_code, 200) payload = response.json() cases_by_key = {item["case_key"]: item for item in payload["cases"]} completed_labels = [action["label"] for action in cases_by_key["CASE-001"]["actions"]] pending_labels = [action["label"] for action in cases_by_key["CASE-002"]["actions"]] self.assertIn("Ver epicrisis", completed_labels) self.assertIn("Generar epicrisis", pending_labels) async def test_delete_case_document_route_returns_contract(self) -> None: service = self.app.state.services.case_deletion_service async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.delete("/api/casos/documentos/doc-123") self.assertEqual(response.status_code, 200) self.assertEqual(service.last_document_request, {"username": "auditor", "document_id": "doc-123"}) self.assertEqual(response.json()["deleted_document_id"], "doc-123") async def test_delete_case_document_route_maps_business_error(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.delete("/api/casos/documentos/derived-doc") self.assertEqual(response.status_code, 400) self.assertEqual(response.json()["detail"], "Documento derivado no permitido.") async def test_delete_case_route_returns_contract(self) -> None: service = self.app.state.services.case_deletion_service async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.delete("/api/casos/CASE-001") self.assertEqual(response.status_code, 200) self.assertEqual(service.last_case_request, {"username": "auditor", "case_key": "CASE-001"}) self.assertEqual(response.json()["deleted_documents"], 2) class UploadAdditionalDocumentRouteTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(documents.router) self.app.include_router(history.router) self.service = CapturingClinicalDocumentService() async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") self.app.dependency_overrides[get_current_user] = override_current_user self.app.state.services = SimpleNamespace( clinical_document_service=self.service, mongo_storage=SimpleNamespace(collection=FakeMongoStorageCollection()), ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_upload_documento_adicional_accepts_case_key(self) -> None: with patch("app.routes.documents.ReadPdf.read", return_value="contenido pdf"): async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/upload_documento_adicional", data={"tipo_documento": "factura", "case_key": "CASE-900"}, files={"file": ("factura.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 200) self.assertIsNotNone(self.service.last_request) self.assertEqual(self.service.last_request.case_key, "CASE-900") self.assertEqual(self.service.last_request.provided_patient_name, "") async def test_procesar_ultimo_pdf_returns_case_identity_fields(self) -> None: self.app.state.services.mongo_storage = SimpleNamespace( collection=FakeMongoStorageCollection( { "contenido_pdf": "contenido pdf", "nombre_archivo": "historia.pdf", "usuario": "auditor", } ) ) self.service.response = { "mensaje": "ok", "tipo_documento": "historia_clinica", "case_key": "CASE-HC-001", "case_number": "HC-001", "patient_id": "111222333", "nombre_paciente": "Paciente HC", "case_resolution_status": "confirmed", "review_required": False, "review_messages": [], } async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/procesar_ultimo_pdf") self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload["case_key"], "CASE-HC-001") self.assertEqual(payload["case_resolution_status"], "confirmed") self.assertFalse(payload["review_required"]) class HistoryRoutesStructuredFallbackTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(history.router) self.retriever = CapturingCie10Retriever() async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") self.app.dependency_overrides[get_current_user] = override_current_user self.app.state.services = SimpleNamespace(cie10_retriever=self.retriever) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_asignar_codigos_accepts_analysis_structured_payload(self) -> None: structured = HistoriaClinicaStructured( patient_name="PACIENTE UNO", resumen_clinico="Resumen", diagnosticos=[{"codigo": "A00", "descripcion": "Colera"}], procedimientos=[], medicamentos=[], ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/asignar_codigos", json={"document_type": "historia_clinica", "analysis_structured": structured}, ) self.assertEqual(response.status_code, 200) self.assertIn("Colera", self.retriever.last_html or "") self.assertEqual(response.json()["codigos"][0]["codigo"], "A00") class HistoryDocumentProjectionRouteTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(history.router) self.document_id = ObjectId() self.quirurgico_id = ObjectId() structured = HistoriaClinicaStructured( patient_name="PACIENTE ROUTE", numero_caso="CASE-900", identificacion_paciente="CC 123456789", resumen_clinico="Resumen proyectado desde structured", diagnosticos=[], procedimientos=[], medicamentos=[], ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True) async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") self.app.dependency_overrides[get_current_user] = override_current_user self.app.state.services = SimpleNamespace( mongo_analyses=SimpleNamespace( collection=FakeOwnedDocumentCollection( [ { "_id": self.document_id, "usuario": "auditor", "tipo_documento": "historia_clinica", "nombre_archivo": "historia.pdf", "nombre_paciente": "PACIENTE ROUTE", "case_number": "CASE-900", "patient_id": "123456789", "fecha_analisis": datetime.fromisoformat("2026-05-22T10:42:23-05:00"), "analysis_structured": structured, "analisis_html": "", "codigos_cie10": [], }, { "_id": self.quirurgico_id, "usuario": "auditor", "tipo_documento": "quirurgico", "nombre_archivo": "cirugia.pdf", "nombre_paciente": "PACIENTE ROUTE", "fecha_analisis": datetime.fromisoformat("2026-05-23T10:42:23-05:00"), "analysis_structured": DocumentoQuirurgicoStructured( patient_name="PACIENTE ROUTE", resumen_clinico="Resumen quirúrgico persistido y proyectado.", ).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True), "analisis_html": "", "codigos_cie10": [], }, ] ) ) ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_api_documentos_derives_html_from_structured_payload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get(f"/api/documentos/{self.document_id}") self.assertEqual(response.status_code, 200) payload = response.json() self.assertIn("PACIENTE ROUTE", payload["analisis_html"]) self.assertIn("CASE-900", payload["analisis_html"]) self.assertGreater(len(payload["analisis_html"]), 0) self.assertEqual(payload["_id"], str(self.document_id)) self.assertEqual(payload["case_number"], "CASE-900") self.assertEqual(payload["patient_id"], "123456789") async def test_api_documentos_exposes_persisted_surgical_summary(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get(f"/api/documentos/{self.quirurgico_id}") self.assertEqual(response.status_code, 200) payload = response.json() self.assertIn("Resumen quirúrgico persistido y proyectado.", payload["analisis_html"]) self.assertLess(payload["analisis_html"].index("Resumen"), payload["analisis_html"].index("Complicaciones")) if __name__ == "__main__": unittest.main()