"""Tests for the 271 (Eligibility Benefit Response) Pydantic models. SP3 Phase 4 (T19): minimal typed shape needed by the parser (T21) and ``/api/eligibility/parse-271`` endpoint (T24). Plus the inline ``SERVICE_TYPE_CODES`` dict + ``service_type_description`` helper. """ from __future__ import annotations import json import re from datetime import date from decimal import Decimal import pytest from pydantic import ValidationError from cyclone.parsers.models import BatchSummary, Envelope from cyclone.parsers.models_271 import ( LAST_UPDATED, SERVICE_TYPE_CODES, BenefitAdditionalInfo, CoverageBenefit, InformationReceiver271, InformationSource271, ParseResult271, Patient271, Subscriber271, service_type_description, ) def _build_result() -> ParseResult271: return ParseResult271( envelope=Envelope( sender_id="PAYERID", receiver_id="PROVIDERID", control_number="000000002", transaction_date=date(2024, 1, 1), implementation_guide="005010X279A1", ), information_source=InformationSource271( name="PAYER NAME", id="SKCO0", ), information_receiver=InformationReceiver271( name="PROVIDER NAME", ), subscriber=Subscriber271( member_id="MEMBER123", first_name="JOHN", last_name="DOE", ), coverage_benefits=[ CoverageBenefit( service_type_code="1", coverage_level="IND", in_plan_network=True, authorization_required=False, plan_coverage_description="Patient has active coverage", ), CoverageBenefit( service_type_code="88", coverage_level="IND", benefit_amount=Decimal("100"), in_plan_network=False, authorization_required=True, ), ], summary=BatchSummary( input_file="minimal_271.txt", control_number="000000002", transaction_date=date(2024, 1, 1), total_claims=2, passed=2, failed=0, ), ) def test_parse_result_271_minimal_round_trip(): """Build a minimal ParseResult271 and round-trip via JSON.""" r = _build_result() blob = json.loads(r.model_dump_json()) assert blob["envelope"]["control_number"] == "000000002" assert blob["envelope"]["implementation_guide"] == "005010X279A1" assert blob["information_source"]["name"] == "PAYER NAME" assert blob["subscriber"]["first_name"] == "JOHN" assert blob["patient"] is None assert len(blob["coverage_benefits"]) == 2 # Decimal serializes as a string in Pydantic v2 by default; verify # the round-trip preserves the value (not the str type). assert blob["coverage_benefits"][1]["benefit_amount"] == "100" # Round-trip back to the model — Decimal must survive. r2 = ParseResult271.model_validate(blob) assert r2.coverage_benefits[1].benefit_amount == Decimal("100") assert r2.coverage_benefits[0].in_plan_network is True assert r2.coverage_benefits[1].authorization_required is True def test_service_type_description_known(): """``service_type_description("1")`` returns ``"Medical Care"``.""" assert service_type_description("1") == "Medical Care" # Numeric codes with leading zero should not exist (X12 omits the # leading zero) but the lookup still works for whatever the caller # hands in. assert service_type_description("88") == "Pharmacy" assert service_type_description("AL") == "Vision (Optometry)" assert service_type_description("MH") == "Mental Health" assert service_type_description("UC") == "Urgent Care" # Case-insensitive: the parser may pass "1" or any case variant. assert service_type_description("mh") == "Mental Health" assert service_type_description("uc") == "Urgent Care" def test_service_type_description_unknown_returns_none(): """Unknown codes (and the empty string) return ``None``.""" assert service_type_description("ZZZ") is None assert service_type_description("9999") is None assert service_type_description("") is None # Whitespace-only also returns None. assert service_type_description(" ") is None def test_service_type_codes_count_ge_50(): """The dict ships at least 50 service-type codes.""" assert len(SERVICE_TYPE_CODES) >= 50 # Every value must be a non-empty string. for code, label in SERVICE_TYPE_CODES.items(): assert isinstance(code, str) assert code, "empty key" assert isinstance(label, str) assert label, f"empty label for {code!r}" def test_last_updated_constant(): """LAST_UPDATED is a YYYY-MM-DD string used to track snapshot refresh.""" assert isinstance(LAST_UPDATED, str) assert re.match(r"^\d{4}-\d{2}-\d{2}$", LAST_UPDATED), ( f"LAST_UPDATED must be YYYY-MM-DD, got {LAST_UPDATED!r}" ) def test_coverage_benefit_resolves_description_on_construction(): """``CoverageBenefit`` auto-populates ``service_type_description`` from ``service_type_code`` when the caller omits it. """ cb = CoverageBenefit(service_type_code="1") assert cb.service_type_description == "Medical Care" # Caller-supplied description wins. cb2 = CoverageBenefit(service_type_code="1", service_type_description="Custom") assert cb2.service_type_description == "Custom" # Unknown code → description is None. cb3 = CoverageBenefit(service_type_code="ZZZ") assert cb3.service_type_description is None def test_coverage_benefit_round_trip_preserves_description(): """Round-tripping JSON does not overwrite a caller-supplied description.""" r = _build_result() blob = json.loads(r.model_dump_json()) # After JSON round-trip, the description we built (e.g. "Medical Care") survives. r2 = ParseResult271.model_validate(blob) assert r2.coverage_benefits[0].service_type_description == "Medical Care" assert r2.coverage_benefits[1].service_type_description == "Pharmacy" def test_required_service_type_code_on_coverage_benefit(): """``service_type_code`` is the only required field on CoverageBenefit.""" with pytest.raises(ValidationError) as exc: CoverageBenefit() # type: ignore[call-arg] assert "service_type_code" in str(exc.value) # All other fields are optional / defaulted. cb = CoverageBenefit(service_type_code="1") assert cb.coverage_level is None assert cb.benefit_amount is None assert cb.in_plan_network is None assert cb.authorization_required is None assert cb.plan_coverage_description is None assert cb.time_qualifier is None assert cb.benefit_date is None assert cb.additional_info == [] def test_benefit_additional_info_codes_default(): """``BenefitAdditionalInfo.codes`` defaults to an empty list.""" bai = BenefitAdditionalInfo(description="Patient has active coverage") assert bai.codes == [] bai2 = BenefitAdditionalInfo( description="Pharmacy copay $20", codes=["H0019", "H0020"], ) assert bai2.codes == ["H0019", "H0020"] def test_patient_271_optional_fields(): """Patient271's fields are all optional; relationship is the only one with a v1 use (HL/INS qualifiers). """ p = Patient271() blob = json.loads(p.model_dump_json()) assert blob == { "first_name": None, "last_name": None, "relationship": None, } p2 = Patient271(first_name="JANE", last_name="DOE", relationship="19") blob2 = json.loads(p2.model_dump_json()) assert blob2["relationship"] == "19" def test_service_type_codes_contains_required_keys(): """Spot-check the codes the spec calls out as required.""" required = { "1": "Medical Care", "2": "Surgical", "33": "Chiropractic", "35": "Dental Care", "47": "Hospital", "48": "Hospital - Inpatient", "50": "Hospital - Outpatient", "86": "Emergency Services", "88": "Pharmacy", "98": "Professional (Physician) Visit - Outpatient", "99": "Professional (Physician) Visit - Other", "AL": "Vision (Optometry)", "MH": "Mental Health", "UC": "Urgent Care", } for code, expected in required.items(): assert SERVICE_TYPE_CODES.get(code) == expected, ( f"missing or mismatched code {code!r}: got {SERVICE_TYPE_CODES.get(code)!r}, expected {expected!r}" )