from __future__ import annotations
import tempfile
import unittest
import zipfile
from io import BytesIO
from datetime import timezone
from pathlib import Path
from openpyxl import load_workbook
from app.batch_processing.application.use_cases import (
GenerateBatchEpicrisisExcelUseCase,
GetBatchStatusUseCase,
ListBatchFilesUseCase,
ListUserBatchesUseCase,
MaterializeBatchFileUseCase,
PrepareBatchUseCase,
QueueBatchEpicrisisUseCase,
QueueBatchEpicrisisExcelUseCase,
RecomputeBatchBulkEpicrisisUseCase,
RefreshBatchCasesUseCase,
RecomputeBatchTotalsUseCase,
ResolveFileAssociationUseCase,
)
from app.batch_processing.domain.models import ArchiveExtractionResult, ExtractedArchiveEntry
from app.batch_processing.infrastructure.heuristic_association import (
HeuristicCaseAssociationService,
)
from app.batch_processing.infrastructure.batch_epicrisis_excel import (
BatchEpicrisisExcelWorkbookBuilder,
)
from app.batch_processing.infrastructure.heuristic_classifier import (
HeuristicDocumentClassifier,
)
from app.batch_processing.infrastructure.local_artifacts import (
LocalBatchArtifactCleaner,
LocalBatchExcelReportStore,
)
from app.batch_processing.infrastructure.zip_archive import SafeZipArchiveExtractor
class SafeZipArchiveExtractorTest(unittest.TestCase):
def test_rejects_path_traversal_and_non_pdf_files(self) -> None:
with tempfile.TemporaryDirectory() as tmp_dir:
archive_path = Path(tmp_dir) / "lote.zip"
with zipfile.ZipFile(archive_path, "w") as zf:
zf.writestr("../escape.pdf", b"fake")
zf.writestr("folder/notas.txt", b"nope")
zf.writestr("folder/ok.pdf", b"%PDF-1.4\nfake")
extractor = SafeZipArchiveExtractor(Path(tmp_dir))
result = extractor.extract_archive(str(archive_path), "batch-test")
self.assertEqual(len(result.entries), 1)
self.assertEqual(result.entries[0].original_name, "ok.pdf")
self.assertEqual(len(result.rejected_entries), 2)
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 HeuristicDocumentClassifierTest(unittest.TestCase):
def test_detects_factura_from_text(self) -> None:
classifier = HeuristicDocumentClassifier()
detected_type = classifier.classify(
"documento.pdf", "Factura de venta. Valor total y copago."
)
self.assertEqual(detected_type, "factura")
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_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_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"},
)
class FakeBatchRepository:
def __init__(self, batch_id: str) -> None:
self.batch_id = batch_id
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)]
def create_batch(self, payload: dict) -> str:
raise NotImplementedError
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]
class FakeBatchFileRepository:
def __init__(self, batch_id: str) -> None:
self.batch_id = batch_id
self.next_id = 2
self.files = {
"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
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 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
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] = []
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")
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 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 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=timezone.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=timezone.utc,
)
result = use_case.execute(batch_id)
self.assertEqual(result, ["file-2"])
self.assertEqual(cleaner.deleted_archives, ["/tmp/demo.zip"])
self.assertEqual(batch_repo.get_batch(batch_id).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=timezone.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=timezone.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=timezone.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=timezone.utc,
)
with self.assertLogs("app.batch_processing.application.command_use_cases", level="WARNING"):
result = use_case.execute(batch_id)
self.assertEqual(result, ["file-2"])
self.assertEqual(batch_repo.get_batch(batch_id).get("archive_path"), "/tmp/demo.zip")
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=timezone.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=clinical_service,
artifact_cleaner=cleaner,
colombia_tz=timezone.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",
"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"])
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")
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=timezone.utc,
)
refresh_cases_use_case = RefreshBatchCasesUseCase(
batch_repository=batch_repo,
batch_file_repository=file_repo,
batch_case_repository=case_repo,
)
use_case = ResolveFileAssociationUseCase(
batch_file_repository=file_repo,
totals_use_case=totals_use_case,
refresh_cases_use_case=refresh_cases_use_case,
colombia_tz=timezone.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")
self.assertEqual(batch_repo.get_batch(batch_id).get("associated_files"), 1)
self.assertEqual(batch_repo.get_batch(batch_id).get("pending_validation_files"), 0)
self.assertEqual(len(case_repo.list_cases(batch_id)), 1)
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=timezone.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=timezone.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=timezone.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.",
},
"procedimientos_hc": ["Lavado quirúrgico", "Cierre por planos"],
"medicamentos_hc_display": ["Código no identificado - Cefazolina 1 g IV"],
"hallazgos_quirurgicos": "Hallazgos sin complicaciones.",
"descripcion_procedimiento": "Se realizó procedimiento sin eventos adversos.",
"procedimientos_factura": [{"codigo_soat": "12345", "descripcion": "Curación compleja"}],
"soat_resultados": [{"codigo_soat": "67890", "descripcion": "Lavado de herida"}],
"codigos_desde_soat": [{"codigo": "S91.3", "descripcion": "Herida del pie"}],
"glosa_analisis": "Sin glosas relevantes.
",
}
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=case_service,
colombia_tz=timezone.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")
if __name__ == "__main__":
unittest.main()