"""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"}], }