Tyler
1942a22629
Adds pure local validators for the 10-digit NPI Luhn checksum (CMS- published algorithm with the '80840' NPPES prefix) and 9-digit EIN format (rejects reserved prefixes 00/07/80-89). No NPPES round-trip, no IRS e-file lookup — catches the 99% typo case at parse time. Surface: - cyclone.npi.is_valid_npi / is_valid_tax_id / normalize_tax_id - CLI: 'cyclone validate-npi <npi>' and 'cyclone validate-tax-id <ein>' - API: GET /api/admin/validate-provider?npi=&tax_id= - Parser validator: new R021_npi_checksum rule (warning, not error, to keep test fixtures with placeholder NPIs ingestible) - minimal_837p.txt fixture NPI updated from '1234567890' to the Luhn-valid '1993999998' so strict-mode CLI parses still pass Tests: - test_npi.py — 27 cases (Luhn math, valid/invalid NPIs, EIN cases, normalize_tax_id edge cases) - test_api_validate_provider.py — 4 cases (both valid, both invalid, omitted NPI, omitted tax_id) - test_cli_validate.py — 8 cases (valid/invalid for both subcommands, exit codes, malformed inputs) - test_validator.py — 4 new R021 cases (valid Luhn silent, bad Luhn warning, skipped when format bad, skipped when NPI missing) Total: 923 tests pass.
363 lines
12 KiB
Python
363 lines
12 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)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# R021 — NPI Luhn checksum (SP20)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_r021_npi_checksum_valid_passes_silently():
|
|
"""A valid Luhn NPI (CMS-published 1234567893) yields no R021 issue."""
|
|
cfg = PayerConfig.co_medicaid()
|
|
claim = _build_claim()
|
|
claim.billing_provider.npi = "1234567893"
|
|
report = validate(claim, cfg)
|
|
assert not any(i.rule == "R021_npi_checksum" for i in report.errors + report.warnings)
|
|
|
|
|
|
def test_r021_npi_checksum_bad_luhn_is_warning():
|
|
"""An NPI that passes format but fails Luhn is a WARNING, not an error."""
|
|
cfg = PayerConfig.co_medicaid()
|
|
claim = _build_claim()
|
|
claim.billing_provider.npi = "1234567890" # right length, wrong check digit
|
|
report = validate(claim, cfg)
|
|
assert report.passed is True # WARNINGs don't fail the report
|
|
assert any(i.rule == "R021_npi_checksum" and i.severity == "warning" for i in report.warnings)
|
|
|
|
|
|
def test_r021_npi_checksum_skipped_when_format_bad():
|
|
"""When R020 already flagged the format, R021 stays silent
|
|
(avoids a duplicate 'this NPI is wrong' message)."""
|
|
cfg = PayerConfig.co_medicaid()
|
|
claim = _build_claim()
|
|
claim.billing_provider.npi = "12345" # wrong length
|
|
report = validate(claim, cfg)
|
|
# R020 errors as expected; R021 stays quiet.
|
|
assert any(i.rule == "R020_npi_format" for i in report.errors)
|
|
assert not any(i.rule == "R021_npi_checksum" for i in report.errors + report.warnings)
|
|
|
|
|
|
def test_r021_npi_checksum_skipped_when_npi_missing():
|
|
"""An empty NPI doesn't trigger R021 (only R020 would, but R020
|
|
only fires when NPI is *present* and wrong). R021 must stay silent
|
|
when NPI is empty so we don't double-fire."""
|
|
cfg = PayerConfig.co_medicaid()
|
|
claim = _build_claim()
|
|
claim.billing_provider.npi = ""
|
|
report = validate(claim, cfg)
|
|
assert not any(i.rule == "R021_npi_checksum" for i in report.errors + report.warnings)
|
|
|
|
|
|
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
|
|
)
|