from __future__ import annotations import unittest from io import BytesIO from openpyxl import load_workbook from app.batch_processing.infrastructure.batch_epicrisis_excel import ( BatchEpicrisisExcelCaseRow, BatchEpicrisisExcelWorkbookBuilder, ) def _find_first_row(case_sheet, value: str) -> int: for row_index in range(1, case_sheet.max_row + 1): if case_sheet.cell(row=row_index, column=1).value == value: return row_index raise AssertionError(f"Value {value!r} not found in first column") def _build_context(*, short: bool = False, with_soat_cards: bool = False) -> dict: context = { "nombre_paciente": "Paciente Demo", "case_number": "1001", "metadatos_hc": { "prestador_servicio": "Clinica Demo", "sexo": "FEMENINO", "edad": "47 AÑOS", "motivo_consulta": "Dolor post trauma.", "resumen": ( "Paciente con evolucion favorable y control del dolor." if short else ( "Paciente con evolucion clinica estable luego de trauma complejo de pie izquierdo. " "Se documenta valoracion por urgencias, control del dolor, inmovilizacion inicial, " "lavado quirurgico y seguimiento ambulatorio con recomendaciones de signos de alarma. " ) * 6 ), }, "historia": { "_id": "hist-1", "fecha_analisis": "2026-03-31T10:00:00-05:00", }, "quirurgico": { "_id": "qx-1", "fecha_analisis": "2026-03-31T10:05:00-05:00", }, "factura": { "_id": "fac-1", "fecha_analisis": "2026-03-31T10:10:00-05:00", }, "procedimientos_hc": [ "873431 - RADIOGRAFIA DE TOBILLO", "902210 - HEMOGRAMA IV", ], "antecedentes_hc": [ "Apendicectomia previa", "Hipertension arterial", ], "medicamentos_hc_display": [ "5214 - ACETAMINOFEN", "5944 - KETOROLACO", ], "hallazgos_quirurgicos": ( "Hallazgos quirurgicos acotados." if short else ( "Tejido inflamado con edema, dolor, compromiso local y necesidad de lavado, " "desbridamiento y posterior inmovilizacion. " ) * 4 ), "descripcion_procedimiento": ( "Procedimiento breve." if short else ( "Se realiza lavado quirurgico, desbridamiento profundo, hemostasia y estabilizacion " "de la extremidad bajo tecnica esteril. " ) * 7 ), "procedimientos_factura": [ {"codigo_soat": "13563", "descripcion": "Reduccion cerrada falanges pie (una a dos)"}, {"codigo_soat": "935304", "descripcion": "Inmovilizacion miembro superior o inferior total o parcial"}, ], "ayudas_diagnosticas": [ { "tipo": "imagen", "nombre": "RX de tobillo", "concepto": "No interpretado", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "fuente": "factura", "documento_nombre": "factura.pdf", "alertas": ["ayuda_no_interpretada"], "exportable": True, } ], "soat_resultados": [ { "codigo_soat": "13563 - Reduccion cerrada falanges pie (una a dos) (p.67)", "procedimiento": "REDUCCION CERRADA DE FRACTURA DE FALANGE PROXIMAL DEL 1ER DEDO EN PIE IZQUIERDO", }, ], "glosa_analisis": ( "Analisis corto." if short else ( "Analisis de riesgo de glosa con recomendaciones de soporte documental, causalidad, " "consistencia diagnostica y completitud quirurgica. " ) * 5 ), "codigos_desde_soat": [], } if with_soat_cards: context["codigos_desde_soat"] = [ { "codigo_soat": "", "descripcion": "Reduccion cerrada falanges pie", "fuente": "soat", "analisis_clinico": ( "Coherencia diagnostico procedimiento y validacion de glosa. " ) * 12, "cie10_principal": { "codigo": "S92.41", "descripcion": "Fractura del dedo gordo del pie izquierdo", }, "cie10_secundarios": [ { "codigo": "V43.6", "descripcion": "Ocupante de automovil lesionado en colision", }, { "codigo": "S92.49", "descripcion": "Otras fracturas de los dedos del pie", }, ], "cups_principal": { "codigo": "770101", "descripcion": ( "Reduccion abierta de fractura de humero con osteosintesis" ), "justificacion": ( "Este es el codigo CUPS mas directo y apropiado para el procedimiento " "de osteosintesis en humero cuando se realiza en la diafisis." ), }, "cups_alternativos": [ { "codigo": "810801", "descripcion": "Osteosintesis de radio y cubito", "justificacion": "Alternativa quirurgica descartada por localizacion anatomica.", }, { "codigo": "791901", "descripcion": "Reduccion abierta de fractura de cubito", "justificacion": "Procedimiento relacionado pero no corresponde al humero.", }, ], }, { "codigo_soat": "935304", "descripcion": "Inmovilizacion miembro inferior", "fuente": "soat", "analisis_clinico": "Soporte clinico de inmovilizacion. " * 3, "cie10_principal": { "codigo": "S52.52", "descripcion": "Fractura de radio distal", }, "cie10_secundarios": [{"codigo": "V49.9", "descripcion": "Accidente de transporte"}], "cups_principal": { "codigo": "890100", "descripcion": "Inmovilizacion de miembro inferior", "justificacion": "Soporte inicial posterior al manejo de la fractura.", }, "cups_alternativos": [ { "codigo": "890200", "descripcion": "Cambio de inmovilizacion", "justificacion": "Alternativa de seguimiento no principal.", } ], }, { "codigo_soat": "QX-HALLAZGOS", "descripcion": "Hallazgos qx relevantes", "fuente": "agente_hallazgos_qx", "analisis_clinico": "Hallazgos qx codificados con soporte. " * 3, "cups_principal": { "codigo": "801101", "descripcion": "Exploracion quirurgica de humero", "justificacion": "Relacion directa con los hallazgos quirurgicos documentados.", }, "cups_alternativos": [ { "codigo": "935600", "descripcion": "Control postoperatorio ortopedico", "justificacion": "Codigo complementario, no principal.", } ], }, ] return context class BatchEpicrisisExcelWorkbookBuilderTest(unittest.TestCase): def test_build_exports_recommendations_before_antecedents_without_trace_fields(self) -> None: context = _build_context(short=True) context["recomendaciones_medicas"] = [ { "categoria": "restriccion", "indicacion": "Evitar cargar peso", "duracion": "6 semanas", "fecha": "2026-07-30", "pagina": 9, "evidencia": "Evidencia clínica reservada", }, {"categoria": "incapacidad", "indicacion": "Incapacidad laboral", "duracion": "15 días"}, ] workbook = load_workbook( BytesIO( BatchEpicrisisExcelWorkbookBuilder().build( batch_id="batch-recommendations", rows=[ BatchEpicrisisExcelCaseRow( case_key="case-recommendations", patient_name="Paciente Demo", patient_id="CC - 1", case_number="1001", procedure_description="Procedimiento", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) ], ) ) ) case_sheet = workbook[workbook.sheetnames[1]] recommendations_row = _find_first_row(case_sheet, "Recomendaciones médicas") antecedentes_row = _find_first_row(case_sheet, "Antecedentes HC") self.assertLess(recommendations_row, antecedentes_row) self.assertEqual( [case_sheet.cell(row=recommendations_row + 1, column=index).value for index in range(1, 5)], ["Categoría", "Indicación", "Duración", "Fecha"], ) self.assertEqual(case_sheet.cell(row=recommendations_row + 2, column=1).value, "Incapacidad") self.assertEqual(case_sheet.cell(row=recommendations_row + 3, column=1).value, "Restricciones") values = [ case_sheet.cell(row=row, column=column).value for row in range(1, case_sheet.max_row + 1) for column in range(1, case_sheet.max_column + 1) ] self.assertNotIn("Evidencia clínica reservada", values) self.assertNotIn("Página", values) self.assertEqual(values.count("Incapacidad"), 1) def test_curated_export_excludes_pending_diagnoses_and_procedures(self) -> None: context = _build_context(short=True) context.update( { "curation_schema_version": "v1", "diagnosticos_curados": [ { "code": "S422", "description": "Fractura proximal", "status": "aprobado", }, { "code": "T870", "description": "Inferencia pendiente", "status": "pendiente_revision", }, ], "procedimientos_curados": [ { "codigo_cups": "793203", "description": "Reducción abierta", "status": "validado_automaticamente", "states": ["realizado"], "classification": "quirurgico", "reconciliation_status": "realizado_y_facturado", }, { "codigo_cups": "836301", "description": "Procedimiento facturado pendiente", "status": "pendiente_revision", "states": ["facturado"], }, ], } ) row = BatchEpicrisisExcelCaseRow( case_key="case-curated", patient_name="Paciente Demo", patient_id="123", case_number="1001", procedure_description="", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) workbook = load_workbook( BytesIO( BatchEpicrisisExcelWorkbookBuilder().build( batch_id="batch-curated", rows=[row], ) ) ) values = [ str(cell.value) for sheet_row in workbook["case-curated"].iter_rows() for cell in sheet_row if cell.value is not None ] rendered = "\n".join(values) self.assertIn("S422", rendered) self.assertIn("793203", rendered) self.assertNotIn("T870", rendered) self.assertNotIn("836301", rendered) def test_build_exports_medications_without_status_column(self) -> None: builder = BatchEpicrisisExcelWorkbookBuilder() context = _build_context(short=True) context["medicamentos_caso"] = [ { "nombre": "Acetaminofén", "presentacion": "Tableta", "posologia": "Oral cada 6 horas", "dosis": "500 mg", "cantidad": 12, "estados": ["administrado", "facturado"], "codigo_referencia": "199999-1", "codigo_facturacion": "005214", "pertinencia": { "estado": "posible_glosa", "motivos": ["No se encontró evidencia asistencial de suministro."], }, }, { "nombre": "Naproxeno", "dosis": "250 mg", "estados": ["formulado"], }, ] workbook = load_workbook( BytesIO( builder.build( batch_id="batch-demo", rows=[ BatchEpicrisisExcelCaseRow( case_key="case-medications", patient_name="Paciente Demo", patient_id="CC - 1", case_number="1001", procedure_description="Procedimiento", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) ], ) ) ) case_sheet = workbook[workbook.sheetnames[1]] section_row = _find_first_row(case_sheet, "Medicamentos") headers = [case_sheet.cell(row=section_row + 1, column=index).value for index in range(1, 6)] self.assertEqual( headers, ["CUM/IUM", "Medicamento", "Posología", "Dosis", "Cantidad"], ) self.assertNotIn("Estado", headers) self.assertEqual(case_sheet.cell(row=section_row + 2, column=1).value, "199999-1") self.assertEqual(case_sheet.cell(row=section_row + 2, column=2).value, "Acetaminofén · Tableta") self.assertEqual(case_sheet.cell(row=section_row + 2, column=5).value, "12") self.assertEqual(case_sheet.cell(row=section_row + 2, column=1).number_format, "@") self.assertIn( f"A{section_row}:E{section_row}", {str(cell_range) for cell_range in case_sheet.merged_cells.ranges}, ) medication_values = [ case_sheet.cell(row=row, column=column).value for row in range(section_row, section_row + 3) for column in range(1, 6) ] self.assertNotIn("Código interno", medication_values) self.assertNotIn("005214", medication_values) self.assertNotIn("formulado", [str(value).lower() for value in headers if value]) self.assertNotIn("Posible glosa", medication_values) self.assertFalse(any("evidencia asistencial" in str(value) for value in medication_values)) self.assertNotIn("Naproxeno", medication_values) def test_build_compacts_case_sheet_spacing_and_preserves_print_setup(self) -> None: builder = BatchEpicrisisExcelWorkbookBuilder() rows = [ BatchEpicrisisExcelCaseRow( case_key="case-short", patient_name="Paciente Corto", patient_id="CC - 1", case_number="1001", procedure_description="Procedimiento corto", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=_build_context(short=True), ), BatchEpicrisisExcelCaseRow( case_key="case-long", patient_name="Paciente Largo", patient_id="CC - 2", case_number="1002", procedure_description="Procedimiento largo", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=_build_context(with_soat_cards=True), ), BatchEpicrisisExcelCaseRow( case_key="case-soat", patient_name="Paciente SOAT", patient_id="CC - 3", case_number="1003", procedure_description="Procedimiento soat", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=_build_context(short=False, with_soat_cards=True), ), ] workbook_bytes = builder.build(batch_id="batch-demo", rows=rows) workbook = load_workbook(BytesIO(workbook_bytes)) self.assertEqual(workbook.sheetnames[0], "Resumen") self.assertEqual(len(workbook.sheetnames), 4) summary = workbook["Resumen"] self.assertEqual(str(summary.page_setup.paperSize), str(summary.PAPERSIZE_LETTER)) self.assertEqual(summary.page_setup.orientation, summary.ORIENTATION_LANDSCAPE) case_sheet = workbook[workbook.sheetnames[2]] 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.freeze_panes, "A4") self.assertEqual(case_sheet["A3"].value, "Identificación") self.assertEqual(case_sheet["A4"].value, "Case key") exported_values = [cell.value for row in case_sheet.iter_rows() for cell in row] self.assertNotIn("Fuentes documentales", exported_values) self.assertNotIn("Procedimientos de factura", exported_values) self.assertNotIn("Resultados SOAT", exported_values) widths = {column: case_sheet.column_dimensions[column].width for column in ("A", "B", "C", "D")} self.assertLess(widths["A"], 22.0) self.assertGreater(widths["B"] + widths["C"] + widths["D"], 80.0) self.assertGreater(sum(widths.values()), 95.0) self.assertEqual(case_sheet.page_margins.left, 0.25) self.assertEqual(case_sheet.page_margins.right, 0.25) blank_rows: list[int] = [] consecutive_blank_streak = 0 max_blank_streak = 0 for row_index in range(1, case_sheet.max_row + 1): values = [ case_sheet.cell(row=row_index, column=column_index).value for column_index in range(1, case_sheet.max_column + 1) ] if any(value not in (None, "") for value in values): consecutive_blank_streak = 0 continue blank_rows.append(row_index) consecutive_blank_streak += 1 max_blank_streak = max(max_blank_streak, consecutive_blank_streak) self.assertLessEqual(max_blank_streak, 2) self.assertIn(9, blank_rows) self.assertGreaterEqual(len(blank_rows), 2) self.assertLessEqual(case_sheet.row_dimensions[1].height, 22.0) self.assertLessEqual(case_sheet.row_dimensions[4].height, 18.0) antecedentes_section_row = _find_first_row(case_sheet, "Antecedentes HC") resumen_row = _find_first_row(case_sheet, "Resumen") ayudas_section_row = _find_first_row(case_sheet, "Ayudas diagnósticas") all_values = [ case_sheet.cell(row=row, column=column).value for row in range(1, case_sheet.max_row + 1) for column in range(1, case_sheet.max_column + 1) ] self.assertGreater(case_sheet.row_dimensions[resumen_row].height, 100.0) self.assertNotIn("Recomendaciones médicas", all_values) self.assertEqual(case_sheet.cell(row=antecedentes_section_row + 1, column=1).value, "Apendicectomia previa") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 1, column=1).value, "Tipo") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 1, column=2).value, "Ayuda") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 1, column=3).value, "Interpretacion") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 2, column=1).value, "imagen") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 2, column=2).value, "RX de tobillo") self.assertIsNone(case_sheet.cell(row=ayudas_section_row + 2, column=3).value) def test_build_excludes_non_exportable_diagnostic_aids_from_excel(self) -> None: builder = BatchEpicrisisExcelWorkbookBuilder() context = _build_context(short=True) context["ayudas_diagnosticas"] = [ { "tipo": "imagen", "nombre": "RX exportable", "concepto": "No interpretado", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "exportable": True, }, { "tipo": "imagen", "nombre": "RX excluida por draft", "concepto": "No interpretado", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "exportable": False, }, { "tipo": "laboratorio", "nombre": "Laboratorio glosado", "concepto": "Pendiente", "ordenado": True, "interpretado": False, "facturado": True, "glosado": True, "estado_interpretacion": "excluido", "exportable": False, }, ] workbook = load_workbook( BytesIO( builder.build( batch_id="batch-demo", rows=[ BatchEpicrisisExcelCaseRow( case_key="case-exportable", patient_name="Paciente Demo", patient_id="CC - 1", case_number="1001", procedure_description="Procedimiento", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) ], ) ) ) case_sheet = workbook[workbook.sheetnames[1]] ayudas_section_row = _find_first_row(case_sheet, "Ayudas diagnósticas") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 2, column=2).value, "RX exportable") self.assertEqual(case_sheet.cell(row=ayudas_section_row + 3, column=1).value, None) sheet_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.assertNotIn("RX excluida por draft", sheet_values) self.assertNotIn("Laboratorio glosado", sheet_values) def test_build_exports_invoice_diagnostic_aid_description_as_label(self) -> None: builder = BatchEpicrisisExcelWorkbookBuilder() context = _build_context(short=True) context["ayudas_diagnosticas"] = [ { "tipo": "imagen", "nombre": "21602", "concepto": "Portátiles con fluoroscopia y/o intensificador de imagen", "ordenado": True, "interpretado": False, "facturado": True, "glosado": False, "estado_interpretacion": "no_interpretado", "fuente": "factura", "exportable": True, } ] workbook = load_workbook( BytesIO( builder.build( batch_id="batch-demo", rows=[ BatchEpicrisisExcelCaseRow( case_key="case-aid-description", patient_name="Paciente Demo", patient_id="CC - 1", case_number="1001", procedure_description="Procedimiento", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=context, ) ], ) ) ) case_sheet = workbook[workbook.sheetnames[1]] ayudas_section_row = _find_first_row(case_sheet, "Ayudas diagnósticas") self.assertEqual( case_sheet.cell(row=ayudas_section_row + 2, column=2).value, "21602 - Portátiles con fluoroscopia y/o intensificador de imagen", ) self.assertIsNone(case_sheet.cell(row=ayudas_section_row + 2, column=3).value) def test_build_keeps_soat_cards_compact_without_blank_rows_between_cards(self) -> None: builder = BatchEpicrisisExcelWorkbookBuilder() rows = [ BatchEpicrisisExcelCaseRow( case_key="case-soat", patient_name="Paciente SOAT", patient_id="CC - 3", case_number="1003", procedure_description="Procedimiento soat", epicrisis_status="completado", included_in_excel=True, omission_reason="", context=_build_context(with_soat_cards=True), ) ] workbook = load_workbook(BytesIO(builder.build(batch_id="batch-demo", rows=rows))) case_sheet = workbook[workbook.sheetnames[1]] card_rows = [ row_index for row_index in range(1, case_sheet.max_row + 1) if case_sheet.cell(row=row_index, column=1).value == "Registro SOAT" ] self.assertEqual(card_rows, []) values = [cell.value for row in case_sheet.iter_rows() for cell in row] self.assertIn("Procedimientos - Quirúrgicos", values) self.assertNotIn("CUPS principal", values) self.assertNotIn("Código SOAT", values)