from __future__ import annotations
import shutil
import subprocess
import unittest
from copy import deepcopy
from datetime import UTC
from pathlib import Path
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import Mock, patch
from bson import ObjectId
from fastapi import FastAPI
from fastapi.templating import Jinja2Templates
from httpx import ASGITransport, AsyncClient
from app.auth import get_current_user
from app.batch_processing.infrastructure.mongo_repositories import MongoBatchCaseRepository
from app.case_epicrisis.application.rules import HistoriaClinicaRequiredRule
from app.case_epicrisis.application.use_cases import (
BuildCaseEpicrisisContextUseCase,
EvaluateEpicrisisRulesUseCase,
GetCaseEpicrisisContextUseCase,
UpdateCaseEpicrisisDraftUseCase,
)
from app.case_epicrisis.application.utils import (
_build_factura_procedimiento,
build_ayudas_diagnosticas,
build_selection_base,
coerce_pdf_draft,
default_pdf_draft,
evaluar_ayuda_diagnostica_exportable,
extract_recomendaciones_medicas,
extraer_imagenes_diagnosticas,
extraer_imagenes_radiologia,
is_compatible_cached_context,
normalize_context_payload,
reconcile_pdf_draft,
refresh_diagnostico_context,
)
from app.case_epicrisis.domain import (
CaseEpicrisisDocuments,
EpicrisisEvidenceSnapshot,
EpicrisisGenerationBlockedError,
EpicrisisRuleEvaluation,
EpicrisisSummaryComposition,
)
from app.case_epicrisis.infrastructure.repositories import MongoCaseDocumentsReader
from app.llm.schemas import (
DocumentoGenericoStructured,
DocumentoQuirurgicoStructured,
HistoriaClinicaStructured,
LaboratorioStructured,
PrescripcionStructured,
RadiologiaStructured,
)
from app.routes import pages
from app.routes.pages.common import EPICRISIS_PDF_CACHE_COLLECTION, _sanitize_legacy_ayudas_diagnosticas_pdf
from app.services.case_epicrisis_runtime_service import CaseEpicrisisRuntimeService
from app.services.factura_html_legacy import extract_factura_procedimientos_from_html
def _base_docs() -> dict:
return {
"historia": {
"_id": "hist-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"analisis_html": "Resumen HC
",
"codigos_cie10": [
{"codigo": "A00", "descripcion": "Diag base"},
],
},
"quirurgico": {
"_id": "qx-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"analisis_html": "Hallazgos QX
",
"codigos_cups": [
{"codigo_cups": "123456", "procedimiento": "Proc base"},
],
},
"factura": {
"_id": "fac-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"factura_json": {
"servicios_procedimientos": {
"procedimientos_quirurgicos": [{"codigo_cups": "555555", "descripcion": "Proc factura"}],
"imagenologia": [{"estudio": "RX de tórax", "descripcion": "Sin lesión aguda"}],
"medicamentos": [],
}
},
},
"radiologia": [
{
"_id": "rx-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"nombre_archivo": "radiografia_torax.pdf",
"tipo_documento": "radiologia",
"analisis_html": (
"Tipo de estudio
RX de tórax
"
"Conclusión
No hay consolidaciones.
"
),
}
],
"laboratorio": [
{
"_id": "lab-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"nombre_archivo": "hemograma.pdf",
"tipo_documento": "laboratorio",
"analisis_html": (
"Examen
Hemograma
"
"Interpretación
Leucocitosis leve.
"
),
}
],
"generico": [
{
"_id": "gen-1",
"nombre_paciente": "Paciente Uno",
"case_number": "1001",
"nombre_archivo": "nota.pdf",
"tipo_documento": "generico",
"descripcion": "Se revisa TAC de cráneo sin reporte definitivo.",
}
],
"prescripcion": [],
}
def _structured_only_docs() -> dict:
docs = _base_docs()
docs["historia"] = {
"_id": "hist-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "historia_clinica",
"analysis_structured": HistoriaClinicaStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"prestador_servicio": "Clinica Central",
"numero_caso": "2002",
"identificacion_paciente": "CC 123",
"sexo": "M",
"edad": "40",
"fecha_ingreso": "2026-05-01",
"fecha_nacimiento": "1986-02-14",
"motivo_consulta": "dolor abdominal",
"resumen_clinico": "Resumen clínico estructurado",
"recomendaciones_medicas": [
{
"categoria": "incapacidad",
"indicacion": "Incapacidad laboral",
"periodo_duracion": "15 días",
"pagina": 4,
"evidencia": "Se expide incapacidad por 15 días",
},
{"categoria": "restriccion", "indicacion": "Evitar cargar peso"},
],
"procedimientos": [{"codigo": "123456", "descripcion": "Lavado quirúrgico"}],
"antecedentes": ["Hipertensión arterial"],
"procedimientos_antecedentes": [{"descripcion": "Apendicectomia previa"}],
"medicamentos": [
{
"codigo": "7701",
"nombre": "Dipirona",
"dosis": "1 g",
"via": "IV",
"frecuencia": "Cada 8 horas",
"cantidad": "1 ampolla",
"tipo_uso": "administrado",
},
{
"nombre": "Acetaminofén",
"dosis": "500 mg",
"frecuencia": "Cada 8 horas",
"duracion": "5 días",
"tipo_uso": "formulado",
},
],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
"codigos_cie10": [{"codigo": "A00", "descripcion": "Diag base"}],
}
docs["quirurgico"] = {
"_id": "qx-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "quirurgico",
"analysis_structured": DocumentoQuirurgicoStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"procedimiento_principal": "Colecistectomia",
"descripcion_procedimiento": "Descripcion estructurada",
"hallazgos": "Hallazgos estructurados",
"procedimientos": [{"codigo": "765432", "descripcion": "Colecistectomia laparoscopica"}],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
"codigos_cups": [{"codigo_cups": "123456", "procedimiento": "Proc base"}],
}
docs["factura"] = {
"_id": "fac-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "factura",
"analysis_structured": {
"dt": "factura",
"pn": "Paciente Estructurado",
"sp": {
"pq": [{"cc": "555555", "d": "Proc factura estructurada"}],
"im": [{"es": "RX de tórax", "d": "Sin lesión aguda"}],
"me": [{"md": "Dipirona", "q": "1 ampolla", "do": "1 g IV"}],
},
},
}
docs["radiologia"] = [
{
"_id": "rx-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "radiologia",
"analysis_structured": RadiologiaStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"tipo_estudio": "RX de tórax",
"conclusion": "No hay consolidaciones.",
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
}
]
docs["laboratorio"] = [
{
"_id": "lab-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "laboratorio",
"analysis_structured": LaboratorioStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"tipo_examen": "Hemograma",
"interpretacion": "Leucocitosis leve.",
"resultados": [],
"valores_alterados": [],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
}
]
docs["generico"] = [
{
"_id": "gen-struct-1",
"nombre_paciente": "Paciente Estructurado",
"case_number": "2002",
"tipo_documento": "generico",
"analysis_structured": DocumentoGenericoStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"tipo_inferido": "nota_clinica",
"resumen": "Seguimiento clínico",
"referencias_diagnosticas": [
{
"tipo": "imagen",
"nombre": "TAC de cráneo",
"concepto": "Pendiente de lectura",
}
],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
}
]
return docs
class FakeDocumentsReader:
def __init__(self, docs: dict):
self.docs = docs
def get_case_documents(self, username: str, case_key: str) -> dict:
return deepcopy(self.docs)
class FakeCacheRepository:
def __init__(self):
self.store: dict[tuple[str, str], dict] = {}
def ensure_indexes(self) -> None:
return None
def get(self, username: str, case_key: str) -> dict | None:
doc = self.store.get((username, case_key))
return deepcopy(doc) if doc else None
def upsert(self, *, username: str, case_key: str, context: dict, regen_requested: bool) -> None:
self.store[(username, case_key)] = {
"usuario": username,
"case_key": case_key,
"regen_requested": regen_requested,
"contexto": deepcopy(context),
}
class FakeCodingGateway:
def __init__(self, *, soat_resultados=None, generated=None, qx_results=None):
self.soat_resultados = soat_resultados or []
self.generated = generated or {}
self.qx_results = qx_results or []
def build_soat_and_glosa(self, *, historia_html: str | None, qx_html: str | None):
return deepcopy(self.soat_resultados), "Glosa
"
def generate_codes_from_descriptions(self, descriptions: list[dict[str, str]]):
output = []
for item in descriptions:
key = str(item.get("descripcion") or "").strip()
output.extend(deepcopy(self.generated.get(key, [])))
return output
def generate_qx_results(self, *, hallazgos: str, descripcion_procedimiento: str):
return deepcopy(self.qx_results)
def refresh_diagnostico_context(self, context: dict):
return refresh_diagnostico_context(deepcopy(context))
class HistoriaClinicaRequiredRuleTest(unittest.TestCase):
def test_passes_when_historia_clinica_is_available(self) -> None:
evaluation = HistoriaClinicaRequiredRule().evaluate(
EpicrisisEvidenceSnapshot(
case_key="case-hc-ok",
document_types_available=["historia_clinica", "factura"],
)
)
self.assertEqual(evaluation.status, "passed")
self.assertTrue(evaluation.generation_allowed)
self.assertEqual(evaluation.missing_documents, [])
def test_blocks_when_historia_clinica_is_missing(self) -> None:
evaluation = HistoriaClinicaRequiredRule().evaluate(
EpicrisisEvidenceSnapshot(
case_key="case-hc-missing",
document_types_available=["factura", "quirurgico"],
)
)
self.assertEqual(evaluation.status, "blocked")
self.assertFalse(evaluation.generation_allowed)
self.assertEqual(evaluation.missing_documents, ["historia_clinica"])
class PdfDraftUtilsTest(unittest.TestCase):
def test_case_epicrisis_documents_derives_html_from_structured_payload(self) -> None:
documents = CaseEpicrisisDocuments.from_payload(
{
"historia": {
"document_type": "historia_clinica",
"analysis_structured": HistoriaClinicaStructured.model_validate(
{
"patient_name": "PACIENTE ESTRUCTURADO",
"resumen_clinico": "Resumen estructurado",
"diagnosticos": [],
"procedimientos": [],
"medicamentos": [],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
},
"quirurgico": {
"document_type": "quirurgico",
"analysis_structured": DocumentoQuirurgicoStructured.model_validate(
{
"patient_name": "PACIENTE ESTRUCTURADO",
"procedimiento_principal": "Colecistectomia",
"diagnosticos": [],
"procedimientos": [],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
},
}
)
self.assertIn("PACIENTE ESTRUCTURADO", documents.historia_html or "")
self.assertIn("Colecistectomia", documents.quirurgico_html or "")
def test_build_case_context_uses_clinical_summary_instead_of_eps_label(self) -> None:
docs = _structured_only_docs()
docs["historia"]["descripcion"] = (
"EPS: SEGUROS COMERCIALES BOLIVAR S.A.\n"
"Resumen\n"
"Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.\n"
)
docs["historia"]["analysis_structured"]["rs"] = "SEGUROS COMERCIALES BOLIVAR S.A."
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
)
context = use_case.execute("auditor", "case-2002")
self.assertEqual(
context["metadatos_hc"]["resumen"],
"Paciente con trauma toracico, dolor intenso y manejo hospitalario inicial.",
)
def test_build_case_context_keeps_integrated_summary_as_canonical_field(self) -> None:
class Composer:
def compose(self, *, username, sources):
return EpicrisisSummaryComposition(
narrative="Historia y cirugía integradas cronológicamente.",
covered_source_ids=[source.identifier for source in sources],
)
docs = _structured_only_docs()
docs["quirurgico"]["analysis_structured"]["rs"] = "Resumen quirúrgico sin pérdida."
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
summary_composer=Composer(),
)
context = use_case.execute("auditor", "case-2002")
self.assertEqual(context["resumen_clinico_integrado"]["estado"], "integrado")
self.assertEqual(
context["metadatos_hc"]["resumen"],
"Historia y cirugía integradas cronológicamente.",
)
def test_build_case_context_consolidates_prescription_medications(self) -> None:
docs = _structured_only_docs()
docs["prescripcion"] = [
{
"_id": "rx-med-1",
"tipo_documento": "prescripcion",
"analysis_structured": PrescripcionStructured.model_validate(
{
"patient_name": "Paciente Estructurado",
"fecha": "2026-05-03",
"medicamentos": [
{
"nombre": "Cefalexina",
"dosis": "500 mg",
"via": "Oral",
"frecuencia": "Cada 6 horas",
"duracion": "Por 5 días",
"cantidad": "20",
}
],
}
).model_dump(by_alias=True, exclude_none=True, exclude_defaults=True),
}
]
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
)
context = use_case.execute("auditor", "case-2002")
prescribed = next(item for item in context["medicamentos_caso"] if item["nombre"] == "Cefalexina")
self.assertIn("formulado", prescribed["estados"])
self.assertIn("prescripcion", prescribed["fuentes"])
def test_medication_clinical_review_survives_context_normalization(self) -> None:
class _Reviewer:
def review(self, *, medications, diagnoses, procedures):
del diagnoses, procedures
evidence_id = medications[0]["evidencias"][0]["evidence_id"]
return [
{
"key": medications[0]["key"],
"estado": "posible_glosa",
"justificacion": "La indicación requiere validación clínica.",
"evidence_ids": [evidence_id],
}
]
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(_structured_only_docs()),
coding_gateway=FakeCodingGateway(),
medication_pertinence_reviewer=_Reviewer(),
)
context = normalize_context_payload(use_case.execute("auditor", "case-2002"))
self.assertEqual(context["medicamentos_caso"][0]["pertinencia"]["origen"], "reglas+llm")
self.assertTrue(is_compatible_cached_context(context))
def test_default_and_coerce_pdf_draft_supports_diagnostic_images(self) -> None:
draft = default_pdf_draft()
self.assertEqual(draft["excluded_imagenes_diagnosticas"], [])
self.assertEqual(draft["manual_imagenes_diagnosticas"], [])
coerced = coerce_pdf_draft(
{
"excluded_imagenes_diagnosticas": [" RX tórax ", "RX tórax", ""],
"manual_imagenes_diagnosticas": [
" TAC cráneo - No interpretado ",
"TAC cráneo - No interpretado",
],
}
)
self.assertEqual(coerced["excluded_imagenes_diagnosticas"], ["RX tórax"])
self.assertEqual(coerced["manual_imagenes_diagnosticas"], ["TAC cráneo - No interpretado"])
def test_reconcile_pdf_draft_filters_obsolete_diagnostic_image_exclusions(self) -> None:
docs = _base_docs()
context = {
"factura": docs["factura"],
"radiologia": docs["radiologia"],
"laboratorio": docs["laboratorio"],
"generico": docs["generico"],
"imagenes_diagnosticas": extraer_imagenes_diagnosticas(docs["factura"], docs["radiologia"]),
}
reconciled = reconcile_pdf_draft(
{
"excluded_imagenes_diagnosticas": [
"RX de tórax - Sin lesión aguda",
"Imagen obsoleta",
],
"manual_imagenes_diagnosticas": ["Ecografía abdomen - No interpretado"],
},
context,
)
self.assertEqual(reconciled["excluded_imagenes_diagnosticas"], ["RX de tórax - Sin lesión aguda"])
self.assertEqual(reconciled["manual_imagenes_diagnosticas"], ["Ecografía abdomen - No interpretado"])
def test_default_and_coerce_pdf_draft_supports_structured_diagnostic_aids(self) -> None:
draft = default_pdf_draft()
self.assertEqual(draft["excluded_ayudas_diagnosticas_keys"], [])
self.assertEqual(draft["manual_ayudas_diagnosticas"], [])
coerced = coerce_pdf_draft(
{
"excluded_ayudas_diagnosticas_keys": [" ayuda:imagen:rx:factura ", "ayuda:imagen:rx:factura"],
"manual_ayudas_diagnosticas": [
{"tipo": "laboratorio", "nombre": "Hemograma", "concepto": "Pendiente"},
{"tipo": "laboratorio", "nombre": "Hemograma", "concepto": "Pendiente"},
],
}
)
self.assertEqual(coerced["excluded_ayudas_diagnosticas_keys"], ["ayuda:imagen:rx:factura"])
self.assertEqual(len(coerced["manual_ayudas_diagnosticas"]), 1)
self.assertEqual(coerced["manual_ayudas_diagnosticas"][0]["tipo"], "laboratorio")
def test_build_ayudas_diagnosticas_combines_sources_and_alerts(self) -> None:
docs = _base_docs()
ayudas = build_ayudas_diagnosticas(
{
**docs,
"hallazgos_quirurgicos": "Se tuvo en cuenta radiografía de tobillo.",
"descripcion_procedimiento": "",
}
)
nombres = [item["nombre"] for item in ayudas]
self.assertIn("RX de tórax - Sin lesión aguda", nombres)
self.assertIn("radiografia_torax.pdf", nombres)
self.assertIn("Hemograma", nombres)
self.assertTrue(any("TAC de cráneo" in item["nombre"] for item in ayudas))
self.assertTrue(any(item["fuente"] == "agente_hallazgos_qx" for item in ayudas))
factura = next(item for item in ayudas if item["fuente"] == "factura" and item["tipo"] == "imagen")
self.assertIn("factura_sin_soporte_clinico_asociado", factura["alertas"])
def test_build_ayudas_diagnosticas_prefers_structured_radiologia_laboratorio_and_generico(self) -> None:
docs = _structured_only_docs()
docs["radiologia"][0]["analisis_html"] = (
"Tipo de estudio
NO USAR
Conclusión
NO USAR
"
)
docs["laboratorio"][0]["analisis_html"] = (
"Tipo de examen
NO USAR
Interpretación
NO USAR
"
)
docs["generico"][0]["analisis_html"] = (
"Referencias diagnósticas
"
)
ayudas = build_ayudas_diagnosticas(
{
**docs,
"hallazgos_quirurgicos": "",
"descripcion_procedimiento": "",
}
)
radiologia = next(item for item in ayudas if item["fuente"] == "radiologia")
laboratorio = next(item for item in ayudas if item["fuente"] == "laboratorio")
generico = next(item for item in ayudas if item["fuente"] == "generico")
self.assertEqual(radiologia["nombre"], "RX de tórax")
self.assertEqual(radiologia["concepto"], "No hay consolidaciones.")
self.assertEqual(laboratorio["nombre"], "Hemograma")
self.assertEqual(laboratorio["concepto"], "Leucocitosis leve.")
self.assertEqual(generico["nombre"], "TAC de cráneo")
self.assertEqual(generico["concepto"], "Pendiente de lectura")
def test_radiologia_and_laboratorio_legacy_html_use_standard_field_parsing(self) -> None:
docs = _base_docs()
docs["radiologia"][0].pop("nombre_archivo", None)
docs["laboratorio"][0].pop("nombre_archivo", None)
ayudas = build_ayudas_diagnosticas(
{
**docs,
"hallazgos_quirurgicos": "",
"descripcion_procedimiento": "",
}
)
radiologia = next(item for item in ayudas if item["fuente"] == "radiologia")
laboratorio = next(item for item in ayudas if item["fuente"] == "laboratorio")
self.assertEqual(radiologia["nombre"], "RX de tórax")
self.assertEqual(radiologia["concepto"], "No hay consolidaciones.")
self.assertEqual(laboratorio["nombre"], "Hemograma")
self.assertEqual(laboratorio["concepto"], "Leucocitosis leve.")
self.assertEqual(
extraer_imagenes_radiologia(docs["radiologia"]), ["RX de tórax - No hay consolidaciones."]
)
def test_generico_without_structure_keeps_textual_heuristic_without_html_field_parsing(self) -> None:
ayudas = build_ayudas_diagnosticas(
{
"factura": None,
"radiologia": [],
"laboratorio": [],
"historia": None,
"quirurgico": None,
"generico": [
{
"_id": "gen-html-1",
"tipo_documento": "generico",
"nombre_archivo": "nota_html.pdf",
"analisis_html": (
"Resumen
"
"Paciente con solicitud de resonancia cerebral y laboratorio de control.
"
),
}
],
"hallazgos_quirurgicos": "",
"descripcion_procedimiento": "",
}
)
self.assertTrue(any(item["fuente"] == "generico" for item in ayudas))
generico = next(item for item in ayudas if item["fuente"] == "generico")
self.assertIn("resonancia cerebral", generico["nombre"].lower())
self.assertEqual(generico["concepto"], "Referencia textual sin interpretación estructurada.")
self.assertIn("referencia_textual_pendiente_revision", generico["alertas"])
def test_export_rule_hides_glosada_or_excluded_diagnostic_aid(self) -> None:
ayuda = {"key": "ayuda:imagen:rx:factura", "glosado": False}
self.assertTrue(evaluar_ayuda_diagnostica_exportable(ayuda))
self.assertFalse(evaluar_ayuda_diagnostica_exportable(ayuda, ["ayuda:imagen:rx:factura"]))
self.assertFalse(evaluar_ayuda_diagnostica_exportable({**ayuda, "glosado": True}))
def test_normalize_context_payload_rehydrates_history_selection_bases(self) -> None:
docs = _base_docs()
docs["historia"]["analisis_html"] = (
"Procedimientos
- Lavado quirúrgico
"
"Medicamentos administrados
"
"- 7701 - 1 ampolla - Dipirona - 1 g IV
"
)
normalized = normalize_context_payload(
{
"historia": docs["historia"],
"quirurgico": docs["quirurgico"],
"factura": docs["factura"],
"radiologia": docs["radiologia"],
"laboratorio": docs["laboratorio"],
"generico": docs["generico"],
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"pdf_draft": {
"manual_diagnosticos": ["DX manual"],
"hallazgos_extra": "nota previa",
},
"procedimientos_hc": [],
"medicamentos_hc": [],
"medicamentos_hc_display": [],
"procedimientos_factura": [],
"diagnosticos_consolidados": [],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [],
"imagenes_diagnosticas": [],
}
)
self.assertEqual(normalized["procedimientos_hc"], ["Lavado quirúrgico"])
self.assertTrue(normalized["medicamentos_hc"])
self.assertTrue(normalized["medicamentos_hc_display"])
self.assertIn("Dipirona", normalized["medicamentos_hc_display"][0])
self.assertEqual(normalized["pdf_draft"]["manual_diagnosticos"], ["DX manual"])
self.assertEqual(normalized["pdf_draft"]["hallazgos_extra"], "nota previa")
def test_normalize_context_payload_rehydrates_factura_procedures_and_medications(self) -> None:
docs = _base_docs()
docs["factura"]["factura_json"]["servicios_procedimientos"]["medicamentos"] = [
{
"codigo_medicamento": "7701",
"medicamento": "Dipirona",
"cantidad": "1 ampolla",
"dosis": "1 g IV",
}
]
normalized = normalize_context_payload(
{
"historia": docs["historia"],
"quirurgico": docs["quirurgico"],
"factura": docs["factura"],
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
"procedimientos_hc": [],
"medicamentos_hc": [],
"medicamentos_hc_display": [],
"procedimientos_factura": [],
"diagnosticos_consolidados": [],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [],
"imagenes_diagnosticas": [],
}
)
self.assertEqual(normalized["procedimientos_factura"][0]["descripcion"], "Proc factura")
medicamentos_factura = normalized["factura"]["factura_json"]["servicios_procedimientos"][
"medicamentos"
]
self.assertEqual(medicamentos_factura[0]["medicamento"], "Dipirona")
base = build_selection_base(normalized)
self.assertTrue(any("Proc factura" in item for item in base["procedimientos"]))
self.assertTrue(any("Dipirona" in item for item in base["medicamentos"]))
def test_build_factura_procedimiento_rejects_placeholder_headers(self) -> None:
self.assertIsNone(_build_factura_procedimiento("Código CUPS", "Descripción"))
self.assertIsNone(_build_factura_procedimiento("123456", "Descripcion"))
self.assertIsNone(_build_factura_procedimiento("", "Procedimiento"))
def test_normalize_context_payload_discards_placeholder_factura_procedures(self) -> None:
docs = _base_docs()
normalized = normalize_context_payload(
{
"historia": docs["historia"],
"quirurgico": docs["quirurgico"],
"factura": docs["factura"],
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
"procedimientos_hc": [],
"medicamentos_hc": [],
"medicamentos_hc_display": [],
"procedimientos_factura": [
{"codigo_soat": "Código CUPS", "descripcion": "Descripción"},
{"codigo_soat": "555555", "descripcion": "Proc factura"},
],
"diagnosticos_consolidados": [],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [],
"imagenes_diagnosticas": [],
}
)
self.assertEqual(
normalized["procedimientos_factura"], [{"codigo_soat": "555555", "descripcion": "Proc factura"}]
)
def test_extract_factura_procedimientos_from_html_skips_header_like_rows(self) -> None:
html = """
Procedimientos quirúrgicos
| Concepto |
Código CUPS |
Descripción |
Cantidad |
Valor unitario |
Total |
| Concepto | Código CUPS | Descripción | Cantidad | Valor unitario | Total |
| Cirugía | 555555 | Proc factura | 1 | 1000 | 1000 |
"""
self.assertEqual(
extract_factura_procedimientos_from_html(html),
[{"codigo_soat": "555555", "descripcion": "Proc factura"}],
)
def test_normalize_context_payload_prefers_analysis_structured_without_html(self) -> None:
docs = _structured_only_docs()
normalized = normalize_context_payload(
{
"historia": docs["historia"],
"quirurgico": docs["quirurgico"],
"factura": docs["factura"],
"radiologia": docs["radiologia"],
"laboratorio": docs["laboratorio"],
"generico": docs["generico"],
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
"procedimientos_hc": [],
"medicamentos_hc": [],
"medicamentos_hc_display": [],
"procedimientos_factura": [],
"diagnosticos_consolidados": [],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [],
"imagenes_diagnosticas": [],
}
)
self.assertEqual(normalized["procedimientos_hc"], ["123456 Lavado quirúrgico"])
self.assertEqual(normalized["antecedentes_hc"], ["Hipertensión arterial", "Apendicectomia previa"])
self.assertNotIn("Apendicectomia previa", normalized["procedimientos_hc"])
self.assertEqual(normalized["medicamentos_hc"][0]["medicamento"], "Dipirona")
self.assertTrue(
any(
item["descripcion"] == "Proc factura estructurada"
for item in normalized["procedimientos_factura"]
)
)
self.assertTrue(any(item["fuente"] == "radiologia" for item in normalized["ayudas_diagnosticas"]))
self.assertTrue(any(item["fuente"] == "laboratorio" for item in normalized["ayudas_diagnosticas"]))
self.assertTrue(any(item["fuente"] == "generico" for item in normalized["ayudas_diagnosticas"]))
def test_normalize_context_payload_builds_primary_pdf_sections_without_soat_noise(self) -> None:
docs = _base_docs()
docs["quirurgico"]["codigos_cups"] = [{"codigo_cups": "123456", "procedimiento": "Proc QX"}]
docs["factura"]["factura_json"]["servicios_procedimientos"]["medicamentos"] = [
{"codigo_medicamento": "2801", "medicamento": "No especificado", "cantidad": "", "dosis": ""}
]
normalized = normalize_context_payload(
{
"historia": docs["historia"],
"quirurgico": docs["quirurgico"],
"factura": docs["factura"],
"radiologia": docs["radiologia"],
"laboratorio": docs["laboratorio"],
"generico": docs["generico"],
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [
{
"codigo_soat": "QX-HALLAZGOS",
"descripcion": "Hallazgos qx",
"cie10_principal": {"codigo": "C00", "descripcion": "Dx qx"},
"cups_principal": {"codigo": "777777", "descripcion": "Proc qx soat"},
}
],
"codigos_desde_soat": [
{
"codigo_soat": "S0",
"descripcion": "Proc soat",
"cie10_principal": {"codigo": "B00", "descripcion": "Dx soat"},
"cups_principal": {"codigo": "654321", "descripcion": "Proc soat cups"},
}
],
"pdf_draft": {},
"procedimientos_hc": ["Proc HC"],
"medicamentos_hc": [
{
"codigo": "7701",
"medicamento": "Dipirona",
"cantidad": "1 ampolla",
"dosis": "1 g IV",
"tipo_uso": "administrado",
}
],
"medicamentos_hc_display": [],
"procedimientos_factura": [{"codigo_soat": "555555", "descripcion": "Proc factura"}],
"diagnosticos_consolidados": [
{
"key": "cie10:A00",
"codigo": "A00",
"descripcion": "Diag base",
"estado_codificacion": "codificado",
"fuentes": ["historia_clinica"],
"texto_presentacion": "A00 - Diag base",
},
{
"key": "cie10:B00",
"codigo": "B00",
"descripcion": "Dx soat",
"estado_codificacion": "codificado",
"fuentes": ["auto_soat"],
"texto_presentacion": "B00 - Dx soat",
},
],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [
{
"key": "ayuda:imagen:rx-factura:factura",
"tipo": "imagen",
"nombre": "RX de tórax",
"concepto": "Sin lesión aguda",
"fuente": "factura",
"facturado": True,
"exportable": True,
},
{
"key": "ayuda:imagen:resonancia:generico",
"tipo": "imagen",
"nombre": "Resonancia cerebral",
"concepto": "Referencia textual sin interpretación estructurada.",
"fuente": "generico",
"alertas": ["referencia_textual_pendiente_revision"],
"exportable": True,
},
],
"imagenes_diagnosticas": [],
}
)
self.assertEqual(
[item["texto_presentacion"] for item in normalized["pdf_primary_diagnosticos"]],
["A00 - Diag base"],
)
self.assertTrue(any(item == "Proc HC" for item in normalized["pdf_primary_procedimientos"]))
self.assertTrue(any(item == "123456 - Proc QX" for item in normalized["pdf_primary_procedimientos"]))
self.assertFalse(any("Proc factura" in item for item in normalized["pdf_primary_procedimientos"]))
self.assertFalse(any("Proc soat" in item for item in normalized["pdf_primary_procedimientos"]))
self.assertEqual(len(normalized["pdf_primary_ayudas_diagnosticas"]), 1)
self.assertEqual(normalized["pdf_primary_ayudas_diagnosticas"][0]["fuente"], "factura")
self.assertTrue(
any(item["medicamento"] == "Dipirona" for item in normalized["pdf_primary_medicamentos"])
)
self.assertFalse(
any(item["medicamento"] == "No especificado" for item in normalized["pdf_primary_medicamentos"])
)
def test_incompatible_cache_contract_requires_selection_bases(self) -> None:
legacy_like_context = {
"historia": {"_id": "hist-1"},
"quirurgico": {"_id": "qx-1"},
"factura": {"_id": "fac-1"},
"manual_soat_results": [],
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
"procedimientos_factura": [],
"diagnosticos_consolidados": [],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
"ayudas_diagnosticas": [],
}
self.assertFalse(is_compatible_cached_context(legacy_like_context))
class BuildCaseEpicrisisContextUseCaseTest(unittest.TestCase):
def test_builds_context_with_consolidated_soat_codes_for_case_flow(self) -> None:
docs = _base_docs()
gateway = FakeCodingGateway(
soat_resultados=[{"codigo_soat": "S1", "descripcion": "Proc soat"}],
generated={
"Proc soat": [
{
"codigo_soat": "S1",
"descripcion": "Proc soat",
"cie10_principal": {"codigo": "B00", "descripcion": "Diag soat"},
"cups_principal": {"codigo": "654321", "descripcion": "Proc soat cups"},
}
],
"Proc factura": [
{
"codigo_soat": "F1",
"descripcion": "Proc factura",
"cie10_principal": {"codigo": "C00", "descripcion": "Diag factura"},
"cups_principal": {"codigo": "111111", "descripcion": "Proc factura cups"},
}
],
},
qx_results=[
{
"codigo_soat": "QX-HALLAZGOS",
"descripcion": "Hallazgos quirurgicos",
"cie10_principal": {"codigo": "D00", "descripcion": "Diag qx"},
"cups_principal": {"codigo": "222222", "descripcion": "Proc qx"},
}
],
)
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
context = use_case.execute("tester", "case-1")
self.assertEqual(context["case_key"], "case-1")
self.assertEqual(len(context["auto_soat_results"]), 2)
self.assertEqual(len(context["qx_agent_results"]), 1)
self.assertEqual(len(context["codigos_desde_soat"]), 3)
self.assertEqual(context["procedimientos_factura"][0]["descripcion"], "Proc factura")
self.assertIn("RX de tórax - Sin lesión aguda", context["imagenes_diagnosticas"])
self.assertIn("radiografia_torax.pdf - No hay consolidaciones.", context["imagenes_diagnosticas"])
self.assertTrue(any(item["tipo"] == "laboratorio" for item in context["ayudas_diagnosticas"]))
self.assertTrue(any(item["fuente"] == "generico" for item in context["ayudas_diagnosticas"]))
self.assertTrue(context["diagnosticos_consolidados"])
def test_builds_context_from_analysis_structured_without_persisted_html(self) -> None:
docs = _structured_only_docs()
gateway = FakeCodingGateway(
soat_resultados=[{"codigo_soat": "S1", "descripcion": "Proc soat"}],
generated={
"Proc factura estructurada": [
{
"codigo_soat": "F1",
"descripcion": "Proc factura estructurada",
"cie10_principal": {"codigo": "C00", "descripcion": "Diag factura"},
}
]
},
)
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
context = use_case.execute("tester", "case-struct-1")
self.assertEqual(context["metadatos_hc"]["nombre_paciente"], "Paciente Estructurado")
self.assertEqual(context["procedimientos_hc"], ["123456 Lavado quirúrgico"])
self.assertEqual(context["medicamentos_hc"][0]["medicamento"], "Dipirona")
self.assertEqual(
[item["categoria"] for item in context["recomendaciones_medicas"]],
["incapacidad", "restriccion", "tratamiento_cuidado"],
)
self.assertEqual(context["recomendaciones_medicas"][0]["pagina"], 4)
self.assertEqual(
[item["nombre"] for item in context["pdf_primary_medicamentos"]],
["Dipirona"],
)
self.assertEqual(context["hallazgos_quirurgicos"], "Hallazgos estructurados")
self.assertEqual(context["descripcion_procedimiento"], "Descripcion estructurada")
self.assertTrue(
any(
item["descripcion"] == "Proc factura estructurada"
for item in context["procedimientos_factura"]
)
)
self.assertTrue(any(item["fuente"] == "radiologia" for item in context["ayudas_diagnosticas"]))
self.assertIn("RX de tórax - No hay consolidaciones.", context["imagenes_diagnosticas"])
def test_builds_context_from_support_persisted_with_effective_document_type(self) -> None:
docs = _structured_only_docs()
docs["radiologia"][0]["selected_document_type"] = "factura"
docs["radiologia"][0]["detected_document_type"] = "radiologia"
docs["radiologia"][0]["effective_document_type"] = "radiologia"
gateway = FakeCodingGateway()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
context = use_case.execute("tester", "case-effective-rx")
self.assertTrue(any(item["fuente"] == "radiologia" for item in context["ayudas_diagnosticas"]))
self.assertTrue(any(item["tipo"] == "imagen" for item in context["ayudas_diagnosticas"]))
self.assertIn("RX de tórax - No hay consolidaciones.", context["imagenes_diagnosticas"])
def test_evaluate_rules_use_case_without_rules_returns_passed(self) -> None:
docs = _structured_only_docs()
docs["quirurgico"] = None
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
)
evaluation = EvaluateEpicrisisRulesUseCase(rules=[]).execute(
use_case._build_evidence_snapshot(CaseEpicrisisDocuments.from_payload(docs), "case-no-rules")
)
self.assertEqual(evaluation.status, "passed")
self.assertTrue(evaluation.generation_allowed)
self.assertFalse(evaluation.missing_documents)
def test_build_context_allows_generation_without_qx_document_when_historia_exists(self) -> None:
docs = _structured_only_docs()
docs["quirurgico"] = None
class GuardedCodingGateway(FakeCodingGateway):
def __init__(self) -> None:
super().__init__()
self.build_called = False
def build_soat_and_glosa(self, *, historia_html: str | None, qx_html: str | None):
self.build_called = True
return super().build_soat_and_glosa(historia_html=historia_html, qx_html=qx_html)
gateway = GuardedCodingGateway()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
context = use_case.execute("tester", "case-qx-allowed")
self.assertTrue(gateway.build_called)
self.assertTrue(context["epicrisis_generation_allowed"])
self.assertEqual(context["rule_evaluation"]["status"], "passed")
self.assertEqual(context["missing_documents"], [])
def test_get_case_context_persists_passed_rule_context_without_qx_document(self) -> None:
docs = _structured_only_docs()
docs["quirurgico"] = None
cache_repo = FakeCacheRepository()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
)
get_use_case = GetCaseEpicrisisContextUseCase(
cache_repository=cache_repo,
build_case_context_use_case=use_case,
)
context = get_use_case.execute("tester", "case-qx-cache", regen=False)
cached = cache_repo.get("tester", "case-qx-cache")
self.assertIsNotNone(cached)
if cached is None:
self.fail("Expected cached epicrisis context for case-qx-cache")
self.assertTrue(context["epicrisis_generation_allowed"])
self.assertEqual(cached["contexto"]["rule_evaluation"]["status"], "passed")
self.assertEqual(cached["contexto"]["missing_documents"], [])
self.assertEqual(cached["contexto"]["recomendaciones_medicas"][0]["pagina"], 4)
self.assertEqual(
cached["contexto"]["recomendaciones_medicas"][0]["evidencia"],
"Se expide incapacidad por 15 días",
)
def test_blocks_generation_when_historia_is_missing_even_with_other_documents(self) -> None:
docs = _structured_only_docs()
docs["historia"] = None
class GuardedCodingGateway(FakeCodingGateway):
def __init__(self) -> None:
super().__init__()
self.build_called = False
def build_soat_and_glosa(self, *, historia_html: str | None, qx_html: str | None):
self.build_called = True
raise AssertionError(
"No debió construir snapshot de codificación cuando falta la historia clínica."
)
gateway = GuardedCodingGateway()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
rules_evaluator=EvaluateEpicrisisRulesUseCase(rules=[HistoriaClinicaRequiredRule()]),
)
with self.assertRaises(EpicrisisGenerationBlockedError) as exc:
use_case.execute("tester", "case-hc-blocked")
self.assertFalse(gateway.build_called)
self.assertEqual(exc.exception.evaluation.status, "blocked")
self.assertEqual(exc.exception.context["missing_documents"], ["historia_clinica"])
self.assertEqual(
exc.exception.context["blocking_reason"], "No se puede generar la epicrisis sin historia clínica."
)
def test_get_case_context_persists_blocked_rule_context_when_historia_is_missing(self) -> None:
docs = _structured_only_docs()
docs["historia"] = None
cache_repo = FakeCacheRepository()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
rules_evaluator=EvaluateEpicrisisRulesUseCase(rules=[HistoriaClinicaRequiredRule()]),
)
get_use_case = GetCaseEpicrisisContextUseCase(
cache_repository=cache_repo,
build_case_context_use_case=use_case,
)
with self.assertRaises(EpicrisisGenerationBlockedError):
get_use_case.execute("tester", "case-hc-cache", regen=False)
cached = cache_repo.get("tester", "case-hc-cache")
self.assertIsNotNone(cached)
if cached is None:
self.fail("Expected cached epicrisis context for case-hc-cache")
self.assertEqual(cached["contexto"]["rule_evaluation"]["status"], "blocked")
self.assertEqual(cached["contexto"]["missing_documents"], ["historia_clinica"])
self.assertEqual(
cached["contexto"]["blocking_reason"], "No se puede generar la epicrisis sin historia clínica."
)
def test_consolidates_equivalent_and_pending_diagnoses(self) -> None:
docs = _base_docs()
docs["historia"]["codigos_cie10"] = [
{
"diagnostico": "Trauma pie izquierdo",
"codigo": "S92.9",
"descripcion": "Fractura de pie izquierdo",
},
{
"diagnostico": "Trauma pie izquierdo",
"codigo": "Pendiente",
"descripcion": "No se pudo codificar automaticamente este diagnostico.",
},
]
docs["quirurgico"]["codigos_cie10"] = [
{
"diagnostico": "POP fractura de pie izquierdo",
"codigo": "S92.9",
"descripcion": "Fractura de pie izquierdo",
},
{"diagnostico": "Herida en dedo pie izquierdo", "codigo": "", "descripcion": ""},
]
gateway = FakeCodingGateway()
use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
context = use_case.execute("tester", "case-1")
textos = [item["texto_presentacion"] for item in context["diagnosticos_consolidados"]]
self.assertEqual(sum(1 for item in textos if "S92.9" in item), 1)
self.assertTrue(
any(
item["estado_codificacion"] == "pendiente_validacion"
for item in context["diagnosticos_consolidados"]
)
)
self.assertTrue(context["diagnosticos_consolidacion_hallazgos"])
def test_preserves_meaningful_pending_diagnoses_as_pdf_selection_candidates(self) -> None:
historia_diagnosticos = [
{
"diagnostico": "SUFRIMIENTO DE COLGAJO LOCAL PIERNA DERECHA",
"codigo": "Pendiente",
"descripcion": "No se pudo codificar automáticamente este diagnostico.",
},
{
"diagnostico": "ULCERA POR DEHISCENCIA DE HERIDA",
"codigo": "Sin codigo",
"descripcion": "No se encontró codigo CIE-10 correspondiente",
},
{
"diagnostico": "INFECCIÓN DEL SITIO OPERATORIO (RAFI DE TIBIA Y PERONÉ)",
"codigo": "Sin codigo",
"descripcion": "No se encontró codigo CIE-10 correspondiente",
},
{
"diagnostico": "pseudoartrosis de tibia derecha",
"codigo": "Sin codigo",
"descripcion": "No se encontró codigo CIE-10 correspondiente",
},
]
context = refresh_diagnostico_context(
{
"historia": {"codigos_cie10": historia_diagnosticos},
"quirurgico": None,
"factura": None,
"auto_soat_results": [],
"qx_agent_results": [],
"manual_soat_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
}
)
catalog_texts = {
item["texto_presentacion"] for item in context["diagnosticos_consolidados"]
}
selection_texts = {
item["texto_presentacion"] for item in context["pdf_primary_diagnosticos"]
}
self.assertEqual(len(catalog_texts), 4)
self.assertEqual(selection_texts, catalog_texts)
self.assertIn("PENDIENTE - ULCERA POR DEHISCENCIA DE HERIDA", selection_texts)
self.assertIn("PENDIENTE - pseudoartrosis de tibia derecha", selection_texts)
self.assertFalse(any("No se encontró codigo CIE-10" in item for item in catalog_texts))
def test_pdf_diagnostic_candidates_require_clinical_source_and_valid_content(self) -> None:
context = refresh_diagnostico_context(
{
"historia": {
"codigos_cie10": [
{
"diagnostico": "Diagnóstico clínico pendiente",
"codigo": "Pendiente",
"descripcion": "Codificación automática no disponible.",
}
]
},
"quirurgico": None,
"factura": None,
"auto_soat_results": [
{
"codigo_soat": "S1",
"descripcion": "Resultado administrativo",
"cie10_principal": {
"codigo": "B00",
"descripcion": "Diagnóstico inferido desde SOAT",
},
}
],
"qx_agent_results": [],
"manual_soat_results": [],
"codigos_desde_soat": [],
"pdf_draft": {},
}
)
selection_texts = [
item["texto_presentacion"] for item in context["pdf_primary_diagnosticos"]
]
self.assertEqual(selection_texts, ["PENDIENTE - Diagnóstico clínico pendiente"])
self.assertTrue(
any(item["texto_presentacion"] == "B00 - Diagnóstico inferido desde SOAT" for item in context["diagnosticos_consolidados"])
)
def test_regen_preserves_manual_and_reconciles_pdf_draft(self) -> None:
docs = _base_docs()
cache_repo = FakeCacheRepository()
cache_repo.store[("tester", "case-1")] = {
"usuario": "tester",
"case_key": "case-1",
"contexto": {
"manual_soat_results": [
{
"codigo_soat": "M1",
"descripcion": "Manual soat",
"cie10_principal": {"codigo": "E00", "descripcion": "Manual dx"},
}
],
"pdf_draft": {
"excluded_diagnosticos": ["A00 - Diag base", "ZZZ - Obsoleto"],
"excluded_diagnostico_keys": ["cie10:A00", "cie10:ZZZ"],
"manual_diagnosticos": ["DX agregado"],
"manual_imagenes_diagnosticas": ["Imagen manual"],
"hallazgos_extra": "nota previa",
},
"auto_soat_results": [],
"qx_agent_results": [],
"codigos_desde_soat": [],
"procedimientos_factura": [],
"diagnosticos_consolidados": [
{
"key": "cie10:A00",
"codigo": "A00",
"descripcion": "Diag base",
"diagnostico": "Diag base",
"estado_codificacion": "codificado",
"fuentes": ["historia_clinica"],
"tipo_contexto": "no_clasificado",
"texto_presentacion": "A00 - Diag base",
}
],
"diagnosticos_pendientes_validacion": [],
"diagnosticos_consolidacion_hallazgos": [],
},
}
gateway = FakeCodingGateway(
soat_resultados=[{"codigo_soat": "S2", "descripcion": "Proc soat nuevo"}],
generated={
"Proc soat nuevo": [
{
"codigo_soat": "S2",
"descripcion": "Proc soat nuevo",
"cie10_principal": {"codigo": "B10", "descripcion": "Nuevo dx"},
}
],
"Proc factura": [],
},
)
build_use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=gateway,
)
get_use_case = GetCaseEpicrisisContextUseCase(
cache_repository=cache_repo,
build_case_context_use_case=build_use_case,
)
context = get_use_case.execute("tester", "case-1", regen=True)
self.assertEqual(context["manual_soat_results"][0]["descripcion"], "Manual soat")
self.assertEqual(context["auto_soat_results"][0]["descripcion"], "Proc soat nuevo")
self.assertEqual(context["pdf_draft"]["excluded_diagnosticos"], ["A00 - Diag base"])
self.assertEqual(context["pdf_draft"]["excluded_diagnostico_keys"], ["cie10:A00"])
self.assertEqual(context["pdf_draft"]["manual_diagnosticos"], ["DX agregado"])
self.assertEqual(context["pdf_draft"]["manual_imagenes_diagnosticas"], ["Imagen manual"])
self.assertEqual(context["pdf_draft"]["hallazgos_extra"], "nota previa")
def test_update_draft_persists_manual_results_and_pdf_delta(self) -> None:
docs = _base_docs()
cache_repo = FakeCacheRepository()
build_use_case = BuildCaseEpicrisisContextUseCase(
documents_reader=FakeDocumentsReader(docs),
coding_gateway=FakeCodingGateway(),
)
get_use_case = GetCaseEpicrisisContextUseCase(
cache_repository=cache_repo,
build_case_context_use_case=build_use_case,
)
update_use_case = UpdateCaseEpicrisisDraftUseCase(
cache_repository=cache_repo,
get_case_context_use_case=get_use_case,
)
context = update_use_case.execute(
"tester",
"case-1",
payload={
"manual_soat_results": [
{
"codigo_soat": "M2",
"descripcion": "Manual nuevo",
"cie10_principal": {"codigo": "F00", "descripcion": "Dx manual"},
}
],
"pdf_draft": {
"manual_procedimientos": ["PROC agregado"],
"manual_ayudas_diagnosticas": [
{"tipo": "laboratorio", "nombre": "Laboratorio manual", "concepto": "Pendiente"}
],
"manual_imagenes_diagnosticas": ["RX manual - No interpretado"],
"hallazgos_extra": "nota persistida",
},
},
)
self.assertEqual(context["manual_soat_results"][0]["descripcion"], "Manual nuevo")
self.assertEqual(context["pdf_draft"]["manual_procedimientos"], ["PROC agregado"])
self.assertEqual(
context["pdf_draft"]["manual_ayudas_diagnosticas"][0]["nombre"], "Laboratorio manual"
)
self.assertEqual(
context["pdf_draft"]["manual_imagenes_diagnosticas"], ["RX manual - No interpretado"]
)
self.assertEqual(context["pdf_draft"]["hallazgos_extra"], "nota persistida")
cached = cache_repo.get("tester", "case-1")
self.assertIsNotNone(cached)
if cached is None:
self.fail("Expected cached epicrisis context for case-1")
self.assertEqual(cached["contexto"]["manual_soat_results"][0]["descripcion"], "Manual nuevo")
class FakeCaseRepo:
def __init__(self, case: dict | None = None):
self.cases: dict[str, dict] = {}
if case:
case_key = str(case.get("case_key") or "").strip() or "case-1"
self.cases[case_key] = deepcopy(case)
def get_user_case(self, username: str, case_key: str):
case = self.cases.get(case_key)
return deepcopy(case) if case else None
def upsert_user_case(self, username: str, case_key: str, payload: dict) -> dict:
current = deepcopy(self.cases.get(case_key) or {})
current.update(
{
"usuario": username,
"case_key": case_key,
}
)
current.update(deepcopy(payload))
self.cases[case_key] = current
return deepcopy(current)
def update_case(self, batch_id: str, case_key: str, payload: dict) -> None:
current = deepcopy(self.cases.get(case_key) or {"batch_id": batch_id, "case_key": case_key})
current.update(deepcopy(payload))
if batch_id and "batch_id" not in current:
current["batch_id"] = batch_id
self.cases[case_key] = current
class FakeMongoCollection:
def __init__(self, docs: list[dict] | None = None):
self.docs = [deepcopy(doc) for doc in (docs or [])]
def find_one(self, query: dict, sort=None):
matches: list[dict[str, Any]] = []
for doc in self.docs:
if all(doc.get(key) == value for key, value in query.items()):
matches.append(deepcopy(doc))
if not matches:
return None
if sort:
field, direction = sort[0]
reverse = int(direction) < 0
matches.sort(key=lambda item: str(item.get(str(field)) or ""), reverse=reverse)
return matches[0]
class MongoCaseDocumentsReaderTest(unittest.TestCase):
class _CollectionWithFind(FakeMongoCollection):
def find(self, query: dict, sort=None):
matches: list[dict[str, Any]] = []
for doc in self.docs:
if all(doc.get(key) == value for key, value in query.items()):
matches.append(deepcopy(doc))
if sort:
field, direction = sort[0]
reverse = int(direction) < 0
matches.sort(key=lambda item: str(item.get(str(field)) or ""), reverse=reverse)
return matches
def test_prefers_latest_non_low_confidence_primary_document(self) -> None:
collection = self._CollectionWithFind(
[
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "factura",
"fecha_analisis": "2026-07-02T10:00:00-05:00",
"analysis_quality": {"status": "ok", "low_confidence": False},
"nombre_paciente": "PACIENTE ESTABLE",
},
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "factura",
"fecha_analisis": "2026-07-02T11:00:00-05:00",
"analysis_quality": {"status": "degraded", "low_confidence": True},
"nombre_paciente": "PACIENTE RUIDOSO",
},
]
)
reader = MongoCaseDocumentsReader(SimpleNamespace(collection=collection))
documents = reader.get_case_documents("tester", "case-1")
self.assertEqual(documents["factura"]["nombre_paciente"], "PACIENTE ESTABLE")
def test_selects_latest_history_as_recommendation_source(self) -> None:
collection = self._CollectionWithFind(
[
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "historia_clinica",
"fecha_analisis": "2026-07-01T10:00:00-05:00",
"analysis_structured": {
"dt": "historia_clinica",
"pn": "Paciente",
"rs": "Resumen anterior",
"rm": [{"ct": "otra", "i": "Recomendación anterior"}],
},
},
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "historia_clinica",
"fecha_analisis": "2026-07-02T10:00:00-05:00",
"analysis_structured": {
"dt": "historia_clinica",
"pn": "Paciente",
"rs": "Resumen reciente",
"rm": [{"ct": "otra", "i": "Recomendación reciente"}],
},
},
]
)
documents = MongoCaseDocumentsReader(
SimpleNamespace(collection=collection)
).get_case_documents("tester", "case-1")
recommendations = extract_recomendaciones_medicas(documents["historia"])
self.assertEqual(recommendations[0]["indicacion"], "Recomendación reciente")
def test_loads_all_prescriptions_for_medication_reconciliation(self) -> None:
collection = self._CollectionWithFind(
[
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "prescripcion",
"fecha_analisis": "2026-07-02T10:00:00-05:00",
"nombre_archivo": "formula-1.pdf",
},
{
"usuario": "tester",
"case_key": "case-1",
"tipo_documento": "prescripcion",
"fecha_analisis": "2026-07-02T11:00:00-05:00",
"nombre_archivo": "formula-2.pdf",
},
]
)
reader = MongoCaseDocumentsReader(SimpleNamespace(collection=collection))
documents = reader.get_case_documents("tester", "case-1")
self.assertEqual(
[item["nombre_archivo"] for item in documents["prescripcion"]],
["formula-2.pdf", "formula-1.pdf"],
)
class FakeCaseEpicrisisService:
def __init__(self, context: dict):
self.base_context = refresh_diagnostico_context(deepcopy(context))
self.cache: dict[str, dict] = {}
def get_cached_case_context(self, username: str, case_key: str):
context = self.cache.get(case_key)
if not context:
return None
return {"contexto": refresh_diagnostico_context(deepcopy(context))}
def cache_case_context(self, username: str, case_key: str, *, regen: bool = False):
context = refresh_diagnostico_context(deepcopy(self.base_context))
context["case_key"] = case_key
context["regen_url"] = f"/epicrisis?case_key={case_key}®en=1"
self.cache[case_key] = context
return deepcopy(context)
def update_case_draft(self, username: str, case_key: str, *, payload: dict):
context = deepcopy(self.cache.get(case_key) or self.base_context)
context["case_key"] = case_key
if "manual_soat_results" in payload:
context["manual_soat_results"] = deepcopy(payload["manual_soat_results"])
draft = deepcopy(context.get("pdf_draft") or {})
draft.update(deepcopy(payload.get("pdf_draft") or {}))
context["pdf_draft"] = draft
context["codigos_desde_soat"] = (
list(context.get("auto_soat_results", []))
+ list(context.get("qx_agent_results", []))
+ list(context.get("manual_soat_results", []))
)
context = refresh_diagnostico_context(context)
self.cache[case_key] = context
return deepcopy(context)
class FakeCaseEpicrisisRuntimeService:
def __init__(self, case_repo: FakeCaseRepo, cache_service: FakeCaseEpicrisisService):
self.case_repo = case_repo
self.cache_service = cache_service
def queue_case_epicrisis(self, username: str, case_key: str, *, regen: bool = False):
cached = None if regen else self.cache_service.get_cached_case_context(username, case_key)
if cached and isinstance(cached.get("contexto"), dict):
if not bool(cached["contexto"].get("epicrisis_generation_allowed", True)):
rule_eval = cached["contexto"].get("rule_evaluation") or {}
blocking_reason = str(cached["contexto"].get("blocking_reason") or "")
self.case_repo.upsert_user_case(
username,
case_key,
{
"ready_for_epicrisis": False,
"epicrisis_status": "fallido",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
"epicrisis_error": blocking_reason,
"epicrisis_rule_status": str(rule_eval.get("status") or "blocked"),
"epicrisis_rule_findings": deepcopy(rule_eval.get("findings") or []),
"epicrisis_blocking_reason": blocking_reason,
"epicrisis_missing_documents": deepcopy(
cached["contexto"].get("missing_documents") or []
),
},
)
return {
"job_id": "",
"status": "fallido",
"reused": True,
"epicrisis_url": f"/epicrisis?case_key={case_key}",
}
case = self.case_repo.upsert_user_case(
username,
case_key,
{
"ready_for_epicrisis": True,
"epicrisis_status": "completado",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
"epicrisis_error": "",
},
)
return {
"job_id": str(case.get("epicrisis_job_id") or ""),
"status": "completado",
"reused": True,
"epicrisis_url": f"/epicrisis?case_key={case_key}",
}
case = self.case_repo.get_user_case(username, case_key)
if case and not bool(case.get("ready_for_epicrisis", True)):
raise ValueError("El caso aún no esta listo para generar epicrisis.")
if not case:
case = self.case_repo.upsert_user_case(
username,
case_key,
{
"case_number": "1001",
"ready_for_epicrisis": True,
"epicrisis_status": "pendiente",
"epicrisis_job_id": "",
"epicrisis_error": "",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
},
)
current_status = str(case.get("epicrisis_status") or "").strip().lower()
current_job_id = str(case.get("epicrisis_job_id") or "").strip()
if not regen and current_status in {"en_cola", "procesando"} and current_job_id:
return {
"job_id": current_job_id,
"status": current_status,
"reused": True,
"epicrisis_url": f"/epicrisis?case_key={case_key}",
}
job_id = f"job-{case_key}"
self.case_repo.upsert_user_case(
username,
case_key,
{
"ready_for_epicrisis": True,
"epicrisis_status": "en_cola",
"epicrisis_job_id": job_id,
"epicrisis_error": "",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
},
)
return {
"job_id": job_id,
"status": "en_cola",
"reused": False,
"epicrisis_url": f"/epicrisis?case_key={case_key}",
}
def get_case_epicrisis_status(self, username: str, case_key: str) -> dict:
cached = self.cache_service.get_cached_case_context(username, case_key)
case = self.case_repo.get_user_case(username, case_key)
if cached and isinstance(cached.get("contexto"), dict):
contexto = cached["contexto"]
if not bool(contexto.get("epicrisis_generation_allowed", True)):
rule_eval = contexto.get("rule_evaluation") or {}
return {
"case_key": case_key,
"case_number": str(
(case or {}).get("case_number") or contexto.get("case_number") or "1001"
),
"ready_for_epicrisis": False,
"epicrisis_status": "fallido",
"epicrisis_job_id": str((case or {}).get("epicrisis_job_id") or ""),
"epicrisis_error": str(contexto.get("blocking_reason") or ""),
"epicrisis_url": f"/epicrisis?case_key={case_key}",
"epicrisis_rule_status": str(rule_eval.get("status") or "blocked"),
"epicrisis_rule_findings": deepcopy(rule_eval.get("findings") or []),
"epicrisis_blocking_reason": str(contexto.get("blocking_reason") or ""),
"epicrisis_missing_documents": deepcopy(contexto.get("missing_documents") or []),
}
return {
"case_key": case_key,
"case_number": str((case or {}).get("case_number") or contexto.get("case_number") or "1001"),
"ready_for_epicrisis": True,
"epicrisis_status": "completado",
"epicrisis_job_id": str((case or {}).get("epicrisis_job_id") or ""),
"epicrisis_error": "",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
"epicrisis_rule_status": "",
"epicrisis_rule_findings": [],
"epicrisis_blocking_reason": "",
"epicrisis_missing_documents": [],
}
if not case:
raise ValueError("Caso no encontrado")
return {
"case_key": case_key,
"case_number": str(case.get("case_number") or "1001"),
"ready_for_epicrisis": bool(case.get("ready_for_epicrisis", False)),
"epicrisis_status": str(case.get("epicrisis_status") or "pendiente"),
"epicrisis_job_id": str(case.get("epicrisis_job_id") or ""),
"epicrisis_error": str(case.get("epicrisis_error") or ""),
"epicrisis_url": str(case.get("epicrisis_url") or f"/epicrisis?case_key={case_key}"),
"epicrisis_rule_status": str(case.get("epicrisis_rule_status") or ""),
"epicrisis_rule_findings": deepcopy(case.get("epicrisis_rule_findings") or []),
"epicrisis_blocking_reason": str(case.get("epicrisis_blocking_reason") or ""),
"epicrisis_missing_documents": deepcopy(case.get("epicrisis_missing_documents") or []),
}
def run_generation_job(
self,
username: str,
case_key: str,
*,
regen: bool = False,
job_id: str = "",
audit_context: dict | None = None,
) -> str:
self.cache_service.cache_case_context(username, case_key, regen=regen)
self.case_repo.upsert_user_case(
username,
case_key,
{
"ready_for_epicrisis": True,
"epicrisis_status": "completado",
"epicrisis_job_id": job_id or f"job-{case_key}",
"epicrisis_error": "",
"epicrisis_url": f"/epicrisis?case_key={case_key}",
},
)
return case_key
class CaseEpicrisisRuntimeServiceRulesTest(unittest.TestCase):
@staticmethod
def _manual_case_context() -> dict[str, object]:
return {
"case_key": "case-manual",
"case_number": "1001",
"nombre_paciente": "Paciente Manual",
"epicrisis_generation_allowed": True,
"rule_evaluation": {"status": "passed", "findings": []},
}
def test_marks_case_as_failed_with_rule_metadata_when_generation_is_blocked(self) -> None:
context = {
"case_key": "case-blocked",
"case_number": "2002",
"epicrisis_generation_allowed": False,
"missing_documents": ["historia_clinica"],
"blocking_reason": "No se puede generar la epicrisis sin historia clínica.",
"rule_evaluation": {
"status": "blocked",
"findings": [
{
"rule_code": "EPICRISIS-HC-001",
"severity": "error",
"message": "No se puede generar la epicrisis sin historia clínica.",
"document_types_checked": ["factura", "quirurgico", "generico"],
"blocking": True,
}
],
"missing_documents": ["historia_clinica"],
"required_documents": ["historia_clinica"],
"generation_allowed": False,
"ready_for_epicrisis": False,
"evaluated_at": "2026-05-27T10:00:00+00:00",
"evaluation_mode": "deterministic_only",
},
}
cached_context: dict[str, object] = {}
class BlockingService:
def get_cached_case_context(self, username: str, case_key: str):
if cached_context:
return {"contexto": deepcopy(cached_context)}
return None
def cache_case_context(self, username: str, case_key: str, *, regen: bool = False):
cached_context.update(deepcopy(context))
raise EpicrisisGenerationBlockedError(
"No se puede generar la epicrisis sin historia clínica.",
evaluation=EpicrisisRuleEvaluation.model_validate(
{
"status": "blocked",
"generation_allowed": False,
"ready_for_epicrisis": False,
"evaluated_at": "2026-05-27T10:00:00+00:00",
"missing_documents": ["historia_clinica"],
"required_documents": ["historia_clinica"],
}
),
context=deepcopy(context),
)
runtime = CaseEpicrisisRuntimeService(
case_repository=cast(MongoBatchCaseRepository, FakeCaseRepo()),
case_epicrisis_service=BlockingService(),
clinical_document_service=SimpleNamespace(
get_user_case_context=lambda username, case_key: {
"case_number": "2002",
"patient_name": "Paciente Bloqueado",
"patient_id": "CC123",
}
),
colombia_tz=UTC,
)
with self.assertRaises(EpicrisisGenerationBlockedError):
runtime.run_generation_job("tester", "case-blocked", job_id="job-blocked")
case = runtime.case_repository.get_user_case("tester", "case-blocked")
self.assertIsNotNone(case)
if case is None:
self.fail("Expected persisted case state for blocked epicrisis generation")
self.assertFalse(case["ready_for_epicrisis"])
self.assertEqual(case["epicrisis_status"], "fallido")
self.assertEqual(case["epicrisis_rule_status"], "blocked")
self.assertEqual(case["epicrisis_missing_documents"], ["historia_clinica"])
status_payload = runtime.get_case_epicrisis_status("tester", "case-blocked")
self.assertEqual(status_payload["epicrisis_status"], "fallido")
self.assertEqual(status_payload["epicrisis_blocking_reason"], context["blocking_reason"])
def test_queue_case_epicrisis_resets_stale_manual_block_without_touching_batch_logic(self) -> None:
runtime = CaseEpicrisisRuntimeService(
case_repository=cast(
MongoBatchCaseRepository,
FakeCaseRepo(
{
"case_key": "case-manual",
"case_number": "1001",
"patient_name": "Paciente Manual",
"patient_id": "CC123",
"runtime_origin": "manual",
"ready_for_epicrisis": False,
"epicrisis_status": "fallido",
"epicrisis_error": "No se puede generar la epicrisis sin historia clínica.",
"epicrisis_rule_status": "blocked",
"epicrisis_rule_findings": [{"rule_code": "EPICRISIS-HC-001"}],
"epicrisis_blocking_reason": "No se puede generar la epicrisis sin historia clínica.",
"epicrisis_missing_documents": ["historia_clinica"],
}
),
),
case_epicrisis_service=FakeCaseEpicrisisService(deepcopy(self._manual_case_context())),
clinical_document_service=SimpleNamespace(
get_user_case_context=lambda username, case_key: {
"case_number": "1001",
"patient_name": "Paciente Manual",
"patient_id": "CC123",
}
),
colombia_tz=UTC,
)
with (
patch.dict("os.environ", {"BATCH_DISPATCHER": "inprocess"}),
patch.object(runtime, "_spawn_inprocess_job", return_value=None),
):
result = runtime.queue_case_epicrisis("tester", "case-manual")
self.assertEqual(result["status"], "en_cola")
case = runtime.case_repository.get_user_case("tester", "case-manual")
self.assertIsNotNone(case)
if case is None:
self.fail("Expected persisted case state for stale manual runtime")
self.assertTrue(case["ready_for_epicrisis"])
self.assertEqual(case["epicrisis_status"], "en_cola")
self.assertEqual(case["epicrisis_rule_status"], "")
self.assertEqual(case["epicrisis_rule_findings"], [])
self.assertEqual(case["epicrisis_blocking_reason"], "")
self.assertEqual(case["epicrisis_missing_documents"], [])
def test_get_case_epicrisis_status_degrades_stale_manual_block_to_pending(self) -> None:
runtime = CaseEpicrisisRuntimeService(
case_repository=cast(
MongoBatchCaseRepository,
FakeCaseRepo(
{
"case_key": "case-manual",
"case_number": "1001",
"patient_name": "Paciente Manual",
"patient_id": "CC123",
"runtime_origin": "manual",
"ready_for_epicrisis": False,
"epicrisis_status": "fallido",
"epicrisis_error": "No se puede generar la epicrisis sin historia clínica.",
"epicrisis_rule_status": "blocked",
"epicrisis_rule_findings": [{"rule_code": "EPICRISIS-HC-001"}],
"epicrisis_blocking_reason": "No se puede generar la epicrisis sin historia clínica.",
"epicrisis_missing_documents": ["historia_clinica"],
}
),
),
case_epicrisis_service=FakeCaseEpicrisisService(deepcopy(self._manual_case_context())),
clinical_document_service=SimpleNamespace(get_user_case_context=lambda username, case_key: {}),
colombia_tz=UTC,
)
status_payload = runtime.get_case_epicrisis_status("tester", "case-manual")
self.assertEqual(status_payload["epicrisis_status"], "pendiente")
self.assertTrue(status_payload["ready_for_epicrisis"])
self.assertEqual(status_payload["epicrisis_rule_status"], "")
self.assertEqual(status_payload["epicrisis_blocking_reason"], "")
self.assertEqual(status_payload["epicrisis_missing_documents"], [])
def test_get_case_epicrisis_status_preserves_batch_failure_without_cache(self) -> None:
runtime = CaseEpicrisisRuntimeService(
case_repository=cast(
MongoBatchCaseRepository,
FakeCaseRepo(
{
"case_key": "case-batch",
"batch_id": "batch-1",
"case_number": "1002",
"patient_name": "Paciente Batch",
"patient_id": "CC999",
"ready_for_epicrisis": False,
"epicrisis_status": "fallido",
"epicrisis_error": "Timeout worker",
"epicrisis_rule_status": "blocked",
"epicrisis_blocking_reason": "No se puede generar la epicrisis sin historia clínica.",
"epicrisis_missing_documents": ["historia_clinica"],
}
),
),
case_epicrisis_service=FakeCaseEpicrisisService(deepcopy(self._manual_case_context())),
clinical_document_service=SimpleNamespace(get_user_case_context=lambda username, case_key: {}),
colombia_tz=UTC,
)
status_payload = runtime.get_case_epicrisis_status("tester", "case-batch")
self.assertEqual(status_payload["epicrisis_status"], "fallido")
self.assertFalse(status_payload["ready_for_epicrisis"])
self.assertEqual(status_payload["epicrisis_rule_status"], "blocked")
class CaseEpicrisisRoutesTest(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
self.base_context = {
"nombre_paciente": "Paciente Uno",
"case_key": "case-1",
"case_number": "1001",
"historia": {
"_id": "hist-1",
"analisis_html": (
"Procedimientos
- Proc HC desde cache
"
"Medicamentos administrados
- 7701 - 1 ampolla - Dipirona - 1 g IV
"
),
"codigos_cie10": [{"codigo": "A00", "descripcion": "Diag base"}],
},
"quirurgico": {
"_id": "qx-1",
"codigos_cups": [{"codigo_cups": "123456", "procedimiento": "Proc base"}],
},
"factura": {
"_id": "fac-1",
"factura_json": {
"servicios_procedimientos": {
"medicamentos": [
{
"codigo_medicamento": "7701",
"medicamento": "Dipirona",
"cantidad": "1 ampolla",
"dosis": "1 g IV",
}
],
"procedimientos_quirurgicos": [
{"codigo_cups": "555555", "descripcion": "Proc factura"}
],
"imagenologia": [{"estudio": "RX rodilla", "descripcion": "No interpretado"}],
}
},
},
"radiologia": [],
"metadatos_hc": {"nombre_paciente": "Paciente Uno", "caso": "1001", "resumen": "Resumen"},
"procedimientos_hc": ["Proc HC"],
"medicamentos_hc": [
{
"codigo_referencia": "7701",
"nombre": "Dipirona",
"cantidad": 1,
"unidad_cantidad": "ampolla",
"dosis": "1 g IV",
"estados": ["administrado"],
}
],
"medicamentos_hc_display": [],
"imagenes_diagnosticas": ["RX rodilla - No interpretado"],
"hallazgos_quirurgicos": "Hallazgos",
"descripcion_procedimiento": "Descripcion",
"soat_resultados": [{"codigo_soat": "S0", "descripcion": "Proc soat"}],
"glosa_analisis": "Glosa
",
"procedimientos_factura": [],
"auto_soat_results": [
{
"codigo_soat": "S0",
"descripcion": "Proc soat",
"cie10_principal": {"codigo": "B00", "descripcion": "Dx soat"},
}
],
"qx_agent_results": [
{
"codigo_soat": "QX-HALLAZGOS",
"descripcion": "Hallazgos quirurgicos",
"cie10_principal": {"codigo": "C00", "descripcion": "Dx qx"},
}
],
"manual_soat_results": [],
"pdf_draft": {
"excluded_diagnosticos": [],
"excluded_diagnostico_keys": [],
"manual_diagnosticos": [],
"excluded_procedimientos": [],
"manual_procedimientos": [],
"excluded_medicamentos": [],
"manual_medicamentos": [],
"excluded_imagenes_diagnosticas": [],
"manual_imagenes_diagnosticas": [],
"hallazgos_extra": "",
"descripcion_procedimiento_extra": "",
},
"codigos_desde_soat": [
{
"codigo_soat": "S0",
"descripcion": "Proc soat",
"cie10_principal": {"codigo": "B00", "descripcion": "Dx soat"},
},
{
"codigo_soat": "QX-HALLAZGOS",
"descripcion": "Hallazgos quirurgicos",
"cie10_principal": {"codigo": "C00", "descripcion": "Dx qx"},
},
],
"regen_url": "/epicrisis?case_key=case-1®en=1",
"epicrisis_cached": False,
}
self.fake_service = FakeCaseEpicrisisService(self.base_context)
self.fake_case_repo = FakeCaseRepo()
self.fake_runtime_service = FakeCaseEpicrisisRuntimeService(self.fake_case_repo, self.fake_service)
self.fake_services = SimpleNamespace(
templates=Jinja2Templates(directory="web"),
mongo_analyses=SimpleNamespace(collection=None),
colombia_tz=UTC,
client_groq=None,
client_gemini=None,
soat_retriever=None,
cie10_retriever=None,
case_epicrisis_service=self.fake_service,
case_epicrisis_runtime_service=self.fake_runtime_service,
)
self.app = FastAPI()
self.app.include_router(pages.router)
self.app.state.services = self.fake_services
async def _override_user():
return SimpleNamespace(username="tester")
self.app.dependency_overrides[get_current_user] = _override_user
self.transport = ASGITransport(app=self.app)
def tearDown(self) -> None:
self.app.dependency_overrides.clear()
async def test_post_then_get_case_epicrisis_uses_canonical_context(self) -> None:
async with AsyncClient(transport=self.transport, base_url="http://testserver") as client:
with patch(
"app.routes.pages._find_latest_case_document",
return_value={"case_number": "1001"},
):
post_response = await client.post("/epicrisis", params={"case_key": "case-1"})
self.assertEqual(post_response.status_code, 202)
self.assertEqual(post_response.json()["epicrisis_status"], "en_cola")
self.fake_runtime_service.run_generation_job("tester", "case-1", job_id="job-case-1")
get_response = await client.get("/epicrisis", params={"case_key": "case-1"})
self.assertEqual(get_response.status_code, 200)
self.assertIn("/static/css/epicrisis.css", get_response.text)
self.assertIn('id="epicrisis_scroll_top_btn"', get_response.text)
self.assertIn("/static/js/user-menu.js", get_response.text)
self.assertIn("/static/js/epicrisis/scroll-top.js", get_response.text)
self.assertIn("/static/js/epicrisis/index.js", get_response.text)
self.assertIn("Epicrisis operativa", get_response.text)
self.assertIn("Volver a casos", get_response.text)
self.assertIn("Operar RIPS", get_response.text)
self.assertIn("Regenerar epicrisis", get_response.text)
self.assertIn("Generar PDF", get_response.text)
self.assertIn("data-user-menu-root", get_response.text)
self.assertIn("Mi perfil", get_response.text)
self.assertIn("Cerrar sesión", get_response.text)
self.assertIn("Proc soat", get_response.text)
self.assertIn("manualSoatResults", get_response.text)
self.assertIn("diagnosticosConsolidados", get_response.text)
self.assertIn("pdfPrimaryDiagnosticos", get_response.text)
self.assertIn("pdfPrimaryProcedimientos", get_response.text)
self.assertIn("pdfPrimaryMedicamentos", get_response.text)
self.assertIn(">Medicamentos", get_response.text)
self.assertIn("Posología | ", get_response.text)
self.assertIn("Cantidad | ", get_response.text)
self.assertIn("medication-matrix", get_response.text)
self.assertNotIn("no_identificado", get_response.text)
self.assertIn("pdfPrimaryAyudasDiagnosticas", get_response.text)
self.assertIn("procedimientosFactura", get_response.text)
self.assertIn("soatResultados", get_response.text)
self.assertIn("hasQuirurgicoDocument: true", get_response.text)
self.assertIn("Análisis de Posibles Glosas", get_response.text)
self.assertNotIn("fuentesDocumentales", get_response.text)
self.assertNotIn("glosaAnalisis", get_response.text)
self.assertNotIn('