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

fix(sp41): trailing newline + unused imports + defensive filename handling + a/b test coverage

Browse Source
This commit is contained in:
Nora
2026-07-07 23:32:40 -06:00
parent b0d03bc10e
commit b0ad0c27b8
2 changed files with 218 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/rebill/run.py
+41 -12
View File
@@ -255,12 +255,24 @@ def run_rebill(
qp = quarantine_dir / f"{claim.claim_id}.837"
qp.write_bytes(body)
else:
base = build_outbound_filename(tpid, "837P")
# Disambiguate same-millisecond collisions across claims.
stem, dot_ext = base.rsplit(".", 1)
p = a_dir / f"{stem}-{claim.claim_id}.{dot_ext}"
p.write_bytes(body)
a_files.append(p)
try:
base = build_outbound_filename(tpid, "837P")
# Disambiguate same-millisecond collisions across claims.
stem, dot_ext = base.rsplit(".", 1)
p = a_dir / f"{stem}-{claim.claim_id}.{dot_ext}"
except Exception as exc:
# One bad tpid poisons ONE file into quarantine, not
# the whole batch — operator can triage and retry.
_log.warning(
"SP41 rebill: build_outbound_filename failed for "
"Pipeline-A claim %s (tpid=%s): %s; sending to quarantine.",
claim.claim_id, tpid, exc,
)
qp = quarantine_dir / f"{claim.claim_id}.837"
qp.write_bytes(body)
else:
p.write_bytes(body)
a_files.append(p)
# 6) Build + emit Pipeline B batches. The Task 14 overload
# ``serialize_member_week_batch`` takes a ``MemberWeekBatch`` and
@@ -289,11 +301,23 @@ def run_rebill(
qp = quarantine_dir / f"{batch_key}.837"
qp.write_bytes(body)
else:
base = build_outbound_filename(tpid, "837P")
stem, dot_ext = base.rsplit(".", 1)
p = b_dir / f"{stem}-{batch_key}.{dot_ext}"
p.write_bytes(body)
b_files.append(p)
try:
base = build_outbound_filename(tpid, "837P")
stem, dot_ext = base.rsplit(".", 1)
p = b_dir / f"{stem}-{batch_key}.{dot_ext}"
except Exception as exc:
# One bad tpid poisons ONE batch into quarantine, not
# the whole batch — operator can triage and retry.
_log.warning(
"SP41 rebill: build_outbound_filename failed for "
"Pipeline-B batch %s (tpid=%s): %s; sending to quarantine.",
batch_key, tpid, exc,
)
qp = quarantine_dir / f"{batch_key}.837"
qp.write_bytes(body)
else:
p.write_bytes(body)
b_files.append(p)
# 7) Summary CSV — one row per non-PAID visit, regardless of pipeline.
summary_path = out_dir_p / "summary.csv"
@@ -383,7 +407,12 @@ def _validate_or_skip(body: bytes, *, claim_id: str) -> str:
Returns:
"ok" — Edifabric reports ``Status in {"success", "warning"}``;
emit to the pipeline dir.
emit to the pipeline dir. Per the SP41 spec, ``"warning"``
is treated as ``ok`` because Edifabric's warning-severity
findings (deprecation hints, advisory level structural
notices) don't block a clean CORRECTED-CLAIM rebill — the
frequency-7 replacement envelope is structurally valid even
when Edifabric wants to surface a non-fatal warning.
"quarantine" — Edifabric reports ``Status == "error"``; emit
to ``<out>/quarantine/`` for operator triage.
"skip" — Edifabric was unreachable (no API key, network error,
backend/tests/test_rebill_run.py
+176 -3
View File
@@ -24,9 +24,7 @@ from decimal import Decimal
from pathlib import Path
from unittest.mock import patch
import pytest
from cyclone.rebill.run import RunResult, run_rebill
from cyclone.rebill.run import run_rebill
# --------------------------------------------------------------------------- #
@@ -326,3 +324,178 @@ def test_run_rebill_emits_real_files_for_pipeline_a(tmp_path):
# Quarantine empty.
q_files = list((out_dir / "quarantine").glob("*.837"))
assert q_files == [], q_files
def test_run_rebill_quarantines_pipeline_a_on_edifabric_error(tmp_path):
"""Pipeline A path on Edifabric Status='error' → quarantine, NOT
pipeline-a/.
The existing ``test_run_rebill_quarantines_on_edifabric_error`` uses
an empty 835 so all visits land in Pipeline B; this one builds a
real 835 with a denied SVC matching the visit so the claim routes
to Pipeline A, then asserts the rejected file lands in
``quarantine/`` and ``pipeline-a/`` stays empty.
"""
svc_date = date(2026, 6, 27)
visits_path = _write_visits_csv(tmp_path / "visits.csv", [
(svc_date.strftime("%m/%d/%Y"), "J813715", "T1019", "$100.00"),
])
ingest_dir = tmp_path / "ingest"
_stub_835_with_denied_svc(
ingest_dir,
member="J813715",
procedure="T1019",
svc_date=svc_date,
claim_id="ORIG-ERROR-1",
carc_reason="29", # CO-29 → REBILL (not EXCLUDED)
)
out_dir = tmp_path / "out"
fake_result = {
"Status": "error",
"Details": [
{"SegmentId": "PER", "Message": "PER-04 is required",
"Status": "error"},
],
"LastIndex": 5,
}
with patch(
"cyclone.rebill.run._edifabric.validate_edi",
return_value=fake_result,
) as mock_validate:
result = run_rebill(
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
override_filing=False,
visits_csv_path=str(visits_path),
ingest_dir=str(ingest_dir),
out_dir=str(out_dir),
)
# Pipeline-A path rejected → 0 files in pipeline-a/.
assert result.pipeline_a_files == [], (
f"Pipeline A rejected file should NOT be in pipeline-a/, got "
f"{[str(p) for p in result.pipeline_a_files]}"
)
a_files = list((out_dir / "pipeline-a").glob("*"))
assert a_files == [], a_files
# File quarantined under the original claim_id key.
q_files = list((out_dir / "quarantine").glob("*.837"))
assert len(q_files) == 1, q_files
assert "ORIG-ERROR-1" in q_files[0].name, q_files[0].name
assert q_files[0].stat().st_size > 0, "quarantined file must be non-empty"
# Gate was called for the one Pipeline-A claim.
assert mock_validate.call_count == 1
def test_run_rebill_mixed_pipelines_route_correctly(tmp_path):
"""Mixed-pipeline run: one DENIED visit (Pipeline A) + one
NOT_IN_835 visit (Pipeline B) in the same window.
Asserts:
- pipeline_a_files has exactly 1 entry (the denied visit)
- pipeline_b_files has exactly 1 entry (the unmatched visit)
- summary.csv has 2 rows (one per non-PAID visit)
"""
svc_date = date(2026, 6, 27)
# Visit 1: matches a denied SVC → Pipeline A.
# Visit 2: different member entirely → NOT_IN_835 → Pipeline B.
visits_path = _write_visits_csv(tmp_path / "visits.csv", [
(svc_date.strftime("%m/%d/%Y"), "J813715", "T1019", "$100.00"),
(svc_date.strftime("%m/%d/%Y"), "J999999", "T1019", "$2.32"),
])
ingest_dir = tmp_path / "ingest"
_stub_835_with_denied_svc(
ingest_dir,
member="J813715",
procedure="T1019",
svc_date=svc_date,
claim_id="ORIG-MIX-1",
carc_reason="29",
)
out_dir = tmp_path / "out"
fake_result = {"Status": "success", "Details": [], "LastIndex": 30}
with patch(
"cyclone.rebill.run._edifabric.validate_edi",
return_value=fake_result,
) as mock_validate:
result = run_rebill(
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
override_filing=False,
visits_csv_path=str(visits_path),
ingest_dir=str(ingest_dir),
out_dir=str(out_dir),
)
# One file each, the two pipelines route independently.
assert len(result.pipeline_a_files) == 1, result.pipeline_a_files
assert len(result.pipeline_b_files) == 1, result.pipeline_b_files
# Files on disk match the returned paths.
a_path = result.pipeline_a_files[0]
b_path = result.pipeline_b_files[0]
assert a_path.exists() and a_path.stat().st_size > 0
assert b_path.exists() and b_path.stat().st_size > 0
# Each gate call hit validate_edi exactly once (one per claim/batch).
assert mock_validate.call_count == 2
# summary.csv has one row per non-PAID visit.
summary_path = out_dir / "summary.csv"
assert summary_path.exists(), summary_path
with summary_path.open(newline="") as f:
rows = list(csv.DictReader(f))
assert len(rows) == 2, rows
dispositions = {r["disposition"] for r in rows}
assert dispositions == {"REBILLED_A", "REBILLED_B"}, dispositions
def test_run_rebill_fail_open_logs_warning(tmp_path, caplog):
"""The Edifabric-unavailable fail-open path MUST log a WARNING so
the operator knows the validation gate didn't actually run.
Per SP40/SP41 contract: when ``validate_edi`` raises
``EdifabricError``, the file still emits to the pipeline dir but a
WARNING is logged. This test pins that contract so a future
refactor doesn't accidentally silence it.
"""
import logging
from cyclone.edifabric import EdifabricError
caplog.set_level(logging.WARNING, logger="cyclone.rebill.run")
visits_path = _write_visits_csv(tmp_path / "visits.csv", [
("06/27/2026", "J813715", "T1019", "$2.32"),
])
ingest_dir = _stub_empty_835(tmp_path / "ingest")
out_dir = tmp_path / "out"
with patch(
"cyclone.rebill.run._edifabric.validate_edi",
side_effect=EdifabricError(0, "API key not configured"),
):
result = run_rebill(
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
override_filing=False,
visits_csv_path=str(visits_path),
ingest_dir=str(ingest_dir),
out_dir=str(out_dir),
)
# File still emits (fail-open semantics).
assert len(result.pipeline_b_files) == 1
# And a WARNING was logged about the unrun gate.
warnings = [r for r in caplog.records if r.levelno >= logging.WARNING]
assert warnings, "expected at least one WARNING record"
assert any(
"edifabric" in r.message.lower() or "validation" in r.message.lower()
for r in warnings
), f"expected WARNING mentioning edifabric/validation, got {[r.message for r in warnings]}"