test(sp41): parametrized test driving shipped SP40 validator on real visit fixture

2026-07-08 01:45:08 -06:00
parent ce61b36504
commit 1978e696d7
2 changed files with 109 additions and 0 deletions
@@ -336,6 +336,109 @@ def test_claim_passes_shipped_sp40_local_validator():
    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"])
+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 -----------------------------------------------------------