test(sp41): end-to-end smoke test for run_rebill()

2026-07-07 23:46:33 -06:00
parent 467f53046e
commit 8790e0a439
1 changed files with 172 additions and 0 deletions
@@ -0,0 +1,172 @@
+"""SP41 Task 17 — end-to-end smoke test for ``run_rebill``.
+
+Wires the full SP41 pipeline (835 SVC reparse → reconcile → CARC
+filter → timely-filing gate → pipeline A → pipeline B → summary CSV)
+end-to-end on synthetic inputs and pins the summary.csv shape +
+counts.
+
+No mocks for ``parse_835_svc`` or ``validate_837`` — the goal is to
+exercise the real pipeline. The autouse conftest handles Edifabric
+fail-open (no API key in CI), and since this run produces only
+quarantined dispositions (no pipeline A/B emissions), ``validate_edi``
+is never called anyway.
+"""
+from __future__ import annotations
+
+import csv
+from datetime import date
+from pathlib import Path
+
+from cyclone.rebill.run import run_rebill
+
+
+# --------------------------------------------------------------------------- #
+# Fixture builders
+# --------------------------------------------------------------------------- #
+
+
+def _write_visits_csv(path: Path) -> Path:
+    """Two visits: one in-window (will match a denied SVC), one past the
+    120-day timely-filing window (no matching SVC).
+
+    DOS column is MM/DD/YYYY per AxisCare's actual export format.
+    """
+    with path.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow([
+            "Visit Date", "Member ID", "Procedure Code",
+            "Billable Amount", "Authorized",
+        ])
+        w.writerow(["06/27/2026", "J813715", "T1019", "2.32", "Y"])
+        w.writerow(["01/01/2026", "OLD001", "T1019", "5.00", "Y"])
+    return path
+
+
+def _write_835_with_denied_svc(ingest_dir: Path) -> Path:
+    """Write a real 835 that ``parse_835_svc`` walks end-to-end.
+
+    Contains exactly one CLP block for member J813715 with:
+      - CLP02 = 4 (Denied)
+      - CLP03 = 2.32 (charge)
+      - SVC*HC:T1019*2.32*0** (paid = $0)
+      - DTM*472*20260627 (service date)
+      - CAS*CO*45*2.32 (triggers ``EXCLUDED_CARC``)
+
+    Returns the **directory** path so the caller can pass it straight to
+    ``run_rebill(ingest_dir=...)`` — run_rebill does its own
+    ``ingest_dir.glob("*.835")`` walk. (Earlier draft returned the
+    inner file path, which made glob come up empty.)
+
+    Shape mirrors the existing ``tests/fixtures/835_sample_svc_with_member.txt``
+    (segments concatenated without ``\\n`` separators — ``parse_835_svc``
+    splits on ``~`` only, and a leading newline makes ``elems[0]`` an empty
+    string that the segment-name match drops silently).
+    """
+    ingest_dir.mkdir(exist_ok=True)
+    segs = [
+        "ISA*00*          *00*          *ZZ*CYCLONE       *ZZ*GAINWELL     *260627*1200*^*00501*000000001*0*P*:~",
+        "GS*HC*CYCLONE*GAINWELL*20260627*1200*1*X*005010X221A1~",
+        "ST*835*0001~",
+        "BPR*I*0*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~",
+        "TRN*1*TRACE01*1512345678~",
+        "DTM*405*20260118~",
+        "N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~",
+        "N3*PO BOX 1100~",
+        "N4*DENVER*CO*80202~",
+        "LX*1~",
+        # CLP02=4 (Denied); CLP03=2.32 (charge); CLP04=0 (paid).
+        "CLP*DENIED-CLM*4*2.32*0**MC*111*11*1~",
+        # NM1*QC.NM109 carries the member_id forward to SVC rows.
+        "NM1*QC*1*DOE*JANE****MR*J813715~",
+        # SVC composite qualifier:procedure (HC:T1019); 4-arg form is fine.
+        "SVC*HC:T1019*2.32*0**~",
+        # DTM*472 carries the service date (matches parse_835_svc's lookup).
+        "DTM*472*20260627~",
+        # CO-45 is in EXCLUDED_CARCS → CarcDecision.EXCLUDED → EXCLUDED_CARC.
+        "CAS*CO*45*2.32~",
+        "SE*14*0001~",
+        "GE*1*1~",
+        "IEA*1*000000001~",
+    ]
+    p = ingest_dir / "x.835"
+    p.write_text("".join(segs))
+    return ingest_dir
+
+
+# --------------------------------------------------------------------------- #
+# Smoke test
+# --------------------------------------------------------------------------- #
+
+
+def test_run_rebill_end_to_end_excluded_dispositions(tmp_path):
+    """End-to-end: 2 visits → 2 quarantined dispositions, no pipeline files.
+
+    Pins the contract that ``run_rebill`` returns a ``RunResult`` whose
+    ``summary.csv`` row-shapes match the visit-side input (one row per
+    non-PAID visit, with the right disposition per row).
+    """
+    visits_csv = _write_visits_csv(tmp_path / "visits.csv")
+    ingest = _write_835_with_denied_svc(tmp_path / "ingest")
+    out_dir = tmp_path / "rebills"
+
+    # ``as_of=2026-07-07`` pins the timely-filing gate so the test is
+    # deterministic — OLD001 (DOS 2026-01-01) is 187 days old, well past
+    # the 120-day HCPF window, so it's EXCLUDED_TIMELY_FILING.
+    result = run_rebill(
+        window_start=date(2026, 1, 1),
+        window_end=date(2026, 6, 27),
+        override_filing=False,
+        visits_csv_path=str(visits_csv),
+        ingest_dir=str(ingest),
+        out_dir=str(out_dir),
+        as_of=date(2026, 7, 7),
+    )
+
+    # --- summary.csv exists and the right text is in it. --- #
+    assert result.summary_path.exists(), result.summary_path
+    text = result.summary_path.read_text()
+    assert "J813715" in text
+    assert "OLD001" in text
+    assert "EXCLUDED_CARC" in text
+    assert "EXCLUDED_TIMELY_FILING" in text
+
+    # --- counts match expectations (one of each excluded disposition). --- #
+    assert result.counts["EXCLUDED_CARC"] == 1, result.counts
+    assert result.counts["EXCLUDED_TIMELY_FILING"] == 1, result.counts
+    # Both visits are excluded — no pipeline emissions.
+    assert result.counts["REBILLED_A"] == 0, result.counts
+    assert result.counts["REBILLED_B"] == 0, result.counts
+
+    # --- pin the per-row shape (stronger than substring checks). --- #
+    with result.summary_path.open(newline="") as f:
+        rows = list(csv.DictReader(f))
+    assert len(rows) == 2, rows
+
+    j_row = next(r for r in rows if r["member_id"] == "J813715")
+    old_row = next(r for r in rows if r["member_id"] == "OLD001")
+
+    # J813715: matched the DENIED SVC, CARC CO-45 is in EXCLUDED_CARCS.
+    assert j_row["disposition"] == "EXCLUDED_CARC", j_row
+    assert j_row["procedure"] == "T1019", j_row
+    assert j_row["dos"] == "2026-06-27", j_row
+    assert j_row["billed"] == "2.32", j_row
+    assert "CO-45" in j_row["cas_reasons"], j_row
+
+    # OLD001: no matching SVC, DOS 187 days old → timely-filing exclusion.
+    assert old_row["disposition"] == "EXCLUDED_TIMELY_FILING", old_row
+    assert old_row["procedure"] == "T1019", old_row
+    assert old_row["dos"] == "2026-01-01", old_row
+    assert old_row["billed"] == "5.00", old_row
+    # No SVC match → empty cas_reasons column.
+    assert old_row["cas_reasons"] == "", old_row
+
+    # --- no pipeline files were emitted (both rows are excluded). --- #
+    assert result.pipeline_a_files == [], result.pipeline_a_files
+    assert result.pipeline_b_files == [], result.pipeline_b_files
+    # The pipeline dirs exist but are empty.
+    assert (out_dir / "pipeline-a").is_dir()
+    assert (out_dir / "pipeline-b").is_dir()
+    assert list((out_dir / "pipeline-a").iterdir()) == []
+    assert list((out_dir / "pipeline-b").iterdir()) == []
+    # No quarantined files either — these dispositions don't emit at all.
+    assert list((out_dir / "quarantine").iterdir()) == []