Sin contenido clínico visible.
", "analysis_structured": {}, "nombre_paciente": "LUISA FERNANDA IBARRA CANO", "case_number": "203922", } service._ensure_prefactura_historia_visibility(payload) self.assertIn("Texto clínico consolidado", payload["analisis_html"]) class LocalBatchArtifactCleanerTest(unittest.TestCase): def test_delete_archive_prunes_batch_dir_but_keeps_root(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: cleaner = LocalBatchArtifactCleaner(Path(tmp_dir)) archive_dir = Path(tmp_dir) / "_batch_uploads" / "batch-1" archive_dir.mkdir(parents=True, exist_ok=True) archive_path = archive_dir / "demo.zip" archive_path.write_bytes(b"zip") cleaner.delete_archive(str(archive_path)) self.assertFalse(archive_path.exists()) self.assertFalse(archive_dir.exists()) self.assertTrue((Path(tmp_dir) / "_batch_uploads").exists()) def test_delete_extracted_file_prunes_nested_dirs_but_keeps_root(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: cleaner = LocalBatchArtifactCleaner(Path(tmp_dir)) nested_dir = Path(tmp_dir) / "_batch_extracted" / "batch-1" / "folder" nested_dir.mkdir(parents=True, exist_ok=True) extracted_path = nested_dir / "doc.pdf" extracted_path.write_bytes(b"pdf") cleaner.delete_extracted_file(str(extracted_path)) self.assertFalse(extracted_path.exists()) self.assertFalse(nested_dir.exists()) self.assertTrue((Path(tmp_dir) / "_batch_extracted").exists()) class LocalBatchWorkingFileStoreTest(unittest.TestCase): def test_persists_single_manual_pdf_inside_batch_extracted_root(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: store = LocalBatchWorkingFileStore(Path(tmp_dir)) stored_path = Path(store.save_file("batch-1", "../../historia final.PDF", b"%PDF-1.4 fake")) self.assertTrue(stored_path.exists()) self.assertEqual(stored_path.suffix, ".pdf") self.assertTrue(str(stored_path).startswith(str(Path(tmp_dir) / "_batch_extracted" / "batch-1"))) class HeuristicDocumentClassifierTest(unittest.TestCase): def test_detects_factura_from_text(self) -> None: classifier = HeuristicDocumentClassifier() detected_type = classifier.classify( "documento.pdf", "Factura Electrónica de Venta. Número de factura FV-100. Valor total de la factura.", ) self.assertEqual(detected_type, "factura") def test_detects_prefactura_with_case_number_without_patient_id(self) -> None: classifier = HeuristicDocumentClassifier() detected_type = classifier.classify( "prefactura.pdf", "Prefactura de servicios\nCaso No. 176654\nDetalle preliminar de servicios", ) self.assertEqual(detected_type, "prefactura") def test_keeps_real_invoice_as_factura(self) -> None: classifier = HeuristicDocumentClassifier() detected_type = classifier.classify( "factura.pdf", "Factura Electrónica de Venta\nCaso No. 176654\nCC 66781911\nNúmero de factura FV-001", ) self.assertEqual(detected_type, "factura") def test_classifies_page_titles_for_prefactura_supports(self) -> None: classifier = HeuristicDocumentClassifier() self.assertEqual( classifier.describe("doc.pdf", "Hoja de Drogas\nPaciente: LUISA FERNANDA IBARRA CANO"), ("prescripcion", "Hoja de Drogas"), ) self.assertEqual( classifier.describe( "doc.pdf", "Ordenes de Paraclínicos Generadas en Historias Clinicas\nRADIOLOGIA" ), ("radiologia", "Órdenes de Paraclínicos Generadas en Historias Clínicas"), ) self.assertEqual( classifier.describe("doc.pdf", "NOTA OBSERVACIÓN\nCaso: 203922"), ("historia_clinica", "Nota Observación"), ) self.assertEqual( classifier.describe("doc.pdf", "EVOLUCIÓN MÉDICA\nCaso: 203922"), ("historia_clinica", "Evolución Médica"), ) self.assertEqual( classifier.describe("doc.pdf", PREF203922_ORDEN_MEDICA), ("prescripcion", "Órdenes Médicas Generadas en Historias Clínicas"), ) self.assertEqual( classifier.describe("doc.pdf", PREF203922_ORDEN_GENERAL), ("prescripcion", "Órdenes Generadas en Historias Clínicas"), ) self.assertEqual( classifier.describe("doc.pdf", PREF203922_ORDEN_MEDICAMENTOS), ("prescripcion", "Órdenes de Medicamentos Generadas en Historias Clínicas"), ) self.assertEqual( classifier.describe("doc.pdf", PREF203922_SALA), ("historia_clinica", "Notas Sala Procedimientos"), ) def test_falls_back_to_generico(self) -> None: classifier = HeuristicDocumentClassifier() detected_type = classifier.classify("misterioso.pdf", "contenido neutro sin señales clínicas claras") self.assertEqual(detected_type, "generico") class HeuristicAssociationServiceTest(unittest.TestCase): BLANCA_QUIRURGICO = """ DESCRIPCIÓN QUIRÚRGICA NOMBRE PACIENTE BLANCA LIJIA RENGIFO AGUIRRE TIPO Y N° DOCUMENTO CC 66781911SEXO F F. NACIMIENTO 24/11/1976N° CASO 176654 PROCEDIMIENTOS REALIZADOS LAVADO + DESBRIDAMIENTO DE LESION DE TEJIDO PROFUNDO DE HERIDA EN PIE IZQUIERDO """ BLANCA_FACTURA = """ Factura Electrónica de Venta Caso No. 176654 Fecha Ingreso: 17/03/2024 06:24 Paciente: BLANCA LIJIA RENGIFO AGUIRRE Fecha Egreso: 18/03/2024 10:54 CC: 66781911 17-03-2024 13563 Reducción cerrada falanges pie (una a dos) """ BLANCA_HC = """ Caso: 176654 Telefonos: 3147759658 - Ciudad: CALI (SANTIAGO DE CALI)Dirección: DIAGONAL 72 C #26 J 55CC - 66781911Identificación FEMENINOSexo BLANCA LIJIA RENGIFO AGUIRRENombre del Paciente 176654No. de Caso: Telefonos: 3147759658 """ MONICA_QUIRURGICO = """ DESCRIPCIÓN QUIRÚRGICA NOMBRE PACIENTE MONICA LORENA PARRA GIRALDO TIPO Y N° DOCUMENTO CC 29465123SEXO F F. NACIMIENTO 19/07/1982N° CASO 178550 N° ADMISIÓN 298498 EDAD 41 AÑOS """ MONICA_HC = """ Caso: 178550 NO. ADMISION: 298353Page 1 of 2 HISTORIA CLINICA DE URGENCIAS Inversiones Médicas Valle Salud S.A.S Nit: 900631361 6 Valle Salud CALI)Dirección: CALLE 72 # 7 F 46CC - 29465123Identificación FEMENINOSexo MONICA LORENA PARRA GIRALDONombre del Paciente 178550No. de Caso: """ MONICA_FACTURA = """ Factura Electrónica de Venta Caso No. 178550 Fecha Ingreso: 09/05/2024 09:09 Paciente: MONICA LORENA PARRA GIRALDO Fecha Egreso: 13/05/2024 10:37 CC: 29465123 """ JULIO_QUIRURGICO = """ DESCRIPCIÓN QUIRÚRGICA NOMBRE PACIENTE JULIO CESAR ACOSTA GUEVARA TIPO Y N° DOCUMENTO CC 93387170SEXO M F. NACIMIENTO 02/02/1990N° CASO 178474 FECHA Y HORA INICIO 11/05/2024 12:30 """ JULIO_FACTURA = """ Factura Electrónica de Venta Caso No. 178474 Fecha Ingreso: 11/05/2024 12:30 Paciente: JULIO CESAR ACOSTA GUEVARA Fecha Egreso: 13/05/2024 10:54 CC: 93387170 """ JULIO_HC = """ Caso: 178474 JULIO CESAR ACOSTA GUEVARANombre del Paciente 178474No. de Caso: CC - 93387170Identificación """ def test_extract_signals_ignores_phone_number_as_patient_id(self) -> None: service = HeuristicCaseAssociationService() text = ( "Paciente: ANA MARIA LOPEZ\n" "Telefono de contacto: 3001234567\n" "Sin identificación clínica explícita." ) signals = service.extract_signals("factura.pdf", text, "factura") self.assertEqual(signals.patient_id, "") self.assertEqual(signals.patient_name, "ANA MARIA LOPEZ") def test_extract_signals_discards_generic_patient_name(self) -> None: service = HeuristicCaseAssociationService() text = "Paciente: Fecha Egreso\nTelefono: 3101234567" signals = service.extract_signals("factura_fecha_egreso.pdf", text, "factura") self.assertEqual(signals.patient_name, "") def test_extract_signals_preserves_masked_identity_without_numeric_fallback(self) -> None: service = HeuristicCaseAssociationService() text = """ No. Historia: CCxxxxxxxxxx - Admision: 1382076 - Paciente: xxxxxxxxxxxxxxxxx1 de 130 No. H. C. CCxxxxxxxxxxxx - 1382076 PACIENTE Xxxxxxxxxxxxxxxxx DOC. ID. CC - xxxxxxxxxxxxxxxx TELEFONO 3175809671 NIT 891380054-1 """ signals = service.extract_signals("HC13820761.pdf", text, "historia_clinica") self.assertEqual(signals.case_number, "1382076") self.assertEqual(signals.patient_id, "") self.assertEqual(signals.patient_name, "") self.assertEqual( set(signals.redacted_identity_fields), {"patient_id", "patient_name"}, ) self.assertIn("identificacion_censurada", signals.evidence) self.assertIn("nombre_paciente_censurado", signals.evidence) def test_extract_signals_from_blanca_documents(self) -> None: service = HeuristicCaseAssociationService() quirurgico = service.extract_signals( "CIRUGIABLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_QUIRURGICO, "quirurgico", ) factura = service.extract_signals( "FACTURA BLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_FACTURA, "factura", ) historia = service.extract_signals( "HISTORIA CLINICA BLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_HC, "historia_clinica", ) self.assertEqual(quirurgico.patient_name, "BLANCA LIJIA RENGIFO AGUIRRE") self.assertEqual(quirurgico.patient_id, "66781911") self.assertEqual(quirurgico.case_number, "176654") self.assertEqual(quirurgico.procedure_code, "") self.assertTrue(quirurgico.procedure_description.startswith("LAVADO + DESBRIDAMIENTO")) self.assertEqual(factura.patient_name, "BLANCA LIJIA RENGIFO AGUIRRE") self.assertEqual(factura.patient_id, "66781911") self.assertEqual(factura.case_number, "176654") self.assertEqual(factura.procedure_code, "") self.assertEqual(historia.patient_name, "BLANCA LIJIA RENGIFO AGUIRRE") self.assertEqual(historia.patient_id, "66781911") self.assertEqual(historia.case_number, "176654") def test_extract_case_number_from_header_variants(self) -> None: service = HeuristicCaseAssociationService() quirurgico = service.extract_signals( "CIRUGIA MONICA LORENA PARRA GIRALDO.pdf", self.MONICA_QUIRURGICO, "quirurgico", ) historia = service.extract_signals( "HISTORIA CLINICA MONICA LORENA PARRA GIRALDO.pdf", self.MONICA_HC, "historia_clinica", ) self.assertEqual(quirurgico.case_number, "178550") self.assertEqual(historia.case_number, "178550") def test_associates_matching_documents_with_score_breakdown(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "1", "batch_id": "batch-1", "original_name": "historia_ana.pdf", "patient_name": "Ana Perez", "patient_id": "12345678", "case_number": "CASE01", "service_date": "01/03/2026", "procedure_code": "", "procedure_description": "", "evidence": ["identificacion", "numero_caso"], }, { "_id": "2", "batch_id": "batch-1", "original_name": "factura_ana.pdf", "patient_name": "Ana Perez", "patient_id": "12345678", "case_number": "CASE01", "service_date": "01/03/2026", "procedure_code": "123456", "procedure_description": "Procedimiento", "evidence": ["identificacion", "numero_caso"], }, ] result = service.associate(files) self.assertEqual(len(result.cases), 1) self.assertEqual({decision.status for decision in result.decisions}, {"asociado"}) second_decision = next(item for item in result.decisions if item.file_id == "2") self.assertIn("identificacion", second_decision.score_breakdown) self.assertIn("numero_caso", second_decision.score_breakdown) self.assertEqual(second_decision.association_source, "auto") self.assertGreaterEqual(second_decision.confidence, 0.0) self.assertLessEqual(second_decision.confidence, 1.0) def test_marks_ambiguous_document_with_top_candidates(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "1", "batch_id": "batch-1", "original_name": "doc_ana.pdf", "patient_name": "Ana Perez", "patient_id": "1111", "case_number": "CASEA", "service_date": "", "procedure_code": "123456", "procedure_description": "", "evidence": ["identificacion", "numero_caso"], }, { "_id": "2", "batch_id": "batch-1", "original_name": "doc_beto.pdf", "patient_name": "Beto Diaz", "patient_id": "2222", "case_number": "CASEB", "service_date": "", "procedure_code": "123456", "procedure_description": "", "evidence": ["identificacion", "numero_caso"], }, { "_id": "3", "batch_id": "batch-1", "original_name": "doc_ambiguous.pdf", "patient_name": "", "patient_id": "", "case_number": "", "service_date": "", "procedure_code": "123456", "procedure_description": "", "evidence": ["codigo_procedimiento"], }, ] result = service.associate(files) ambiguous = next(item for item in result.decisions if item.file_id == "3") self.assertEqual(ambiguous.status, "pendiente_validacion") self.assertGreaterEqual(len(ambiguous.top_candidates), 1) self.assertEqual(ambiguous.score_breakdown, {}) self.assertEqual(ambiguous.confidence, 0.0) self.assertTrue( all(0.0 <= float(candidate.get("score", 0.0)) <= 1.0 for candidate in ambiguous.top_candidates) ) def test_does_not_seed_cluster_with_name_only(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "1", "batch_id": "batch-1", "original_name": "historia_ana_perez.pdf", "patient_name": "Ana Perez", "patient_id": "", "case_number": "", "service_date": "", "procedure_code": "", "procedure_description": "", "evidence": ["nombre_paciente"], }, ] result = service.associate(files) self.assertEqual(len(result.cases), 0) self.assertEqual(result.decisions[0].status, "pendiente_validacion") def test_marks_pending_when_anchor_score_below_high_precision_threshold(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "1", "batch_id": "batch-1", "original_name": "doc_seed.pdf", "patient_name": "Ana Perez", "patient_id": "12345678", "case_number": "CASE01", "service_date": "", "procedure_code": "", "procedure_description": "", "evidence": ["identificacion", "numero_caso", "nombre_paciente"], }, { "_id": "2", "batch_id": "batch-1", "original_name": "doc_second.pdf", "patient_name": "", "patient_id": "12345678", "case_number": "CASE01", "service_date": "", "procedure_code": "", "procedure_description": "", "evidence": ["identificacion", "numero_caso"], }, ] result = service.associate(files) decision = next(item for item in result.decisions if item.file_id == "2") self.assertEqual(decision.status, "asociado") self.assertAlmostEqual(decision.confidence, 0.8, places=4) self.assertGreaterEqual(len(decision.top_candidates), 1) def test_associates_blanca_batch_into_single_case(self) -> None: service = HeuristicCaseAssociationService() docs = [ ( "1", "CIRUGIABLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_QUIRURGICO, "quirurgico", ), ( "2", "FACTURA BLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_FACTURA, "factura", ), ( "3", "HISTORIA CLINICA BLANCA LIJIA RENGIFO AGUIRRE.pdf", self.BLANCA_HC, "historia_clinica", ), ] files = [] for file_id, original_name, text, detected_type in docs: signals = service.extract_signals(original_name, text, detected_type) files.append( { "_id": file_id, "batch_id": "batch-blanca", "original_name": original_name, "patient_name": signals.patient_name, "patient_id": signals.patient_id, "case_number": signals.case_number, "service_date": signals.service_date, "procedure_code": signals.procedure_code, "procedure_description": signals.procedure_description, "evidence": signals.evidence, } ) result = service.associate(files) self.assertEqual(len(result.cases), 1) self.assertEqual({item.status for item in result.decisions}, {"asociado"}) self.assertEqual( {item.patient_id for item in result.decisions}, {"66781911"}, ) self.assertEqual( {item.case_number for item in result.decisions}, {"176654"}, ) def test_seeds_new_cluster_when_best_candidate_has_only_non_anchor_match(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "1", "batch_id": "batch-1", "original_name": "julio.pdf", "patient_name": "JULIO CESAR ACOSTA GUEVARA", "patient_id": "93387170", "case_number": "178474", "service_date": "11/05/2024", "procedure_code": "", "procedure_description": "", "evidence": ["identificacion", "numero_caso", "nombre_paciente"], }, { "_id": "2", "batch_id": "batch-1", "original_name": "monica.pdf", "patient_name": "MONICA LORENA PARRA GIRALDO", "patient_id": "29465123", "case_number": "178550", "service_date": "11/05/2024", "procedure_code": "", "procedure_description": "", "evidence": ["identificacion", "numero_caso", "nombre_paciente"], }, ] result = service.associate(files) self.assertEqual(len(result.cases), 2) self.assertEqual({item.status for item in result.decisions}, {"asociado"}) self.assertEqual( {item.case_number for item in result.decisions}, {"178474", "178550"}, ) def test_associates_prefactura_by_case_number_without_patient_id(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "prefactura-1", "batch_id": "batch-prefactura", "original_name": "prefactura.pdf", "patient_name": "", "patient_id": "", "case_number": "176654", "service_date": "", "procedure_code": "", "procedure_description": "", "detected_type": "prefactura", "evidence": ["numero_caso", "prefactura_sin_identificacion"], }, { "_id": "soporte-1", "batch_id": "batch-prefactura", "original_name": "laboratorio.pdf", "patient_name": "", "patient_id": "", "case_number": "176654", "service_date": "", "procedure_code": "", "procedure_description": "", "detected_type": "laboratorio", "evidence": ["numero_caso"], }, ] result = service.associate(files) self.assertEqual(len(result.cases), 1) self.assertEqual({item.status for item in result.decisions}, {"asociado"}) self.assertEqual({item.case_number for item in result.decisions}, {"176654"}) def test_associates_monica_batch_into_single_case(self) -> None: service = HeuristicCaseAssociationService() docs = [ ( "1", "CIRUGIA MONICA LORENA PARRA GIRALDO.pdf", self.MONICA_QUIRURGICO, "quirurgico", ), ( "2", "FACTURA MONICA LORENA PARRA GIRALDO.pdf", self.MONICA_FACTURA, "factura", ), ( "3", "HISTORIA CLINICA MONICA LORENA PARRA GIRALDO.pdf", self.MONICA_HC, "historia_clinica", ), ] files = [] for file_id, original_name, text, detected_type in docs: signals = service.extract_signals(original_name, text, detected_type) files.append( { "_id": file_id, "batch_id": "batch-monica", "original_name": original_name, "patient_name": signals.patient_name, "patient_id": signals.patient_id, "case_number": signals.case_number, "service_date": signals.service_date, "procedure_code": signals.procedure_code, "procedure_description": signals.procedure_description, "evidence": signals.evidence, } ) result = service.associate(files) self.assertEqual(len(result.cases), 1) self.assertEqual({item.status for item in result.decisions}, {"asociado"}) self.assertEqual( {item.case_number for item in result.decisions}, {"178550"}, ) def test_associates_mixed_batch_into_three_cases(self) -> None: service = HeuristicCaseAssociationService() docs = [ ("1", "CIRUGIABLANCA.pdf", self.BLANCA_QUIRURGICO, "quirurgico"), ("2", "FACTURABLANCA.pdf", self.BLANCA_FACTURA, "factura"), ("3", "HCBLANCA.pdf", self.BLANCA_HC, "historia_clinica"), ("4", "CIRUGIAMONICA.pdf", self.MONICA_QUIRURGICO, "quirurgico"), ("5", "FACTURAMONICA.pdf", self.MONICA_FACTURA, "factura"), ("6", "HCMONICA.pdf", self.MONICA_HC, "historia_clinica"), ("7", "CIRUGIAJULIO.pdf", self.JULIO_QUIRURGICO, "quirurgico"), ("8", "FACTURAJULIO.pdf", self.JULIO_FACTURA, "factura"), ("9", "HCJULIO.pdf", self.JULIO_HC, "historia_clinica"), ] files = [] for file_id, original_name, text, detected_type in docs: signals = service.extract_signals(original_name, text, detected_type) files.append( { "_id": file_id, "batch_id": "batch-mixto", "original_name": original_name, "patient_name": signals.patient_name, "patient_id": signals.patient_id, "case_number": signals.case_number, "service_date": signals.service_date, "procedure_code": signals.procedure_code, "procedure_description": signals.procedure_description, "evidence": signals.evidence, } ) result = service.associate(files) self.assertEqual(len(result.cases), 3) self.assertEqual({item.status for item in result.decisions}, {"asociado"}) self.assertEqual( {case["case_number"] for case in result.cases}, {"176654", "178550", "178474"}, ) def test_prefers_selected_case_when_signals_are_consistent(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "support-1", "batch_id": "batch-manual", "original_name": "factura-blanca.pdf", "patient_name": "BLANCA LIJIA RENGIFO AGUIRRE", "patient_id": "66781911", "case_number": "176654", "service_date": "", "procedure_code": "", "procedure_description": "Factura soporte", "preferred_case_key": "case-blanca", "preferred_case_number": "176654", "preferred_patient_id": "66781911", "preferred_patient_name": "BLANCA LIJIA RENGIFO AGUIRRE", "preferred_association_mode": "validate_and_warn", "evidence": ["identificacion", "numero_caso"], } ] result = service.associate(files) self.assertEqual(len(result.cases), 1) decision = result.decisions[0] self.assertEqual(decision.status, "asociado") self.assertEqual(decision.case_key, "case-blanca") self.assertEqual(decision.association_source, ASSOCIATION_SOURCE_MANUAL) self.assertIn("case_key_preseleccionado", decision.evidence) self.assertFalse(decision.review_required) def test_marks_pending_validation_when_selected_case_contradicts_extracted_signals(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "support-2", "batch_id": "batch-manual", "original_name": "factura-conflictiva.pdf", "patient_name": "Paciente Documento", "patient_id": "99999", "case_number": "HC-999", "service_date": "", "procedure_code": "", "procedure_description": "Factura soporte", "preferred_case_key": "case-real", "preferred_case_number": "HC-001", "preferred_patient_id": "11111", "preferred_patient_name": "Paciente Real", "preferred_association_mode": "validate_and_warn", "evidence": ["identificacion", "numero_caso"], } ] result = service.associate(files) self.assertEqual(len(result.cases), 0) decision = result.decisions[0] self.assertEqual(decision.status, "pendiente_validacion") self.assertEqual(decision.case_key, "case-real") self.assertEqual(decision.association_source, ASSOCIATION_SOURCE_MANUAL) self.assertTrue(decision.review_required) self.assertTrue(decision.review_messages) def test_prefactura_compound_keeps_selected_case_when_only_name_differs(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "support-3", "batch_id": "batch-prefactura", "original_name": "hc-prefactura.pdf", "patient_name": "PACIENTE DOCUMENTO", "patient_id": "", "case_number": "203922", "service_date": "", "procedure_code": "", "procedure_description": "Historia clínica", "preferred_case_key": "203922", "preferred_case_number": "203922", "preferred_patient_id": "1088025550", "preferred_patient_name": "LUISA FERNANDA IBARRA CANO", "preferred_association_mode": "validate_and_warn", "parent_prefactura_batch_id": "batch-prefactura", "evidence": ["numero_caso"], } ] result = service.associate(files) self.assertEqual(len(result.cases), 1) decision = result.decisions[0] self.assertEqual(decision.status, "asociado") self.assertEqual(decision.case_key, "203922") self.assertTrue(decision.review_required) self.assertIn("advertencia_case_key_preseleccionado", decision.evidence) def test_prefactura_compound_blocks_when_patient_id_conflicts_with_anchor(self) -> None: service = HeuristicCaseAssociationService() files = [ { "_id": "support-4", "batch_id": "batch-prefactura", "original_name": "hc-prefactura.pdf", "patient_name": "LUISA FERNANDA IBARRA CANO", "patient_id": "99999", "case_number": "203922", "service_date": "", "procedure_code": "", "procedure_description": "Historia clínica", "preferred_case_key": "203922", "preferred_case_number": "203922", "preferred_patient_id": "1088025550", "preferred_patient_name": "LUISA FERNANDA IBARRA CANO", "preferred_association_mode": "validate_and_warn", "parent_prefactura_batch_id": "batch-prefactura", "evidence": ["identificacion", "numero_caso"], } ] result = service.associate(files) self.assertEqual(len(result.cases), 0) decision = result.decisions[0] self.assertEqual(decision.status, "pendiente_validacion") self.assertEqual(decision.case_key, "203922") self.assertTrue(decision.review_required) class FakeBatchRepository: def __init__(self, batch_id: str) -> None: self.batch_id = batch_id self.created_payloads: list[dict] = [] self.batch = { "_id": batch_id, "usuario": "tester", "nombre_archivo": "demo.zip", "status": "procesando", "created_at": "2026-03-14T10:00:00-05:00", "updated_at": "2026-03-14T10:00:00-05:00", "total_files": 1, "processed_files": 0, "failed_files": 0, "pending_validation_files": 1, "associated_files": 0, "clinical_processed_files": 0, "clinical_failed_files": 0, "clinical_pending_files": 0, "bulk_epicrisis_status": "pendiente", "bulk_epicrisis_job_id": "", "bulk_epicrisis_requested_at": "", "bulk_epicrisis_total_target": 0, "bulk_epicrisis_completed_count": 0, "bulk_epicrisis_failed_count": 0, "bulk_epicrisis_skipped_count": 0, "bulk_epicrisis_case_keys": [], "excel_epicrisis_status": "pendiente", "excel_epicrisis_job_id": "", "excel_epicrisis_requested_at": "", "excel_epicrisis_generated_at": "", "excel_epicrisis_error": "", "excel_epicrisis_filename": "", "excel_epicrisis_download_url": "", "excel_epicrisis_included_count": 0, "excel_epicrisis_omitted_count": 0, "excel_epicrisis_path": "", "archive_path": "", } self.user_batches = [dict(self.batch)] self._cases_refresh_lock_owner = "" self._cases_refresh_locked_at = 0.0 self.deleted_batch_ids: list[str] = [] def create_batch(self, payload: dict) -> str: self.created_payloads.append(dict(payload)) self.batch = {"_id": self.batch_id, **payload} self.user_batches = [dict(self.batch)] return self.batch_id def update_batch(self, batch_id: str, payload: dict) -> None: if batch_id != self.batch_id: return self.batch.update(payload) if self.user_batches: self.user_batches[0].update(payload) def get_batch(self, batch_id: str) -> dict | None: if batch_id != self.batch_id: return None return dict(self.batch) def list_user_batches(self, username: str, *, limit: int = 20) -> list[dict]: items = [dict(item) for item in self.user_batches if item.get("usuario") == username] items.sort(key=lambda item: item.get("updated_at", ""), reverse=True) return items[:limit] def delete_batch(self, batch_id: str) -> int: if batch_id != self.batch_id: return 0 self.deleted_batch_ids.append(batch_id) self.batch = {} self.user_batches = [] return 1 def acquire_cases_refresh_lock(self, batch_id: str, owner: str, ttl_seconds: int) -> bool: if batch_id != self.batch_id or not owner: return False now = time.monotonic() lock_is_expired = bool(self._cases_refresh_lock_owner) and ( now - self._cases_refresh_locked_at ) >= max(int(ttl_seconds or 0), 1) if not self._cases_refresh_lock_owner or self._cases_refresh_lock_owner == owner or lock_is_expired: self._cases_refresh_lock_owner = owner self._cases_refresh_locked_at = now return True return False def release_cases_refresh_lock(self, batch_id: str, owner: str) -> None: if batch_id != self.batch_id or self._cases_refresh_lock_owner != owner: return self._cases_refresh_lock_owner = "" self._cases_refresh_locked_at = 0.0 class FakeBatchFileRepository: def __init__(self, batch_id: str) -> None: self.batch_id = batch_id self.next_id = 2 self.files: dict[str, dict[str, Any]] = { "file-1": { "_id": "file-1", "batch_id": batch_id, "status": "pendiente_validacion", "original_name": "ambiguous.pdf", "stored_path": "", "patient_name": "", "patient_id": "", "case_number": "", "service_date": "", "procedure_code": "", "procedure_description": "", "case_key": "", "associated_user": "", "association_source": "auto", "confidence": 0.34, "evidence": ["nombre_archivo"], "score_breakdown": {}, "top_candidates": [], "manual_resolution": {}, "clinical_status": "pendiente", "clinical_job_id": "", "clinical_error": "", "clinical_processed_at": "", "analysis_document_id": "", "error": "", "created_at": "2026-03-14T10:00:00-05:00", "updated_at": "2026-03-14T10:00:00-05:00", } } def create_file(self, payload: dict) -> str: file_id = f"file-{self.next_id}" self.next_id += 1 self.files[file_id] = {"_id": file_id, **payload} return file_id def get_file(self, file_id: str) -> dict | None: item = self.files.get(file_id) return dict(item) if item else None def update_file(self, file_id: str, payload: dict) -> None: if file_id not in self.files: return self.files[file_id].update(payload) def list_files(self, batch_id: str, *, status: str | None = None) -> list[dict]: items = [dict(item) for item in self.files.values() if item.get("batch_id") == batch_id] if status: items = [item for item in items if item.get("status") == status] return items def delete_files_by_batch(self, batch_id: str) -> int: keys_to_delete = [key for key, item in self.files.items() if item.get("batch_id") == batch_id] for key in keys_to_delete: self.files.pop(key, None) return len(keys_to_delete) class FakeBatchCaseRepository: def __init__(self, batch_id: str) -> None: self.batch_id = batch_id self.cases = [] def replace_cases(self, batch_id: str, cases: list[dict]) -> None: if batch_id != self.batch_id: return self.cases = [dict(item) for item in cases] def list_cases(self, batch_id: str) -> list[dict]: if batch_id != self.batch_id: return [] return [dict(item) for item in self.cases] def get_case(self, batch_id: str, case_key: str) -> dict | None: if batch_id != self.batch_id: return None for item in self.cases: if item.get("case_key") == case_key: return dict(item) return None def get_user_case(self, username: str, case_key: str) -> dict | None: for item in self.cases: if item.get("case_key") == case_key and item.get("usuario", "tester") == username: return dict(item) return None def list_user_cases(self, username: str, *, limit: int = 50) -> list[dict]: items = [ dict(item) for item in self.cases if item.get("usuario", "tester") == username ] return items[: max(1, int(limit or 50))] def update_case(self, batch_id: str, case_key: str, payload: dict) -> None: if batch_id != self.batch_id: return for item in self.cases: if item.get("case_key") == case_key: item.update(payload) return def delete_cases_by_batch(self, batch_id: str) -> int: if batch_id != self.batch_id: return 0 deleted = len(self.cases) self.cases = [] return deleted class ConcurrentSensitiveBatchCaseRepository(FakeBatchCaseRepository): def __init__(self, batch_id: str) -> None: super().__init__(batch_id) self._replace_guard = threading.Lock() self.max_parallel_replacements = 0 self._current_parallel_replacements = 0 def replace_cases(self, batch_id: str, cases: list[dict]) -> None: with self._replace_guard: self._current_parallel_replacements += 1 self.max_parallel_replacements = max( self.max_parallel_replacements, self._current_parallel_replacements, ) try: time.sleep(0.03) super().replace_cases(batch_id, cases) finally: with self._replace_guard: self._current_parallel_replacements -= 1 class FakeBatchArtifactCleaner: def __init__(self, *, fail_archive: bool = False, fail_extracted: bool = False) -> None: self.fail_archive = fail_archive self.fail_extracted = fail_extracted self.deleted_archives: list[str] = [] self.deleted_files: list[str] = [] self.deleted_batch_artifacts: list[str] = [] def delete_archive(self, archive_path: str) -> None: self.deleted_archives.append(archive_path) if self.fail_archive: raise OSError("archive cleanup failed") def delete_extracted_file(self, file_path: str) -> None: self.deleted_files.append(file_path) if self.fail_extracted: raise OSError("file cleanup failed") def delete_batch_artifacts(self, batch_id: str) -> None: self.deleted_batch_artifacts.append(batch_id) class FakeBatchWorkingFileStore: def __init__(self) -> None: self.saved_files: list[tuple[str, str, bytes]] = [] def save_file(self, batch_id: str, filename: str, data: bytes) -> str: self.saved_files.append((batch_id, filename, data)) return f"/tmp/{batch_id}/{filename}" class FakePrefacturaPdfExtractor: def __init__(self, pages: list[PrefacturaPdfPage]) -> None: self.pages = pages self.ranges: list[tuple[int, int]] = [] self.page_sets: list[list[int]] = [] def extract_pages(self, contents: bytes) -> list[PrefacturaPdfPage]: return list(self.pages) def build_range_pdf(self, contents: bytes, page_start: int, page_end: int) -> bytes: self.ranges.append((page_start, page_end)) return f"%PDF range {page_start}-{page_end}".encode() def build_pages_pdf(self, contents: bytes, page_numbers: list[int]) -> bytes: self.page_sets.append(list(page_numbers)) if page_numbers: self.ranges.append((page_numbers[0], page_numbers[-1])) return f"%PDF pages {','.join(str(page) for page in page_numbers)}".encode() class FakePrefacturaClassificationAdvisor: def __init__(self, responses: dict[int, PrefacturaPageClassification | None]) -> None: self.responses = dict(responses) self.calls: list[int] = [] def classify_page( self, *, filename: str, page_number: int, page_text: str, detected_type: str, classifier_title: str, candidate_types: list[str], ) -> PrefacturaPageClassification | None: _ = filename, page_text, detected_type, classifier_title, candidate_types self.calls.append(page_number) return self.responses.get(page_number) class FakeTextExtractor: def __init__(self, text_by_path: dict[str, str]) -> None: self.text_by_path = dict(text_by_path) def extract_text(self, file_path: str) -> str: return str(self.text_by_path.get(file_path, "")) def extract(self, file_path: str) -> PdfExtractionResult: return PdfExtractionResult(text=self.extract_text(file_path)) class FakeArchiveExtractor: def __init__( self, *, result: ArchiveExtractionResult | None = None, exc: Exception | None = None, ) -> None: self.result = result or ArchiveExtractionResult() self.exc = exc def extract_archive(self, archive_path: str, batch_id: str) -> ArchiveExtractionResult: if self.exc: raise self.exc return self.result class FakeClinicalDocumentService: def __init__(self, *, result: dict | None = None, exc: Exception | None = None) -> None: self.result = result or {"id_documento": "analysis-1"} self.exc = exc self.requests = [] def process_and_persist(self, request) -> dict: self.requests.append(request) if self.exc: raise self.exc return dict(self.result) class FakeCaseDeletionService: def __init__(self) -> None: self.deleted_cases: list[tuple[str, str]] = [] self.purged_documents: list[tuple[str, str]] = [] self.purged_batch_file_ids: list[tuple[str, tuple[str, ...]]] = [] def delete_case(self, *, username: str, case_key: str) -> dict: self.deleted_cases.append((username, case_key)) return { "deleted_documents": 2, "deleted_derived_artifacts": 1, } def purge_document_record(self, *, username: str, document_id: str) -> bool: self.purged_documents.append((username, document_id)) return True def purge_documents_by_batch_file_ids(self, *, username: str, batch_file_ids: list[str]) -> int: normalized = tuple(sorted(value for value in batch_file_ids if value)) self.purged_batch_file_ids.append((username, normalized)) return len(normalized) class FakeBatchJobDispatcher: def __init__(self) -> None: self.dispatched_batch_ids: list[str] = [] def dispatch(self, batch_id: str) -> None: self.dispatched_batch_ids.append(batch_id) class FakeCaseEpicrisisService: def __init__( self, *, cached_contexts: dict[str, dict] | None = None, rebuilt_contexts: dict[str, dict] | None = None, rebuild_failures: dict[str, Exception] | None = None, ) -> None: self.cached_contexts = dict(cached_contexts or {}) self.rebuilt_contexts = dict(rebuilt_contexts or {}) self.rebuild_failures = dict(rebuild_failures or {}) self.rebuild_requests: list[tuple[str, str, bool]] = [] def get_cached_case_context(self, username: str, case_key: str) -> dict | None: context = self.cached_contexts.get(case_key) if context is None: return None return {"contexto": dict(context)} def cache_case_context( self, username: str, case_key: str, *, regen: bool = False, ) -> dict: self.rebuild_requests.append((username, case_key, regen)) if case_key in self.rebuild_failures: raise self.rebuild_failures[case_key] context = self.rebuilt_contexts.get(case_key) if context is None: raise ValueError("Sin contexto reconstruible") self.cached_contexts[case_key] = dict(context) return dict(context) class CreateManualBatchUploadUseCaseTest(unittest.TestCase): def test_creates_manual_historia_batch_with_single_pending_file(self) -> None: batch_repo = FakeBatchRepository("batch-manual") file_repo = FakeBatchFileRepository("batch-manual") file_repo.files = {} file_store = FakeBatchWorkingFileStore() job_dispatcher = FakeBatchJobDispatcher() use_case = CreateManualBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=file_store, job_dispatcher=job_dispatcher, colombia_tz=UTC, ) result = use_case.execute( filename="historia.pdf", contents=b"%PDF-1.4 fake", username="tester", ingestion_mode=INGESTION_MODE_MANUAL_HISTORIA, detected_type="historia_clinica", ) self.assertEqual(result["batch_id"], "batch-manual") self.assertEqual(job_dispatcher.dispatched_batch_ids, ["batch-manual"]) batch = batch_repo.get_batch("batch-manual") if batch is None: raise AssertionError("Batch manual no fue creado") self.assertEqual(batch["ingestion_mode"], INGESTION_MODE_MANUAL_HISTORIA) saved_file = next(iter(file_repo.files.values())) self.assertEqual(saved_file["status"], "pendiente") self.assertEqual(saved_file["detected_type"], "historia_clinica") self.assertEqual(saved_file["preferred_case_key"], "") self.assertEqual(file_store.saved_files[0][0], "batch-manual") def test_support_batch_resolves_selected_case_context_before_dispatch(self) -> None: batch_repo = FakeBatchRepository("batch-support") file_repo = FakeBatchFileRepository("batch-support") file_repo.files = {} file_store = FakeBatchWorkingFileStore() use_case = CreateManualBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=file_store, job_dispatcher=FakeBatchJobDispatcher(), colombia_tz=UTC, case_context_resolver=lambda username, case_key: { "case_key": case_key, "case_number": "HC-100", "patient_id": "12345", "patient_name": "Paciente Demo", }, ) use_case.execute( filename="factura.pdf", contents=b"%PDF-1.4 fake", username="tester", ingestion_mode=INGESTION_MODE_MANUAL_SOPORTE, detected_type="factura", case_key="CASE-100", ) saved_file = next(iter(file_repo.files.values())) self.assertEqual(saved_file["preferred_case_key"], "CASE-100") self.assertEqual(saved_file["preferred_case_number"], "HC-100") self.assertEqual(saved_file["preferred_patient_id"], "12345") self.assertEqual(saved_file["preferred_patient_name"], "Paciente Demo") class CreatePrefacturaBatchUploadUseCaseTest(unittest.TestCase): def test_creates_prefactura_batch_with_internal_segments_and_case_anchor(self) -> None: batch_repo = FakeBatchRepository("batch-prefactura") file_repo = FakeBatchFileRepository("batch-prefactura") file_repo.files = {} file_store = FakeBatchWorkingFileStore() extractor = FakePrefacturaPdfExtractor( [ PrefacturaPdfPage( page_number=1, text=( "PreFactura de Servicios\nCaso No. 176654\n" "Paciente : LUISA FERNANDA IBARRA CANO\nPROC001 Curación compleja $1000" ), ), PrefacturaPdfPage( page_number=2, text="Resultado de laboratorio\nCaso No. 176654\nHemograma normal", ), ] ) job_dispatcher = FakeBatchJobDispatcher() use_case = CreatePrefacturaBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=file_store, prefactura_pdf_extractor=extractor, prefactura_orchestrator=build_prefactura_orchestrator(), job_dispatcher=job_dispatcher, colombia_tz=UTC, ) result = use_case.execute( filename="paquete-prefactura.pdf", contents=b"%PDF-1.4 fake", username="tester", ) self.assertEqual(result["batch_id"], "batch-prefactura") self.assertEqual(job_dispatcher.dispatched_batch_ids, ["batch-prefactura"]) batch = batch_repo.get_batch("batch-prefactura") if batch is None: raise AssertionError("Batch de prefactura no fue creado") self.assertEqual(batch["ingestion_mode"], INGESTION_MODE_PREFACTURA_PDF) self.assertEqual(batch["total_files"], 3) files = sorted(file_repo.files.values(), key=lambda item: item["source_page_start"]) self.assertEqual( [item["detected_type"] for item in files], ["prefactura", "historia_clinica", "laboratorio"] ) self.assertEqual(files[0]["document_title"], "Prefactura") self.assertEqual(files[0]["document_key"], "prefactura") self.assertEqual(files[0]["document_reference"], "") self.assertEqual({item["preferred_case_key"] for item in files}, {"176654"}) self.assertEqual({item["preferred_case_number"] for item in files}, {"176654"}) self.assertEqual({item["preferred_patient_name"] for item in files}, {"LUISA FERNANDA IBARRA CANO"}) self.assertEqual({item["patient_name"] for item in files}, {"LUISA FERNANDA IBARRA CANO"}) self.assertIn("prefactura_sin_identificacion", files[0]["evidence"]) self.assertEqual(extractor.ranges, [(1, 1), (2, 2), (2, 2)]) self.assertEqual(len(file_store.saved_files), 3) def test_segments_prefactura_pdf_by_document_title_and_page_reset(self) -> None: batch_repo = FakeBatchRepository("batch-prefactura") file_repo = FakeBatchFileRepository("batch-prefactura") file_repo.files = {} extractor = FakePrefacturaPdfExtractor( [ PrefacturaPdfPage( page_number=1, text="PreFactura de Servicios\nCaso No. 203922\nPaciente : LUISA FERNANDA IBARRA CANO", ), PrefacturaPdfPage( page_number=2, text="HISTORIA CLINICA DE URGENCIAS\nPage 1 of 2\nCaso: 203922\nNombre del Paciente\nLUISA FERNANDA IBARRA CANO", ), PrefacturaPdfPage( page_number=3, text="HISTORIA CLINICA DE URGENCIAS\nPage 2 of 2\nPACIENTE: CC - 1088025550 - LUISA FERNANDA IBARRA CANO", ), PrefacturaPdfPage( page_number=4, text="Hoja de Drogas\nPaciente: LUISA FERNANDA IBARRA CANO\nCaso N. 203922", ), PrefacturaPdfPage( page_number=5, text="Ordenes de Paraclínicos Generadas en Historias Clinicas\nRADIOLOGIA\nCaso: 203922", ), ] ) use_case = CreatePrefacturaBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=FakeBatchWorkingFileStore(), prefactura_pdf_extractor=extractor, prefactura_orchestrator=build_prefactura_orchestrator(), job_dispatcher=FakeBatchJobDispatcher(), colombia_tz=UTC, ) use_case.execute(filename="prefactura.pdf", contents=b"%PDF-1.4 fake", username="tester") files = sorted(file_repo.files.values(), key=lambda item: item["source_page_start"]) self.assertEqual( [(item["detected_type"], item["document_title"]) for item in files], [ ("prefactura", "Prefactura"), ("historia_clinica", "Historia clínica"), ], ) self.assertEqual( [(item["source_page_start"], item["source_page_end"]) for item in files], [(1, 1), (2, 5)], ) self.assertEqual({item["preferred_case_number"] for item in files}, {"203922"}) self.assertEqual({item["patient_name"] for item in files}, {"LUISA FERNANDA IBARRA CANO"}) self.assertEqual(extractor.page_sets, [[1], [2, 3, 4, 5]]) def test_segments_realistic_prefactura_bundle_into_internal_documents(self) -> None: batch_repo = FakeBatchRepository("batch-prefactura") file_repo = FakeBatchFileRepository("batch-prefactura") file_repo.files = {} extractor = FakePrefacturaPdfExtractor(build_prefactura_203922_pages()) use_case = CreatePrefacturaBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=FakeBatchWorkingFileStore(), prefactura_pdf_extractor=extractor, prefactura_orchestrator=build_prefactura_orchestrator(), job_dispatcher=FakeBatchJobDispatcher(), colombia_tz=UTC, ) use_case.execute( filename="CASO 203922 - prefactura completa.pdf", contents=b"%PDF-1.4 fake", username="tester", ) files = sorted(file_repo.files.values(), key=lambda item: item["source_page_start"]) self.assertEqual(len(files), 2) self.assertEqual({item["preferred_case_number"] for item in files}, {"203922"}) self.assertEqual({item["preferred_patient_name"] for item in files}, {"LUISA FERNANDA IBARRA CANO"}) self.assertIn("1088025550", {str(item["patient_id"]) for item in files}) self.assertEqual( [ ( item["detected_type"], item["document_title"], item["source_page_start"], item["source_page_end"], ) for item in files ], [ ("prefactura", "Prefactura", 1, 1), ("historia_clinica", "Historia clínica", 2, 23), ], ) self.assertEqual(files[0]["document_key"], "prefactura") self.assertEqual(files[1]["document_key"], "historia_clinica") self.assertEqual(files[0]["document_reference"], "") self.assertEqual(files[1]["document_reference"], "") self.assertEqual(extractor.page_sets, [[1], list(range(2, 24))]) self.assertNotIn("prescripcion", {item["detected_type"] for item in files}) def test_marks_internal_files_for_review_when_case_number_is_missing(self) -> None: batch_repo = FakeBatchRepository("batch-prefactura") file_repo = FakeBatchFileRepository("batch-prefactura") file_repo.files = {} extractor = FakePrefacturaPdfExtractor( [PrefacturaPdfPage(page_number=1, text="Prefactura sin número de caso")] ) use_case = CreatePrefacturaBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=FakeBatchWorkingFileStore(), prefactura_pdf_extractor=extractor, prefactura_orchestrator=build_prefactura_orchestrator(), job_dispatcher=FakeBatchJobDispatcher(), colombia_tz=UTC, ) use_case.execute(filename="prefactura.pdf", contents=b"%PDF-1.4 fake", username="tester") saved_file = next(iter(file_repo.files.values())) self.assertTrue(saved_file["review_required"]) self.assertEqual(saved_file["preferred_case_key"], "") self.assertIn("No se identificó número de caso", saved_file["review_messages"][0]) def test_uses_gemini_advisor_only_for_ambiguous_pages(self) -> None: batch_repo = FakeBatchRepository("batch-prefactura") file_repo = FakeBatchFileRepository("batch-prefactura") file_repo.files = {} extractor = FakePrefacturaPdfExtractor( [ PrefacturaPdfPage( page_number=1, text="PreFactura de Servicios\nCaso No. 203922\nPaciente : LUISA FERNANDA IBARRA CANO", ), PrefacturaPdfPage( page_number=2, text="Hallazgos de imagen diagnóstica\nCaso: 203922\nEstudio del hombro derecho", ), ] ) advisor = FakePrefacturaClassificationAdvisor( { 2: PrefacturaPageClassification( page_number=2, document_type="radiologia", document_title="Radiología", confidence=0.92, evidence=["gemini", "hallazgos_imagenologicos"], reason="La página corresponde a imagenología diagnóstica.", source="gemini", is_autonomous=True, page_text="Hallazgos de imagen diagnóstica", ) } ) use_case = CreatePrefacturaBatchUploadUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, working_file_store=FakeBatchWorkingFileStore(), prefactura_pdf_extractor=extractor, prefactura_orchestrator=build_prefactura_orchestrator(advisor=advisor), job_dispatcher=FakeBatchJobDispatcher(), colombia_tz=UTC, ) use_case.execute(filename="prefactura.pdf", contents=b"%PDF-1.4 fake", username="tester") self.assertEqual(advisor.calls, [2]) files = sorted(file_repo.files.values(), key=lambda item: item["source_page_start"]) self.assertEqual( [(item["detected_type"], item["source_page_start"], item["source_page_end"]) for item in files], [ ("prefactura", 1, 1), ("historia_clinica", 2, 2), ("radiologia", 2, 2), ], ) class PrepareBatchUseCaseCleanupTest(unittest.TestCase): def test_deletes_zip_and_clears_archive_path_after_successful_extraction(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch(batch_id, {"archive_path": "/tmp/demo.zip"}) file_repo = FakeBatchFileRepository(batch_id) file_repo.files = {} extractor = FakeArchiveExtractor( result=ArchiveExtractionResult( entries=[ ExtractedArchiveEntry( original_name="doc.pdf", relative_path="doc.pdf", extracted_path="/tmp/doc.pdf", ) ] ) ) cleaner = FakeBatchArtifactCleaner() totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) use_case = PrepareBatchUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, archive_extractor=extractor, artifact_cleaner=cleaner, totals_use_case=totals_use_case, colombia_tz=UTC, ) result = use_case.execute(batch_id) self.assertEqual(result, ["file-2"]) self.assertEqual(cleaner.deleted_archives, ["/tmp/demo.zip"]) batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(batch.get("archive_path"), "") def test_deletes_zip_and_clears_archive_path_after_extraction_failure(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch(batch_id, {"archive_path": "/tmp/demo.zip"}) file_repo = FakeBatchFileRepository(batch_id) file_repo.files = {} extractor = FakeArchiveExtractor(exc=RuntimeError("zip corrupto")) cleaner = FakeBatchArtifactCleaner() totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) use_case = PrepareBatchUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, archive_extractor=extractor, artifact_cleaner=cleaner, totals_use_case=totals_use_case, colombia_tz=UTC, ) result = use_case.execute(batch_id) self.assertEqual(result, []) batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(batch.get("status"), "fallido") self.assertEqual(batch.get("archive_path"), "") self.assertEqual(cleaner.deleted_archives, ["/tmp/demo.zip"]) def test_logs_and_preserves_archive_path_when_zip_cleanup_fails(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch(batch_id, {"archive_path": "/tmp/demo.zip"}) file_repo = FakeBatchFileRepository(batch_id) file_repo.files = {} extractor = FakeArchiveExtractor( result=ArchiveExtractionResult( entries=[ ExtractedArchiveEntry( original_name="doc.pdf", relative_path="doc.pdf", extracted_path="/tmp/doc.pdf", ) ] ) ) cleaner = FakeBatchArtifactCleaner(fail_archive=True) totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) use_case = PrepareBatchUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, archive_extractor=extractor, artifact_cleaner=cleaner, totals_use_case=totals_use_case, colombia_tz=UTC, ) with self.assertLogs("app.batch_processing.application.command_use_cases", level="WARNING"): result = use_case.execute(batch_id) self.assertEqual(result, ["file-2"]) batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(batch.get("archive_path"), "/tmp/demo.zip") class ProcessBatchFileUseCaseTest(unittest.TestCase): def test_prefactura_segments_keep_orchestrated_document_type(self) -> None: batch_id = "batch-prefactura" file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "stored_path": "/tmp/prefactura-segment.pdf", "status": "pendiente", "detected_type": "prefactura", "document_title": "PreFactura de Servicios", "parent_prefactura_batch_id": batch_id, "patient_name": "", "patient_id": "", "case_number": "203922", }, ) use_case = ProcessBatchFileUseCase( batch_repository=FakeBatchRepository(batch_id), batch_file_repository=file_repo, text_extractor=FakeTextExtractor( { "/tmp/prefactura-segment.pdf": ( "PreFactura de Servicios\nCaso No. CM - 203922\n" "Paciente : LUISA FERNANDA IBARRA CANO\nTotal de la Factura\n413,200" ) } ), document_classifier=HeuristicDocumentClassifier(), association_service=HeuristicCaseAssociationService(), colombia_tz=UTC, ) use_case.execute(batch_id, "file-1") updated = file_repo.get_file("file-1") or {} self.assertEqual(updated.get("status"), "clasificado") self.assertEqual(updated.get("detected_type"), "prefactura") self.assertEqual(updated.get("patient_name"), "LUISA FERNANDA IBARRA CANO") self.assertEqual(updated.get("case_number"), "203922") class MaterializeBatchFileUseCaseCleanupTest(unittest.TestCase): def _build_use_case( self, *, batch_repo: FakeBatchRepository, file_repo: FakeBatchFileRepository, case_repo: FakeBatchCaseRepository, clinical_service: FakeClinicalDocumentService, cleaner: FakeBatchArtifactCleaner, ) -> MaterializeBatchFileUseCase: totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) refresh_cases_use_case = RefreshBatchCasesUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, ) return MaterializeBatchFileUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, totals_use_case=totals_use_case, refresh_cases_use_case=refresh_cases_use_case, clinical_document_service=cast(ClinicalDocumentService, clinical_service), case_deletion_service=FakeCaseDeletionService(), artifact_cleaner=cleaner, colombia_tz=UTC, ) def test_deletes_extracted_pdf_and_clears_stored_path_on_success(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "stored_path": "/tmp/doc.pdf", "detected_type": "factura", "document_title": "PreFactura de Servicios", "document_key": "prefactura|prefactura-de-servicios", "document_reference": "reporte:prefactura", "text_preview": "contenido", "extracted_text": "contenido", }, ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() clinical_service = FakeClinicalDocumentService(result={"id_documento": "analysis-1"}) use_case = self._build_use_case( batch_repo=batch_repo, file_repo=file_repo, case_repo=case_repo, clinical_service=clinical_service, cleaner=cleaner, ) result = use_case.execute(batch_id, "file-1") updated = file_repo.get_file("file-1") or {} self.assertEqual(result["clinical_status"], "completado") self.assertEqual(updated.get("analysis_document_id"), "analysis-1") self.assertEqual(updated.get("stored_path"), "") self.assertEqual(cleaner.deleted_files, ["/tmp/doc.pdf"]) self.assertEqual(clinical_service.requests[0].document_title, "PreFactura de Servicios") self.assertEqual(clinical_service.requests[0].document_key, "prefactura|prefactura-de-servicios") self.assertEqual(clinical_service.requests[0].document_reference, "reporte:prefactura") def test_does_not_delete_extracted_pdf_when_clinical_processing_fails(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "stored_path": "/tmp/doc.pdf", "detected_type": "factura", "text_preview": "contenido", "extracted_text": "contenido", }, ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() clinical_service = FakeClinicalDocumentService(exc=RuntimeError("boom")) use_case = self._build_use_case( batch_repo=batch_repo, file_repo=file_repo, case_repo=case_repo, clinical_service=clinical_service, cleaner=cleaner, ) result = use_case.execute(batch_id, "file-1") updated = file_repo.get_file("file-1") or {} self.assertEqual(result["clinical_status"], "fallido") self.assertEqual(updated.get("stored_path"), "/tmp/doc.pdf") self.assertEqual(cleaner.deleted_files, []) def test_cleans_stale_stored_path_when_file_was_already_completed(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "clinical_status": "completado", "analysis_document_id": "analysis-1", "stored_path": "/tmp/doc.pdf", }, ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() clinical_service = FakeClinicalDocumentService() use_case = self._build_use_case( batch_repo=batch_repo, file_repo=file_repo, case_repo=case_repo, clinical_service=clinical_service, cleaner=cleaner, ) result = use_case.execute(batch_id, "file-1") updated = file_repo.get_file("file-1") or {} self.assertEqual(result["clinical_status"], "completado") self.assertEqual(updated.get("stored_path"), "") self.assertEqual(cleaner.deleted_files, ["/tmp/doc.pdf"]) def test_logs_and_preserves_completed_status_when_pdf_cleanup_fails(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "stored_path": "/tmp/doc.pdf", "detected_type": "factura", "text_preview": "contenido", "extracted_text": "contenido", }, ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner(fail_extracted=True) clinical_service = FakeClinicalDocumentService(result={"id_documento": "analysis-1"}) use_case = self._build_use_case( batch_repo=batch_repo, file_repo=file_repo, case_repo=case_repo, clinical_service=clinical_service, cleaner=cleaner, ) with self.assertLogs("app.batch_processing.application.command_use_cases", level="WARNING"): result = use_case.execute(batch_id, "file-1") updated = file_repo.get_file("file-1") or {} self.assertEqual(result["clinical_status"], "completado") self.assertEqual(updated.get("clinical_status"), "completado") self.assertEqual(updated.get("stored_path"), "/tmp/doc.pdf") def test_rebuilds_case_aggregate_after_multiple_materializations(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch(batch_id, {"status": "completado"}) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "case_key": "case-main", "patient_name": "Paciente Uno", "patient_id": "1001", "case_number": "HC-1", "stored_path": "/tmp/doc-1.pdf", "detected_type": "historia_clinica", "text_preview": "contenido 1", "extracted_text": "contenido 1", }, ) file_repo.create_file( { "batch_id": batch_id, "original_name": "doc-2.pdf", "relative_path": "doc-2.pdf", "stored_path": "/tmp/doc-2.pdf", "status": "asociado", "created_at": "2026-03-14T10:00:01-05:00", "updated_at": "2026-03-14T10:00:01-05:00", "case_key": "case-main", "patient_name": "Paciente Uno", "patient_id": "1001", "case_number": "HC-1", "detected_type": "factura", "text_preview": "contenido 2", "extracted_text": "contenido 2", } ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() clinical_service = FakeClinicalDocumentService() use_case = self._build_use_case( batch_repo=batch_repo, file_repo=file_repo, case_repo=case_repo, clinical_service=clinical_service, cleaner=cleaner, ) use_case.execute(batch_id, "file-1") use_case.execute(batch_id, "file-2") cases = case_repo.list_cases(batch_id) self.assertEqual(len(cases), 1) self.assertEqual(cases[0].get("case_key"), "case-main") self.assertEqual(cases[0].get("document_count"), 2) self.assertEqual(cases[0].get("processed_document_count"), 2) self.assertEqual(cases[0].get("clinical_failed_count"), 0) def test_rolls_back_materialized_document_when_batch_is_marked_for_deletion(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "stored_path": "/tmp/doc.pdf", "detected_type": "factura", "text_preview": "contenido", "extracted_text": "contenido", }, ) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() clinical_service = FakeClinicalDocumentService(result={"id_documento": "analysis-99"}) deletion_service = FakeCaseDeletionService() totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) refresh_cases_use_case = RefreshBatchCasesUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, ) class RaceBatchRepository(FakeBatchRepository): def __init__(self, source: FakeBatchRepository) -> None: self.__dict__ = source.__dict__ self._read_count = 0 def get_batch(self, batch_id: str) -> dict | None: batch = super().get_batch(batch_id) self._read_count += 1 if self._read_count >= 2 and batch: batch["status"] = "eliminando" batch["deletion_status"] = "requested" return batch race_repo = RaceBatchRepository(batch_repo) use_case = MaterializeBatchFileUseCase( batch_repository=race_repo, batch_file_repository=file_repo, totals_use_case=totals_use_case, refresh_cases_use_case=refresh_cases_use_case, clinical_document_service=cast(ClinicalDocumentService, clinical_service), case_deletion_service=deletion_service, artifact_cleaner=cleaner, colombia_tz=UTC, ) result = use_case.execute(batch_id, "file-1") self.assertEqual(result, {}) updated = file_repo.get_file("file-1") or {} self.assertEqual(updated.get("analysis_document_id"), "") self.assertEqual(deletion_service.purged_documents, [("tester", "analysis-99")]) class ResolveAssociationUseCaseTest(unittest.TestCase): def test_manual_resolution_updates_file_and_batch_totals(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) case_repo = FakeBatchCaseRepository(batch_id) totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) refresh_cases_use_case = RefreshBatchCasesUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, ) use_case = ResolveFileAssociationUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, totals_use_case=totals_use_case, refresh_cases_use_case=refresh_cases_use_case, colombia_tz=UTC, ) result = use_case.execute( batch_id=batch_id, file_id="file-1", resolved_by="tester", case_key="__new__", patient_name="Paciente Demo", patient_id="10203040", procedure_code="998877", reason="Documento ambiguo resuelto manualmente", ) self.assertEqual(result["status"], "asociado") updated = file_repo.get_file("file-1") or {} self.assertEqual(updated.get("association_source"), "manual") self.assertEqual(updated.get("associated_user"), "10203040") self.assertIn("validacion_manual", updated.get("evidence", [])) self.assertEqual(updated.get("manual_resolution", {}).get("resolved_by"), "tester") batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(batch.get("associated_files"), 1) self.assertEqual(batch.get("pending_validation_files"), 0) self.assertEqual(len(case_repo.list_cases(batch_id)), 1) def test_rejects_manual_resolution_when_batch_is_being_deleted(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch(batch_id, {"status": "eliminando", "deletion_status": "requested"}) file_repo = FakeBatchFileRepository(batch_id) case_repo = FakeBatchCaseRepository(batch_id) totals_use_case = RecomputeBatchTotalsUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, colombia_tz=UTC, ) refresh_cases_use_case = RefreshBatchCasesUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, ) use_case = ResolveFileAssociationUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, totals_use_case=totals_use_case, refresh_cases_use_case=refresh_cases_use_case, colombia_tz=UTC, ) with self.assertRaisesRegex(ValueError, "eliminado"): use_case.execute( batch_id=batch_id, file_id="file-1", resolved_by="tester", case_key="__new__", patient_name="Paciente Demo", patient_id="10203040", procedure_code="998877", reason="Documento ambiguo resuelto manualmente", ) class DeleteBatchUseCaseTest(unittest.TestCase): def test_deletes_orphaned_clinical_documents_by_batch_file_id(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) case_repo = FakeBatchCaseRepository(batch_id) cleaner = FakeBatchArtifactCleaner() deletion_service = FakeCaseDeletionService() use_case = DeleteBatchUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, artifact_cleaner=cleaner, case_deletion_service=deletion_service, colombia_tz=UTC, ) result = use_case.execute( batch_id=batch_id, username="tester", confirmation_batch_id=batch_id, confirmation_phrase="ELIMINAR LOTE", ) self.assertEqual(result["orphaned_batch_documents"], 1) self.assertEqual(deletion_service.purged_batch_file_ids, [("tester", ("file-1",))]) def test_deletes_batch_records_artifacts_and_cascades_case_deletion(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file("file-1", {"case_key": "CASE-1"}) file_repo.create_file( { "batch_id": batch_id, "original_name": "doc-2.pdf", "relative_path": "doc-2.pdf", "stored_path": "/tmp/doc-2.pdf", "status": "asociado", "case_key": "CASE-2", "created_at": "2026-03-14T10:00:01-05:00", "updated_at": "2026-03-14T10:00:01-05:00", } ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.cases = [{"case_key": "CASE-2"}, {"case_key": "CASE-3"}] cleaner = FakeBatchArtifactCleaner() deletion_service = FakeCaseDeletionService() use_case = DeleteBatchUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, artifact_cleaner=cleaner, case_deletion_service=deletion_service, colombia_tz=UTC, ) result = use_case.execute( batch_id=batch_id, username="tester", confirmation_batch_id=batch_id, confirmation_phrase="ELIMINAR LOTE", ) self.assertEqual(result["deleted_case_keys"], 3) self.assertEqual(file_repo.list_files(batch_id), []) self.assertEqual(case_repo.list_cases(batch_id), []) self.assertEqual(cleaner.deleted_batch_artifacts, [batch_id]) self.assertEqual(batch_repo.deleted_batch_ids, [batch_id]) self.assertEqual( deletion_service.deleted_cases, [("tester", "CASE-1"), ("tester", "CASE-2"), ("tester", "CASE-3")], ) def test_rejects_invalid_confirmation(self) -> None: use_case = DeleteBatchUseCase( batch_repository=FakeBatchRepository("batch-1"), batch_file_repository=FakeBatchFileRepository("batch-1"), batch_case_repository=FakeBatchCaseRepository("batch-1"), artifact_cleaner=FakeBatchArtifactCleaner(), case_deletion_service=FakeCaseDeletionService(), colombia_tz=UTC, ) with self.assertRaises(BatchDeletionError) as ctx: use_case.execute( batch_id="batch-1", username="tester", confirmation_batch_id="otro", confirmation_phrase="ELIMINAR LOTE", ) self.assertEqual(ctx.exception.status_code, 400) class RefreshBatchCasesUseCaseConcurrencyTest(unittest.TestCase): def test_serializes_refreshes_for_the_same_batch(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) file_repo = FakeBatchFileRepository(batch_id) file_repo.update_file( "file-1", { "status": "asociado", "case_key": "case-main", "patient_name": "Paciente Uno", "patient_id": "1001", "case_number": "HC-1", "clinical_status": "completado", }, ) file_repo.create_file( { "batch_id": batch_id, "original_name": "doc-2.pdf", "relative_path": "doc-2.pdf", "stored_path": "/tmp/doc-2.pdf", "status": "asociado", "created_at": "2026-03-14T10:00:01-05:00", "updated_at": "2026-03-14T10:00:01-05:00", "case_key": "case-main", "patient_name": "Paciente Uno", "patient_id": "1001", "case_number": "HC-1", "clinical_status": "fallido", } ) case_repo = ConcurrentSensitiveBatchCaseRepository(batch_id) use_case = RefreshBatchCasesUseCase( batch_repository=batch_repo, batch_file_repository=file_repo, batch_case_repository=case_repo, lock_retry_interval_seconds=0.01, lock_wait_timeout_seconds=0.5, ) barrier = threading.Barrier(2) errors: list[Exception] = [] def run_refresh() -> None: try: barrier.wait(timeout=1) use_case.execute(batch_id) except Exception as exc: errors.append(exc) first = threading.Thread(target=run_refresh) second = threading.Thread(target=run_refresh) first.start() second.start() first.join(timeout=2) second.join(timeout=2) self.assertFalse(errors) self.assertLessEqual(case_repo.max_parallel_replacements, 1) cases = case_repo.list_cases(batch_id) self.assertEqual(len(cases), 1) self.assertEqual(cases[0].get("case_key"), "case-main") self.assertEqual(cases[0].get("document_count"), 2) self.assertEqual(cases[0].get("processed_document_count"), 1) self.assertEqual(cases[0].get("clinical_failed_count"), 1) self.assertEqual(batch_repo._cases_refresh_lock_owner, "") class ListBatchFilesUseCaseTest(unittest.TestCase): def test_serializer_exposes_case_number(self) -> None: repository = FakeBatchFileRepository("batch-1") repository.update_file( "file-1", { "case_number": "176654", "case_key": "66781911-176654-blanca-lijia-rengifo-aguirre", }, ) use_case = ListBatchFilesUseCase(repository) result = use_case.execute("batch-1") self.assertEqual(result[0]["case_number"], "176654") self.assertEqual( result[0]["case_key"], "66781911-176654-blanca-lijia-rengifo-aguirre", ) class ListUserBatchesUseCaseTest(unittest.TestCase): def test_returns_recent_batches_for_user_with_limit(self) -> None: repository = FakeBatchRepository("batch-1") repository.user_batches = [ { "_id": "batch-old", "usuario": "tester", "nombre_archivo": "old.zip", "status": "completado", "created_at": "2026-03-14T08:00:00-05:00", "updated_at": "2026-03-14T09:00:00-05:00", "total_files": 4, "processed_files": 4, "associated_files": 4, "pending_validation_files": 0, "failed_files": 0, "error": "", }, { "_id": "batch-new", "usuario": "tester", "nombre_archivo": "new.zip", "status": "completado_con_errores", "created_at": "2026-03-15T08:00:00-05:00", "updated_at": "2026-03-15T10:00:00-05:00", "total_files": 9, "processed_files": 9, "associated_files": 7, "pending_validation_files": 2, "failed_files": 0, "error": "", }, { "_id": "batch-foreign", "usuario": "other-user", "nombre_archivo": "other.zip", "status": "fallido", "created_at": "2026-03-16T08:00:00-05:00", "updated_at": "2026-03-16T10:00:00-05:00", "total_files": 1, "processed_files": 1, "associated_files": 0, "pending_validation_files": 0, "failed_files": 1, "error": "boom", }, ] use_case = ListUserBatchesUseCase(repository) result = use_case.execute("tester", limit=1) self.assertEqual(len(result), 1) self.assertEqual(result[0]["batch_id"], "batch-new") self.assertEqual(result[0]["nombre_archivo"], "new.zip") self.assertEqual(result[0]["pending_validation_files"], 2) self.assertEqual(result[0]["failed_files"], 0) class QueueBatchEpicrisisUseCaseTest(unittest.TestCase): def test_queues_only_ready_pending_cases_and_persists_bulk_summary(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-ready", "ready_for_epicrisis": True, "epicrisis_status": "pendiente", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-done", "ready_for_epicrisis": True, "epicrisis_status": "completado", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-running", "ready_for_epicrisis": True, "epicrisis_status": "procesando", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-blocked", "ready_for_epicrisis": False, "epicrisis_status": "pendiente", }, ], ) recompute = RecomputeBatchBulkEpicrisisUseCase(batch_repo, case_repo) use_case = QueueBatchEpicrisisUseCase( batch_repository=batch_repo, batch_case_repository=case_repo, recompute_bulk_epicrisis_use_case=recompute, colombia_tz=UTC, ) result = use_case.execute(batch_id, job_id="job-123") self.assertEqual(result["queued_count"], 1) self.assertEqual(result["skipped_completed_count"], 1) self.assertEqual(result["skipped_inflight_count"], 1) self.assertEqual(result["skipped_not_ready_count"], 1) batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(batch.get("bulk_epicrisis_status"), "en_cola") self.assertEqual(batch.get("bulk_epicrisis_job_id"), "job-123") self.assertEqual(batch.get("bulk_epicrisis_total_target"), 1) self.assertEqual(batch.get("bulk_epicrisis_skipped_count"), 3) self.assertEqual(batch.get("bulk_epicrisis_case_keys"), ["case-ready"]) def test_get_batch_status_recomputes_bulk_progress_from_cases(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, "bulk_epicrisis_status": "en_cola", "bulk_epicrisis_job_id": "job-123", "bulk_epicrisis_total_target": 2, "bulk_epicrisis_skipped_count": 1, "bulk_epicrisis_case_keys": ["case-a", "case-b"], }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-a", "ready_for_epicrisis": True, "epicrisis_status": "completado", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-b", "ready_for_epicrisis": True, "epicrisis_status": "fallido", }, ], ) recompute = RecomputeBatchBulkEpicrisisUseCase(batch_repo, case_repo) use_case = GetBatchStatusUseCase( batch_repository=batch_repo, recompute_bulk_epicrisis_use_case=recompute, ) result = use_case.execute(batch_id) or {} self.assertEqual(result["bulk_epicrisis_status"], "completado") self.assertEqual(result["bulk_epicrisis_completed_count"], 1) self.assertEqual(result["bulk_epicrisis_failed_count"], 1) self.assertEqual(result["bulk_epicrisis_skipped_count"], 1) class QueueBatchEpicrisisExcelUseCaseTest(unittest.TestCase): def test_persists_excel_queue_metadata_when_batch_is_ready(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-a", "epicrisis_status": "completado", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-b", "epicrisis_status": "fallido", }, ], ) use_case = QueueBatchEpicrisisExcelUseCase( batch_repository=batch_repo, batch_case_repository=case_repo, colombia_tz=UTC, ) result = use_case.execute(batch_id, job_id="excel-job-1", persist=True) batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(result["eligible_count"], 1) self.assertEqual(batch.get("excel_epicrisis_status"), "en_cola") self.assertEqual(batch.get("excel_epicrisis_job_id"), "excel-job-1") self.assertEqual(batch.get("excel_epicrisis_filename"), "epicrisis_lote_batch-1.xlsx") self.assertEqual(batch.get("excel_epicrisis_download_url"), "") def test_rejects_queue_when_any_epicrisis_is_inflight(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-a", "epicrisis_status": "procesando", } ], ) use_case = QueueBatchEpicrisisExcelUseCase( batch_repository=batch_repo, batch_case_repository=case_repo, colombia_tz=UTC, ) with self.assertRaisesRegex(ValueError, "en curso"): use_case.execute(batch_id, persist=False) class GenerateBatchEpicrisisExcelUseCaseTest(unittest.TestCase): def _build_context(self, case_key: str, patient_name: str, case_number: str) -> dict: return { "nombre_paciente": patient_name, "case_key": case_key, "case_number": case_number, "historia": { "_id": f"hist-{case_key}", "tipo_documento": "historia_clinica", "fecha_analisis": "2026-03-18T10:00:00-05:00", "nombre_paciente": patient_name, "case_number": case_number, "analisis_html": "HTML oculto
", }, "quirurgico": { "_id": f"qx-{case_key}", "tipo_documento": "quirurgico", "fecha_analisis": "2026-03-18T10:05:00-05:00", "nombre_paciente": patient_name, "case_number": case_number, }, "factura": { "_id": f"fac-{case_key}", "tipo_documento": "factura", "fecha_analisis": "2026-03-18T10:10:00-05:00", "nombre_paciente": patient_name, "case_number": case_number, }, "metadatos_hc": { "nombre_paciente": patient_name, "caso": case_number, "datos_identificacion_paciente": "CC 123456789", "resumen": ( "Paciente con evolución favorable, control del dolor, seguimiento clínico y " "recomendaciones de signos de alarma. " ) * 5, }, "procedimientos_hc": ["Lavado quirúrgico", "Cierre por planos"], "medicamentos_hc_display": ["Código no identificado - Cefazolina 1 g IV"], "hallazgos_quirurgicos": ( "Hallazgos quirúrgicos sin complicaciones, con adecuada viabilidad tisular y " "control local posterior al lavado y desbridamiento. " ) * 4, "descripcion_procedimiento": ( "Se realizó procedimiento sin eventos adversos, con lavado, hemostasia, " "desbridamiento y cierre por planos bajo técnica estéril. " ) * 6, "procedimientos_factura": [{"codigo_soat": "12345", "descripcion": "Curación compleja"}], "soat_resultados": [{"codigo_soat": "67890", "descripcion": "Lavado de herida"}], "codigos_desde_soat": [ { "codigo_soat": "67890", "descripcion": "Lavado de herida", "cie10_principal": { "codigo": "S91.3", "descripcion": "Herida del pie", }, "cie10_secundarios": [ { "codigo": "V43.6", "descripcion": "Accidente de tránsito", } ], "cups_principal": { "codigo": "123456", "descripcion": "Lavado quirúrgico", }, "cups_alternativos": [ { "codigo": "654321", "descripcion": "Curación compleja", } ], "analisis_clinico": "----------------------------\n• Coherencia diagnóstica\n• Prevención de glosas", "respuesta_completa": "Texto crudo del LLM que no debe exportarse.", }, { "codigo_soat": "QX-HALLAZGOS", "fuente": "agente_hallazgos_qx", "descripcion": "Hallazgos quirúrgicos relevantes", "cie10_principal": { "codigo": "M24.01", "descripcion": "Cuerpo libre articular", }, "cups_principal": { "codigo": "222222", "descripcion": "Procedimiento QX", }, "analisis_clinico": "• Hallazgo codificado desde agente QX", "respuesta_completa": "Texto crudo del agente QX que no debe exportarse.", }, ], "glosa_analisis": ( "Sin glosas relevantes, con soporte documental suficiente y consistencia " "diagnóstica para la auditoría.
" ) * 6, } def test_generates_workbook_with_summary_and_case_sheets(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-a-muy-largo-con-caracteres-prohibidos/[1]", "patient_name": "Paciente A", "patient_id": "CC 123456789", "case_number": "1001", "procedure_description": "Lavado quirúrgico", "epicrisis_status": "completado", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-b", "patient_name": "Paciente B", "patient_id": "CC 987654321", "case_number": "1002", "procedure_description": "Curación compleja", "epicrisis_status": "completado", }, { "batch_id": batch_id, "usuario": "tester", "case_key": "case-c", "patient_name": "Paciente C", "patient_id": "CC 555555555", "case_number": "1003", "procedure_description": "Caso omitido", "epicrisis_status": "fallido", }, ], ) cached_contexts = { "case-a-muy-largo-con-caracteres-prohibidos/[1]": self._build_context( "case-a-muy-largo-con-caracteres-prohibidos/[1]", "Paciente A", "1001", ) } rebuilt_contexts = { "case-b": self._build_context("case-b", "Paciente B", "1002"), } case_service = FakeCaseEpicrisisService( cached_contexts=cached_contexts, rebuilt_contexts=rebuilt_contexts, ) with tempfile.TemporaryDirectory() as tmp_dir: report_store = LocalBatchExcelReportStore(Path(tmp_dir)) use_case = GenerateBatchEpicrisisExcelUseCase( batch_repository=batch_repo, batch_case_repository=case_repo, report_store=report_store, workbook_builder=BatchEpicrisisExcelWorkbookBuilder(), case_epicrisis_service=cast(CaseEpicrisisService, case_service), colombia_tz=UTC, ) result = use_case.execute(batch_id, job_id="excel-job-1") batch = batch_repo.get_batch(batch_id) or {} self.assertEqual(result["status"], "completado_con_errores") self.assertEqual(batch.get("excel_epicrisis_status"), "completado_con_errores") self.assertEqual(batch.get("excel_epicrisis_included_count"), 2) self.assertEqual(batch.get("excel_epicrisis_omitted_count"), 1) self.assertEqual( batch.get("excel_epicrisis_download_url"), f"/api/lotes/{batch_id}/excel-epicrisis/descarga", ) self.assertEqual(case_service.rebuild_requests, [("tester", "case-b", False)]) report_path = Path(str(batch.get("excel_epicrisis_path") or "")) self.assertTrue(report_path.exists()) workbook = load_workbook(filename=BytesIO(report_path.read_bytes())) self.assertEqual(workbook.sheetnames[0], "Resumen") self.assertEqual(len(workbook.sheetnames), 3) self.assertTrue(all(len(name) <= 31 for name in workbook.sheetnames)) summary = workbook["Resumen"] summary_values = [summary["A5"].value, summary["A6"].value, summary["A7"].value] self.assertIn("case-a-muy-largo-con-caracteres-prohibidos/[1]", summary_values) self.assertIn("case-b", summary_values) self.assertIn("case-c", summary_values) self.assertEqual(summary["C5"].value, "CC 123456789") self.assertEqual(summary["G7"].value, "No") self.assertIsNotNone(summary["A5"].hyperlink) self.assertEqual(summary["A5"].hyperlink.target, f"#'{workbook.sheetnames[1]}'!A1") self.assertEqual(str(summary.page_setup.paperSize), str(summary.PAPERSIZE_LETTER)) self.assertEqual(summary.page_setup.orientation, summary.ORIENTATION_LANDSCAPE) self.assertEqual(summary.page_setup.fitToWidth, 1) self.assertEqual(summary.page_setup.fitToHeight, 0) self.assertEqual(summary.print_area, "'Resumen'!$A$1:$H$7") case_sheet = workbook[workbook.sheetnames[1]] self.assertEqual(str(case_sheet.page_setup.paperSize), str(case_sheet.PAPERSIZE_LETTER)) self.assertEqual(case_sheet.page_setup.orientation, case_sheet.ORIENTATION_PORTRAIT) self.assertEqual(case_sheet.page_setup.fitToWidth, 1) self.assertEqual(case_sheet.page_setup.fitToHeight, 0) self.assertRegex(str(case_sheet.print_area), r"!\$A\$1:\$[DH]\$") self.assertIsNotNone(case_sheet.row_dimensions[1].height) def test_omits_non_exportable_diagnostic_aids_from_case_workbook(self) -> None: batch_id = "batch-1" batch_repo = FakeBatchRepository(batch_id) batch_repo.update_batch( batch_id, { "status": "completado", "pending_validation_files": 0, "clinical_pending_files": 0, }, ) case_repo = FakeBatchCaseRepository(batch_id) case_repo.replace_cases( batch_id, [ { "batch_id": batch_id, "usuario": "tester", "case_key": "case-a", "patient_name": "Paciente A", "patient_id": "CC 123456789", "case_number": "1001", "procedure_description": "Lavado quirúrgico", "epicrisis_status": "completado", } ], ) context = self._build_context("case-a", "Paciente A", "1001") context["ayudas_diagnosticas"] = [ { "tipo": "imagen", "nombre": "RX visible", "concepto": "No interpretado", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "exportable": True, }, { "tipo": "imagen", "nombre": "RX excluida", "concepto": "No interpretado", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "exportable": False, }, ] case_service = FakeCaseEpicrisisService(cached_contexts={"case-a": context}) with tempfile.TemporaryDirectory() as tmp_dir: report_store = LocalBatchExcelReportStore(Path(tmp_dir)) use_case = GenerateBatchEpicrisisExcelUseCase( batch_repository=batch_repo, batch_case_repository=case_repo, report_store=report_store, workbook_builder=BatchEpicrisisExcelWorkbookBuilder(), case_epicrisis_service=cast(CaseEpicrisisService, case_service), colombia_tz=UTC, ) use_case.execute(batch_id, job_id="excel-job-1") batch = batch_repo.get_batch(batch_id) or {} workbook = load_workbook(filename=BytesIO(Path(str(batch["excel_epicrisis_path"])).read_bytes())) case_sheet = workbook[workbook.sheetnames[1]] values = [ case_sheet.cell(row=row_index, column=column_index).value for row_index in range(1, case_sheet.max_row + 1) for column_index in range(1, case_sheet.max_column + 1) ] self.assertIn("RX visible", values) self.assertNotIn("RX excluida", values) self.assertGreater(case_sheet.row_dimensions[1].height, 18) def find_row(label: str) -> int: for row_index in range(1, case_sheet.max_row + 1): if case_sheet.cell(row=row_index, column=1).value == label: return row_index raise AssertionError(f"Label {label!r} not found") self.assertGreater(case_sheet.row_dimensions[find_row("Resumen")].height, 60.0) self.assertNotIn("Procedimientos de factura", values) self.assertNotIn("Resultados SOAT", values) self.assertNotIn("Glosa analítica", values) values = [ cell for row in case_sheet.iter_rows(min_row=1, max_row=case_sheet.max_row, min_col=1, max_col=4) for cell in [row[0].value, row[1].value, row[2].value, row[3].value] if isinstance(cell, str) ] self.assertNotIn("Código SOAT", values) self.assertNotIn("CIE10 principal", values) self.assertNotIn("CUPS alternativos", values) self.assertNotIn("Hallazgos quirúrgicos codificados", values) self.assertNotIn("Agente hallazgos QX", values) self.assertNotIn("67890", values) self.assertNotIn("QX-HALLAZGOS", values) self.assertNotIn("Respuesta completa", values) self.assertNotIn("Texto crudo del LLM que no debe exportarse.", values) self.assertNotIn("Texto crudo del agente QX que no debe exportarse.", values) self.assertNotIn("----------------------------", values) if __name__ == "__main__": unittest.main()