from __future__ import annotations import hashlib import json from copy import deepcopy from decimal import Decimal from io import BytesIO from pathlib import Path from unittest.mock import patch from zipfile import ZipFile import pytest from app.core.heavy_resources import HeavyResourceRegistry from app.llm.renderers.clinical_documents import render_factura_html from app.llm.schemas import dump_clinical_model from app.scripts.soat_catalog import ( DEFAULT_SOURCE_DIR, SourceSpec, catalog_bytes, download_sources, extract_header_unit_value, extract_surgical_procedures, extract_tariff_rows, extract_uvb_value, load_manifest, validate_source, validate_sources, ) from app.services.clinical_document_projection import render_document_analysis_html from app.services.factura_deterministic import build_factura_structured from app.soat_tariffs.application.surgical_costs import associate_surgical_costs from app.soat_tariffs.application.valuation import round_to_hundred, valuate_factura_soat from app.soat_tariffs.domain.models import CatalogSource, SoatCatalog, SoatCatalogEntry, SoatTariffComponent from app.soat_tariffs.domain.surgical_costs import CostoQuirurgicoAsociado from app.soat_tariffs.infrastructure.json_catalog import JsonSoatTariffCatalog FIXTURE = Path("test/fixtures/factura_julio_cesar_acosta.txt") class InMemoryCatalog: def __init__(self, catalogs: dict[int, SoatCatalog]) -> None: self.catalogs = catalogs self.queries: list[tuple[int, str]] = [] def load(self, year: int) -> SoatCatalog: if year not in self.catalogs: raise FileNotFoundError return self.catalogs[year] def find_exact(self, year: int, code: str) -> SoatCatalogEntry | None: self.queries.append((year, code)) return next((entry for entry in self.load(year).entries if entry.code == code), None) def validate_all(self) -> dict[int, str]: return {year: "ready" for year in self.catalogs} def _catalog(year: int = 2024) -> SoatCatalog: source = CatalogSource( url="https://example.test/official.pdf", sha256="a" * 64, page_table="p. 1, tabla de prueba", ) entry = SoatCatalogEntry( code="13170", description="Procedimiento exacto", surgical_group=7, base_coefficient=Decimal("5.50"), components=[ SoatTariffComponent(name="cirujano", code="39005", coefficient=Decimal("1.01")), SoatTariffComponent(name="anestesia", code="39105", coefficient=Decimal("1.02")), SoatTariffComponent(name="ayudantia", code="39118", coefficient=Decimal("1.03")), SoatTariffComponent(name="sala", code="39209", coefficient=Decimal("1.20")), SoatTariffComponent(name="materiales", code="39303", coefficient=Decimal("1.24")), ], source=source, ) return SoatCatalog( year=year, unit="UVT", unit_value=Decimal("1000"), version=f"test-{year}", generated_from=[source], entries=[entry], ) def _invoice(*, service_date: str = "2024-05-09", code: str = "13170") -> dict: return { "informacion_factura": {"fecha_emision": "2024-05-27"}, "lineas_canonicas": [ { "fecha_servicio": service_date, "codigo_facturacion": code, "codigo_referencia": "793101", "descripcion": "Texto secundario", "valor_unitario": "$123.00", "total": "$123.00", } ], "servicios_procedimientos": { "procedimientos_quirurgicos": [ { "fecha_servicio": service_date, "codigo_facturacion": code, "codigo_referencia": "793101", "descripcion": "Texto secundario", "valor_unitario": "$123.00", "total": "$123.00", } ] }, "analisis_financiero": {"valor_total_factura": "$123.00"}, } def test_rounding_and_component_conversion_use_hundreds() -> None: assert round_to_hundred(Decimal("149")) == 100 assert round_to_hundred(Decimal("150")) == 200 result = valuate_factura_soat(_invoice(), InMemoryCatalog({2024: _catalog()})) valuation = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"] assert valuation["valor_base"] == 5_400 assert valuation["componentes"] == { "cirujano": 1_000, "anestesia": 1_000, "ayudantia": 1_000, "sala": 1_200, "materiales": 1_200, } def test_service_year_is_strict_and_issue_year_is_recorded_as_fallback() -> None: catalog = InMemoryCatalog({2024: _catalog()}) unavailable = valuate_factura_soat(_invoice(service_date="2023-05-09"), catalog) assert ( unavailable["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"][ "regla_aplicada" ] == "vigencia_no_disponible" ) assert catalog.queries == [] fallback = valuate_factura_soat(_invoice(service_date=""), catalog) evidence = fallback["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"][ "evidencia" ] assert any("emision" in item for item in evidence) def test_unknown_code_never_uses_description_or_semantic_search_for_price() -> None: catalog = InMemoryCatalog({2024: _catalog()}) result = valuate_factura_soat(_invoice(code="99999"), catalog) valuation = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"] assert catalog.queries == [(2024, "99999")] assert valuation["estado"] == "pendiente_revision" assert valuation.get("valor_base") is None assert "similitud semantica" in valuation["evidencia"][0] def test_single_surgery_uses_full_base_and_preserves_original_amounts() -> None: invoice = _invoice() original = deepcopy(invoice) result = valuate_factura_soat(invoice, InMemoryCatalog({2024: _catalog()})) valuation = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"] assert valuation["estado"] == "liquidada" assert valuation["valor_liquidado"] == valuation["valor_base"] == 5_400 assert result["lineas_canonicas"] == original["lineas_canonicas"] assert result["analisis_financiero"] == original["analisis_financiero"] def test_multiple_surgeries_without_unique_principal_or_components_remain_pending() -> None: year_catalog = _catalog() year_catalog.entries.append( year_catalog.entries[0].model_copy(update={"code": "18211", "description": "Segundo procedimiento"}) ) invoice = _invoice() second = deepcopy(invoice["servicios_procedimientos"]["procedimientos_quirurgicos"][0]) second["codigo_facturacion"] = "18211" invoice["servicios_procedimientos"]["procedimientos_quirurgicos"].append(second) result = valuate_factura_soat(invoice, InMemoryCatalog({2024: year_catalog})) valuations = [ item["valoracion_soat"] for item in result["servicios_procedimientos"]["procedimientos_quirurgicos"] ] assert all(item["estado"] == "pendiente_revision" for item in valuations) assert all(item["valor_liquidado"] == item["valor_base"] for item in valuations) def test_anonymized_2024_fixture_matches_expected_multiple_surgery_valuation() -> None: model, factura_json = build_factura_structured( raw_text=FIXTURE.read_text(encoding="utf-8"), current_structured=None ) procedures = factura_json["servicios_procedimientos"]["procedimientos_quirurgicos"] values = { item["codigo_facturacion"]: ( item["codigo_referencia"], item["grupo_quirurgico_soat"], item["valoracion_soat"]["valor_base"], item["valoracion_soat"]["valor_liquidado"], ) for item in procedures } assert values == { "13170": ("793101", 7, 1_794_600, 344_200), "14332": ("817204", 10, 3_005_600, 598_800), "18211": ("765401", 7, 1_794_600, 344_200), "18215": ("818307", 11, 3_228_300, 3_228_300), } assert factura_json["resumen_tarifario_soat"]["total_base"] == 9_823_100 assert factura_json["resumen_tarifario_soat"]["total_liquidado"] == 4_515_500 assert procedures[0]["valoracion_soat"]["componentes"] == { "cirujano": 371_300, "anestesia": 218_900, "ayudantia": 97_900, "sala": 697_000, "materiales": 409_500, } assert procedures[1]["valoracion_soat"]["componentes"] == { "cirujano": 648_600, "anestesia": 372_300, "ayudantia": 176_500, "sala": 1_159_200, "materiales": 649_000, } assert ( factura_json["analisis_financiero"]["valor_total_factura"] == model.resumen_financiero.valor_total_factura ) def test_anonymized_fixture_associates_complete_surgical_charges_without_double_counting() -> None: _, factura_json = build_factura_structured( raw_text=FIXTURE.read_text(encoding="utf-8"), current_structured=None ) procedures = factura_json["servicios_procedimientos"]["procedimientos_quirurgicos"] costs = { item["codigo_cups"]: Decimal(item["costo_quirurgico_asociado"]["valor_final"]) for item in procedures } assert costs == { "818307": Decimal("3228300.00"), "817204": Decimal("598800.00"), "793101": Decimal("344200.00"), "765401": Decimal("344200.00"), } assert sum(costs.values()) == Decimal("4515500.00") assert all( item["costo_quirurgico_asociado"]["estado"] in {"directa", "proporcional"} for item in procedures ) def test_renderer_is_central_and_marks_valuation_as_reference() -> None: model, _ = build_factura_structured(raw_text=FIXTURE.read_text(encoding="utf-8"), current_structured=None) rendered = render_factura_html(model) assert "Cálculo de procedimientos quirúrgicos" in rendered assert "no modifica el total declarado" in rendered assert " None: factura = { "lineas_canonicas": [ {"codigo_facturacion": "11111", "fecha_servicio": "2024-01-01", "total": "10.00"}, {"codigo_facturacion": "22222", "fecha_servicio": "2024-01-01", "total": "20.00"}, {"codigo_facturacion": "39005", "fecha_servicio": "2024-01-01", "total": "100.01"}, ], "servicios_procedimientos": { "procedimientos_quirurgicos": [ { "codigo_cups": "010101", "fecha_servicio": "2024-01-01", "valoracion_soat": { "indice_linea_canonica": 0, "componentes_liquidados": {"cirujano": 100}, "codigos_componentes": {"cirujano": "39005"}, }, }, { "codigo_cups": "010102", "fecha_servicio": "2024-01-01", "valoracion_soat": { "indice_linea_canonica": 1, "componentes_liquidados": {"cirujano": 100}, "codigos_componentes": {"cirujano": "39005"}, }, }, ] }, } result = associate_surgical_costs(factura) costs = [ item["costo_quirurgico_asociado"] for item in result["servicios_procedimientos"]["procedimientos_quirurgicos"] ] assert [cost["valor_final"] for cost in costs] == ["60.01", "70.00"] assert sum(Decimal(cost["componentes"][0]["valor_cobrado"]) for cost in costs) == Decimal("100.01") assert [cost["estado"] for cost in costs] == ["proporcional", "proporcional"] assert costs[0]["componentes"][0]["indices_lineas_canonicas"] == [2] def test_associated_cost_model_round_trips_compact_and_expanded_payloads() -> None: expanded = { "estado": "directa", "valor_directo": "25.50", "componentes": [], "valor_final": "25.50", "metodo": "asignacion_directa", "formula": "valor directo + suma de componentes cobrados asignados", } model = CostoQuirurgicoAsociado.model_validate(expanded) compact = model.model_dump(by_alias=True, mode="json") assert compact["vf"] == "25.50" assert CostoQuirurgicoAsociado.model_validate(compact) == model def test_associated_cost_reports_unmatched_charge_and_does_not_reuse_direct_line() -> None: factura = { "lineas_canonicas": [ {"fecha_servicio": "2024-01-01", "codigo_facturacion": "11111", "total": "50"}, {"fecha_servicio": "2024-01-01", "codigo_facturacion": "39399", "total": "75"}, ], "servicios_procedimientos": { "procedimientos_quirurgicos": [ {"fecha_servicio": "2024-01-01", "valoracion_soat": {"indice_linea_canonica": 0}}, {"fecha_servicio": "2024-01-01", "valoracion_soat": {"indice_linea_canonica": 0}}, ] }, } result = associate_surgical_costs(factura) costs = [ item["costo_quirurgico_asociado"] for item in result["servicios_procedimientos"]["procedimientos_quirurgicos"] ] assert [cost["valor_directo"] for cost in costs] == ["50", "0"] assert costs[0]["saldo_no_asignado"] == "75" assert costs[0]["indices_saldo_no_asignado"] == [1] assert all(cost["estado"] == "incompleta" for cost in costs) def test_catalog_rejects_missing_and_corrupt_vigencies(tmp_path: Path) -> None: adapter = JsonSoatTariffCatalog(tmp_path) with pytest.raises(FileNotFoundError): adapter.load(2024) (tmp_path / "soat-2024.json").write_text(json.dumps({"year": 2024}), encoding="utf-8") with pytest.raises(ValueError, match="corrupto"): adapter.load(2024) def test_corrupt_tariff_catalog_prevents_soat_warmup_ready(tmp_path: Path) -> None: registry = HeavyResourceRegistry(tmp_path) index_dir = tmp_path / "soat_faiss" index_dir.mkdir() registry.set_soat_index_dir(index_dir) with patch.object(JsonSoatTariffCatalog, "validate_all", side_effect=ValueError("corrupto")): assert registry.soat_available() is False assert registry.warmup("soat")["soat"] != "ready" def test_generated_catalogs_have_complete_shared_surgical_universe() -> None: adapter = JsonSoatTariffCatalog() universes: list[set[str]] = [] for year in range(2022, 2027): catalog = adapter.load(year) codes = {entry.code for entry in catalog.entries} universes.append(codes) assert len(catalog.entries) == len(codes) == 1771 assert {entry.surgical_group for entry in catalog.entries} == { *range(2, 14), *range(20, 24), } assert all(entry.description.strip() for entry in catalog.entries) assert catalog.version == f"soat-tarifas-{year}-v2" assert all(universe == universes[0] for universe in universes[1:]) assert "01101" in universes[0] def test_every_surgical_group_has_representative_components() -> None: catalog = JsonSoatTariffCatalog().load(2024) representatives = { group: next(entry for entry in catalog.entries if entry.surgical_group == group) for group in (*range(2, 14), *range(20, 24)) } for group, entry in representatives.items(): names = {component.name for component in entry.components} assert "ayudantia" not in names if group <= 5 else "ayudantia" in names assert "materiales" in names if group <= 13 else "materiales" not in names def test_four_digit_code_is_zero_filled_and_malformed_codes_are_rejected() -> None: adapter = JsonSoatTariffCatalog() assert adapter.find_exact(2024, "1101").code == "01101" assert adapter.find_exact(2024, "01101").code == "01101" assert adapter.find_exact(2024, "01-101") is None assert adapter.find_exact(2024, "101") is None assert adapter.find_exact(2024, "001101") is None result = valuate_factura_soat(_invoice(code="1101"), adapter) valuation = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"] assert valuation["estado"] == "liquidada" assert result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["codigo_soat"] == "01101" def test_official_sources_extract_xls_xlsx_and_pdf_values() -> None: sources = {source.id: source for source in load_manifest()} validate_sources(sources.values()) with ZipFile(DEFAULT_SOURCE_DIR / sources["tariffs-2022"].local_file) as archive: xls = archive.read(sources["tariffs-2022"].archive_member) assert len(extract_surgical_procedures(xls)) == 1771 rows_2022 = extract_tariff_rows(xls, year=2022, archive_member=sources["tariffs-2022"].archive_member) assert len(rows_2022) == 64 assert extract_header_unit_value( xls, year=2022, archive_member=sources["tariffs-2022"].archive_member ) == Decimal("33333.33") with ZipFile(DEFAULT_SOURCE_DIR / sources["tariffs-2024"].local_file) as archive: xlsx = archive.read(sources["tariffs-2024"].archive_member) rows_2024 = extract_tariff_rows(xlsx, year=2024, archive_member=sources["tariffs-2024"].archive_member) assert len(rows_2024) == 64 assert rows_2024["39005"]["official_value"] == 371_300 assert extract_header_unit_value( xlsx, year=2024, archive_member=sources["tariffs-2024"].archive_member ) == Decimal("47065") assert extract_uvb_value( (DEFAULT_SOURCE_DIR / sources["uvb-2025"].local_file).read_bytes(), 2025 ) == Decimal("11552") assert extract_uvb_value( (DEFAULT_SOURCE_DIR / sources["uvb-2026"].local_file).read_bytes(), 2026 ) == Decimal("12110") def test_representative_components_use_official_values_and_2026_rounding() -> None: adapter = JsonSoatTariffCatalog() expected_group_7 = { 2022: [300_000, 176_700, 79_000, 562_700, 330_700], 2023: [334_600, 197_200, 88_200, 628_100, 369_000], 2024: [371_300, 218_900, 97_900, 697_000, 409_500], 2025: [391_700, 230_900, 103_300, 735_300, 432_000], } for year, expected in expected_group_7.items(): entry = adapter.find_exact(year, "13170") assert [component.official_value for component in entry.components] == expected assert entry.official_base_value == sum(expected) entry_2026 = adapter.find_exact(2026, "13170") calculated = [ round_to_hundred(component.coefficient * Decimal("12110")) for component in entry_2026.components ] assert calculated == [410_700, 242_000, 108_300, 770_800, 452_800] result = valuate_factura_soat(_invoice(service_date="2026-05-09"), adapter) valuation = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"] assert list(valuation["componentes"].values()) == calculated assert valuation["valor_base"] == sum(calculated) def test_generated_from_is_complete_and_exposed_as_valuation_evidence() -> None: adapter = JsonSoatTariffCatalog() catalog = adapter.load(2025) assert len(catalog.generated_from) == 4 result = valuate_factura_soat(_invoice(service_date="2025-05-09"), adapter) evidence = result["servicios_procedimientos"]["procedimientos_quirurgicos"][0]["valoracion_soat"][ "evidencia" ] source_evidence = [item for item in evidence if item.startswith("Fuente:")] assert len(source_evidence) == len(catalog.generated_from) assert all("sha256=" in item and "origen=" in item for item in source_evidence) def test_download_is_atomic_hash_checked_and_network_free_when_present(tmp_path: Path) -> None: content = b"official-source" digest = hashlib.sha256(content).hexdigest() source = SourceSpec( id="test-source", url="https://example.test/source.bin", local_file="source.bin", sha256=digest, role="annual_tariffs", applies_to=[2022], page_table="fixture", ) manifest = tmp_path / "manifest.json" manifest.write_text(json.dumps({"sources": [source.model_dump(mode="json")]}), encoding="utf-8") source_dir = tmp_path / "sources" source_dir.mkdir() calls: list[str] = [] def opener(url: str) -> BytesIO: calls.append(url) return BytesIO(content) assert download_sources(manifest_path=manifest, source_dir=source_dir, opener=opener) == [ "source.bin: downloaded" ] assert (source_dir / "source.bin").read_bytes() == content assert not list(source_dir.glob("*.tmp")) assert download_sources(manifest_path=manifest, source_dir=source_dir, opener=opener) == [ "source.bin: present" ] assert calls == [source.url] (source_dir / "source.bin").write_bytes(b"damaged") with pytest.raises(ValueError, match="Hash descargado invalido"): download_sources( manifest_path=manifest, source_dir=source_dir, opener=lambda _url: BytesIO(b"wrong-download"), ) assert (source_dir / "source.bin").read_bytes() == b"damaged" assert not list(source_dir.glob("*.tmp")) def test_source_validation_rejects_missing_corrupt_and_wrong_zip_member(tmp_path: Path) -> None: source = SourceSpec( id="missing", url="https://example.test/source.zip", local_file="source.zip", sha256="a" * 64, role="annual_tariffs", applies_to=[2022], page_table="fixture", archive_member="missing.xls", ) with pytest.raises(FileNotFoundError): validate_source(source, tmp_path) (tmp_path / "source.zip").write_bytes(b"corrupt") with pytest.raises(ValueError, match="Hash invalido"): validate_source(source, tmp_path) archive_buffer = BytesIO() with ZipFile(archive_buffer, "w") as archive: archive.writestr("present.xls", b"fixture") archive_content = archive_buffer.getvalue() (tmp_path / "source.zip").write_bytes(archive_content) wrong_member_source = source.model_copy(update={"sha256": hashlib.sha256(archive_content).hexdigest()}) with pytest.raises(ValueError, match="Miembro ZIP ausente"): validate_source(wrong_member_source, tmp_path) def test_catalog_generation_is_byte_deterministic_and_matches_versioned_artifact() -> None: first = catalog_bytes(2024) second = catalog_bytes(2024) assert first == second assert first == Path("app/soat_tariffs/catalogs/soat-2024.json").read_bytes() def test_adapter_rejects_structurally_valid_but_truncated_catalog(tmp_path: Path) -> None: payload = json.loads(Path("app/soat_tariffs/catalogs/soat-2024.json").read_text(encoding="utf-8")) payload["entries"] = payload["entries"][:1] (tmp_path / "soat-2024.json").write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(ValueError, match="1.771"): JsonSoatTariffCatalog(tmp_path).load(2024)