tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(sp41): AppleDouble skip + visits CSV validation + CLI test coverage

Browse Source
This commit is contained in:
Nora
2026-07-07 22:50:36 -06:00
parent 30aee61956
commit cc41c73d2a
2 changed files with 311 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_cli_rebill.py
+275 -3
View File
@@ -7,15 +7,61 @@ Three smoke tests:
3. End-to-end run against a 2-row visits CSV (one in-window, one
ancient) and a zero-SVC 835 fixture produces a summary.csv with
the in-window visit classified as REBILLED_B.
Plus five disposition-coverage tests added during the Task 10 review:
4. EXCLUDED_CARC: CO-45 on a denied visit → not rebilled.
5. REBILLED_A: CO-97 on a denied visit → pipeline-a freq-7.
6. PAID visits are filtered out of the summary.
7. EXCLUDED_TIMELY_FILING vs REBILLED_B via --override-filing (two runs).
8. --status on a real summary.csv prints the per-disposition tally.
"""
from __future__ import annotations
import csv
from datetime import date as _date
from decimal import Decimal
from pathlib import Path
from unittest import mock
from click.testing import CliRunner
from cyclone.cli import main
from cyclone.rebill.parse_835_svc import SvcRow
from cyclone.rebill.reconcile import VisitRow
def _write_visits_csv(path: Path, rows: list[tuple[str, str, str, str]]) -> Path:
"""rows: list of (dos_mmddyyyy, member, procedure, billed_str)."""
with path.open("w", newline="") as f:
w = csv.writer(f)
w.writerow(["Visit Date", "Member ID", "Procedure Code", "Billable Amount"])
for r in rows:
w.writerow(r)
return path
def _read_summary(out_dir: Path) -> list[dict[str, str]]:
with (out_dir / "summary.csv").open(newline="") as f:
return list(csv.DictReader(f))
def _stub_svc(monkeypatch, svcs: list[SvcRow]) -> None:
"""Replace cli.parse_835_svc's underlying cyclonic call with a fixed list."""
# cli.rebill_from_835 imports parse_835_svc locally, so patch the
# source module's symbol — the local `from ... import parse_835_svc`
# binds the symbol at call time on each invocation.
monkeypatch.setattr(
"cyclone.rebill.parse_835_svc.parse_835_svc",
lambda path, _svcs=svcs: iter(_svcs),
)
def _make_835_placeholder(ingest_dir: Path, name: str = "x.835") -> Path:
"""A real *.835 file present in the glob; parse_835_svc is mocked so content doesn't matter."""
ingest_dir.mkdir(exist_ok=True)
p = ingest_dir / name
p.write_text("ST*835*0001~SE*0*0001~")
return p
def test_rebill_from_835_help():
@@ -74,8 +120,7 @@ def test_rebill_from_835_runs(tmp_path: Path):
summary_path = out_dir / "summary.csv"
assert summary_path.exists(), summary_path
with summary_path.open(newline="") as f:
rows = list(csv.DictReader(f))
rows = _read_summary(out_dir)
# Two visits in, two summary rows out (one REBILLED_B, one
# EXCLUDED_TIMELY_FILING). PAID outcomes would be dropped, but
# neither of these is PAID — the 835 has no SVCs.
@@ -88,4 +133,231 @@ def test_rebill_from_835_runs(tmp_path: Path):
ancient = by_member["ANCIENT"]
assert ancient["disposition"] == "EXCLUDED_TIMELY_FILING", ancient
assert ancient["dos"] == "2020-01-01", ancient
assert ancient["dos"] == "2020-01-01", ancient
# ---------------------------------------------------------------------------
# Disposition coverage (Task 10 review)
# ---------------------------------------------------------------------------
def test_rebill_carc_excluded_visit_landed_as_excluded_carc(tmp_path: Path, monkeypatch):
"""A denied visit whose CAS list contains CO-45 (excluded) must land as EXCLUDED_CARC, not REBILLED_A."""
visits = _write_visits_csv(tmp_path / "visits.csv", [
("06/27/2026", "J813715", "T1019", "$16.24"),
])
ingest = tmp_path / "ingest"
_make_835_placeholder(ingest)
out_dir = tmp_path / "out"
# A denied SVC for the visit (paid == 0) with CO-45 in CAS.
svc = SvcRow(
src_file="x.835",
claim_id="T1",
member_id="J813715",
status="4", # denied
procedure="T1019",
modifiers="",
charge=Decimal("16.24"),
paid=Decimal("0"),
units=Decimal("1"),
svc_date=_date(2026, 6, 27),
cas_reasons=("CO-45",),
pay_date=None,
)
_stub_svc(monkeypatch, [svc])
runner = CliRunner()
result = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits),
"--ingest", str(ingest),
"--out", str(out_dir),
"--as-of", "2026-07-07",
])
assert result.exit_code == 0, (result.output, result.exception)
rows = _read_summary(out_dir)
assert len(rows) == 1, rows
assert rows[0]["disposition"] == "EXCLUDED_CARC"
assert rows[0]["member_id"] == "J813715"
assert "CO-45" in rows[0]["cas_reasons"]
def test_rebill_denied_rebill_visit_landed_as_rebilled_a(tmp_path: Path, monkeypatch):
"""A denied visit whose CAS list contains CO-97 (rebillable) must land as REBILLED_A."""
visits = _write_visits_csv(tmp_path / "visits.csv", [
("06/27/2026", "J813715", "T1019", "$16.24"),
])
ingest = tmp_path / "ingest"
_make_835_placeholder(ingest)
out_dir = tmp_path / "out"
svc = SvcRow(
src_file="x.835",
claim_id="T1",
member_id="J813715",
status="4", # denied
procedure="T1019",
modifiers="",
charge=Decimal("16.24"),
paid=Decimal("0"),
units=Decimal("1"),
svc_date=_date(2026, 6, 27),
cas_reasons=("CO-97",), # rebillable (not in EXCLUDED_CARCS or REVIEW_CARCS)
pay_date=None,
)
_stub_svc(monkeypatch, [svc])
runner = CliRunner()
result = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits),
"--ingest", str(ingest),
"--out", str(out_dir),
"--as-of", "2026-07-07",
])
assert result.exit_code == 0, (result.output, result.exception)
rows = _read_summary(out_dir)
assert len(rows) == 1, rows
assert rows[0]["disposition"] == "REBILLED_A"
assert rows[0]["file_path"] == "pipeline-a/"
assert "CO-97" in rows[0]["cas_reasons"]
def test_rebill_paid_visit_excluded_from_summary(tmp_path: Path, monkeypatch):
"""PAID visits (paid >= 95% of billed) must not appear in summary.csv — only NOT_IN_835 should."""
visits = _write_visits_csv(tmp_path / "visits.csv", [
("06/27/2026", "PAID-MEMBER", "T1019", "$100.00"),
("06/27/2026", "MISSING-MEMBER", "T1019", "$2.32"),
])
ingest = tmp_path / "ingest"
_make_835_placeholder(ingest)
out_dir = tmp_path / "out"
# One fully-paid SVC for PAID-MEMBER; nothing matches MISSING-MEMBER.
paid_svc = SvcRow(
src_file="x.835",
claim_id="T1",
member_id="PAID-MEMBER",
status="1",
procedure="T1019",
modifiers="",
charge=Decimal("100.00"),
paid=Decimal("100.00"),
units=Decimal("1"),
svc_date=_date(2026, 6, 27),
cas_reasons=(),
pay_date=None,
)
_stub_svc(monkeypatch, [paid_svc])
runner = CliRunner()
result = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits),
"--ingest", str(ingest),
"--out", str(out_dir),
"--as-of", "2026-07-07",
])
assert result.exit_code == 0, (result.output, result.exception)
rows = _read_summary(out_dir)
assert len(rows) == 1, rows
assert rows[0]["member_id"] == "MISSING-MEMBER"
assert rows[0]["disposition"] == "REBILLED_B"
def test_rebill_override_filing_makes_ancient_visit_rebilled_b(tmp_path: Path, monkeypatch):
"""A 2020 visit without --override-filing → EXCLUDED_TIMELY_FILING; with it → REBILLED_B."""
# No SVCs at all — both visits fall into NOT_IN_835.
visits = _write_visits_csv(tmp_path / "visits.csv", [
("01/01/2020", "ANCIENT", "T1019", "$2.32"),
])
ingest = tmp_path / "ingest"
_make_835_placeholder(ingest)
out_dir = tmp_path / "out"
# Stub the (local) parse_835_svc import with an empty generator so
# the CLI sees no SVCs and every visit is NOT_IN_835.
monkeypatch.setattr(
"cyclone.rebill.parse_835_svc.parse_835_svc",
lambda path: iter([]),
)
runner = CliRunner()
# Run 1: as-of 2026-07-07 minus 2020-01-01 = ~6.5 years > 120 days → EXCLUDED.
r1 = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits),
"--ingest", str(ingest),
"--out", str(out_dir),
"--as-of", "2026-07-07",
])
assert r1.exit_code == 0, (r1.output, r1.exception)
rows1 = _read_summary(out_dir)
assert len(rows1) == 1, rows1
assert rows1[0]["disposition"] == "EXCLUDED_TIMELY_FILING", rows1
# Run 2: --override-filing bypasses the 120-day gate → REBILLED_B.
r2 = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits),
"--ingest", str(ingest),
"--out", str(out_dir),
"--as-of", "2026-07-07",
"--override-filing",
])
assert r2.exit_code == 0, (r2.output, r2.exception)
rows2 = _read_summary(out_dir)
assert len(rows2) == 1, rows2
assert rows2[0]["disposition"] == "REBILLED_B", rows2
assert rows2[0]["file_path"] == "pipeline-b/"
def test_rebill_status_with_real_summary_prints_tally(tmp_path: Path):
"""--status against a real summary.csv must print both REBILLED_B and EXCLUDED_TIMELY_FILING rows."""
from cyclone.rebill.reconcile import VisitRow
from cyclone.rebill.summary import (
EXCLUDED_TIMELY_FILING,
REBILLED_B,
SummaryRow,
write_summary_csv,
)
out_dir = tmp_path / "dev" / "rebills" / "2026-07-07"
out_dir.mkdir(parents=True, exist_ok=True)
summary_path = out_dir / "summary.csv"
rows = [
SummaryRow(
visit=VisitRow(
date=_date(2026, 6, 27), member_id="MEM-A",
procedure="T1019", billed=Decimal("2.32"),
),
disposition=REBILLED_B,
unpaid=Decimal("2.32"),
cas_reasons=(),
file_path="pipeline-b/",
),
SummaryRow(
visit=VisitRow(
date=_date(2026, 6, 27), member_id="MEM-B",
procedure="T1019", billed=Decimal("2.32"),
),
disposition=EXCLUDED_TIMELY_FILING,
unpaid=Decimal("2.32"),
cas_reasons=(),
file_path="",
),
]
write_summary_csv(rows, summary_path)
runner = CliRunner()
result = runner.invoke(main, [
"rebill-from-835",
"--status",
"--out", str(out_dir),
])
assert result.exit_code == 0, (result.output, result.exception)
assert "REBILLED_B" in result.output
assert "EXCLUDED_TIMELY_FILING" in result.output
assert "TOTAL" in result.output