from __future__ import annotations import unittest from copy import deepcopy from app.rda.application.builders import build_rda_emergency_summary, build_rda_patient_summary from app.rda.application.fhir_projection import build_rda_fhir_projection from app.rda.application.fhir_validation import ( ValidateRdaFhirProjectionUseCase, _resource_type_for_gap, _validate_bundle, _validate_composition, _validate_encounter, _validate_expected_organizations, _validate_patient, _warnings_from_projection_gaps, ) from app.rda.domain.fhir_projection import RdaFhirProjection from app.rda.domain.fhir_validation import ( FhirLikeResourceType, RDA_FHIR_PROJECTION_MATRIX, RdaValidationFinding, RdaValidationResult, ValidationSeverity, ) from app.rda.domain.models import RdaArtifact, RdaArtifactType from test.fixtures.rips_cases import base_rips_context def _projection_rules_for(artifact_type: RdaArtifactType) -> list[tuple[FhirLikeResourceType, bool, str | None]]: return [ (item.resource_type, item.required, item.activation_note) for item in RDA_FHIR_PROJECTION_MATRIX if item.artifact_type == artifact_type ] def _base_context() -> dict: context = base_rips_context() context.update( { "case_key": "CASE-001", "case_number": "CASE-001", "nombre_paciente": "Paciente Demo", "metadatos_hc": { **context.get("metadatos_hc", {}), "prestador_servicio": "Clinica Central", "motivo_consulta": "Dolor abdominal agudo", "triage": "II", "profesional_tratante": "Dra. Ana Torres", }, "medicamentos_hc_display": ["CEFALEXINA 500 MG"], "procedimientos_factura": [{"codigo_cups": "895100", "descripcion": "Sutura simple"}], "ayudas_diagnosticas": [{"nombre": "RX tórax", "tipo": "imagen"}], } ) context["factura"]["factura_json"]["proveedor"]["nombre"] = "Clinica Central" return context def _artifact_from_context(artifact_type: RdaArtifactType, context: dict) -> RdaArtifact: if artifact_type == RdaArtifactType.PATIENT: payload, section_trace, missing_fields = build_rda_patient_summary(context) else: payload, section_trace, missing_fields = build_rda_emergency_summary(context) return RdaArtifact( case_key=context["case_key"], artifact_type=artifact_type, version=1, schema_version=1, mapper_version="2026-05-rda-v1", completeness_status="partial" if missing_fields else "complete", missing_fields=missing_fields, section_trace=section_trace, payload=payload, source_context_fingerprint="fingerprint-demo", ) class RdaFhirValidationContractTest(unittest.TestCase): def test_result_is_valid_when_findings_do_not_contain_errors(self) -> None: result = RdaValidationResult.from_findings( artifact_type=RdaArtifactType.PATIENT, findings=[ RdaValidationFinding( severity=ValidationSeverity.WARNING, resource_type=FhirLikeResourceType.DOCUMENT_REFERENCE, resource_path="bundle.entry[3].resource", rule_code="RDA-FHIR-WARN-001", message="DocumentReference sigue pendiente en v1.", observed_value=None, ) ], ) self.assertTrue(result.is_valid) self.assertEqual(result.error_count, 0) self.assertEqual(result.warning_count, 1) self.assertEqual(len(result.findings), 1) def test_result_counts_are_rejected_when_summary_is_incoherent(self) -> None: with self.assertRaisesRegex(ValueError, "error_count"): RdaValidationResult( artifact_type=RdaArtifactType.EMERGENCY, is_valid=False, error_count=0, warning_count=0, findings=[ RdaValidationFinding( severity=ValidationSeverity.ERROR, resource_type=FhirLikeResourceType.ENCOUNTER, resource_path="bundle.entry[2].resource", rule_code="RDA-FHIR-ERR-001", message="Encounter es obligatorio para emergency.", observed_value=None, ) ], ) def test_patient_projection_matrix_does_not_require_encounter(self) -> None: patient_rules = _projection_rules_for(RdaArtifactType.PATIENT) resource_types = {resource_type for resource_type, _required, _note in patient_rules} self.assertIn(FhirLikeResourceType.BUNDLE, resource_types) self.assertIn(FhirLikeResourceType.COMPOSITION, resource_types) self.assertIn(FhirLikeResourceType.PATIENT, resource_types) self.assertIn(FhirLikeResourceType.ORGANIZATION, resource_types) self.assertNotIn(FhirLikeResourceType.ENCOUNTER, resource_types) def test_emergency_projection_matrix_requires_encounter_and_marks_practitioner_optional(self) -> None: emergency_rules = _projection_rules_for(RdaArtifactType.EMERGENCY) required_by_type = {resource_type: required for resource_type, required, _note in emergency_rules} note_by_type = {resource_type: note for resource_type, _required, note in emergency_rules} self.assertTrue(required_by_type[FhirLikeResourceType.ENCOUNTER]) self.assertFalse(required_by_type[FhirLikeResourceType.PRACTITIONER]) self.assertIn("responsable", note_by_type[FhirLikeResourceType.PRACTITIONER] or "") def test_findings_are_kept_as_flat_list(self) -> None: findings = [ RdaValidationFinding( severity=ValidationSeverity.ERROR, resource_type=FhirLikeResourceType.COMPOSITION, resource_path="bundle.entry[0].resource", rule_code="RDA-FHIR-ERR-010", message="Composition debe referenciar al Patient.", observed_value={"subject": None}, ), RdaValidationFinding( severity=ValidationSeverity.WARNING, resource_type=FhirLikeResourceType.DOCUMENT_REFERENCE, resource_path="bundle.entry[4].resource", rule_code="RDA-FHIR-WARN-010", message="DocumentReference no fue proyectado en v1.", observed_value=None, ), ] result = RdaValidationResult.from_findings( artifact_type=RdaArtifactType.EMERGENCY, findings=findings, ) self.assertEqual([item.rule_code for item in result.findings], ["RDA-FHIR-ERR-010", "RDA-FHIR-WARN-010"]) self.assertEqual(result.error_count, 1) self.assertEqual(result.warning_count, 1) self.assertFalse(result.is_valid) class RdaFhirProjectionTest(unittest.TestCase): def test_patient_projection_builds_bundle_composition_patient_and_two_organizations(self) -> None: artifact = _artifact_from_context(RdaArtifactType.PATIENT, _base_context()) projection = build_rda_fhir_projection(artifact) self.assertIsInstance(projection, RdaFhirProjection) self.assertIsNone(projection.encounter) self.assertIsNone(projection.practitioner) self.assertIsNone(projection.document_reference) self.assertEqual(projection.patient.identifier_value, "123456789") self.assertEqual([item.role for item in projection.organizations], ["provider", "payer"]) self.assertEqual( [entry.reference.reference for entry in projection.bundle.entries], [ f"Patient/{projection.patient.id}", f"Composition/{projection.composition.id}", f"Organization/{projection.organizations[0].id}", f"Organization/{projection.organizations[1].id}", ], ) self.assertEqual(projection.composition.subject.reference, f"Patient/{projection.patient.id}") self.assertIn("document_reference:not_projected_in_v1", projection.projection_gaps) self.assertIn("diagnoses:no_condition_resource_v1", projection.projection_gaps) def test_patient_projection_can_omit_organizations_when_provider_and_payer_are_missing(self) -> None: context = _base_context() context["factura"]["factura_json"]["proveedor"]["nombre"] = "" context["factura"]["factura_json"]["proveedor"]["nit"] = "" context["factura"]["factura_json"]["pagador"]["aseguradora_eps"] = "" context["factura"]["factura_json"]["pagador"]["nit_pagador"] = "" context["metadatos_hc"]["prestador_servicio"] = "" projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.PATIENT, context)) self.assertEqual(projection.organizations, []) self.assertEqual(projection.composition.organization_references, []) self.assertEqual(len(projection.bundle.entries), 2) def test_emergency_projection_builds_encounter_and_practitioner_when_source_exists(self) -> None: artifact = _artifact_from_context(RdaArtifactType.EMERGENCY, _base_context()) projection = build_rda_fhir_projection(artifact) self.assertIsNotNone(projection.encounter) self.assertIsNotNone(projection.practitioner) self.assertEqual(projection.encounter.subject.reference, f"Patient/{projection.patient.id}") self.assertEqual(projection.composition.subject.reference, f"Patient/{projection.patient.id}") self.assertEqual( projection.composition.encounter.reference if projection.composition.encounter else None, f"Encounter/{projection.encounter.id}", ) self.assertEqual(projection.practitioner.name, "Dra. Ana Torres") self.assertEqual([item.role for item in projection.organizations], ["provider", "payer"]) self.assertIsNone(projection.document_reference) self.assertNotIn("practitioner:missing_source_data", projection.projection_gaps) def test_emergency_projection_keeps_practitioner_absence_as_gap_without_breaking_projection(self) -> None: context = _base_context() context["metadatos_hc"]["profesional_tratante"] = "" context["metadatos_hc"]["medico_responsable"] = "" projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.EMERGENCY, context)) self.assertIsNotNone(projection.encounter) self.assertIsNone(projection.practitioner) self.assertIsNone(projection.encounter.practitioner_reference) self.assertIsNone(projection.composition.practitioner_reference) self.assertIn("practitioner:missing_source_data", projection.projection_gaps) self.assertIn("document_reference:not_projected_in_v1", projection.projection_gaps) class ValidateRdaFhirProjectionUseCaseTest(unittest.TestCase): def setUp(self) -> None: self.use_case = ValidateRdaFhirProjectionUseCase() def test_patient_validates_with_warnings_but_without_errors(self) -> None: artifact = _artifact_from_context(RdaArtifactType.PATIENT, _base_context()) result = self.use_case.execute(artifact) self.assertTrue(result.is_valid) self.assertEqual(result.error_count, 0) self.assertGreaterEqual(result.warning_count, 1) warning_codes = [item.rule_code for item in result.findings if item.severity == ValidationSeverity.WARNING] self.assertTrue(all(code.startswith("RDA-FHIR-WARN-") for code in warning_codes)) self.assertTrue(any(item.observed_value == "document_reference:not_projected_in_v1" for item in result.findings)) def test_patient_without_identifier_type_or_number_produces_error(self) -> None: context = _base_context() context["metadatos_hc"]["datos_identificacion_paciente"] = "" context["factura"]["factura_json"]["informacion_paciente"]["numero_identificacion"] = "" context["historia"]["patient_id"] = "" result = self.use_case.execute(_artifact_from_context(RdaArtifactType.PATIENT, context)) self.assertFalse(result.is_valid) self.assertTrue(any(item.rule_code == "RDA-FHIR-ERR-003" for item in result.findings)) def test_emergency_validates_encounter_and_composition_references(self) -> None: artifact = _artifact_from_context(RdaArtifactType.EMERGENCY, _base_context()) result = self.use_case.execute(artifact) self.assertTrue(result.is_valid) self.assertFalse(any(item.rule_code == "RDA-FHIR-ERR-004" for item in result.findings)) self.assertTrue(any(item.observed_value == "document_reference:not_projected_in_v1" for item in result.findings)) def test_emergency_without_professional_emits_warning_not_error(self) -> None: context = _base_context() context["metadatos_hc"]["profesional_tratante"] = "" context["metadatos_hc"]["medico_responsable"] = "" result = self.use_case.execute(_artifact_from_context(RdaArtifactType.EMERGENCY, context)) self.assertTrue(result.is_valid) self.assertFalse(any(item.rule_code == "RDA-FHIR-ERR-006" for item in result.findings)) self.assertTrue(any(item.observed_value == "practitioner:missing_source_data" for item in result.findings)) def test_expected_organization_roles_are_derived_from_payload(self) -> None: artifact = _artifact_from_context(RdaArtifactType.PATIENT, _base_context()) result = self.use_case.execute(artifact) self.assertTrue(result.is_valid) self.assertFalse(any(item.rule_code == "RDA-FHIR-ERR-005" for item in result.findings)) class RdaFhirValidationHelpersTest(unittest.TestCase): def test_validate_bundle_reports_missing_patient_or_composition(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.PATIENT, _base_context())) broken_entries = projection.bundle.entries[1:] broken_projection = projection.model_copy( update={"bundle": projection.bundle.model_copy(update={"entries": broken_entries})} ) findings = _validate_bundle(broken_projection) self.assertEqual([item.rule_code for item in findings], ["RDA-FHIR-ERR-001"]) self.assertEqual(findings[0].resource_path, "bundle.entries") self.assertIn(f"Patient/{projection.patient.id}", findings[0].observed_value) def test_validate_composition_reports_missing_patient_subject(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.PATIENT, _base_context())) broken_composition = projection.composition.model_copy(update={"subject": projection.bundle.entries[1].reference}) broken_projection = projection.model_copy(update={"composition": broken_composition}) findings = _validate_composition(broken_projection) self.assertTrue(any(item.rule_code == "RDA-FHIR-ERR-002" for item in findings)) self.assertTrue(any(item.resource_path == "composition.subject.reference" for item in findings)) def test_validate_composition_reports_missing_encounter_reference_for_emergency(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.EMERGENCY, _base_context())) broken_composition = projection.composition.model_copy(update={"encounter": None}) broken_projection = projection.model_copy(update={"composition": broken_composition}) findings = _validate_composition(broken_projection) self.assertTrue(any(item.rule_code == "RDA-FHIR-ERR-004" for item in findings)) self.assertTrue(any(item.resource_path == "composition.encounter.reference" for item in findings)) def test_validate_composition_reports_missing_organization_references(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.PATIENT, _base_context())) broken_composition = projection.composition.model_copy(update={"organization_references": []}) broken_projection = projection.model_copy(update={"composition": broken_composition}) findings = _validate_composition(broken_projection) self.assertTrue(any(item.rule_code == "RDA-FHIR-ERR-005" for item in findings)) self.assertTrue(any(item.resource_path == "composition.organization_references" for item in findings)) def test_validate_composition_reports_missing_practitioner_reference(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.EMERGENCY, _base_context())) broken_composition = projection.composition.model_copy(update={"practitioner_reference": None}) broken_projection = projection.model_copy(update={"composition": broken_composition}) findings = _validate_composition(broken_projection) self.assertTrue(any(item.rule_code == "RDA-FHIR-ERR-006" for item in findings)) self.assertTrue(any(item.resource_path == "composition.practitioner_reference.reference" for item in findings)) def test_validate_patient_reports_missing_identifier_fields(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.PATIENT, _base_context())) broken_patient = projection.patient.model_copy(update={"identifier_type": None, "identifier_value": None}) broken_projection = projection.model_copy(update={"patient": broken_patient}) findings = _validate_patient(broken_projection) self.assertEqual([item.rule_code for item in findings], ["RDA-FHIR-ERR-003"]) self.assertEqual(findings[0].resource_path, "patient.identifier") def test_validate_encounter_reports_wrong_subject_reference(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.EMERGENCY, _base_context())) broken_encounter = projection.encounter.model_copy(update={"subject": projection.bundle.entries[1].reference}) broken_projection = projection.model_copy(update={"encounter": broken_encounter}) findings = _validate_encounter(broken_projection) self.assertEqual([item.rule_code for item in findings], ["RDA-FHIR-ERR-004"]) self.assertEqual(findings[0].resource_path, "encounter.subject.reference") def test_validate_expected_organizations_reports_missing_role_projection(self) -> None: artifact = _artifact_from_context(RdaArtifactType.PATIENT, _base_context()) projection = build_rda_fhir_projection(artifact) broken_projection = projection.model_copy(update={"organizations": [projection.organizations[0]]}) findings = _validate_expected_organizations(artifact, broken_projection) self.assertEqual([item.rule_code for item in findings], ["RDA-FHIR-ERR-005"]) self.assertEqual(findings[0].resource_path, "organizations") self.assertEqual(findings[0].observed_value, ["payer"]) def test_gap_warnings_cover_document_reference_practitioner_and_clinical_gaps(self) -> None: projection = build_rda_fhir_projection(_artifact_from_context(RdaArtifactType.EMERGENCY, _base_context())) projection = projection.model_copy( update={ "projection_gaps": [ "document_reference:not_projected_in_v1", "practitioner:missing_source_data", "diagnoses:no_condition_resource_v1", "procedures:no_procedure_resource_v1", "medications:no_medication_resource_v1", "diagnostic_support:no_diagnostic_report_resource_v1", "clinical_background:no_explicit_resource_v1", ] } ) findings = _warnings_from_projection_gaps(projection) self.assertEqual(len(findings), 7) self.assertTrue(all(item.severity == ValidationSeverity.WARNING for item in findings)) self.assertEqual(findings[0].resource_type, FhirLikeResourceType.DOCUMENT_REFERENCE) self.assertEqual(findings[1].resource_type, FhirLikeResourceType.PRACTITIONER) self.assertEqual(findings[2].resource_type, FhirLikeResourceType.COMPOSITION) self.assertEqual(findings[3].resource_type, FhirLikeResourceType.ENCOUNTER) self.assertTrue(all(item.rule_code.startswith("RDA-FHIR-WARN-") for item in findings)) def test_resource_type_for_gap_maps_current_gap_taxonomy(self) -> None: self.assertEqual(_resource_type_for_gap("document_reference:not_projected_in_v1"), FhirLikeResourceType.DOCUMENT_REFERENCE) self.assertEqual(_resource_type_for_gap("practitioner:missing_source_data"), FhirLikeResourceType.PRACTITIONER) self.assertEqual(_resource_type_for_gap("diagnoses:no_condition_resource_v1"), FhirLikeResourceType.COMPOSITION) self.assertEqual(_resource_type_for_gap("clinical_background:no_explicit_resource_v1"), FhirLikeResourceType.COMPOSITION) self.assertEqual(_resource_type_for_gap("procedures:no_procedure_resource_v1"), FhirLikeResourceType.ENCOUNTER) self.assertEqual(_resource_type_for_gap("medications:no_medication_resource_v1"), FhirLikeResourceType.ENCOUNTER) self.assertEqual(_resource_type_for_gap("diagnostic_support:no_diagnostic_report_resource_v1"), FhirLikeResourceType.ENCOUNTER)