tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
07ceeaf4cd402b7ef50bb4682bddc1c913b0403a
cyclone/backend/tests/test_validator_835.py
T

261 lines
9.9 KiB
Python
Raw Blame History

"""Tests for the 835 ERA validator rules."""
from datetime import date
from decimal import Decimal
from pathlib import Path
import pytest
from cyclone.parsers.models import Address, BatchSummary, Envelope
from cyclone.parsers.models_835 import (
ClaimAdjustment,
ClaimPayment,
FinancialInfo,
ParseResult835,
Payer835,
Payee835,
ReassociationTrace,
ServicePayment,
)
from cyclone.parsers.payer import PayerConfig835
from cyclone.parsers.validator_835 import validate
# --------------------------------------------------------------------------- #
# Builders
# --------------------------------------------------------------------------- #
def _build_result(**overrides) -> ParseResult835:
"""Build a passing CO Medicaid 835 result; override fields to break rules."""
financial = FinancialInfo(
handling_code="C",
paid_amount=Decimal("85.00"),
credit_debit_flag="C",
payment_method="ACH",
payer_tax_id="81-1725341",
payment_date=date(2026, 6, 17),
)
trace = ReassociationTrace(
trace_type_code="1",
trace_number="TPID001",
originating_company_id="TPID001",
payer_id="81-1725341",
)
payer = Payer835(
name="CO_TXIX",
id="7912900843",
address=Address(line1="PO BOX 11000", city="DENVER", state="CO", zip="802110000"),
)
payee = Payee835(
name="TOC, Inc.",
npi="1881068062",
address=Address(line1="1100 East Main St", city="Montrose", state="CO", zip="814014063"),
)
claim = ClaimPayment(
payer_claim_control_number="CLM001",
status_code="1",
status_label="Processed as Primary",
total_charge=Decimal("100.00"),
total_paid=Decimal("85.00"),
patient_responsibility=Decimal("15.00"),
claim_filing_indicator="MC",
original_claim_id="CLM001",
facility_type="12",
frequency_code="1",
per_diem_covered_days=None,
ref_benefit_plan=None,
service_payments=[
ServicePayment(
line_number=1,
procedure_qualifier="HC",
procedure_code="S5130",
modifiers=["U2"],
charge=Decimal("60.00"),
payment=Decimal("50.00"),
units=Decimal("7.73"),
unit_type="UN",
service_date=date(2026, 6, 2),
adjustments=[],
ref_benefit_plan=None,
),
ServicePayment(
line_number=2,
procedure_qualifier="HC",
procedure_code="S5130",
modifiers=["U2"],
charge=Decimal("40.00"),
payment=Decimal("35.00"),
units=Decimal("7.73"),
unit_type="UN",
service_date=date(2026, 6, 3),
adjustments=[],
ref_benefit_plan=None,
),
],
raw_segments=[],
)
summary = BatchSummary(input_file="f", total_claims=1, passed=1, failed=0)
result = ParseResult835(
envelope=Envelope(
sender_id="COMEDASSISTPROG",
receiver_id="TPID001",
control_number="991102984",
transaction_date=date(2026, 6, 17),
),
financial_info=financial,
trace=trace,
payer=payer,
payee=payee,
claims=[claim],
summary=summary,
)
if overrides:
result = result.model_copy(update=overrides)
return result
# --------------------------------------------------------------------------- #
# Sanity: balanced claim passes
# --------------------------------------------------------------------------- #
def test_balanced_claim_passes_all_rules():
cfg = PayerConfig835.co_medicaid_835()
report = validate(_build_result(), cfg)
assert report.passed is True
assert report.errors == []
# --------------------------------------------------------------------------- #
# BPR rules
# --------------------------------------------------------------------------- #
def test_r835_bpr01_handling_code_disallowed():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"handling_code": "Z"}))
report = validate(result, cfg)
assert any(i.rule == "R835_BPR01_handling_code_allowed" for i in report.errors)
assert report.passed is False
def test_r835_bpr02_positive():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("0.00")}))
report = validate(result, cfg)
assert any(i.rule == "R835_BPR02_positive" for i in report.errors)
def test_r835_bal_bpr_vs_clp04_unbalanced():
"""BPR02 (85.00) must equal sum of CLP04 (85.00). Off by 0.50 → error."""
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("85.50")}))
report = validate(result, cfg)
assert any(i.rule == "R835_BAL_BPR_vs_CLP04" for i in report.errors)
assert report.passed is False
def test_r835_bal_bpr_vs_clp04_within_tolerance_passes():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("85.005")}))
report = validate(result, cfg)
# 0.005 difference is below tolerance (0.01) — no error.
assert not any(i.rule == "R835_BAL_BPR_vs_CLP04" for i in report.errors)
# --------------------------------------------------------------------------- #
# CLP rules
# --------------------------------------------------------------------------- #
def test_r835_bal_clp04_vs_svc03_unbalanced():
cfg = PayerConfig835.co_medicaid_835()
# The default fixture has CLP04=85.00 and SVC03 sum = 50+35 = 85.00.
# Change one SVC payment to 40 → sum = 40+35 = 75, off by 10.
base = _build_result()
svc = base.claims[0].service_payments[0].model_copy(update={"payment": Decimal("40.00")})
result = base.model_copy(update={"claims": [base.claims[0].model_copy(update={"service_payments": [svc, base.claims[0].service_payments[1]]})]})
report = validate(result, cfg)
assert any(i.rule == "R835_BAL_CLP04_vs_SVC03" for i in report.errors)
assert "CLM001" in [i.message for i in report.errors if i.rule == "R835_BAL_CLP04_vs_SVC03"][0]
def test_r835_clp01_present():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"payer_claim_control_number": ""})])
report = validate(result, cfg)
assert any(i.rule == "R835_CLP01_present" for i in report.errors)
def test_r835_clp02_status_disallowed():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"status_code": "9", "status_label": None})])
report = validate(result, cfg)
assert any(i.rule == "R835_CLP02_status_allowed" for i in report.errors)
def test_r835_clp03_present():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"total_charge": Decimal("0.00")})])
report = validate(result, cfg)
assert any(i.rule == "R835_CLP03_present" for i in report.errors)
# --------------------------------------------------------------------------- #
# Payer / payee rules
# --------------------------------------------------------------------------- #
def test_r835_payer_tax_id_mismatch_warning():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": "99-9999999"}))
report = validate(result, cfg)
assert any(i.rule == "R835_PAYER_TAX_ID_matches" and i.severity == "warning" for i in report.warnings)
# Warning only — report still passes.
assert report.passed is True
def test_r835_payer_tax_id_skipped_when_set_empty():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": None}))
report = validate(result, cfg)
# When BPR10 is empty, the rule is skipped silently (per spec wording).
assert not any(i.rule == "R835_PAYER_TAX_ID_matches" for i in report.errors + report.warnings)
def test_r835_payer_health_plan_id_mismatch_warning():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(payer=_build_result().payer.model_copy(update={"id": "12345"}))
report = validate(result, cfg)
assert any(i.rule == "R835_PAYER_HEALTH_PLAN_ID_matches" and i.severity == "warning" for i in report.warnings)
assert report.passed is True
def test_r835_payee_npi_format():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(payee=_build_result().payee.model_copy(update={"npi": "12345"}))
report = validate(result, cfg)
assert any(i.rule == "R835_PAYEE_NPI_format" for i in report.errors)
assert report.passed is False
def test_r835_trn02_originating_company_id_present():
cfg = PayerConfig835.co_medicaid_835()
result = _build_result(trace=_build_result().trace.model_copy(update={"originating_company_id": ""}))
report = validate(result, cfg)
assert any(i.rule == "R835_REGN01_trn02_present" for i in report.errors)
# --------------------------------------------------------------------------- #
# Optional: warnings don't fail the report
# --------------------------------------------------------------------------- #
def test_warnings_do_not_set_passed_false():
cfg = PayerConfig835.co_medicaid_835()
# A wrong payer tax id is a warning; report should still pass.
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": "BAD"}))
report = validate(result, cfg)
assert report.passed is True
assert any(i.severity == "warning" for i in report.warnings)
Reference in New Issue View Git Blame Copy Permalink