from __future__ import annotations import unittest from dataclasses import dataclass from datetime import UTC from pathlib import Path from types import SimpleNamespace from typing import Any, cast from fastapi import FastAPI from httpx import ASGITransport, AsyncClient from app.auth import get_current_user from app.batch_processing.infrastructure.heuristic_association import HeuristicCaseAssociationService from app.clinical_pipeline.domain.errors import HistoriaSummaryCoverageError from app.clinical_pipeline.domain.models import ( DocumentClassificationDecision, PdfExtractionMetadata, PdfExtractionResult, ) from app.individual_ingestion.application.support_identity_matching import SupportIdentityMatchEngine from app.individual_ingestion.application.use_cases import ( CancelIndividualUploadUseCase, CloseIndividualUploadSessionUseCase, ConfirmIndividualUploadUseCase, CreateIndividualUploadUseCase, GetActiveIndividualUploadSessionUseCase, GetIndividualUploadSessionUseCase, GetIndividualUploadStatusUseCase, ListIndividualUploadSessionsUseCase, RetryIndividualUploadUseCase, RunIndividualMaterializationUseCase, RunIndividualPrecheckUseCase, ) from app.individual_ingestion.domain.models import ( PENDING_ACTION_CONFIRM_CASE_IDENTITY, PENDING_ACTION_CONFIRM_CASE_OVERRIDE, PENDING_ACTION_CONFIRM_TYPE_OVERRIDE, UPLOAD_STATUS_BLOCKED_BASE_FAILED, UPLOAD_STATUS_CANCELADO, UPLOAD_STATUS_COMPLETADO, UPLOAD_STATUS_ELIMINADO, UPLOAD_STATUS_ESPERANDO_CONFIRMACION, UPLOAD_STATUS_FALLIDO, UPLOAD_STATUS_LISTO_PARA_PROCESAR, UPLOAD_STATUS_PRECHECK_EN_COLA, UPLOAD_STATUS_PROCESANDO, UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION, ) from app.routes import individual_ingestion from app.services.clinical_document_service import ClinicalDocumentRequest JULIO_FACTURA_DIRTY_TEXT = Path("test/fixtures/factura_julio_cesar_acosta_dirty.txt").read_text( encoding="utf-8" ) class FakeRepository: def __init__(self) -> None: self.records: dict[str, dict] = {} self.next_id = 1 def create_upload(self, payload: dict) -> str: upload_id = str(self.next_id) self.next_id += 1 self.records[upload_id] = {"_id": upload_id, **payload} return upload_id def update_upload(self, upload_id: str, payload: dict) -> None: self.records[upload_id].update(payload) def update_upload_if_status(self, upload_id: str, expected_statuses: set[str], payload: dict) -> bool: record = self.records.get(upload_id) if not record or record.get("status") not in expected_statuses: return False record.update(payload) return True def claim_retry(self, upload_id: str, username: str, payload: dict) -> bool: record = self.records.get(upload_id) if not record or record.get("usuario") != username or record.get("status") != UPLOAD_STATUS_FALLIDO: return False record.update(payload) record["retry_count"] = int(record.get("retry_count") or 0) + 1 return True def get_upload(self, upload_id: str) -> dict | None: record = self.records.get(upload_id) return dict(record) if record else None def find_latest_active(self, username: str) -> dict | None: for record in reversed(list(self.records.values())): if record.get("usuario") == username and record.get("status") != UPLOAD_STATUS_COMPLETADO: return dict(record) return None def find_latest_by_hash(self, username: str, source_file_hash: str) -> dict | None: for record in reversed(list(self.records.values())): if record.get("usuario") == username and record.get("source_file_hash") == source_file_hash: return dict(record) return None def get_active_session(self, username: str) -> dict | None: sessions = self.list_sessions(username) return sessions[0] if sessions else None def get_session(self, username: str, session_id: str) -> dict | None: uploads = self.list_session_uploads(username, session_id) if not uploads: return None latest = uploads[-1] return self._build_session(latest, uploads) def list_sessions(self, username: str, *, include_terminal: bool = False) -> list[dict]: sessions: dict[str, list[dict]] = {} for record in self.records.values(): if record.get("usuario") != username or not record.get("session_id"): continue session_id = str(record.get("session_id") or "") sessions.setdefault(session_id, []).append(dict(record)) built = [] for uploads in sessions.values(): ordered = sorted( uploads, key=lambda item: str(item.get("updated_at") or item.get("created_at") or "") ) latest = ordered[-1] session_status = str(latest.get("session_status") or "") is_terminal = session_status in {"completed", "cancelled"} if is_terminal and not include_terminal: continue built.append(self._build_session(latest, ordered)) built.sort(key=lambda item: str(item.get("updated_at") or ""), reverse=True) return built def list_session_uploads(self, username: str, session_id: str) -> list[dict]: return [ dict(record) for record in self.records.values() if record.get("usuario") == username and record.get("session_id") == session_id ] def list_dependent_uploads(self, username: str, base_upload_id: str) -> list[dict]: return [ dict(record) for record in self.records.values() if record.get("usuario") == username and record.get("depends_on_upload_id") == base_upload_id ] def _build_session(self, latest: dict, uploads: list[dict]) -> dict: base_upload_id = latest.get("base_upload_id") or uploads[0].get("_id") or "" base_upload = next((item for item in uploads if item.get("_id") == base_upload_id), None) return { "session_id": str(latest.get("session_id") or ""), "active_case_key": latest.get("session_case_key") or latest.get("case_key") or "", "base_upload_id": base_upload_id, "session_status": latest.get("session_status") or "", "base_ready": bool(base_upload and base_upload.get("status") == UPLOAD_STATUS_COMPLETADO), "blocked_by_base_failure": bool( base_upload and base_upload.get("status") in {UPLOAD_STATUS_BLOCKED_BASE_FAILED, "fallido", "cancelado"} ), "updated_at": str(latest.get("updated_at") or latest.get("created_at") or ""), "uploads": uploads, } class FakeFileStore: def __init__(self) -> None: self.saved: dict[str, bytes] = {} self.deleted: list[str] = [] def save_file(self, upload_id: str, filename: str, contents: bytes) -> str: path = f"/tmp/{upload_id}-{filename}" self.saved[path] = contents return path def delete_file(self, stored_path: str) -> None: if stored_path: self.deleted.append(stored_path) self.saved.pop(stored_path, None) class FakeDispatcher: def __init__(self) -> None: self.precheck_calls: list[str] = [] self.materialization_calls: list[str] = [] def dispatch_precheck(self, upload_id: str) -> None: self.precheck_calls.append(upload_id) def dispatch_materialization(self, upload_id: str) -> None: self.materialization_calls.append(upload_id) class FakeTextExtractor: def __init__(self, text: str) -> None: self.text = text def extract_text(self, file_path: str) -> str: return self.text def extract(self, file_path: str) -> PdfExtractionResult: return PdfExtractionResult( text=self.text, metadata=PdfExtractionMetadata( total_pages=1, pages_with_text=1 if self.text.strip() else 0, character_count=len(self.text), pages_without_text=[] if self.text.strip() else [1], ), ) class FakeClassifier: def __init__(self, detected_type: str) -> None: self.detected_type = detected_type def describe(self, filename: str, text: str) -> tuple[str, str]: return self.detected_type, self.detected_type.replace("_", " ").title() def classify(self, filename: str, text: str) -> str: return self.detected_type def inspect(self, filename: str, text: str) -> DocumentClassificationDecision: return DocumentClassificationDecision( document_type=self.detected_type, title=self.detected_type.replace("_", " ").title(), confidence=1.0, reasons=["test_fixture"], ) @dataclass class FakeResolution: case_number: str = "HC-001" patient_id: str = "111222333" patient_name: str = "Paciente Uno" review_messages: list[str] | None = None redacted_identity_fields: tuple[str, ...] = () class FakeClinicalDocumentService: def __init__(self) -> None: self.persist_calls: list[ClinicalDocumentRequest] = [] self.case_context = { "case_key": "CASE-1", "case_number": "HC-001", "patient_id": "111222333", "patient_name": "Paciente Uno", } def resolve_case_identity(self, request): return FakeResolution(review_messages=["Identidad provisional inferida."]) def get_user_case_context(self, username: str, case_key: str) -> dict: if case_key == "CASE-1": return dict(self.case_context) return {} def process_and_persist(self, request): self.persist_calls.append(request) return {"id_documento": "doc-1", "case_key": request.case_key or "CASE-1"} class MissingIdentityClinicalDocumentService(FakeClinicalDocumentService): def resolve_case_identity(self, request): return FakeResolution( case_number="203922", patient_id="", patient_name="desconocido", review_messages=[], ) class RedactedIdentityClinicalDocumentService(FakeClinicalDocumentService): def resolve_case_identity(self, request): return FakeResolution( case_number="1390913", patient_id="", patient_name="desconocido", review_messages=[], redacted_identity_fields=("patient_id", "patient_name"), ) class FailingClinicalDocumentService(FakeClinicalDocumentService): def process_and_persist(self, request): self.persist_calls.append(request) raise HistoriaSummaryCoverageError( "No fue posible resumir todos los bloques de la historia clínica.", failed_chunks=[2], stage="historia_chunk_summary", attempts=3, provider="gemini", model="gemini-2.5-flash-lite", error_kind="transient", retryable=True, source="external_provider", ) class FakeCaseAssociationService: def __init__( self, *, patient_id: str = "", case_number: str = "", patient_name: str = "", contradictions=None ) -> None: self.signals = SimpleNamespace( patient_id=patient_id, case_number=case_number, patient_name=patient_name, ) self._contradictions = list(contradictions or []) def extract_signals(self, filename: str, text: str, detected_type: str): return self.signals def associate(self, file_records: list[dict]): return SimpleNamespace( associated=[], pending=[], warnings=[], ) def _detect_preferred_case_contradictions(self, payload: dict) -> list[str]: return list(self._contradictions) class SupportIdentityMatchEngineTest(unittest.TestCase): def test_matches_numeric_case_number_after_normalization(self) -> None: engine = SupportIdentityMatchEngine() decision = engine.evaluate( filename="factura.pdf", raw_text="Caso No. 178474N\nCC: 93387170", detected_type="factura", preferred_case_context={ "case_number": "178474", "patient_id": "93387170", "patient_name": "JULIO CESAR ACOSTA GUEVARA", }, extracted_signals=SimpleNamespace( case_number="178474N", patient_id="93387170", patient_name="", ), ) self.assertEqual(decision.case_number_result.status, "match") self.assertTrue(decision.case_number_result.matched_by_normalization) self.assertEqual(decision.contradictions, []) def test_prefers_real_patient_name_over_admin_entity_on_invoice(self) -> None: engine = SupportIdentityMatchEngine() decision = engine.evaluate( filename="FACTURA JULIO CESAR ACOSTA GUEVARA.pdf", raw_text=JULIO_FACTURA_DIRTY_TEXT, detected_type="factura", preferred_case_context={ "case_number": "178474", "patient_id": "93387170", "patient_name": "JULIO CESAR ACOSTA GUEVARA", }, extracted_signals=SimpleNamespace( case_number="178474N", patient_id="93387170", patient_name="COLOMBIA ENTIDAD COOPERATIVA", ), ) self.assertEqual(decision.patient_name_result.status, "match") self.assertEqual(decision.patient_name_result.extracted_value, "JULIO CESAR ACOSTA GUEVARA") self.assertEqual(decision.contradictions, []) def test_flags_low_confidence_when_only_admin_entity_is_present(self) -> None: engine = SupportIdentityMatchEngine() decision = engine.evaluate( filename="factura.pdf", raw_text="Paciente:\nCOLOMBIA ENTIDAD COOPERATIVA\nConvenio: SOAT", detected_type="factura", preferred_case_context={ "case_number": "178474", "patient_id": "93387170", "patient_name": "JULIO CESAR ACOSTA GUEVARA", }, extracted_signals=SimpleNamespace(case_number="", patient_id="", patient_name=""), ) self.assertEqual(decision.patient_name_result.status, "low_confidence") self.assertIn( "Se detectó una entidad administrativa donde se esperaba el nombre del paciente.", decision.warnings, ) self.assertEqual(decision.contradictions, []) class CreateIndividualUploadUseCaseTest(unittest.TestCase): def test_create_upload_deduplicates_active_hash(self) -> None: repository = FakeRepository() file_store = FakeFileStore() dispatcher = FakeDispatcher() use_case = CreateIndividualUploadUseCase( repository=repository, file_store=file_store, dispatcher=dispatcher, colombia_tz=UTC, ) first = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) second = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) self.assertEqual(first["status"], UPLOAD_STATUS_PRECHECK_EN_COLA) self.assertTrue(second["duplicate"]) self.assertEqual(second["message"], "El documento ya está en curso.") self.assertEqual(dispatcher.precheck_calls, ["1"]) def test_create_upload_reuses_recoverable_failed_hash(self) -> None: repository = FakeRepository() file_store = FakeFileStore() dispatcher = FakeDispatcher() use_case = CreateIndividualUploadUseCase( repository=repository, file_store=file_store, dispatcher=dispatcher, colombia_tz=UTC, ) first = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) repository.update_upload( first["upload_id"], { "status": UPLOAD_STATUS_FALLIDO, "extracted_text": "texto conservado", "failure": { "retryable": True, "source": "external_provider", }, }, ) duplicate = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) self.assertEqual(duplicate["upload_id"], first["upload_id"]) self.assertEqual(duplicate["status"], UPLOAD_STATUS_FALLIDO) self.assertTrue(duplicate["duplicate"]) self.assertEqual(len(repository.records), 1) self.assertEqual(dispatcher.precheck_calls, ["1"]) def test_create_upload_allows_same_hash_after_tombstone(self) -> None: repository = FakeRepository() dispatcher = FakeDispatcher() use_case = CreateIndividualUploadUseCase( repository=repository, file_store=FakeFileStore(), dispatcher=dispatcher, colombia_tz=UTC, ) first = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) repository.update_upload( first["upload_id"], { "status": UPLOAD_STATUS_ELIMINADO, "stored_path": "", "extracted_text": "", "deleted_at": "2026-08-18T12:00:00+00:00", "deletion_reason": "case_deleted", }, ) second = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) self.assertFalse(second["duplicate"]) self.assertNotEqual(second["upload_id"], first["upload_id"]) self.assertEqual(dispatcher.precheck_calls, ["1", "2"]) def test_story_upload_creates_reactive_session_metadata(self) -> None: repository = FakeRepository() use_case = CreateIndividualUploadUseCase( repository=repository, file_store=FakeFileStore(), dispatcher=FakeDispatcher(), colombia_tz=UTC, ) result = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 uno", username="auditor", selected_document_type="historia_clinica", ) record = cast(dict[str, Any], repository.get_upload(result["upload_id"])) self.assertIsNotNone(record) self.assertTrue(record["session_id"]) self.assertEqual(record["base_upload_id"], result["upload_id"]) self.assertEqual(record["session_status"], "active") def test_support_upload_reuses_active_session(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "session_case_key": "CASE-1", "session_status": "active", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, } use_case = CreateIndividualUploadUseCase( repository=repository, file_store=FakeFileStore(), dispatcher=FakeDispatcher(), colombia_tz=UTC, ) result = use_case.execute( filename="factura.pdf", contents=b"%PDF-1.4 soporte", username="auditor", selected_document_type="factura", provided_case_key="CASE-1", session_id="manual-1", lock_case_selection=True, ) record = cast(dict[str, Any], repository.get_upload(result["upload_id"])) self.assertIsNotNone(record) self.assertEqual(record["session_id"], "manual-1") self.assertEqual(record["depends_on_upload_id"], "1") self.assertTrue(record["case_selection_locked"]) class IndividualPrecheckUseCaseTest(unittest.TestCase): def test_precheck_does_not_process_tombstone(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ELIMINADO, "stored_path": "/tmp/1-historia.pdf", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", } use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("texto que no debe extraerse"), classifier=FakeClassifier("historia_clinica"), clinical_document_service=FakeClinicalDocumentService(), case_association_service=FakeCaseAssociationService(), dispatcher=FakeDispatcher(), colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_ELIMINADO) def test_precheck_projects_document_error_without_retry_for_pdf_without_text(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-historia.pdf", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", } use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor(""), classifier=FakeClassifier("historia_clinica"), clinical_document_service=FakeClinicalDocumentService(), case_association_service=FakeCaseAssociationService(), dispatcher=FakeDispatcher(), colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_FALLIDO) self.assertEqual(result["failure"]["source"], "document") self.assertEqual(result["failure"]["stage"], "pdf_extraction") self.assertFalse(result["can_retry"]) self.assertIn("Selecciona otro archivo", result["error"]) def test_historia_precheck_requires_identity_confirmation(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-historia.pdf", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", "provided_case_key": "", } dispatcher = FakeDispatcher() use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("texto hc"), classifier=FakeClassifier("historia_clinica"), clinical_document_service=FakeClinicalDocumentService(), case_association_service=FakeCaseAssociationService(), dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_ESPERANDO_CONFIRMACION) self.assertEqual(result["pending_actions"], [PENDING_ACTION_CONFIRM_CASE_IDENTITY]) self.assertEqual(result["provisional_case_number"], "HC-001") self.assertEqual(result["provisional_patient_id"], "111222333") self.assertEqual(result["provisional_patient_name"], "Paciente Uno") self.assertEqual(result["effective_document_type"], "historia_clinica") self.assertEqual(dispatcher.materialization_calls, []) def test_historia_precheck_leaves_ambiguous_identity_fields_empty(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-historia.pdf", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", "provided_case_key": "", } dispatcher = FakeDispatcher() use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("texto hc"), classifier=FakeClassifier("historia_clinica"), clinical_document_service=MissingIdentityClinicalDocumentService(), case_association_service=FakeCaseAssociationService(), dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_ESPERANDO_CONFIRMACION) self.assertEqual(result["provisional_case_number"], "203922") self.assertEqual(result["provisional_patient_id"], "") self.assertEqual(result["provisional_patient_name"], "") self.assertIn( "No fue posible identificar con confianza la identificación del paciente.", result["warnings"], ) self.assertIn( "No fue posible identificar con confianza el nombre del paciente.", result["warnings"], ) def test_historia_precheck_projects_redacted_identity_fields(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-historia.pdf", "selected_document_type": "historia_clinica", "original_name": "historia.pdf", "provided_case_key": "", } use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("texto hc censurado"), classifier=FakeClassifier("historia_clinica"), clinical_document_service=RedactedIdentityClinicalDocumentService(), case_association_service=FakeCaseAssociationService(), dispatcher=FakeDispatcher(), colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["provisional_case_number"], "1390913") self.assertEqual(result["provisional_patient_id"], "") self.assertEqual(result["provisional_patient_name"], "") self.assertEqual( set(result["redacted_identity_fields"]), {"patient_id", "patient_name"}, ) self.assertIn("La identificación del paciente está censurada", " ".join(result["warnings"])) self.assertIn("El nombre del paciente está censurado", " ".join(result["warnings"])) def test_support_precheck_stops_on_type_and_case_contradictions(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-soporte.pdf", "selected_document_type": "factura", "original_name": "soporte.pdf", "provided_case_key": "CASE-1", } dispatcher = FakeDispatcher() use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("texto soporte"), classifier=FakeClassifier("radiologia"), clinical_document_service=FakeClinicalDocumentService(), case_association_service=FakeCaseAssociationService( patient_id="999", case_number="HC-999", patient_name="Paciente Dos", contradictions=["La identificación del documento no coincide con el caso destino."], ), dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_ESPERANDO_CONFIRMACION) self.assertIn(PENDING_ACTION_CONFIRM_TYPE_OVERRIDE, result["pending_actions"]) self.assertIn(PENDING_ACTION_CONFIRM_CASE_OVERRIDE, result["pending_actions"]) self.assertEqual(result["effective_document_type"], "radiologia") self.assertTrue(result["impact_summary"]) self.assertEqual(dispatcher.materialization_calls, []) def test_factura_precheck_uses_invoice_identity_without_false_patient_mismatch(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-factura.pdf", "selected_document_type": "factura", "original_name": "FACTURA JULIO CESAR ACOSTA GUEVARA.pdf", "provided_case_key": "CASE-1", } dispatcher = FakeDispatcher() clinical_service = FakeClinicalDocumentService() clinical_service.case_context = { "case_key": "CASE-1", "case_number": "178474", "patient_id": "93387170", "patient_name": "JULIO CESAR ACOSTA GUEVARA", } use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor(JULIO_FACTURA_DIRTY_TEXT), classifier=FakeClassifier("factura"), clinical_document_service=clinical_service, case_association_service=HeuristicCaseAssociationService(), dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertEqual(result["pending_actions"], []) self.assertEqual(result["contradictions"], []) self.assertFalse(any("Nombre de paciente extraído" in message for message in result["warnings"])) def test_support_precheck_accepts_numeric_case_match_after_normalization(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PRECHECK_EN_COLA, "stored_path": "/tmp/1-factura.pdf", "selected_document_type": "factura", "original_name": "factura-178474N.pdf", "provided_case_key": "CASE-1", } dispatcher = FakeDispatcher() clinical_service = FakeClinicalDocumentService() clinical_service.case_context = { "case_key": "CASE-1", "case_number": "178474", "patient_id": "93387170", "patient_name": "JULIO CESAR ACOSTA GUEVARA", } use_case = RunIndividualPrecheckUseCase( repository=repository, file_store=FakeFileStore(), text_extractor=FakeTextExtractor("Caso No. 178474N\nCC: 93387170"), classifier=FakeClassifier("factura"), clinical_document_service=clinical_service, case_association_service=FakeCaseAssociationService( patient_id="93387170", case_number="178474N", patient_name="", ), dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertEqual(result["pending_actions"], []) self.assertEqual(result["contradictions"], []) class IndividualConfirmAndCancelUseCaseTest(unittest.TestCase): def test_confirm_accepts_empty_fields_when_source_identity_is_redacted(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "historia_clinica", "detected_document_type": "historia_clinica", "pending_actions": [PENDING_ACTION_CONFIRM_CASE_IDENTITY], "provisional_case_number": "1390913", "provisional_patient_id": "", "provisional_patient_name": "", "redacted_identity_fields": ["patient_id", "patient_name"], } dispatcher = FakeDispatcher() use_case = ConfirmIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute(upload_id="1", username="auditor") self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertEqual(result["confirmed_case_number"], "1390913") self.assertEqual(result["confirmed_patient_id"], "") self.assertEqual(result["confirmed_patient_name"], "") self.assertEqual( set(result["redacted_identity_fields"]), {"patient_id", "patient_name"}, ) self.assertEqual(dispatcher.materialization_calls, ["1"]) def test_confirm_still_requires_case_number_for_redacted_identity(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "historia_clinica", "detected_document_type": "historia_clinica", "pending_actions": [PENDING_ACTION_CONFIRM_CASE_IDENTITY], "redacted_identity_fields": ["patient_id", "patient_name"], } dispatcher = FakeDispatcher() use_case = ConfirmIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) with self.assertRaisesRegex(ValueError, "número de caso"): use_case.execute(upload_id="1", username="auditor") self.assertEqual(dispatcher.materialization_calls, []) def test_confirm_accepts_manual_identity_without_losing_redaction_trace(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "historia_clinica", "detected_document_type": "historia_clinica", "pending_actions": [PENDING_ACTION_CONFIRM_CASE_IDENTITY], "provisional_case_number": "1390913", "redacted_identity_fields": ["patient_id", "patient_name"], } dispatcher = FakeDispatcher() use_case = ConfirmIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute( upload_id="1", username="auditor", confirmed_patient_id="123456789", confirmed_patient_name="Paciente Confirmado", ) self.assertEqual(result["confirmed_patient_id"], "123456789") self.assertEqual(result["confirmed_patient_name"], "Paciente Confirmado") self.assertEqual( set(result["redacted_identity_fields"]), {"patient_id", "patient_name"}, ) def test_confirm_dispatches_materialization_and_persists_override_audit(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "factura", "detected_document_type": "radiologia", "pending_actions": [PENDING_ACTION_CONFIRM_TYPE_OVERRIDE], "warnings": ["Mismatch."], "contradictions": [], "provisional_case_number": "", "provisional_patient_id": "", "provisional_patient_name": "", } dispatcher = FakeDispatcher() use_case = ConfirmIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute(upload_id="1", username="auditor") self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertFalse(result["selected_type_override_confirmed"]) self.assertEqual(result["effective_document_type"], "radiologia") self.assertEqual(result["override_audit"]["confirmed_by"], "auditor") self.assertEqual(result["override_audit"]["effective_document_type"], "radiologia") self.assertEqual(dispatcher.materialization_calls, ["1"]) def test_confirm_can_force_selected_document_type(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "factura", "detected_document_type": "radiologia", "effective_document_type": "radiologia", "pending_actions": [PENDING_ACTION_CONFIRM_TYPE_OVERRIDE], "warnings": ["Mismatch."], "contradictions": [], "provisional_case_number": "", "provisional_patient_id": "", "provisional_patient_name": "", } dispatcher = FakeDispatcher() use_case = ConfirmIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute( upload_id="1", username="auditor", confirmed_effective_document_type="factura", ) self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertTrue(result["selected_type_override_confirmed"]) self.assertEqual(result["effective_document_type"], "factura") self.assertEqual(result["override_audit"]["effective_document_type"], "factura") self.assertEqual(dispatcher.materialization_calls, ["1"]) def test_cancel_clears_temp_state_and_marks_audit_only(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "stored_path": "/tmp/1-soporte.pdf", "extracted_text": "texto", "selected_document_type": "factura", } file_store = FakeFileStore() use_case = CancelIndividualUploadUseCase( repository=repository, file_store=file_store, colombia_tz=UTC, ) result = use_case.execute(upload_id="1", username="auditor", reason="cancelled") self.assertEqual(result["status"], UPLOAD_STATUS_CANCELADO) self.assertEqual(result["stored_path"], "") self.assertEqual(result["extracted_text"], "") self.assertTrue(result["audit_only"]) self.assertEqual(file_store.deleted, ["/tmp/1-soporte.pdf"]) class IndividualMaterializationUseCaseTest(unittest.TestCase): def test_worker_does_not_resurrect_document_deleted_during_processing(self) -> None: class DeleteBeforeCompletionRepository(FakeRepository): def update_upload_if_status( self, upload_id: str, expected_statuses: set[str], payload: dict ) -> bool: if payload.get("status") == UPLOAD_STATUS_COMPLETADO: self.records[upload_id].update( { "status": UPLOAD_STATUS_ELIMINADO, "deleted_at": "2026-08-18T12:00:00+00:00", "deletion_reason": "case_deleted", } ) return False return super().update_upload_if_status(upload_id, expected_statuses, payload) class FakeDocumentCleanup: def __init__(self) -> None: self.purged: list[tuple[str, str]] = [] def purge_document_record(self, *, username: str, document_id: str) -> bool: self.purged.append((username, document_id)) return True repository = DeleteBeforeCompletionRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/1-factura.pdf", "extracted_text": "texto", "selected_document_type": "factura", "detected_document_type": "factura", "effective_document_type": "factura", "source_file_hash": "hash-1", "provided_case_key": "CASE-1", "original_name": "factura.pdf", } cleanup = FakeDocumentCleanup() service = FakeClinicalDocumentService() use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=FakeFileStore(), clinical_document_service=service, colombia_tz=UTC, document_cleanup_service=cleanup, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_ELIMINADO) self.assertEqual(cleanup.purged, [("auditor", "doc-1")]) self.assertEqual(len(service.persist_calls), 1) def test_support_waits_until_base_story_completes(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PROCESANDO, "base_upload_id": "1", "session_id": "manual-1", "session_status": "active", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/2-factura.pdf", "extracted_text": "texto", "selected_document_type": "factura", "detected_document_type": "factura", "effective_document_type": "factura", "source_file_hash": "hash-2", "provided_case_key": "CASE-1", "depends_on_upload_id": "1", "session_id": "manual-1", "session_status": "active", "original_name": "factura.pdf", } use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=FakeFileStore(), clinical_document_service=FakeClinicalDocumentService(), colombia_tz=UTC, ) result = use_case.execute("2") self.assertEqual(result["status"], UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION) self.assertEqual(result["waiting_for_base_upload_id"], "1") def test_story_completion_releases_waiting_supports(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/1-historia.pdf", "extracted_text": "texto base", "selected_document_type": "historia_clinica", "detected_document_type": "historia_clinica", "effective_document_type": "historia_clinica", "source_file_hash": "hash-1", "base_upload_id": "1", "session_id": "manual-1", "session_status": "active", "case_key": "CASE-1", "session_case_key": "CASE-1", "confirmed_case_number": "HC-001", "confirmed_patient_id": "111222333", "confirmed_patient_name": "Paciente Uno", "original_name": "historia.pdf", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "status": UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION, "stored_path": "/tmp/2-factura.pdf", "extracted_text": "texto soporte", "selected_document_type": "factura", "detected_document_type": "factura", "effective_document_type": "factura", "source_file_hash": "hash-2", "provided_case_key": "CASE-1", "depends_on_upload_id": "1", "session_id": "manual-1", "session_status": "active", "confirmed_case_number": "HC-001", "confirmed_patient_id": "111222333", "confirmed_patient_name": "Paciente Uno", "original_name": "factura.pdf", } file_store = FakeFileStore() service = FakeClinicalDocumentService() use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=file_store, clinical_document_service=service, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_COMPLETADO) dependent_record = cast(dict[str, Any], repository.get_upload("2")) self.assertIsNotNone(dependent_record) self.assertEqual(dependent_record["status"], UPLOAD_STATUS_COMPLETADO) self.assertEqual(len(service.persist_calls), 2) def test_story_failure_blocks_waiting_supports(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_CANCELADO, "base_upload_id": "1", "session_id": "manual-1", "session_status": "cancelled", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/2-factura.pdf", "extracted_text": "texto soporte", "selected_document_type": "factura", "detected_document_type": "factura", "effective_document_type": "factura", "source_file_hash": "hash-2", "provided_case_key": "CASE-1", "depends_on_upload_id": "1", "session_id": "manual-1", "session_status": "active", "confirmed_case_number": "HC-001", "confirmed_patient_id": "111222333", "confirmed_patient_name": "Paciente Uno", "original_name": "factura.pdf", } use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=FakeFileStore(), clinical_document_service=FakeClinicalDocumentService(), colombia_tz=UTC, ) result = use_case.execute("2") self.assertEqual(result["status"], UPLOAD_STATUS_BLOCKED_BASE_FAILED) self.assertTrue(result["blocked_by_base_failure"]) def test_materialization_uses_effective_type_and_cleans_file(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/1-soporte.pdf", "extracted_text": "texto", "selected_document_type": "factura", "detected_document_type": "radiologia", "effective_document_type": "radiologia", "source_file_hash": "hash-1", "selected_type_override_confirmed": False, "override_audit": {"confirmed_by": "auditor"}, "provided_case_key": "CASE-1", "confirmed_case_number": "HC-001", "confirmed_patient_id": "111222333", "confirmed_patient_name": "Paciente Uno", "original_name": "factura.pdf", } file_store = FakeFileStore() service = FakeClinicalDocumentService() use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=file_store, clinical_document_service=service, colombia_tz=UTC, ) result = use_case.execute("1") self.assertEqual(result["status"], UPLOAD_STATUS_COMPLETADO) self.assertEqual(result["clinical_document_id"], "doc-1") self.assertEqual(file_store.deleted, ["/tmp/1-soporte.pdf"]) self.assertEqual(service.persist_calls[0].selected_document_type, "factura") self.assertEqual(service.persist_calls[0].detected_document_type, "radiologia") self.assertEqual(service.persist_calls[0].effective_document_type, "radiologia") self.assertEqual(service.persist_calls[0].detected_type, "radiologia") self.assertFalse(service.persist_calls[0].category_override_confirmed) def test_recoverable_story_failure_keeps_base_and_support_artifacts(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_LISTO_PARA_PROCESAR, "stored_path": "/tmp/1-historia.pdf", "extracted_text": "texto base", "selected_document_type": "historia_clinica", "detected_document_type": "historia_clinica", "effective_document_type": "historia_clinica", "source_file_hash": "hash-1", "base_upload_id": "1", "session_id": "manual-1", "session_status": "active", "original_name": "historia.pdf", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "status": UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION, "stored_path": "/tmp/2-factura.pdf", "extracted_text": "texto soporte", "selected_document_type": "factura", "depends_on_upload_id": "1", "session_id": "manual-1", "session_status": "active", "original_name": "factura.pdf", } file_store = FakeFileStore() use_case = RunIndividualMaterializationUseCase( repository=repository, file_store=file_store, clinical_document_service=FailingClinicalDocumentService(), colombia_tz=UTC, ) result = use_case.execute("1") dependent = cast(dict[str, Any], repository.get_upload("2")) self.assertEqual(result["status"], UPLOAD_STATUS_FALLIDO) self.assertTrue(result["can_retry"]) self.assertEqual(result["failure"]["source"], "external_provider") self.assertEqual(result["failure"]["failed_chunks"], [2]) self.assertEqual(result["stored_path"], "/tmp/1-historia.pdf") self.assertEqual(result["extracted_text"], "texto base") self.assertEqual(dependent["status"], UPLOAD_STATUS_BLOCKED_BASE_FAILED) self.assertEqual(dependent["stored_path"], "/tmp/2-factura.pdf") self.assertEqual(dependent["extracted_text"], "texto soporte") self.assertEqual(file_store.deleted, []) class RetryIndividualUploadUseCaseTest(unittest.TestCase): def test_legacy_coverage_failure_is_projected_as_retryable_when_artifacts_remain(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_FALLIDO, "stored_path": "/tmp/1-historia.pdf", "extracted_text": "texto base", "selected_document_type": "historia_clinica", "error": "No fue posible resumir todos los bloques de la historia clínica.", } status = cast( dict[str, Any], GetIndividualUploadStatusUseCase(repository=repository).execute( "1", username="auditor", ), ) self.assertTrue(status["can_retry"]) self.assertEqual(status["failure"]["source"], "external_provider") self.assertEqual(status["failure"]["provider"], "gemini") def test_retry_reuses_same_record_and_restores_dependent_supports(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_FALLIDO, "stored_path": "/tmp/1-historia.pdf", "extracted_text": "texto base", "selected_document_type": "historia_clinica", "base_upload_id": "1", "session_id": "manual-1", "session_status": "blocked", "retry_count": 0, "failure": { "retryable": True, "source": "external_provider", "public_message": "Falla temporal.", }, } repository.records["2"] = { "_id": "2", "usuario": "auditor", "status": UPLOAD_STATUS_BLOCKED_BASE_FAILED, "stored_path": "/tmp/2-factura.pdf", "extracted_text": "texto soporte", "depends_on_upload_id": "1", "session_id": "manual-1", "session_status": "blocked", } dispatcher = FakeDispatcher() use_case = RetryIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) status = cast( dict[str, Any], GetIndividualUploadStatusUseCase(repository=repository).execute( "1", username="auditor", ), ) result = use_case.execute(upload_id="1", username="auditor") dependent = cast(dict[str, Any], repository.get_upload("2")) self.assertTrue(status["can_retry"]) self.assertEqual(status["retry_url"], "/api/cargue-individual/1/reintentar") self.assertEqual(result["status"], UPLOAD_STATUS_LISTO_PARA_PROCESAR) self.assertEqual(result["retry_count"], 1) self.assertEqual(result["failure"], {}) self.assertFalse(result["can_retry"]) self.assertEqual(dispatcher.materialization_calls, ["1"]) self.assertEqual(dependent["status"], UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION) self.assertEqual(dependent["stored_path"], "/tmp/2-factura.pdf") def test_retry_is_idempotent_while_upload_is_processing(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_PROCESANDO, "retry_count": 1, } dispatcher = FakeDispatcher() use_case = RetryIndividualUploadUseCase( repository=repository, dispatcher=dispatcher, colombia_tz=UTC, ) result = use_case.execute(upload_id="1", username="auditor") self.assertEqual(result["status"], UPLOAD_STATUS_PROCESANDO) self.assertEqual(dispatcher.materialization_calls, []) def test_retry_rejects_missing_artifacts(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "status": UPLOAD_STATUS_FALLIDO, "failure": {"retryable": True}, } use_case = RetryIndividualUploadUseCase( repository=repository, dispatcher=FakeDispatcher(), colombia_tz=UTC, ) with self.assertRaisesRegex(ValueError, "no es recuperable"): use_case.execute(upload_id="1", username="auditor") class IndividualSessionUseCaseTest(unittest.TestCase): def test_active_session_view_returns_case_key_and_queue(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "session_case_key": "CASE-1", "session_status": "active", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "selected_document_type": "historia_clinica", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "depends_on_upload_id": "1", "session_case_key": "CASE-1", "session_status": "active", "status": UPLOAD_STATUS_WAITING_BASE_MATERIALIZATION, "selected_document_type": "factura", } use_case = GetActiveIndividualUploadSessionUseCase(repository=repository) result = cast(dict[str, Any], use_case.execute("auditor")) self.assertIsNotNone(result) self.assertEqual(result["active_case_key"], "CASE-1") self.assertEqual(len(result["uploads"]), 2) def test_list_sessions_returns_multiple_active_sessions_sorted_by_recency(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "session_case_key": "CASE-1", "session_status": "active", "status": UPLOAD_STATUS_COMPLETADO, "updated_at": "2026-06-01T10:00:00+00:00", "selected_document_type": "historia_clinica", } repository.records["2"] = { "_id": "2", "usuario": "auditor", "session_id": "manual-2", "base_upload_id": "2", "session_case_key": "CASE-2", "session_status": "active", "status": UPLOAD_STATUS_ESPERANDO_CONFIRMACION, "updated_at": "2026-06-02T10:00:00+00:00", "selected_document_type": "historia_clinica", } result = ListIndividualUploadSessionsUseCase(repository=repository).execute("auditor") self.assertEqual([item["session_id"] for item in result], ["manual-2", "manual-1"]) def test_close_session_requires_terminal_uploads(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "session_case_key": "CASE-1", "session_status": "active", "status": UPLOAD_STATUS_PROCESANDO, } use_case = CloseIndividualUploadSessionUseCase(repository=repository, colombia_tz=UTC) with self.assertRaisesRegex(ValueError, "Debes cancelar o completar"): use_case.execute(username="auditor", session_id="manual-1") def test_get_session_returns_specific_session_view(self) -> None: repository = FakeRepository() repository.records["1"] = { "_id": "1", "usuario": "auditor", "session_id": "manual-1", "base_upload_id": "1", "session_case_key": "CASE-1", "session_status": "completed", "status": UPLOAD_STATUS_COMPLETADO, "updated_at": "2026-06-03T10:00:00+00:00", "selected_document_type": "historia_clinica", } result = GetIndividualUploadSessionUseCase(repository=repository).execute( "auditor", session_id="manual-1", ) self.assertIsNotNone(result) assert result is not None self.assertEqual(result["session_id"], "manual-1") class CapturingCreateUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(kwargs) return { "upload_id": "upload-1", "status": "precheck_en_cola", "created_at": "2026-06-03T00:00:00-05:00", "poll_url": "/api/cargue-individual/upload-1", "duplicate": False, "message": "Documento aceptado para precheck.", } class CapturingConfirmUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(kwargs) return {"status": "listo_para_procesar"} class CapturingCancelUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(kwargs) return {"status": "cancelado"} class CapturingRetryUpload: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(kwargs) return {"status": "listo_para_procesar"} class CapturingCloseSession: def __init__(self) -> None: self.calls: list[dict] = [] def execute(self, **kwargs): self.calls.append(kwargs) return { "session_id": kwargs["session_id"], "active_case_key": "CASE-1", "session_status": "completed", "uploads": [], } class FakeDemoIdentityService: enabled = True def resolve_case_key(self, *, username: str, visible_case_key: str) -> str: return "CASE-1" if visible_case_key == "00001-CASE-1-paciente-01" else visible_case_key def resolve_patient_id(self, *, username: str, visible_patient_id: str) -> str: return "111" if visible_patient_id == "00001" else visible_patient_id def resolve_patient_name(self, *, username: str, visible_patient_name: str) -> str: return "Paciente" if visible_patient_name == "paciente-01" else visible_patient_name def project_case_identity( self, *, username: str, case_key: str, case_number: str = "", patient_id: str = "", patient_name: str = "", ) -> dict[str, str]: return { "case_key": "00001-CASE-1-paciente-01", "case_number": case_number, "patient_id": "00001", "patient_name": "paciente-01", "nombre_paciente": "paciente-01", } class IndividualIngestionRoutesTest(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: self.app = FastAPI() self.app.include_router(individual_ingestion.router) self.create_upload = CapturingCreateUpload() self.confirm_upload = CapturingConfirmUpload() self.cancel_upload = CapturingCancelUpload() self.retry_upload = CapturingRetryUpload() self.close_session = CapturingCloseSession() self.app.state.services = SimpleNamespace( individual_ingestion_runtime=SimpleNamespace( create_upload=self.create_upload, get_latest_pending=SimpleNamespace( execute=lambda username: { "upload_id": "upload-pending", "status": "esperando_confirmacion", } ), get_active_session=SimpleNamespace( execute=lambda username: { "session_id": "manual-1", "active_case_key": "CASE-1", "uploads": [{"upload_id": "upload-pending", "status": "esperando_confirmacion"}], } ), get_session=SimpleNamespace( execute=lambda username, session_id: { "session_id": session_id, "active_case_key": "CASE-1", "session_status": "active", "uploads": [{"upload_id": "upload-pending", "status": "esperando_confirmacion"}], } ), list_sessions=SimpleNamespace( execute=lambda username, include_terminal=False: [ { "session_id": "manual-1", "active_case_key": "CASE-1", "session_status": "active", "uploads": [{"upload_id": "upload-pending", "status": "esperando_confirmacion"}], } ] ), get_upload_status=SimpleNamespace( execute=lambda upload_id, username: {"upload_id": upload_id, "status": "procesando"} ), confirm_upload=self.confirm_upload, cancel_upload=self.cancel_upload, retry_upload=self.retry_upload, close_session=self.close_session, ), demo_identity_service=FakeDemoIdentityService(), ) async def override_current_user(): return SimpleNamespace(username="auditor", role="auditor") self.app.dependency_overrides[get_current_user] = override_current_user self.transport = ASGITransport(app=self.app) async def asyncTearDown(self) -> None: self.app.dependency_overrides.clear() async def test_historia_route_returns_202_and_uses_historia_type(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/cargue-individual/historia", files={"file": ("historia.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) self.assertEqual(self.create_upload.calls[0]["selected_document_type"], "historia_clinica") async def test_soporte_route_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/cargue-individual/soporte", data={"tipo_documento": "factura", "case_key": "CASE-1"}, files={"file": ("factura.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) self.assertEqual(self.create_upload.calls[0]["selected_document_type"], "factura") self.assertEqual(self.create_upload.calls[0]["provided_case_key"], "CASE-1") async def test_soporte_route_accepts_session_context(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/cargue-individual/soporte", data={ "tipo_documento": "factura", "case_key": "CASE-1", "session_id": "manual-1", "lock_case_selection": "true", }, files={"file": ("factura.pdf", b"%PDF-1.4 fake", "application/pdf")}, ) self.assertEqual(response.status_code, 202) self.assertEqual(self.create_upload.calls[-1]["session_id"], "manual-1") self.assertTrue(self.create_upload.calls[-1]["lock_case_selection"]) async def test_pending_latest_returns_payload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/cargue-individual/pending-latest") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["upload"]["upload_id"], "upload-pending") async def test_active_session_route_returns_payload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/cargue-individual/sesion-activa") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["session"]["active_case_key"], "00001-CASE-1-paciente-01") async def test_list_sessions_route_returns_payload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/cargue-individual/sesiones") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["sessions"][0]["session_id"], "manual-1") async def test_get_session_route_returns_payload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.get("/api/cargue-individual/sesiones/manual-1") self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["session"]["session_id"], "manual-1") async def test_confirm_route_passes_identity_fields(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/cargue-individual/upload-1/confirmar", json={ "case_number": "HC-001", "patient_id": "111", "patient_name": "Paciente", "effective_document_type": "radiologia", }, ) self.assertEqual(response.status_code, 202) self.assertEqual(self.confirm_upload.calls[0]["confirmed_case_number"], "HC-001") self.assertEqual(self.confirm_upload.calls[0]["confirmed_patient_id"], "111") self.assertEqual(self.confirm_upload.calls[0]["confirmed_patient_name"], "Paciente") self.assertEqual(self.confirm_upload.calls[0]["confirmed_effective_document_type"], "radiologia") async def test_cancel_route_passes_reason(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post( "/api/cargue-individual/upload-1/cancelar", json={"reason": "cancelled_from_dialog"}, ) self.assertEqual(response.status_code, 200) self.assertEqual(self.cancel_upload.calls[0]["reason"], "cancelled_from_dialog") async def test_retry_route_requeues_same_upload(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post("/api/cargue-individual/upload-1/reintentar") self.assertEqual(response.status_code, 202) self.assertEqual(response.json()["upload_id"], "upload-1") self.assertEqual(self.retry_upload.calls[0]["upload_id"], "upload-1") self.assertEqual(self.retry_upload.calls[0]["username"], "auditor") async def test_close_session_route_passes_session_id(self) -> None: async with AsyncClient(transport=self.transport, base_url="http://testserver") as client: response = await client.post("/api/cargue-individual/sesiones/manual-1/cerrar") self.assertEqual(response.status_code, 200) self.assertEqual(self.close_session.calls[0]["session_id"], "manual-1")