from __future__ import annotations import json import unittest from copy import deepcopy from datetime import UTC, datetime from hashlib import sha256 from types import SimpleNamespace from app.rips.application.errors import RipsGenerationError from app.rips.application.models import GenerateCaseRipsCommand, RipsGenerationOptions from app.rips.application.use_cases import GenerateCaseRipsUseCase from app.rips.application.validation import validate_rips_payload from app.rips.domain.models import RipsPayload from app.rips.infrastructure.repositories import MongoRipsPayloadRepository from test.fixtures.rips_cases import ( base_rips_context, base_rips_options, base_rips_v15_context, base_rips_v15_options, ) def _base_context() -> dict: return base_rips_context() def _base_options() -> dict: return base_rips_options() def _base_context_v15() -> dict: return base_rips_v15_context() def _base_options_v15() -> dict: return base_rips_v15_options() class FakeContextProvider: def __init__(self, context: dict): self.context = context self.calls = 0 def get_case_context(self, username: str, case_key: str, *, regen: bool = False) -> dict: self.calls += 1 return deepcopy(self.context) class FakeRipsRepository: def __init__(self): self.items: dict[tuple[str, str], list[dict]] = {} self.templates: dict[tuple[str, str], list[dict]] = {} def ensure_indexes(self) -> None: return None def get_latest(self, username: str, case_key: str) -> dict | None: versions = self.items.get((username, case_key), []) return deepcopy(versions[-1]) if versions else None def save_version( self, *, username: str, case_key: str, payload: dict, validation_report: dict, field_trace_map: list[dict], source_context: dict, regen_requested: bool, ) -> dict: versions = self.items.setdefault((username, case_key), []) findings = validation_report.get("findings") if isinstance(validation_report.get("findings"), list) else [] source_document_ids = { "historia_id": (source_context.get("historia") or {}).get("_id"), "quirurgico_id": (source_context.get("quirurgico") or {}).get("_id"), "factura_id": (source_context.get("factura") or {}).get("_id"), } document = { "version": len(versions) + 1, "schema_version": 2, "artifact_type": "rips_json", "payload": deepcopy(payload), "validation_report": deepcopy(validation_report), "validation_status": validation_report.get("status"), "ready_for_cuv": bool(validation_report.get("ready_for_cuv")), "finding_count": len(findings), "reject_count": len([item for item in findings if isinstance(item, dict) and item.get("severity") == "reject"]), "notify_count": len([item for item in findings if isinstance(item, dict) and item.get("severity") == "notify"]), "field_trace_map": deepcopy(field_trace_map), "field_trace_count": len(field_trace_map), "source_context": deepcopy(source_context), "source_document_ids": deepcopy(source_document_ids), "source_document_count": len([value for value in source_document_ids.values() if value]), "source_context_fingerprint": sha256( json.dumps(source_context, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") ).hexdigest(), "persistence_metadata": { "schema_version": 2, "artifact_type": "rips_json", "validation_summary": { "status": validation_report.get("status"), "ready_for_cuv": bool(validation_report.get("ready_for_cuv")), "finding_count": len(findings), "reject_count": len( [item for item in findings if isinstance(item, dict) and item.get("severity") == "reject"] ), "notify_count": len( [item for item in findings if isinstance(item, dict) and item.get("severity") == "notify"] ), }, }, "regen_requested": regen_requested, } versions.append(document) return deepcopy(document) def get_template(self, username: str, scope_candidates: list[dict]) -> dict | None: for candidate in scope_candidates: key = str(candidate.get("scope_key") or "") versions = self.templates.get((username, key), []) if versions: return deepcopy(versions[-1]) return None def save_template(self, *, username: str, scope: dict, options: dict) -> dict: key = str(scope.get("scope_key") or "") versions = self.templates.setdefault((username, key), []) document = { "template_scope": deepcopy(scope), "options": deepcopy(options), "version": len(versions) + 1, } versions.append(document) return deepcopy(document) class RipsDomainModelTest(unittest.TestCase): def test_payload_requires_tipo_nota_and_num_nota_together(self) -> None: with self.assertRaises(ValueError): RipsPayload( numDocumentoIdObligado="900123456", numFactura="FV1001", tipoNota="NC", usuarios=[ { "tipoDocumentoIdentificacion": "CC", "numDocumentoIdentificacion": "123456789", "tipoUsuario": "01", "fechaNacimiento": "1990-01-15", "codSexo": "M", "codPaisResidencia": "170", "codMunicipioResidencia": "11001", "codZonaTerritorialResidencia": "01", "incapacidad": "NO", "consecutivo": 1, "servicios": { "procedimientos": [ { "codPrestador": "900123456", "fechaInicioAtencion": "2026-04-20 07:45", "codProcedimiento": "895100", "viaIngresoServicioSalud": "01", "modalidadGrupoServicioTecSal": "01", "grupoServicios": "04", "codServicio": 1, "finalidadTecnologiaSalud": "44", "tipoDocumentoIdentificacion": "CC", "numDocumentoIdentificacion": "123456789", "codDiagnosticoPrincipal": "A080", "vrServicio": 49700, "conceptoRecaudo": "02", "valorPagoModerador": 0, "consecutivo": 1, } ] }, } ], ) class FakeMongoCollection: def __init__(self) -> None: self.documents: list[dict] = [] self.indexes: list[list[tuple[str, int]]] = [] def create_index(self, keys: list[tuple[str, int]]) -> None: self.indexes.append(keys) def find_one(self, query: dict, sort: list[tuple[str, int]] | None = None) -> dict | None: matches = [ document for document in self.documents if all(document.get(key) == value for key, value in query.items()) ] if not matches: return None if sort: for key, direction in reversed(sort): reverse = direction < 0 matches.sort(key=lambda item: item.get(key), reverse=reverse) return deepcopy(matches[0]) def insert_one(self, document: dict) -> None: self.documents.append(deepcopy(document)) class MongoRipsPayloadRepositoryTest(unittest.TestCase): def test_save_version_persists_explicit_metadata_for_traceability(self) -> None: collection = FakeMongoCollection() repository = MongoRipsPayloadRepository( mongo_analyses=SimpleNamespace(collection=collection), colombia_tz=UTC, ) payload = {"numFactura": "FV1001", "usuarios": []} validation_report = { "status": "notified", "ready_for_cuv": True, "findings": [ {"rule_code": "N1", "severity": "notify", "message": "Aviso"}, {"rule_code": "R1", "severity": "reject", "message": "Error"}, ], } field_trace_map = [{"field_path": "numFactura", "source_path": "factura.numero_factura", "transform": "copy"}] source_context = _base_context() saved = repository.save_version( username="tester", case_key="CASE-REPO-001", payload=payload, validation_report=validation_report, field_trace_map=field_trace_map, source_context=source_context, regen_requested=True, ) self.assertEqual(saved["schema_version"], 2) self.assertEqual(saved["artifact_type"], "rips_json") self.assertEqual(saved["validation_status"], "notified") self.assertTrue(saved["ready_for_cuv"]) self.assertEqual(saved["finding_count"], 2) self.assertEqual(saved["reject_count"], 1) self.assertEqual(saved["notify_count"], 1) self.assertEqual(saved["field_trace_count"], 1) self.assertEqual(saved["source_document_count"], 3) self.assertEqual(saved["source_document_ids"]["historia_id"], source_context["historia"]["_id"]) self.assertTrue(saved["source_context_fingerprint"]) self.assertEqual(saved["persistence_metadata"]["validation_summary"]["finding_count"], 2) self.assertIsInstance(saved["fecha_analisis"], datetime) def test_get_latest_returns_highest_version_for_case(self) -> None: collection = FakeMongoCollection() repository = MongoRipsPayloadRepository( mongo_analyses=SimpleNamespace(collection=collection), colombia_tz=UTC, ) source_context = _base_context() repository.save_version( username="tester", case_key="CASE-REPO-002", payload={"numFactura": "FV1001", "usuarios": []}, validation_report={"status": "approved", "ready_for_cuv": True, "findings": []}, field_trace_map=[], source_context=source_context, regen_requested=False, ) latest = repository.save_version( username="tester", case_key="CASE-REPO-002", payload={"numFactura": "FV1002", "usuarios": []}, validation_report={"status": "notified", "ready_for_cuv": True, "findings": []}, field_trace_map=[], source_context=source_context, regen_requested=True, ) loaded = repository.get_latest("tester", "CASE-REPO-002") self.assertIsNotNone(loaded) self.assertEqual(loaded["version"], 2) self.assertEqual(loaded["payload"]["numFactura"], latest["payload"]["numFactura"]) class GenerateCaseRipsUseCaseTest(unittest.TestCase): def test_generates_persisted_rips_payload_from_epicrisis_context(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=RipsGenerationOptions(**_base_options()), regen=False, persist=True, ) ) self.assertEqual(result.version, 1) self.assertEqual(result.source, "generated") self.assertEqual(provider.calls, 1) self.assertEqual(result.payload["numFactura"], "FV1001") self.assertEqual(result.payload["usuarios"][0]["servicios"]["procedimientos"][0]["codProcedimiento"], "895100") self.assertEqual( result.payload["usuarios"][0]["servicios"]["medicamentos"][0]["codTecnologiaSalud"], "44405-3", ) self.assertEqual(result.validation_report.status, "notified") self.assertTrue(result.validation_report.ready_for_cuv) self.assertEqual(result.export_status, "ready") self.assertEqual(result.export_message, "El JSON puede pasar a revisión final.") self.assertTrue(result.field_trace_map) self.assertEqual(result.field_trace_map[0].field_path, "numDocumentoIdObligado") self.assertEqual(result.payload["usuarios"][0]["incapacidad"], "NO") def test_generates_v15_hospitalizacion_and_otros_servicios(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-001", options=RipsGenerationOptions(**_base_options_v15()), regen=False, persist=False, ) ) servicios = result.payload["usuarios"][0]["servicios"] self.assertEqual(len(servicios["hospitalizacion"]), 1) self.assertEqual(servicios["hospitalizacion"][0]["fechaEgreso"], "2026-04-22 12:30") self.assertEqual(len(servicios["otrosServicios"]), 4) self.assertEqual(servicios["otrosServicios"][0]["codTecnologiaSalud"], "LAB001") self.assertEqual(servicios["otrosServicios"][2]["tipoOS"], "03") self.assertEqual(servicios["otrosServicios"][3]["nomTecnologiaSalud"], "Honorarios cirujano") notify_codes = {item.rule_code for item in result.validation_report.findings if item.severity == "notify"} self.assertNotIn("RIPS-V1-NOTIFY-005", notify_codes) def test_generates_v15_from_source_codes_and_section_defaults_without_full_catalog(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options_v15() options["other_service_catalog"] = [] result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-DEFAULTS", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) otros_servicios = result.payload["usuarios"][0]["servicios"]["otrosServicios"] self.assertEqual([item["tipoOS"] for item in otros_servicios], ["04", "04", "03", "04"]) self.assertEqual([item["codTecnologiaSalud"] for item in otros_servicios], ["LAB001", "IMG001", "106002", "HON001"]) def test_generates_v15_otros_servicios_from_explicit_other_service_section(self) -> None: context = _base_context_v15() context["factura"]["factura_json"]["servicios_procedimientos"]["otros_servicios"] = [ { "codigo_facturacion": "MAT001", "descripcion": "Jeringa descartable 10 ml", "cantidad": "2", "valor_unitario": "2500", "total": "5000", } ] provider = FakeContextProvider(context) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options_v15() options["other_service_defaults"].append( { "source_section": "otros_servicios", "tipo_os": "01", } ) options["other_service_catalog"] = [] result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-OTHER-SERVICES", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) otros_servicios = result.payload["usuarios"][0]["servicios"]["otrosServicios"] self.assertEqual(len(otros_servicios), 5) self.assertEqual(otros_servicios[-1]["tipoOS"], "01") self.assertEqual(otros_servicios[-1]["codTecnologiaSalud"], "MAT001") self.assertEqual(otros_servicios[-1]["nomTecnologiaSalud"], "Jeringa descartable 10 ml") def test_returns_cached_payload_when_regen_is_false(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) command = GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=RipsGenerationOptions(**_base_options()), regen=False, persist=True, ) first = use_case.execute(command) second = use_case.execute(command) self.assertEqual(first.version, 1) self.assertEqual(second.version, 1) self.assertEqual(second.source, "cache") self.assertEqual(provider.calls, 1) self.assertEqual(second.validation_report.status, "notified") self.assertEqual(second.export_status, "ready") self.assertTrue(second.field_trace_map) def test_regen_persists_new_version(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) base_options = RipsGenerationOptions(**_base_options()) first = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=base_options, regen=False, persist=True, ) ) second = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=base_options, regen=True, persist=True, ) ) self.assertEqual(first.version, 1) self.assertEqual(second.version, 2) self.assertEqual(second.source, "generated") self.assertEqual(second.export_status, "ready") self.assertEqual(provider.calls, 2) def test_reports_rejected_validation_when_administrative_values_are_invalid(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options() options["procedure"]["concepto_recaudo"] = "99" options["user"]["cod_municipio_residencia"] = "11" result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=RipsGenerationOptions(**options), regen=False, persist=True, ) ) self.assertEqual(result.validation_report.status, "rejected") self.assertFalse(result.validation_report.ready_for_cuv) self.assertEqual(result.export_status, "preliminary") self.assertIn("preliminar", result.export_message.lower()) self.assertTrue(any(item.rule_code == "RIPS-USR-004" for item in result.validation_report.findings)) self.assertTrue(any(item.rule_code == "RIPS-PAY-001" for item in result.validation_report.findings)) def test_normalizes_legacy_lst_si_no_codes_to_catalog_code(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options() options["user"]["incapacidad"] = "02" result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-003", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) self.assertEqual(result.payload["usuarios"][0]["incapacidad"], "NO") self.assertTrue(result.validation_report.ready_for_cuv) def test_notifies_absent_non_v1_segments(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-002", options=RipsGenerationOptions(**_base_options()), regen=False, persist=False, ) ) notify_codes = {item.rule_code for item in result.validation_report.findings if item.severity == "notify"} self.assertTrue({"RIPS-V1-NOTIFY-001", "RIPS-V1-NOTIFY-002", "RIPS-V1-NOTIFY-004"}.issubset(notify_codes)) def test_accepts_full_concepto_recaudo_catalog(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options() options["procedure"]["concepto_recaudo"] = "04" options["procedure"]["valor_pago_moderador"] = 0 result = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-004", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) self.assertFalse(any(item.rule_code == "RIPS-PAY-001" for item in result.validation_report.findings)) def test_requires_hospitalization_config_when_factura_has_hospitalization_source(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options() with self.assertRaises(RipsGenerationError): use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-002", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) def test_requires_other_service_resolution_when_defaults_and_source_codes_are_missing(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options_v15() source_context = _base_context_v15() options["other_service_catalog"] = [] options["other_service_defaults"] = [] source_context["factura"]["factura_json"]["servicios_procedimientos"]["examenes_laboratorio"][0]["codigo"] = "" source_context["factura"]["factura_json"]["servicios_procedimientos"]["imagenologia"][0]["codigo_servicio"] = "" source_context["factura"]["factura_json"]["servicios_procedimientos"]["hospitalizacion"][0]["codigo_servicio"] = "" source_context["factura"]["factura_json"]["servicios_procedimientos"]["honorarios_medicos"][0]["codigo"] = "" provider.context = source_context with self.assertRaises(RipsGenerationError): use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-003", options=RipsGenerationOptions(**options), regen=False, persist=False, ) ) def test_can_store_and_resolve_operational_template_by_provider_and_payer(self) -> None: from app.services.rips_service import RipsService repository = FakeRipsRepository() service = RipsService.__new__(RipsService) service.audit_logger = type("AuditLogger", (), {"business_event": staticmethod(lambda **_: None)})() service.repository = repository service.context_provider = FakeContextProvider(_base_context_v15()) service.generate_case_rips_use_case = None saved = service.save_case_template( "tester", "CASE-V15-TEMPLATE", options=RipsGenerationOptions(**_base_options_v15()), scope_level="provider_payer", ) loaded = service.get_case_template("tester", "CASE-V15-TEMPLATE") self.assertTrue(saved.stored) self.assertEqual(saved.scope.level, "provider_payer") self.assertEqual(saved.version, 1) self.assertEqual(loaded.scope.provider_nit, "900123456") self.assertEqual(loaded.scope.payer_name, "EPS Demo") self.assertEqual(loaded.resolved_from_level, "provider_payer") self.assertIn("hospitalization", loaded.options) def test_rejects_when_service_document_does_not_match_user(self) -> None: payload = RipsPayload( **GenerateCaseRipsUseCase( context_provider=FakeContextProvider(_base_context()), repository=FakeRipsRepository(), ).execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-005", options=RipsGenerationOptions(**_base_options()), regen=False, persist=False, ) ).payload ) mutated = payload.model_copy(deep=True) mutated.usuarios[0].servicios.procedimientos[0].numDocumentoIdentificacion = "999" report = validate_rips_payload(mutated) self.assertEqual(report.status, "rejected") self.assertTrue(any(item.rule_code == "RIPS-REL-002" for item in report.findings)) def test_rejects_when_transaction_minimum_fields_are_missing(self) -> None: payload = RipsPayload( **GenerateCaseRipsUseCase( context_provider=FakeContextProvider(_base_context()), repository=FakeRipsRepository(), ).execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-REQ024-001", options=RipsGenerationOptions(**_base_options()), regen=False, persist=False, ) ).payload ) mutated = payload.model_copy(deep=True) mutated.numDocumentoIdObligado = "" mutated.numFactura = "" report = validate_rips_payload(mutated) self.assertEqual(report.status, "rejected") self.assertFalse(report.ready_for_cuv) self.assertTrue(any(item.rule_code == "RIPS-TXN-001" for item in report.findings)) self.assertTrue(any(item.rule_code == "RIPS-TXN-002" for item in report.findings)) def test_rejects_when_invoice_and_provider_do_not_match_source_context(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) generated = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-REQ024-002", options=RipsGenerationOptions(**_base_options()), regen=False, persist=False, ) ) payload = RipsPayload(**generated.payload) mutated = payload.model_copy(deep=True) mutated.numDocumentoIdObligado = "999999999" mutated.numFactura = "FV9999" report = validate_rips_payload(mutated, source_context=_base_context()) self.assertEqual(report.status, "rejected") self.assertFalse(report.ready_for_cuv) self.assertTrue(any(item.rule_code == "RIPS-REL-010" for item in report.findings)) self.assertTrue(any(item.rule_code == "RIPS-REL-011" for item in report.findings)) def test_rejects_when_factura_relation_totals_do_not_match(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) generated = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-006", options=RipsGenerationOptions(**_base_options()), regen=False, persist=False, ) ) payload = RipsPayload(**generated.payload) source_context = _base_context() source_context["factura"]["factura_json"]["servicios_procedimientos"]["medicamentos"][0]["total"] = "99999" report = validate_rips_payload(payload, source_context=source_context) self.assertEqual(report.status, "rejected") self.assertTrue(any(item.rule_code == "RIPS-REL-015" for item in report.findings)) def test_rejects_when_source_has_hospitalization_but_payload_omits_it(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) generated = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-REQ024-003", options=RipsGenerationOptions(**_base_options_v15()), regen=False, persist=False, ) ) payload = RipsPayload(**generated.payload) mutated = payload.model_copy(deep=True) mutated.usuarios[0].servicios.hospitalizacion = [] report = validate_rips_payload(mutated, source_context=_base_context_v15()) self.assertEqual(report.status, "rejected") self.assertFalse(report.ready_for_cuv) self.assertTrue(any(item.rule_code == "RIPS-REL-019" for item in report.findings)) def test_rejects_when_other_service_estancia_has_no_hospitalization(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) generated = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-004", options=RipsGenerationOptions(**_base_options_v15()), regen=False, persist=False, ) ) payload = RipsPayload(**generated.payload) mutated = payload.model_copy(deep=True) mutated.usuarios[0].servicios.hospitalizacion = [] report = validate_rips_payload(mutated) self.assertEqual(report.status, "rejected") self.assertTrue(any(item.rule_code == "RIPS-REL-016" for item in report.findings)) def test_rejects_when_other_service_source_totals_do_not_match(self) -> None: provider = FakeContextProvider(_base_context_v15()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) generated = use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-V15-005", options=RipsGenerationOptions(**_base_options_v15()), regen=False, persist=False, ) ) payload = RipsPayload(**generated.payload) source_context = _base_context_v15() source_context["factura"]["factura_json"]["servicios_procedimientos"]["imagenologia"][0]["total"] = "99999" report = validate_rips_payload(payload, source_context=source_context) self.assertEqual(report.status, "rejected") self.assertTrue(any(item.rule_code == "RIPS-REL-021" for item in report.findings)) def test_fails_when_medication_mapping_is_missing(self) -> None: provider = FakeContextProvider(_base_context()) repository = FakeRipsRepository() use_case = GenerateCaseRipsUseCase(context_provider=provider, repository=repository) options = _base_options() options["medication_catalog"] = [] with self.assertRaises(RipsGenerationError) as error: use_case.execute( GenerateCaseRipsCommand( username="tester", case_key="CASE-001", options=RipsGenerationOptions(**options), regen=False, persist=True, ) ) self.assertIn("medication_catalog", str(error.exception)) if __name__ == "__main__": unittest.main()