from __future__ import annotations import unittest from types import SimpleNamespace from app.case_epicrisis.application.medication_pertinence import ( MedicationInvoiceState, apply_clinical_pertinence_reviews, build_medication_glosa_findings, evaluate_medication_pertinence, ) from app.case_epicrisis.infrastructure.medication_pertinence import LLMMedicationPertinenceReviewer from app.services.historia_medicamentos import reconcile_case_medications def _medication(*statuses: str, evidence: str = "", discrepancy: bool = False) -> dict: return { "key": "med-1", "nombre": "Cefazolina", "dosis": "2 g", "posologia": "Intravenosa ahora", "estados": list(statuses), "fuentes": ["historia_clinica"], "evidencias": [ { "evidence_id": "medev-clinical-1", "fuente": "historia_clinica", "estado": statuses[0] if statuses else "otro", "role": ( "administracion" if statuses and statuses[0] == "administrado" else "facturacion" if statuses and statuses[0] == "facturado" else "orden" ), "evidencia": evidence, } ], "discrepancia_cantidad": discrepancy, } class MedicationPertinencePolicyTests(unittest.TestCase): def test_complete_documentary_chain_is_pertinent(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "administrado", "facturado")], invoice_available=True, ) self.assertEqual(result[0]["pertinencia"]["estado"], "pertinente") self.assertEqual( result[0]["pertinencia"]["cadena"], {"orden": "presente", "administracion": "presente", "facturacion": "presente"}, ) def test_billed_without_order_or_administration_is_possible_glosa(self) -> None: result = evaluate_medication_pertinence( [_medication("facturado")], invoice_available=True, ) assessment = result[0]["pertinencia"] self.assertEqual(assessment["estado"], "posible_glosa") self.assertIn("MED-GLOSA-ORDEN", assessment["codigos_regla"]) self.assertIn("MED-GLOSA-ADMINISTRACION", assessment["codigos_regla"]) def test_ordered_and_administered_but_not_billed_is_possible_glosa(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "administrado")], invoice_available=True, ) self.assertEqual(result[0]["pertinencia"]["estado"], "posible_glosa") self.assertIn("MED-GLOSA-FACTURACION", result[0]["pertinencia"]["codigos_regla"]) def test_missing_invoice_is_review_not_confirmed_absence(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "administrado")], invoice_available=False, ) self.assertEqual(result[0]["pertinencia"]["estado"], "revisar") self.assertEqual(result[0]["pertinencia"]["cadena"]["facturacion"], "no_evaluable") def test_insufficient_invoice_is_review_not_possible_glosa(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "administrado")], invoice_state=MedicationInvoiceState.INSUFICIENTE, ) assessment = result[0]["pertinencia"] self.assertEqual(assessment["estado"], "revisar") self.assertIn("MED-REV-FACTURA-INSUFICIENTE", assessment["codigos_regla"]) def test_billed_quantity_above_clinical_support_is_possible_glosa(self) -> None: medication = _medication("ordenado", "administrado", "facturado") medication["cantidad_facturada"] = 5 medication["unidad_cantidad"] = "viales" medication["evidencias"] = [ { "evidence_id": "medev-admin", "fuente": "historia_clinica", "estado": "administrado", "role": "administracion", "cantidad": 2, "unidad": "viales", "evidencia": "Se administraron dos viales.", }, { "evidence_id": "medev-billed", "fuente": "factura", "estado": "facturado", "role": "facturacion", "cantidad": 5, "unidad": "unidades", "evidencia": "Se facturaron cinco viales.", }, ] result = evaluate_medication_pertinence([medication], invoice_available=True) self.assertEqual(result[0]["pertinencia"]["estado"], "posible_glosa") self.assertIn("MED-GLOSA-CANTIDAD", result[0]["pertinencia"]["codigos_regla"]) def test_incompatible_quantity_units_require_review_without_asserting_excess(self) -> None: medication = _medication("ordenado", "administrado", "facturado") medication["cantidad_facturada"] = 500 medication["evidencias"] = [ { "evidence_id": "medev-admin-ml", "fuente": "historia_clinica", "estado": "administrado", "role": "administracion", "cantidad": 1, "unidad": "bolsa", "evidencia": "Se administró una bolsa.", }, { "evidence_id": "medev-billed-ml", "fuente": "factura", "estado": "facturado", "role": "facturacion", "cantidad": 500, "unidad": "ml", "evidencia": "Se facturaron 500 ml.", }, ] result = evaluate_medication_pertinence([medication], invoice_available=True) assessment = result[0]["pertinencia"] self.assertEqual(assessment["estado"], "revisar") self.assertIn("MED-REV-CANTIDAD", assessment["codigos_regla"]) self.assertNotIn("MED-GLOSA-CANTIDAD", assessment["codigos_regla"]) def test_discharge_formula_does_not_require_administration_or_invoice(self) -> None: result = evaluate_medication_pertinence( [_medication("formulado", evidence="Fórmula de egreso por 5 días")], invoice_available=True, ) assessment = result[0]["pertinencia"] self.assertEqual(assessment["estado"], "pertinente") self.assertEqual(assessment["cadena"]["administracion"], "no_aplica") self.assertEqual(assessment["cadena"]["facturacion"], "no_aplica") def test_antecedent_is_not_an_administrative_risk(self) -> None: result = evaluate_medication_pertinence( [_medication("antecedente")], invoice_available=True, ) self.assertEqual(result[0]["pertinencia"]["estado"], "pertinente") def test_contextual_surgical_mention_supports_administration(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "facturado")], invoice_available=True, support_documents=[ { "fuente": "quirurgico: nota.pdf", "texto": "Durante el procedimiento se empleó cefazolina sin complicaciones.", } ], ) self.assertEqual(result[0]["pertinencia"]["estado"], "pertinente") self.assertIn("quirurgico: nota.pdf", result[0]["pertinencia"]["fuentes"]) def test_allergy_or_antecedent_mention_does_not_support_administration(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "facturado")], invoice_available=True, support_documents=[ { "fuente": "quirurgico", "texto": "Antecedente de alergia a cefazolina.", } ], ) self.assertEqual(result[0]["pertinencia"]["estado"], "posible_glosa") self.assertEqual(result[0]["pertinencia"]["cadena"]["administracion"], "ausente") def test_recommendation_or_suspension_does_not_support_administration(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "facturado")], invoice_available=True, support_documents=[ { "fuente": "historia_clinica", "texto": "Se recomienda cefazolina. Posteriormente se suspende cefazolina.", } ], ) assessment = result[0]["pertinencia"] self.assertEqual(assessment["estado"], "posible_glosa") self.assertEqual(assessment["cadena"]["administracion"], "ausente") def test_radiology_mention_does_not_support_administration(self) -> None: result = evaluate_medication_pertinence( [_medication("ordenado", "facturado")], invoice_available=True, support_documents=[ { "fuente": "radiologia: informe.pdf", "texto": "El contraste fue administrado junto con cefazolina.", } ], ) self.assertEqual(result[0]["pertinencia"]["cadena"]["administracion"], "ausente") def test_clinical_review_can_raise_risk_only_with_evidence(self) -> None: base = evaluate_medication_pertinence( [_medication("ordenado", "administrado", "facturado")], invoice_available=True, ) without_evidence = apply_clinical_pertinence_reviews( base, [ { "key": "med-1", "estado": "posible_glosa", "justificacion": "No guarda relación con el procedimiento.", "evidencias": [], } ], ) with_unknown_evidence = apply_clinical_pertinence_reviews( base, [ { "key": "med-1", "estado": "posible_glosa", "justificacion": "No guarda relación con el procedimiento.", "evidence_ids": ["medev-invented"], } ], ) with_evidence = apply_clinical_pertinence_reviews( base, [ { "key": "med-1", "estado": "posible_glosa", "justificacion": "No guarda relación con el procedimiento documentado.", "evidence_ids": ["medev-clinical-1"], } ], ) self.assertEqual(without_evidence[0]["pertinencia"]["estado"], "pertinente") self.assertEqual(with_unknown_evidence[0]["pertinencia"]["estado"], "pertinente") self.assertEqual(with_evidence[0]["pertinencia"]["estado"], "posible_glosa") self.assertEqual(with_evidence[0]["pertinencia"]["origen"], "reglas+llm") self.assertEqual(build_medication_glosa_findings(with_evidence)[0]["key"], "med-1") class MedicationPertinenceReviewerTests(unittest.TestCase): def test_structured_reviewer_returns_traceable_review_without_identity(self) -> None: captured = [] class _Router: def generate_structured(self, request): captured.append(request) return SimpleNamespace( content={ "sv": "v2", "it": [ { "k": "med-1", "e": "posible_glosa", "j": "No corresponde al procedimiento documentado.", "ev": ["medev-clinical-1"], } ], } ) reviewer = LLMMedicationPertinenceReviewer(_Router()) reviews = reviewer.review( medications=evaluate_medication_pertinence( [_medication("ordenado", "administrado", "facturado")], invoice_available=True, ), diagnoses=[{"codigo": "S00", "descripcion": "Trauma superficial"}], procedures=[{"codigo_cups": "123", "descripcion": "Curación simple"}], ) self.assertEqual(reviews[0]["estado"], "posible_glosa") self.assertEqual(reviews[0]["evidence_ids"], ["medev-clinical-1"]) self.assertIn('"k": "med-1"', captured[0].prompt) self.assertNotIn("patient", captured[0].prompt.casefold()) def test_invalid_structured_response_falls_back_without_raising(self) -> None: router = SimpleNamespace( generate_structured=lambda request: SimpleNamespace(content={"it": [{"e": "pertinente"}]}) ) reviews = LLMMedicationPertinenceReviewer(router).review( medications=[_medication("facturado")], diagnoses=[], procedures=[], ) self.assertEqual(reviews, []) class MedicationInvoiceReconciliationTests(unittest.TestCase): def test_exact_cum_has_priority_over_name(self) -> None: clinical = [ { "nombre": "Cefazolina", "dosis": "1 g", "posologia": "Intravenosa", "codigo_referencia": "CUM-100", "tipo_uso": "administrado", } ] invoice = [ { "medicamento": "Nombre comercial distinto", "codigo_referencia": "CUM-100", "cantidad": 1, } ] result = reconcile_case_medications(clinical, invoice) self.assertEqual(len(result), 1) self.assertEqual(result[0]["asociacion_factura"], "exacta") self.assertIn("facturado", result[0]["estados"]) def test_ambiguous_invoice_does_not_mark_every_scheme_as_billed(self) -> None: clinical = [ { "nombre": "Cefazolina", "dosis": "1 g", "posologia": "Intravenosa ahora", "tipo_uso": "administrado", }, { "nombre": "Cefazolina", "dosis": "1 g", "posologia": "Intravenosa cada 8 horas", "tipo_uso": "administrado", }, ] invoice = [{"medicamento": "Cefazolina", "cantidad": 4}] result = reconcile_case_medications(clinical, invoice) self.assertEqual(len(result), 2) self.assertTrue(all(item["asociacion_factura"] == "ambigua" for item in result)) self.assertTrue(all("facturado" not in item["estados"] for item in result)) def test_first_word_alone_does_not_associate_different_chlorides(self) -> None: clinical = [ { "nombre": "Cloruro de potasio", "dosis": "10 ml", "posologia": "Intravenosa", "tipo_uso": "administrado", } ] invoice = [{"medicamento": "Cloruro de sodio solución salina 0.9%", "cantidad": 2}] result = reconcile_case_medications(clinical, invoice) self.assertEqual(len(result), 2) billed = next(item for item in result if "facturado" in item["estados"]) self.assertEqual(billed["asociacion_factura"], "sin_coincidencia") if __name__ == "__main__": unittest.main()