Files
cyclone/backend/tests/test_validator.py
T
Tyler 716bfb0a26 feat(parsers): R034 REF*G1 enforcement + R035 BHT06 transaction type code (SP3 P1)
- R034 _r034_ref_g1_required: when payer opts in (require_ref_g1_for_adjustments),
  emit error if frequency_code is 7/8 and no REF*G1 segment appears in raw_segments.
  CO Medicaid stays lenient in v1 (gate is False by default).
- R035 _r035_bht06_allowed: BHT06 transaction type code must be in cfg.allowed_bht06.
  Skipped silently when transaction_type_code is None.
- Add transaction_type_code (str|None) to Envelope and ClaimOutput.
- parse_837: read BHT06 (seg[6]) in _build_envelope; mirror onto each ClaimOutput
  via the existing model_copy(update={...}) call site.
- 9 new tests in test_validator.py: 5 R034 (incl. lenient no-op) + 4 R035.
  213 passed, 1 skipped (prodfile corpus present-conditional).
2026-06-20 07:25:29 -06:00

316 lines
10 KiB
Python

from datetime import date
from decimal import Decimal
import pytest
from cyclone.parsers.models import (
Address,
BillingProvider,
ClaimHeader,
ClaimOutput,
Diagnosis,
Envelope,
Payer,
Procedure,
ServiceLine,
Subscriber,
ValidationReport,
)
from cyclone.parsers.payer import PayerConfig
from cyclone.parsers.validator import validate
def _build_claim(**overrides) -> ClaimOutput:
"""Build a passing claim for tests; override fields to break specific rules."""
base = dict(
claim_id="C1",
control_number="991102977",
transaction_date=date(2026, 6, 11),
billing_provider=BillingProvider(
name="Test Provider",
npi="1234567890",
tax_id="123456789",
address=Address(line1="1 Main", city="X", state="CO", zip="80000"),
),
subscriber=Subscriber(
first_name="John",
last_name="Doe",
member_id="M1",
dob=date(1980, 1, 1),
gender="M",
address=Address(line1="1 Main", city="X", state="CO", zip="80000"),
),
payer=Payer(name="COHCPF", id="SKCO0"),
claim=ClaimHeader(
claim_id="C1",
total_charge=Decimal("100.00"),
place_of_service="12",
frequency_code="1",
provider_signature="Y",
assignment="Y",
release_of_info="Y",
),
diagnoses=[Diagnosis(code="Z00", qualifier="ABK")],
service_lines=[
ServiceLine(
line_number=1,
procedure=Procedure(qualifier="HC", code="99213", modifiers=[]),
charge=Decimal("100.00"),
unit_type="UN",
units=Decimal("1.0"),
service_date=date(2026, 6, 11),
)
],
validation=ValidationReport(passed=True, errors=[], warnings=[]),
raw_segments=[],
)
base.update(overrides)
return ClaimOutput(**base)
def test_validate_passing_claim():
cfg = PayerConfig.co_medicaid()
report = validate(_build_claim(), cfg)
assert report.passed is True
assert report.errors == []
def test_r010_clm01_required():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.claim_id = ""
report = validate(claim, cfg)
assert not report.passed
assert any(i.rule == "R010_clm01_present" for i in report.errors)
def test_r011_total_charge_positive():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.total_charge = Decimal("0.00")
report = validate(claim, cfg)
assert any(i.rule == "R011_total_charge_positive" for i in report.errors)
def test_r020_npi_must_be_ten_digits():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.billing_provider.npi = "12345"
report = validate(claim, cfg)
assert any(i.rule == "R020_npi_format" for i in report.errors)
def test_r030_frequency_allowed():
cfg = PayerConfig.co_medicaid() # only 1, 7, 8
claim = _build_claim()
claim.claim.frequency_code = "5"
report = validate(claim, cfg)
assert any(i.rule == "R030_frequency_allowed" for i in report.errors)
def test_r031_ref_g1_optional_no_error():
"""R031 is informational in v1 — no REF*G1 should not error."""
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
report = validate(claim, cfg)
assert not any(i.rule == "R031_ref_g1_optional" and i.severity == "error" for i in report.errors)
def test_r050_diagnosis_required():
cfg = PayerConfig.co_medicaid()
claim = _build_claim(diagnoses=[])
report = validate(claim, cfg)
assert any(i.rule == "R050_diagnosis_present" for i in report.errors)
def test_r060_service_dates_required():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.service_lines[0].service_date = None
report = validate(claim, cfg)
assert any(i.rule == "R060_service_dates_present" for i in report.errors)
# Now add the date:
claim.service_lines[0].service_date = date(2026, 6, 11)
report = validate(claim, cfg)
assert not any(i.rule == "R060_service_dates_present" for i in report.errors + report.warnings)
def test_r070_charges_sum_warning():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.total_charge = Decimal("999.00") # mismatch
report = validate(claim, cfg)
assert any(i.rule == "R070_charges_sum" and i.severity == "warning" for i in report.warnings)
def test_r100_payer_id_warning_only():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.payer.id = "WRONG"
report = validate(claim, cfg)
assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings)
assert report.passed is True
def test_r032_passes_when_qualifier_b():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.facility_code_qualifier = "B"
report = validate(claim, cfg)
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
def test_r032_errors_when_qualifier_unknown():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.facility_code_qualifier = "X"
report = validate(claim, cfg)
assert any(i.rule == "R032_clm05_2_facility_qualifier" and i.severity == "error" for i in report.errors)
assert report.passed is False
def test_r032_skipped_when_qualifier_none():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
# Default _build_claim leaves facility_code_qualifier unset (None).
assert claim.claim.facility_code_qualifier is None
report = validate(claim, cfg)
assert not any(i.rule == "R032_clm05_2_facility_qualifier" for i in report.errors + report.warnings)
def test_r033_passes_with_valid_pos_code():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.place_of_service = "12" # Office — valid CMS POS
report = validate(claim, cfg)
assert not any(i.rule == "R033_clm05_1_place_of_service_code" for i in report.errors + report.warnings)
def test_r033_errors_with_invalid_pos_code():
cfg = PayerConfig.co_medicaid()
claim = _build_claim()
claim.claim.place_of_service = "999" # not a valid CMS POS code
report = validate(claim, cfg)
assert any(i.rule == "R033_clm05_1_place_of_service_code" and i.severity == "error" for i in report.errors)
assert report.passed is False
# --------------------------------------------------------------------------- #
# R034 — REF*G1 enforcement (SP3 Phase 1)
# --------------------------------------------------------------------------- #
def _strict_cfg() -> PayerConfig:
"""Strict cfg that mirrors the CO Medicaid defaults but turns on R034 enforcement.
We don't modify :meth:`PayerConfig.co_medicaid` — the lenient default stays
in v1. This local fixture lets us exercise the strict path without leaking
config changes across the suite.
"""
return PayerConfig(
name="StrictTest",
require_ref_g1_for_adjustments=True,
allowed_bht06={"CH"},
payer_id="X",
)
def test_r034_ref_g1_required_freq_7_no_ref_g1_errors():
cfg = _strict_cfg()
claim = _build_claim()
claim.claim.frequency_code = "7"
claim.raw_segments = []
report = validate(claim, cfg)
assert any(
i.rule == "R034_ref_g1_required" and i.severity == "error" for i in report.errors
)
assert report.passed is False
def test_r034_ref_g1_required_freq_8_no_ref_g1_errors():
cfg = _strict_cfg()
claim = _build_claim()
claim.claim.frequency_code = "8"
claim.raw_segments = []
report = validate(claim, cfg)
assert any(
i.rule == "R034_ref_g1_required" and i.severity == "error" for i in report.errors
)
assert report.passed is False
def test_r034_ref_g1_required_freq_7_with_ref_g1_passes():
cfg = _strict_cfg()
claim = _build_claim()
claim.claim.frequency_code = "7"
claim.raw_segments = [["REF", "G1", "12345"]]
report = validate(claim, cfg)
assert not any(
i.rule == "R034_ref_g1_required" for i in report.errors + report.warnings
)
def test_r034_ref_g1_required_freq_1_no_ref_g1_passes():
cfg = _strict_cfg()
claim = _build_claim()
claim.claim.frequency_code = "1"
claim.raw_segments = []
report = validate(claim, cfg)
assert not any(
i.rule == "R034_ref_g1_required" for i in report.errors + report.warnings
)
def test_r034_ref_g1_lenient_cfg_never_errors():
cfg = PayerConfig.co_medicaid() # require_ref_g1_for_adjustments=False (lenient v1)
claim = _build_claim()
claim.claim.frequency_code = "7"
claim.raw_segments = []
report = validate(claim, cfg)
assert not any(
i.rule == "R034_ref_g1_required" for i in report.errors + report.warnings
)
# --------------------------------------------------------------------------- #
# R035 — BHT06 transaction type code (SP3 Phase 1)
# --------------------------------------------------------------------------- #
def test_r035_bht06_allowed_ch_passes():
cfg = PayerConfig.co_medicaid()
claim = _build_claim(transaction_type_code="CH")
report = validate(claim, cfg)
assert not any(
i.rule == "R035_bht06_allowed" for i in report.errors + report.warnings
)
def test_r035_bht06_allowed_rp_errors_for_co_medicaid():
cfg = PayerConfig.co_medicaid()
claim = _build_claim(transaction_type_code="RP")
report = validate(claim, cfg)
assert any(
i.rule == "R035_bht06_allowed" and i.severity == "error" for i in report.errors
)
assert report.passed is False
def test_r035_bht06_missing_skips():
cfg = PayerConfig.co_medicaid()
claim = _build_claim(transaction_type_code=None)
assert claim.transaction_type_code is None
report = validate(claim, cfg)
assert not any(
i.rule == "R035_bht06_allowed" for i in report.errors + report.warnings
)
def test_r035_bht06_rp_allowed_for_generic_837p():
cfg = PayerConfig.generic_837p() # allows {"CH", "RP"}
claim = _build_claim(transaction_type_code="RP")
report = validate(claim, cfg)
assert not any(
i.rule == "R035_bht06_allowed" for i in report.errors + report.warnings
)