diff --git a/backend/tests/test_rebill_end_to_end.py b/backend/tests/test_rebill_end_to_end.py new file mode 100644 index 0000000..b04d156 --- /dev/null +++ b/backend/tests/test_rebill_end_to_end.py @@ -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()) == []