from __future__ import annotations import unittest from copy import deepcopy from pathlib import Path from types import SimpleNamespace from unittest.mock import patch from fastapi import FastAPI from fastapi.templating import Jinja2Templates from httpx import ASGITransport, AsyncClient from app.auth import get_current_user from app.routes import batches class BatchPageRenderTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(batches.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")), batch_runtime=SimpleNamespace(dispatcher_name="inprocess"), ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_subir_lote_renders_single_shell_header_and_collapsed_upload_panel(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/subir_lote") self.assertEqual(response.status_code, 200) self.assertIn("Carga masiva por ZIP", response.text) self.assertIn("Historial de lotes", response.text) self.assertNotIn('class="batch-header"', response.text) self.assertIn('data-app-shell-sidebar', response.text) self.assertIn('data-app-shell-toggle', response.text) self.assertIn('/static/js/app-shell.js', response.text) self.assertIn('/static/js/user-menu.js', response.text) self.assertIn('data-case-upload-dialog', response.text) self.assertIn('/static/icons/stack-2.svg', response.text) self.assertIn('/static/icons/menu-2.svg', response.text) self.assertIn('class="ui-header__leading"', response.text) self.assertNotIn('icon-tabler-stack-2', response.text) self.assertIn('aria-current="page"', response.text) self.assertIn('data-batch-upload-toggle', response.text) self.assertIn('aria-expanded="false"', response.text) self.assertIn('data-batch-upload-panel', response.text) self.assertIn('hidden', response.text) self.assertIn('data-batch-rules-trigger', response.text) self.assertIn('role="tooltip"', 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) class CapturingCreateManualBatchUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(dict(kwargs)) return { "batch_id": "batch-manual-001", "status": "recibido", "created_at": "2026-05-18T10:00:00-05:00", "poll_url": "/api/lotes/batch-manual-001", } class CapturingCreatePrefacturaBatchUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(dict(kwargs)) return { "batch_id": "batch-prefactura-001", "status": "procesando", "created_at": "2026-05-18T10:00:00-05:00", "poll_url": "/api/lotes/batch-prefactura-001", } class CapturingDeleteBatch: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(dict(kwargs)) if kwargs.get("confirmation_phrase") != "ELIMINAR LOTE": from app.batch_processing.application.command_use_cases import BatchDeletionError raise BatchDeletionError("La frase de confirmación no coincide.", status_code=400) return { "message": "Lote eliminado correctamente.", "batch_id": kwargs["batch_id"], } class CapturingCancelBatch: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(dict(kwargs)) return {"message": "Lote cancelado correctamente.", "batch_id": kwargs["batch_id"], "status": "cancelado"} class CapturingRetryBatch: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(dict(kwargs)) return {"message": "Trabajo recuperable reencolado correctamente.", "batch_id": kwargs["batch_id"]} class FakeBatchDemoIdentityService: enabled = True visible_case_key = "00001-176654-paciente-01-caso-00001" visible_patient_id = "00001" visible_patient_name = "paciente-01" real_case_key = "CASE-001" real_patient_id = "66781911" real_patient_name = "BLANCA LIJA RENGIFO AGUIRRE" 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 in {self.real_case_key, self.visible_case_key}: return { "case_key": self.visible_case_key, "case_number": case_number, "patient_id": self.visible_patient_id, "patient_name": self.visible_patient_name, "nombre_paciente": self.visible_patient_name, } return { "case_key": case_key, "case_number": case_number, "patient_id": patient_id, "patient_name": patient_name, "nombre_paciente": patient_name, } def resolve_case_key(self, *, username: str, visible_case_key: str) -> str: return self.real_case_key if visible_case_key == self.visible_case_key else visible_case_key def resolve_patient_id(self, *, username: str, visible_patient_id: str) -> str: return self.real_patient_id if visible_patient_id == self.visible_patient_id else visible_patient_id def resolve_patient_name(self, *, username: str, visible_patient_name: str) -> str: return self.real_patient_name if visible_patient_name == self.visible_patient_name else visible_patient_name def project_document(self, *, username: str, document: dict | None): if document is None: return None return self._sanitize_payload(deepcopy(document)) def _sanitize_payload(self, value): if isinstance(value, dict): projected = {key: self._sanitize_payload(item) for key, item in value.items()} if projected.get("case_key") == self.real_case_key: projected["case_key"] = self.visible_case_key if projected.get("patient_id") == self.real_patient_id: projected["patient_id"] = self.visible_patient_id if projected.get("patient_name") == self.real_patient_name: projected["patient_name"] = self.visible_patient_name if projected.get("nombre_paciente") == self.real_patient_name: projected["nombre_paciente"] = self.visible_patient_name return projected if isinstance(value, list): return [self._sanitize_payload(item) for item in value] if isinstance(value, str): return ( value.replace(self.real_case_key, self.visible_case_key) .replace(self.real_patient_id, self.visible_patient_id) .replace(self.real_patient_name, self.visible_patient_name) ) return value class ManualBatchRoutesTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(batches.router) self.create_manual = CapturingCreateManualBatchUpload() self.create_prefactura = CapturingCreatePrefacturaBatchUpload() self.delete_batch = CapturingDeleteBatch() self.cancel_batch = CapturingCancelBatch() self.retry_batch = CapturingRetryBatch() 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")), batch_runtime=SimpleNamespace( dispatcher_name="inprocess", create_manual_batch_upload=self.create_manual, create_prefactura_batch_upload=self.create_prefactura, delete_batch=self.delete_batch, cancel_batch_processing=self.cancel_batch, retry_batch_processing=self.retry_batch, ), ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_manual_historia_endpoint_returns_202_and_uses_historia_mode(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/lotes/manual/historia", files={"file": ("historia.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) payload = response.json() self.assertEqual(payload["batch_id"], "batch-manual-001") self.assertEqual(self.create_manual.calls[0]["ingestion_mode"], "manual_historia") self.assertEqual(self.create_manual.calls[0]["detected_type"], "historia_clinica") async def test_manual_soporte_endpoint_propagates_case_key_and_document_type(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/lotes/manual/soporte", data={"tipo_documento": "factura", "case_key": "CASE-900"}, files={"file": ("factura.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) self.assertEqual(self.create_manual.calls[0]["ingestion_mode"], "manual_soporte") self.assertEqual(self.create_manual.calls[0]["detected_type"], "factura") self.assertEqual(self.create_manual.calls[0]["case_key"], "CASE-900") async def test_prefactura_endpoint_returns_202_and_uses_prefactura_use_case(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/lotes/prefactura", files={"file": ("prefactura.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) payload = response.json() self.assertEqual(payload["batch_id"], "batch-prefactura-001") self.assertEqual(self.create_prefactura.calls[0]["filename"], "prefactura.pdf") self.assertEqual(self.create_prefactura.calls[0]["username"], "auditor") async def test_delete_batch_route_requires_exact_confirmation(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/lotes/batch-manual-001/eliminar", json={ "confirmation_batch_id": "batch-manual-001", "confirmation_phrase": "OTRA FRASE", }, ) self.assertEqual(response.status_code, 400) async def test_delete_batch_route_returns_contract(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/lotes/batch-manual-001/eliminar", json={ "confirmation_batch_id": "batch-manual-001", "confirmation_phrase": "ELIMINAR LOTE", }, ) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["batch_id"], "batch-manual-001") self.assertEqual(self.delete_batch.calls[0]["username"], "auditor") async def test_cancel_batch_route_returns_contract(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post("/api/lotes/batch-manual-001/cancelar") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["status"], "cancelado") self.assertEqual(self.cancel_batch.calls[0]["username"], "auditor") async def test_retry_batch_route_returns_contract(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post("/api/lotes/batch-manual-001/reintentar") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["batch_id"], "batch-manual-001") self.assertEqual(self.retry_batch.calls[0]["username"], "auditor") class BatchTrackingDemoProjectionTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.dispatcher_env = patch.dict("os.environ", {"BATCH_DISPATCHER": "inprocess"}) self.dispatcher_env.start() self.app = FastAPI() self.app.include_router(batches.router) self.demo_identity = FakeBatchDemoIdentityService() self.resolve_file_association_calls: list[dict] = [] async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") def get_batch_status(batch_id: str) -> dict: return { "batch_id": batch_id, "usuario": "auditor", "status": "completado", } def resolve_file_association(**kwargs): self.resolve_file_association_calls.append(dict(kwargs)) return { "_id": kwargs["file_id"], "status": "asociado", "case_key": kwargs["case_key"], "association_source": "manual", "manual_resolution": {"reason": kwargs["reason"]}, "confidence": 1.0, } 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")), demo_identity_service=self.demo_identity, batch_runtime=SimpleNamespace( dispatcher_name="inprocess", get_batch_status=SimpleNamespace(execute=get_batch_status), list_batch_files=SimpleNamespace( execute=lambda batch_id: [ { "file_id": "file-1", "original_name": "CIRUGIA.pdf", "status": "asociado", "detected_type": "quirurgico", "case_key": "CASE-001", "case_number": "176654", "patient_name": "BLANCA LIJA RENGIFO AGUIRRE", "patient_id": "66781911", "associated_user": "66781911", "top_candidates": [{"case_key": "CASE-001", "score": 1.0}], "manual_resolution": { "reason": "Paciente BLANCA LIJA RENGIFO AGUIRRE 66781911 en CASE-001" }, } ] ), list_batch_cases=SimpleNamespace( execute=lambda batch_id: [ { "case_key": "CASE-001", "case_number": "176654", "patient_name": "BLANCA LIJA RENGIFO AGUIRRE", "patient_id": "66781911", "ready_for_epicrisis": True, "epicrisis_status": "pendiente", "epicrisis_url": "/epicrisis?case_key=CASE-001", } ] ), list_pending_associations=SimpleNamespace(execute=lambda batch_id: []), resolve_file_association=SimpleNamespace(execute=resolve_file_association), queue_batch_file_clinical=SimpleNamespace(execute=lambda *args, **kwargs: None), materialize_batch_file=SimpleNamespace(execute=lambda *args, **kwargs: None), ), ) self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() self.dispatcher_env.stop() async def test_batch_tracking_routes_project_demo_aliases(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: files_response = await client.get("/api/lotes/batch-001/documentos") cases_response = await client.get("/api/lotes/batch-001/casos") self.assertEqual(files_response.status_code, 200) self.assertEqual(cases_response.status_code, 200) file_item = files_response.json()["documentos"][0] self.assertEqual(file_item["case_key"], self.demo_identity.visible_case_key) self.assertEqual(file_item["patient_name"], self.demo_identity.visible_patient_name) self.assertEqual(file_item["patient_id"], self.demo_identity.visible_patient_id) self.assertEqual(file_item["top_candidates"][0]["case_key"], self.demo_identity.visible_case_key) self.assertIn(self.demo_identity.visible_patient_name, file_item["manual_resolution"]["reason"]) self.assertNotIn(self.demo_identity.real_patient_name, file_item["manual_resolution"]["reason"]) case_item = cases_response.json()["casos"][0] self.assertEqual(case_item["case_key"], self.demo_identity.visible_case_key) self.assertEqual(case_item["patient_name"], self.demo_identity.visible_patient_name) self.assertEqual(case_item["patient_id"], self.demo_identity.visible_patient_id) self.assertIn(self.demo_identity.visible_case_key, case_item["epicrisis_url"]) async def test_manual_resolution_route_resolves_visible_aliases_before_persisting(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.patch( "/api/lotes/batch-001/documentos/file-1/asociacion", json={ "case_key": self.demo_identity.visible_case_key, "patient_name": self.demo_identity.visible_patient_name, "patient_id": self.demo_identity.visible_patient_id, "procedure_code": "890201", "reason": "Validación manual", }, ) self.assertEqual(response.status_code, 200) self.assertEqual(self.resolve_file_association_calls[0]["case_key"], self.demo_identity.real_case_key) self.assertEqual(self.resolve_file_association_calls[0]["patient_name"], self.demo_identity.real_patient_name) self.assertEqual(self.resolve_file_association_calls[0]["patient_id"], self.demo_identity.real_patient_id) if __name__ == "__main__": unittest.main()