Nora
173 lines
7.0 KiB
Python
173 lines
7.0 KiB
Python
"""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()) == []
|