"""SP9 — R200-R210 validation rule tests. These tests use the in-code PAYER_FACTORIES (which include the live payer_config blocks loaded from YAML) so the rules have a non-empty cfg_block to read. The live ``cyclone.payers`` registry is the source of truth for CO_MAP; the rules fall back to the in-code PayerConfig when the registry is empty. """ from __future__ import annotations from datetime import date from decimal import Decimal import pytest from cyclone import db as db_mod from cyclone import payers as payer_loader from cyclone.parsers.models import ( Address, BillingProvider, ClaimHeader, ClaimOutput, Diagnosis, Payer, Procedure, ServiceLine, Subscriber, ValidationReport, ) from cyclone.parsers.payer import PayerConfig from cyclone.parsers.validator import validate from cyclone.store import store @pytest.fixture(autouse=True) def _seed_db(tmp_path, monkeypatch): """Seed an in-memory DB with 3 providers + CO_TXIX payer, load YAML.""" monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") db_mod._reset_for_tests() db_mod.init_db() payer_loader.load_payer_configs() store.ensure_clearhouse_seeded() yield db_mod._reset_for_tests() payer_loader.reset() def _co_txix_cfg() -> PayerConfig: """Build a PayerConfig matching the live CO_TXIX YAML block.""" return PayerConfig( name="Colorado Medical Assistance Program", sbr09_claim_filing="MC", allowed_claim_frequencies={1, 7, 8}, require_ref_g1_for_adjustments=False, allowed_bht06={"CH", "RP"}, payer_id="CO_TXIX", payer_name="COHCPF", no_patient_loop=True, encounter_claim_in_same_batch=False, ) def _claim(*, npi="1881068062", sbr09=None, prv_code=None, payer_id="CO_TXIX", payer_name="COHCPF", ref_ei="721587149", transaction_type_code="CH", has_pwk=False) -> ClaimOutput: raw = [] if sbr09 is not None: raw.append(["SBR", "P", "18", "", "", "", "", "", "", sbr09]) if prv_code is not None: raw.append(["PRV", "BI", "PXC", f"PXC{prv_code}"]) raw.append(["NM1", "85", "2", "TOC, Inc.", "", "", "", "", "XX", npi]) if ref_ei is not None: raw.append(["REF", "EI", ref_ei]) if has_pwk: raw.append(["PWK"]) return ClaimOutput( claim_id="CLM1", control_number="0001", transaction_date=date(2026, 6, 20), billing_provider=BillingProvider( name="TOC, Inc.", npi=npi, tax_id=ref_ei, address=Address(line1="1100 East Main St", city="Montrose", state="CO", zip="81401"), ), subscriber=Subscriber(first_name="John", last_name="Doe", member_id="M1"), payer=Payer(name=payer_name, id=payer_id), claim=ClaimHeader( claim_id="CLM1", total_charge=Decimal("100.00"), frequency_code="1", place_of_service="11", facility_code_qualifier="B", ), transaction_type_code=transaction_type_code, diagnoses=[Diagnosis(code="R69")], service_lines=[ ServiceLine( line_number=1, procedure=Procedure(qualifier="HC", code="99213"), charge=Decimal("100.00"), units=Decimal("1"), ), ], raw_segments=raw, validation=ValidationReport(passed=True, errors=[], warnings=[]), ) # ----- R200: BHT06 allowed ------------------------------------------------ def test_r200_passes_for_allowed_bht06(): c = _claim(transaction_type_code="CH") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule.startswith("R200")] assert bad == [] def test_r200_fails_for_disallowed_bht06(): c = _claim(transaction_type_code="XX") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R200_bht06_allowed"] assert len(bad) == 1 assert "not in" in bad[0].message def test_r200_skips_when_no_bht06(): c = _claim(transaction_type_code="") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R200_bht06_allowed"] # ----- R201: BHT06 no mixed batch ------------------------------------------ def test_r201_is_a_per_claim_noop(): # Batch-level rule; per-claim the rule yields nothing c = _claim(transaction_type_code="CH") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R201_bht06_no_mixed_batch"] # ----- R202: SBR09 allowed ------------------------------------------------ def test_r202_passes_for_mc(): c = _claim(sbr09="MC") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R202_sbr09_allowed"] def test_r202_fails_for_unknown(): c = _claim(sbr09="AB") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R202_sbr09_allowed"] assert len(bad) == 1 def test_r202_passes_for_zz_mco_encounter(): c = _claim(sbr09="ZZ") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R202_sbr09_allowed"] # ----- R203: PRV matches provider ---------------------------------------- def test_r203_passes_when_prv_matches_provider_taxonomy(): c = _claim(prv_code="251E00000X") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R203_prv_matches_provider"] def test_r203_fails_when_prv_mismatch(): c = _claim(prv_code="999Z00000X") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R203_prv_matches_provider"] assert len(bad) == 1 # ----- R204: NPI in providers table -------------------------------------- def test_r204_passes_for_known_npi(): c = _claim(npi="1881068062") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R204_npi_in_providers_table"] def test_r204_fails_for_unknown_npi(): c = _claim(npi="9999999999") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R204_npi_in_providers_table"] assert len(bad) == 1 # ----- R205: REF*EI matches provider -------------------------------------- def test_r205_passes_when_ref_ei_matches(): c = _claim(ref_ei="721587149") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R205_ref_ei_matches_provider"] def test_r205_fails_when_ref_ei_mismatch(): c = _claim(ref_ei="000000000") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R205_ref_ei_matches_provider"] assert len(bad) == 1 # ----- R206: Payer ID matches --------------------------------------------- def test_r206_passes_for_co_txix(): c = _claim(payer_id="CO_TXIX") report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R206_payer_id_matches"] def test_r206_fails_for_wrong_payer_id(): c = _claim(payer_id="ZZZZZZ") report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R206_payer_id_matches"] assert len(bad) == 1 # ----- R207: no PWK segment ---------------------------------------------- def test_r207_passes_when_no_pwk(): c = _claim(has_pwk=False) report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R207_no_pwk_segment"] def test_r207_fails_when_pwk_present(): c = _claim(has_pwk=True) report = validate(c, _co_txix_cfg()) bad = [i for i in report.errors if i.rule == "R207_no_pwk_segment"] assert len(bad) == 1 # ----- R208: 2320 CAS*PI* group ------------------------------------------ def test_r208_is_a_per_claim_noop(): c = _claim() report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule == "R208_cas_2320_group"] # ----- R209, R210: filename rules (validate at submit / parse time) ---- def test_r209_r210_per_claim_noop(): # These run at file-routing time, not per-claim validate c = _claim() report = validate(c, _co_txix_cfg()) assert not [i for i in report.errors if i.rule in ("R209_outbound_filename", "R210_inbound_filename")]