"""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()) == []