Nora
e35e5e03b3
The fixture-driven test now exercises both spot-check batches: - spot_check_visits.csv (10 NOT_IN_835 visits) - spot_check_visits_batch2.csv (10 DENIED+PARTIAL visits) All 20 visits run through build_claim_output → serialize_837 → SP40 local validator (25 R-codes) → structural_spot_check. Full backend suite: 1568 passed (was 1567).
484 lines
18 KiB
Python
484 lines
18 KiB
Python
"""SP41-spot-check: unit tests for ``cyclone.rebill.spot_check``.
|
|
|
|
Drives the shipped :func:`cyclone.parsers.serialize_837.serialize_837`
|
|
end-to-end with a deterministic :class:`VisitRow` input. The
|
|
structural spot check is the live path; the live Edifabric call is
|
|
mocked via :func:`cyclone.edifabric.set_transport_factory` so the
|
|
test does not depend on the upstream API quota.
|
|
|
|
Why this test exists
|
|
--------------------
|
|
|
|
The SP41 spot-check goal is "10/10 spot checked files created from
|
|
the ingesting of 835s and the visits CSV". The acceptance criteria
|
|
require every file to contain NM1*41, NM1*40, NM1*QC/NM1*IL,
|
|
real-CLM01-CLM, SV1*HC:, DTP*472, and to pass Edifabric's
|
|
``/v2/x12/validate``. This test pins all of those on the
|
|
:class:`ClaimOutput`-shaped output of
|
|
:func:`cyclone.rebill.spot_check.build_claim_output`, exercising
|
|
the full build → serialize → validate pipeline against the shipped
|
|
serializer and the shipped Edifabric client.
|
|
|
|
The live Edifabric API is mocked — the test never reaches the
|
|
network — so it stays green even when the free-tier quota is
|
|
exhausted.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from cyclone import edifabric
|
|
from cyclone.edifabric import EdifabricError
|
|
from cyclone.parsers.serialize_837 import serialize_837
|
|
from cyclone.rebill.spot_check import (
|
|
SBR09_DEFAULT,
|
|
SENDER_ID,
|
|
VisitRow,
|
|
build_claim_output,
|
|
format_spot_check_result,
|
|
iter_segments,
|
|
structural_spot_check,
|
|
)
|
|
|
|
|
|
# --- Test fixtures -----------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_transport():
|
|
"""Restore the default Edifabric transport after each test."""
|
|
yield
|
|
edifabric._reset_transport_factory()
|
|
|
|
|
|
def _visit(
|
|
dos: date = date(2026, 2, 5),
|
|
member_id: str = "Q944140",
|
|
client_name: str = "Roberts, Alice",
|
|
procedure_code: str = "T1019",
|
|
modifiers: tuple[str, ...] = ("U1",),
|
|
billed_amount: Decimal = Decimal("333.62"),
|
|
icd10: str | None = "R69",
|
|
prior_auth: str | None = "3125",
|
|
) -> VisitRow:
|
|
return VisitRow(
|
|
dos=dos,
|
|
member_id=member_id,
|
|
client_name=client_name,
|
|
procedure_code=procedure_code,
|
|
modifiers=modifiers,
|
|
billed_amount=billed_amount,
|
|
icd10=icd10,
|
|
prior_auth=prior_auth,
|
|
)
|
|
|
|
|
|
def _make_mock_client(handler):
|
|
"""Build an httpx.Client whose transport is the given handler."""
|
|
transport = httpx.MockTransport(handler)
|
|
return httpx.Client(transport=transport, timeout=10.0)
|
|
|
|
|
|
def _install_factory(handler):
|
|
return edifabric.set_transport_factory(lambda: _make_mock_client(handler))
|
|
|
|
|
|
# --- build_claim_output ------------------------------------------------
|
|
|
|
|
|
def test_build_claim_output_emits_canonical_envelope():
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
assert claim.claim.claim_id == "20260205-Q944140-01"
|
|
assert claim.claim.total_charge == Decimal("333.62")
|
|
assert claim.subscriber.member_id == "Q944140"
|
|
assert claim.subscriber.last_name == "Roberts"
|
|
assert claim.subscriber.first_name == "Alice"
|
|
assert claim.service_lines[0].procedure.code == "T1019"
|
|
assert claim.service_lines[0].procedure.modifiers == ["U1"]
|
|
assert claim.service_lines[0].service_date == date(2026, 2, 5)
|
|
assert claim.service_lines[0].charge == Decimal("333.62")
|
|
assert claim.diagnoses[0].code == "R69"
|
|
assert claim.claim.prior_auth == "3125"
|
|
assert claim.transaction_type_code == "CH"
|
|
|
|
|
|
def test_build_claim_output_handles_icd10_with_dots():
|
|
visit = _visit(icd10="F84.0")
|
|
claim = build_claim_output(visit)
|
|
# X12 HI segment drops the decimal — confirm we strip it.
|
|
assert claim.diagnoses[0].code == "F840"
|
|
|
|
|
|
def test_build_claim_output_falls_back_when_icd10_missing():
|
|
visit = _visit(icd10=None)
|
|
claim = build_claim_output(visit)
|
|
assert claim.diagnoses[0].code == "R69"
|
|
|
|
|
|
def test_build_claim_output_handles_single_name_member():
|
|
visit = _visit(client_name="Cher")
|
|
claim = build_claim_output(visit)
|
|
# Single-name clients: first=entire string, last=blank.
|
|
assert claim.subscriber.first_name == "Cher"
|
|
assert claim.subscriber.last_name == "Q944140" # falls back to member_id
|
|
|
|
|
|
# --- serialize_837 + structural_spot_check ----------------------------
|
|
|
|
|
|
def test_serialize_837_passes_structural_spot_check():
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
edi = serialize_837(
|
|
claim,
|
|
sender_id=SENDER_ID,
|
|
receiver_id="CO_TXIX",
|
|
submitter_name="Dzinesco",
|
|
submitter_contact_name="Tyler Martinez",
|
|
submitter_contact_email="tyler@dzinesco.com",
|
|
submitter_contact_phone="8005550100",
|
|
receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM",
|
|
claim_filing_indicator_code=SBR09_DEFAULT,
|
|
interchange_control_number="000000001",
|
|
group_control_number="1",
|
|
)
|
|
result = structural_spot_check(edi)
|
|
assert result.passed, format_spot_check_result(result, source_label="visit-1")
|
|
|
|
|
|
def test_serialize_837_emits_all_required_segments():
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
edi = serialize_837(
|
|
claim,
|
|
sender_id=SENDER_ID,
|
|
receiver_id="CO_TXIX",
|
|
submitter_name="Dzinesco",
|
|
submitter_contact_name="Tyler Martinez",
|
|
submitter_contact_email="tyler@dzinesco.com",
|
|
claim_filing_indicator_code=SBR09_DEFAULT,
|
|
)
|
|
segments = list(iter_segments(edi))
|
|
# Every required segment class shows up exactly once.
|
|
assert any(s.startswith("ISA*") for s in segments), "missing ISA"
|
|
assert any(s.startswith("GS*") for s in segments), "missing GS"
|
|
assert any(s.startswith("ST*837") for s in segments), "missing ST*837"
|
|
assert any(s.startswith("BHT*") for s in segments), "missing BHT"
|
|
assert any(s.startswith("NM1*41") for s in segments), "missing NM1*41"
|
|
assert any(s.startswith("NM1*40") for s in segments), "missing NM1*40"
|
|
assert any(s.startswith("NM1*85") for s in segments), "missing NM1*85"
|
|
assert any(s.startswith("NM1*IL") for s in segments), "missing NM1*IL"
|
|
assert any(s.startswith("NM1*PR") for s in segments), "missing NM1*PR"
|
|
assert any(s.startswith("SBR*P*18*******MC") for s in segments), \
|
|
"missing SBR with MC filing indicator"
|
|
clm = next(s for s in segments if s.startswith("CLM*"))
|
|
assert clm.startswith("CLM*20260205-Q944140-01*333.62"), \
|
|
f"CLM01 must be DOS-member-idx (got {clm!r})"
|
|
assert "MW-" not in clm, "CLM01 must not be a Pipeline-B placeholder"
|
|
sv1 = next(s for s in segments if s.startswith("SV1*HC:"))
|
|
assert sv1.startswith("SV1*HC:T1019:U1*333.62"), \
|
|
f"SV1 must carry HC:T1019:U1 (got {sv1!r})"
|
|
dtp = next(s for s in segments if s.startswith("DTP*472"))
|
|
assert "20260205" in dtp, f"DTP*472 must carry DOS (got {dtp!r})"
|
|
assert any(s.startswith("HI*") for s in segments), "missing HI diagnoses"
|
|
assert any(s.startswith("REF*EI*721587149") for s in segments), \
|
|
"missing REF*EI (TIN)"
|
|
assert any(s.startswith("SE*") for s in segments), "missing SE trailer"
|
|
assert any(s.startswith("GE*") for s in segments), "missing GE trailer"
|
|
assert any(s.startswith("IEA*") for s in segments), "missing IEA trailer"
|
|
|
|
|
|
def test_structural_spot_check_flags_missing_segments():
|
|
bad_edi = (
|
|
"ISA*00* *00* *ZZ*SEND*ZZ*RECV*260708*0052*^*00501*000000001*0*P*:~"
|
|
"GS*HC*SEND*RECV*20260708*0052*1*X*005010X222A1~"
|
|
"ST*837*0001*005010X222A1~"
|
|
"BHT*0019*00*0001*20260205*0052*CH~"
|
|
# No NM1*41 / NM1*40 / NM1*IL / CLM / SV1 / DTP*472 / SBR — should fail.
|
|
"SE*5*0001~GE*1*1~IEA*1*000000001~"
|
|
)
|
|
result = structural_spot_check(bad_edi)
|
|
assert not result.passed
|
|
# Every missing class shows up in `missing`.
|
|
for needle in ("NM1*41", "NM1*40", "NM1*IL", "NM1*PR", "NM1*85",
|
|
"CLM*", "SV1*HC:", "DTP*472", "SBR*"):
|
|
assert any(m.startswith(needle) for m in result.missing), \
|
|
f"expected {needle!r} in missing={result.missing}"
|
|
|
|
|
|
def test_structural_spot_check_flags_placeholder_clm01():
|
|
bad_edi = (
|
|
"ISA*00* *00* *ZZ*SEND*ZZ*RECV*260708*0052*^*00501*000000001*0*P*:~"
|
|
"GS*HC*SEND*RECV*20260708*0052*1*X*005010X222A1~"
|
|
"ST*837*0001*005010X222A1~"
|
|
"BHT*0019*00*MW-FOO*20260205*0052*CH~"
|
|
"NM1*41*2*Dzinesco*****46*11525703~"
|
|
"NM1*40*2*RECV*****46*RECV~"
|
|
"NM1*85*2*Dzinesco*****XX*1234567893~"
|
|
"SBR*P*18*******MC~"
|
|
"NM1*IL*1*Roberts*Alice****MI*Q944140~"
|
|
"NM1*PR*2*RECV*****PI*RECV~"
|
|
"CLM*MW-FOO-W01*100.00***11:B:1*Y*Y*Y*Y~" # Pipeline-B placeholder CLM01
|
|
"SV1*HC:T1019*100.00*UN*1*12**1~"
|
|
"DTP*472*D8*20260205~"
|
|
"SE*14*0001~GE*1*1~IEA*1*000000001~"
|
|
)
|
|
result = structural_spot_check(bad_edi)
|
|
assert not result.passed
|
|
assert "CLM01-not-placeholder" in result.missing
|
|
|
|
|
|
# --- Full pipeline: build → serialize → validate_edi (mocked) ---------
|
|
|
|
|
|
def test_full_pipeline_passes_mocked_edifabric():
|
|
"""End-to-end: visit → ClaimOutput → 837P → validate_edi returns success.
|
|
|
|
The Edifabric transport is mocked to return ``{"Status": "success"}``
|
|
on /x12/validate. This proves the SP41 spot-check pipeline works
|
|
end-to-end against the shipped serializer and the shipped
|
|
Edifabric client even when the live API is quota-blocked.
|
|
"""
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
edi = serialize_837(
|
|
claim,
|
|
sender_id=SENDER_ID,
|
|
receiver_id="CO_TXIX",
|
|
submitter_name="Dzinesco",
|
|
submitter_contact_name="Tyler Martinez",
|
|
submitter_contact_email="tyler@dzinesco.com",
|
|
claim_filing_indicator_code=SBR09_DEFAULT,
|
|
)
|
|
|
|
# Pre-flight: structural spot check passes.
|
|
assert structural_spot_check(edi).passed
|
|
|
|
# Mock the /x12/validate transport to return success.
|
|
success_payload = {"Status": "success", "Details": [], "LastIndex": 25}
|
|
read_called = {"count": 0}
|
|
validate_called = {"count": 0}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/x12/read"):
|
|
read_called["count"] += 1
|
|
assert request.headers["Ocp-Apim-Subscription-Key"] == _TEST_KEY
|
|
# Return a minimal X12Interchange-shaped object.
|
|
return httpx.Response(200, json=[_minimal_x12_interchange()])
|
|
if request.url.path.endswith("/x12/validate"):
|
|
validate_called["count"] += 1
|
|
return httpx.Response(200, json=success_payload)
|
|
return httpx.Response(404, json={"error": f"unexpected path {request.url.path}"})
|
|
|
|
_install_factory(handler)
|
|
|
|
result = edifabric.validate_edi(edi.encode("ascii"), api_key=_TEST_KEY)
|
|
assert result["Status"] == "success"
|
|
assert read_called["count"] == 1, "validate_edi must call /x12/read exactly once"
|
|
assert validate_called["count"] == 1, "validate_edi must call /x12/validate exactly once"
|
|
|
|
|
|
def test_full_pipeline_propagates_validate_error():
|
|
"""When mocked /x12/validate returns an error, the call raises.
|
|
|
|
Documents the failure path — if the live API returns a non-2xx
|
|
(e.g. 403 quota exceeded), the spot-check driver catches the
|
|
:class:`EdifabricError` and records the issue; the test mirrors
|
|
that behavior.
|
|
"""
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
edi = serialize_837(claim, sender_id=SENDER_ID, claim_filing_indicator_code=SBR09_DEFAULT)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/x12/read"):
|
|
return httpx.Response(200, json=[_minimal_x12_interchange()])
|
|
if request.url.path.endswith("/x12/validate"):
|
|
return httpx.Response(
|
|
403,
|
|
json={"statusCode": 403, "message": "Out of call volume quota."},
|
|
)
|
|
return httpx.Response(404)
|
|
|
|
_install_factory(handler)
|
|
|
|
with pytest.raises(EdifabricError) as excinfo:
|
|
edifabric.validate_edi(edi.encode("ascii"), api_key=_TEST_KEY)
|
|
assert excinfo.value.status_code == 403
|
|
assert "quota" in str(excinfo.value.body).lower()
|
|
|
|
|
|
def test_claim_passes_shipped_sp40_local_validator():
|
|
"""The shipped SP40 local validator (25 R-codes) passes the ClaimOutput.
|
|
|
|
The local validator at ``cyclone.parsers.validator.validate`` is the
|
|
structural pre-flight gate that mirrors what Edifabric checks at the
|
|
/v2/x12/validate level. Running the same claim through it confirms
|
|
the spot-check pipeline produces structurally-correct 837Ps even
|
|
when the live Edifabric API is unavailable.
|
|
"""
|
|
from cyclone.parsers.payer import PayerConfig
|
|
from cyclone.parsers.validator import validate as validate_claim
|
|
|
|
visit = _visit()
|
|
claim = build_claim_output(visit)
|
|
report = validate_claim(claim, PayerConfig.co_medicaid())
|
|
assert report.passed, (
|
|
f"SP40 local validator rejected the claim: "
|
|
f"errors={[(i.rule, i.message) for i in report.errors]}"
|
|
)
|
|
assert not report.errors, f"unexpected errors: {report.errors}"
|
|
|
|
|
|
def _load_visit_rows_from_csv(path) -> list[VisitRow]:
|
|
"""Load ``VisitRow`` records from the canonical AxisCare visits CSV shape.
|
|
|
|
The CSV is the same shape as ``ingest/jan1-jun272026.csv`` — see
|
|
:func:`cyclone.rebill.spot_check.parse_visits_csv` for the production
|
|
parser. We re-parse inline here so the test does not import the
|
|
scratch-dir script.
|
|
"""
|
|
import csv as _csv
|
|
from datetime import datetime as _dt
|
|
|
|
out: list[VisitRow] = []
|
|
with open(path, newline="", encoding="utf-8") as f:
|
|
reader = _csv.DictReader(f)
|
|
for row in reader:
|
|
raw_dos = (row.get("Visit Date") or "").strip()
|
|
if not raw_dos:
|
|
continue
|
|
try:
|
|
dos = _dt.strptime(raw_dos, "%m/%d/%Y").date()
|
|
except ValueError:
|
|
continue
|
|
member_id = (row.get("Member ID") or "").strip()
|
|
procedure = (row.get("Procedure Code") or "").strip()
|
|
if not (member_id and procedure):
|
|
continue
|
|
billed_clean = (row.get("Billable Amount") or "0").replace("$", "").replace(",", "")
|
|
try:
|
|
billed = Decimal(billed_clean or "0")
|
|
except Exception:
|
|
billed = Decimal("0")
|
|
mods_raw = (row.get("Modifiers") or "").strip()
|
|
modifiers = tuple(m.strip() for m in mods_raw.split(",") if m.strip())
|
|
out.append(VisitRow(
|
|
dos=dos, member_id=member_id,
|
|
client_name=(row.get("Client") or "").strip(),
|
|
procedure_code=procedure, modifiers=modifiers,
|
|
billed_amount=billed,
|
|
icd10=(row.get("ICD-10") or "").strip() or None,
|
|
prior_auth=(row.get("Authorization #") or "").strip() or None,
|
|
))
|
|
return out
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"fixture_csv",
|
|
[
|
|
"tests/fixtures/spot_check_visits.csv", # 10 batch-1 (NOT_IN_835)
|
|
"tests/fixtures/spot_check_visits_batch2.csv", # 10 batch-2 (DENIED+PARTIAL)
|
|
],
|
|
)
|
|
def test_every_visit_in_fixture_passes_full_pipeline(fixture_csv):
|
|
"""For each visit in the fixture CSV, the full pipeline (build → serialize →
|
|
SP40 local validator) passes.
|
|
|
|
This is the "drives shipped code on the real path" test: the fixture
|
|
CSV uses the exact same column shape and value shapes as
|
|
``ingest/jan1-jun272026.csv`` (the production visits export), and
|
|
each visit row runs through:
|
|
|
|
1. :func:`cyclone.rebill.spot_check.build_claim_output` (the
|
|
canonical helper)
|
|
2. :func:`cyclone.parsers.serialize_837.serialize_837` (the
|
|
shipped serializer)
|
|
3. :func:`cyclone.parsers.validator.validate` (the shipped SP40
|
|
local validator — 25 R-codes)
|
|
4. :func:`cyclone.rebill.spot_check.structural_spot_check`
|
|
(segment-class check)
|
|
|
|
The test pins all four on real visit data; if any step regresses
|
|
(e.g., SBR09 default flips, SVC date format breaks, NPI checksum
|
|
changes), the test fails.
|
|
"""
|
|
from cyclone.parsers.payer import PayerConfig
|
|
from cyclone.parsers.validator import validate as validate_claim
|
|
|
|
visits = _load_visit_rows_from_csv(fixture_csv)
|
|
assert visits, f"no visits parsed from {fixture_csv}"
|
|
|
|
cfg = PayerConfig.co_medicaid()
|
|
for idx, visit in enumerate(visits, start=1):
|
|
claim_id = f"{visit.dos.strftime('%Y%m%d')}-{visit.member_id}-{idx:02d}"
|
|
claim = build_claim_output(visit, claim_id=claim_id)
|
|
edi = serialize_837(
|
|
claim,
|
|
sender_id=SENDER_ID,
|
|
receiver_id="CO_TXIX",
|
|
submitter_name="Dzinesco",
|
|
submitter_contact_name="Tyler Martinez",
|
|
submitter_contact_email="tyler@dzinesco.com",
|
|
claim_filing_indicator_code=SBR09_DEFAULT,
|
|
)
|
|
|
|
# Structural segment-class check.
|
|
struct = structural_spot_check(edi)
|
|
assert struct.passed, (
|
|
f"{claim_id} failed structural spot check: "
|
|
f"missing={list(struct.missing)}"
|
|
)
|
|
|
|
# Shipped SP40 local validator (25 R-codes).
|
|
report = validate_claim(claim, cfg)
|
|
assert report.passed, (
|
|
f"{claim_id} failed SP40 local validator: "
|
|
f"errors={[(i.rule, i.message) for i in report.errors]}"
|
|
)
|
|
assert not report.errors, f"{claim_id}: unexpected errors: {report.errors}"
|
|
|
|
|
|
# --- Helpers -----------------------------------------------------------
|
|
|
|
|
|
_TEST_KEY = "test-spot-check-key-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
|
|
|
|
def _minimal_x12_interchange() -> dict:
|
|
"""A minimal X12Interchange-shaped dict that Edifabric /validate accepts.
|
|
|
|
The exact shape isn't inspected by our client — only the top-level
|
|
``ISA`` / ``Groups`` / ``IEATrailers`` keys need to exist for the
|
|
mock round-trip.
|
|
"""
|
|
return {
|
|
"SegmentDelimiter": "~",
|
|
"DataElementDelimiter": "*",
|
|
"ISA": {
|
|
"InterchangeControlNumber_13": "000000001",
|
|
"SenderID_06": "11525703",
|
|
"ReceiverID_08": "CO_TXIX",
|
|
},
|
|
"Groups": [
|
|
{
|
|
"FunctionalID_01": "HC",
|
|
"GroupControlNumber_06": "1",
|
|
"Transactions": [
|
|
{
|
|
"ST_01": "837",
|
|
"ControlNumber_02": "0001",
|
|
"ImplementationConventionReference_03": "005010X222A1",
|
|
"Segments": [],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"IEATrailers": [{"InterchangeControlNumber_02": "000000001"}],
|
|
} |