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

refactor(sp41): split spot-check into committed pipeline + validate

Browse Source 
The previous scratch build_and_validate.py was a single ~800-line
monolith that:
  (1) re-implemented visit-to-835 reconcile instead of using the
      committed visits_store + parse_835_svc shape, and
  (2) had a pyx12 fallback on 403 quota that printed '10/10 passed'
      and wrote passed=true to evidence without ever talking to
      Edifabric — directly contradicting AC5/VP4.

This commit splits it into two committed modules under
cyclone.rebill/:

  spot_check_pipeline.py — generation
    - select_spot_check_visits(session, *, window_start, window_end, n)
      reads visits + service_line_payments + cas_adjustments; drops
      PAID + DENIED_DUPLICATE_NOISE (OA-18 only); returns top-n by
      billed_amount desc.
    - write_spot_check_files(visits, out_dir) — uses committed
      spot_check.build_claim_output + serialize_837 +
      build_outbound_filename; writes segment-per-line with
      uppercase-R CLM01 (so plan's literal grep ^CLM*[A-Z] matches).

  spot_check_validate.py — validation (NO FALLBACK)
    - is_edifabric_pass(result) — True for Status in {success, warning}.
    - validate_spot_check_files(paths) — calls ONLY
      cyclone.edifabric.validate_edi; on EdifabricError re-raises
      (no pyx12 substitute). passed=true requires a real
      OperationResult with Status in {success, warning}.

  test_spot_check_pipeline.py — 13 unit tests
    - 4 pipeline tests: in-window filter, NOT_IN_835 default,
      n-truncation, OA-18 dedup.
    - 2 file-write tests: plan-grep matches all 6 segment classes
      (NM1*41|NM1*40|NM1*QC|^CLM*[A-Z]|SV1*HC:|DTP*472),
      segment-per-line shape.
    - 4 is_edifabric_pass classifier tests (success/warning/error/unknown).
    - 3 validate_spot_check_files tests: success status → passed;
      error status → passed=false; 403 quota → EdifabricError raised
      (no silent substitution).

The scratch driver (build_and_validate.py) is now ~40 lines: imports
the two modules, prints '10/10 passed' only when all 10 entries are
live Edifabric passes.

Also removes tests/test_spot_check_driver.py (subsumed by the
pipeline tests above).
This commit is contained in:
Nora
2026-07-08 03:43:57 -06:00
parent 61e90b64b3
commit 23fc85cad2
4 changed files with 783 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_spot_check_driver.py
-205
View File
@@ -1,205 +0,0 @@
"""SP41-goal: prove the spot-check driver's validate_edi code path
works against the Edifabric /v2/x12/validate OperationResult shape.
The live Edifabric API is quota-blocked (resets next UTC midnight);
until then we can't make real network calls. This test wires the
shipped ``cyclone.edifabric`` MockTransport seam to return a clean
``Status: success`` OperationResult for each of the 10 spot-check
files and asserts the driver's ``passed()`` classifier returns True.
This pins the end-to-end driver path against the real
``cyclone.edifabric.validate_edi`` client — without using a network
call. The test exercises:
1. validate_edi → OperationResult with Status='success' → passed=True
2. validate_edi → OperationResult with Status='warning' → passed=True
3. validate_edi → OperationResult with Status='error' → passed=False
4. validate_edi → 403 quota → EdifabricError raised (caught by driver)
Each scenario runs against the actual 837P files produced by the
driver in this session (loaded from the scratch dir).
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
import httpx
import pytest
# Make the spot-check driver importable.
SCRATCH = Path("/tmp/grok-goal-4336f8e1af6d/implementer")
sys.path.insert(0, str(SCRATCH))
import cyclone.edifabric as edifabric # noqa: E402
import cyclone.secrets as _secrets # noqa: E402
_REAL_GET_SECRET = _secrets.get_secret
_TEST_KEY = "3ecf6b1c5cf34bd797a5f4c57951a1cf"
_X12_INTERCHANGE = {
"SegmentDelimiter": "~",
"DataElementDelimiter": "*",
"ISA": {"InterchangeControlNumber_13": "000000001"},
"Groups": [],
"IEATrailers": [],
}
def _mock_secrets(patched_key: str):
def _fake(name: str):
if name == "edifabric.api_key":
return patched_key
return _REAL_GET_SECRET(name)
_secrets.get_secret = _fake # type: ignore[assignment]
def _install(handler):
transport = httpx.MockTransport(handler)
edifabric.set_transport_factory(lambda: httpx.Client(transport=transport, timeout=10.0))
@pytest.fixture(autouse=True)
def _restore():
yield
edifabric._reset_transport_factory()
_secrets.get_secret = _REAL_GET_SECRET # type: ignore[assignment]
def _first_spot_file() -> Path:
"""Return the first spot-check 837P file from the scratch dir.
Skip if no driver run has produced files yet (the live driver
writes them; this test reads them).
"""
files = sorted((SCRATCH / "spot-check-837Ps").glob("*.x12"))
if not files:
pytest.skip("no spot-check 837P files in scratch dir — run build_and_validate.py first")
return files[0]
@pytest.fixture
def spot_file() -> Path:
return _first_spot_file()
def _make_handler(operation_result: dict):
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(200, json=[_X12_INTERCHANGE])
if request.url.path.endswith("/validate"):
return httpx.Response(200, json=operation_result)
raise AssertionError(f"unexpected path: {request.url.path}")
return handler
# --- The 4 scenarios --------------------------------------------------- #
def test_driver_passed_classifier_on_success_status(spot_file: Path):
"""Status='success' → driver.passed() returns (True, 'success')."""
from build_and_validate import passed # type: ignore
_mock_secrets(_TEST_KEY)
_install(_make_handler({"Status": "success", "Details": [], "LastIndex": 46}))
body = spot_file.read_bytes()
result = edifabric.validate_edi(body)
ok, msg = passed(result)
assert ok is True
assert msg == "success"
def test_driver_passed_classifier_on_warning_status(spot_file: Path):
"""Status='warning' → driver.passed() returns (True, 'warning').
Edifabric 'warning' is advisory (deprecation / best-practice
notices) and does NOT block a structurally valid envelope per the
SP41 contract.
"""
from build_and_validate import passed # type: ignore
_mock_secrets(_TEST_KEY)
_install(_make_handler({
"Status": "warning",
"Details": [{
"Index": 12,
"SegmentId": "CLM",
"Value": "CLM*R20260205-Q944140-01*333.62***12:B:1*Y*A*Y*Y",
"Message": "Best practice advisory: use place-of-service code 11",
"Status": "warning",
}],
"LastIndex": 46,
}))
body = spot_file.read_bytes()
result = edifabric.validate_edi(body)
ok, msg = passed(result)
assert ok is True
assert msg == "warning"
def test_driver_passed_classifier_on_error_status(spot_file: Path):
"""Status='error' → driver.passed() returns (False, error details)."""
from build_and_validate import passed # type: ignore
_mock_secrets(_TEST_KEY)
_install(_make_handler({
"Status": "error",
"Details": [{
"Index": 5,
"SegmentId": "CLM",
"Value": "CLM*X",
"Message": "Segment CLM is missing required data elements",
"Status": "error",
}],
"LastIndex": 5,
}))
body = spot_file.read_bytes()
result = edifabric.validate_edi(body)
ok, msg = passed(result)
assert ok is False
assert "missing required data elements" in msg
def test_driver_passed_classifier_on_403_quota(spot_file: Path):
"""HTTP 403 quota → EdifabricError raised (driver falls back to pyX12)."""
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(
403,
json={"statusCode": 403,
"message": "Out of call volume quota."},
)
raise AssertionError(f"unexpected path: {request.url.path}")
_mock_secrets(_TEST_KEY)
_install(handler)
body = spot_file.read_bytes()
with pytest.raises(edifabric.EdifabricError) as exc_info:
edifabric.validate_edi(body)
assert exc_info.value.status_code == 403
def test_driver_emits_correct_validate_edi_call(spot_file: Path):
"""The driver calls validate_edi (read+validate) — pin the wire shape.
Validates the read path returns the X12Interchange shape and the
validate path receives that same shape (not the raw bytes).
"""
captured_validates: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(200, json=[_X12_INTERCHANGE])
if request.url.path.endswith("/validate"):
captured_validates.append(json.loads(request.content))
return httpx.Response(200, json={"Status": "success", "Details": [], "LastIndex": 46})
raise AssertionError(f"unexpected path: {request.url.path}")
_mock_secrets(_TEST_KEY)
_install(handler)
body = spot_file.read_bytes()
result = edifabric.validate_edi(body)
assert result["Status"] == "success"
assert len(captured_validates) == 1
# The validate body is the X12Interchange, not the raw bytes.
assert captured_validates[0]["ISA"]["InterchangeControlNumber_13"] == "000000001"
backend/tests/test_spot_check_pipeline.py
+383
View File
@@ -0,0 +1,383 @@
"""SP41-goal: unit tests for the committed spot-check pipeline + validate
modules.
The two modules are pure (no I/O outside the DB session + Edifabric
HTTP seam) and unit-testable end-to-end:
- :mod:`cyclone.rebill.spot_check_pipeline` — generation (DB → files).
- :mod:`cyclone.rebill.spot_check_validate` — validation (files →
OperationResult, no fallback).
These tests pin the live-shape contract that the production driver
exercises, so a regression in either module shows up here without
having to run the live driver.
"""
from __future__ import annotations
import csv
import json
import re
import sys
from datetime import date
from decimal import Decimal
from pathlib import Path
import httpx
import pytest
from sqlalchemy import text
import cyclone.edifabric as edifabric
import cyclone.secrets as _secrets
from cyclone import db as db_mod
from cyclone.rebill.spot_check_pipeline import (
select_spot_check_visits,
write_spot_check_files,
)
from cyclone.rebill.spot_check_validate import (
is_edifabric_pass,
validate_spot_check_files,
)
from cyclone.rebill.visits_store import load_visits_csv
_REAL_GET_SECRET = _secrets.get_secret
_TEST_KEY = "3ecf6b1c5cf34bd797a5f4c57951a1cf"
_X12_INTERCHANGE = {
"SegmentDelimiter": "~",
"DataElementDelimiter": "*",
"ISA": {"InterchangeControlNumber_13": "000000001"},
"Groups": [],
"IEATrailers": [],
}
# --- Fixtures --------------------------------------------------------- #
@pytest.fixture
def visits_csv(tmp_path: Path) -> Path:
csv_path = tmp_path / "visits.csv"
with csv_path.open("w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"Client", "Visit Date", "Payer", "Authorized", "Member ID",
"Auth Start Date", "Auth End Date", "Authorization #", "ICD-10",
"Service", "Procedure Code", "Modifiers", "Client Classes",
"Billable Hours", "Billable Amount", "Invoice #", "Claimed",
])
writer.writerow([
"Doe, Jane", "01/15/2026", "COHCPF", "Yes", "R111111",
"01/01/2026", "12/31/2026", "1", "R69",
"Homemaker S5150", "S5150", "U8", "DD Waiver", "1", "$300.00",
"INV-2026-01-15-R111111", "Yes",
])
writer.writerow([
"Roe, Richard", "02/20/2026", "COHCPF", "Yes", "R222222",
"01/01/2026", "12/31/2026", "2", "R69",
"IHSS T1019", "T1019", "KX, SC, U2", "IHSS MR", "1.5", "$250.00",
"INV-2026-02-20-R222222", "Yes",
])
writer.writerow([
"Old, Stale", "12/31/2025", "COHCPF", "Yes", "R999999",
"01/01/2025", "12/31/2025", "0", "R69",
"Old S5150", "S5150", "U8", "DD Waiver", "1", "$100.00",
"INV-2025-12-31-R999999", "Yes", # OUT-OF-WINDOW
])
return csv_path
@pytest.fixture
def loaded_db(tmp_path: Path, visits_csv: Path, monkeypatch):
"""Load visits_csv into a fresh DB and return the session."""
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path / 'test.db'}")
db_mod._reset_for_tests()
db_mod.init_db()
inserted = load_visits_csv(
visits_csv,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
)
assert inserted == 2 # the stale row is out-of-window
SessionLocal = db_mod.SessionLocal()
sess = SessionLocal()
yield sess
sess.close()
@pytest.fixture(autouse=True)
def _restore():
yield
edifabric._reset_transport_factory()
_secrets.get_secret = _REAL_GET_SECRET
def _mock_secrets(patched_key: str):
def _fake(name: str):
if name == "edifabric.api_key":
return patched_key
return _REAL_GET_SECRET(name)
_secrets.get_secret = _fake # type: ignore[assignment]
def _install(handler):
transport = httpx.MockTransport(handler)
edifabric.set_transport_factory(
lambda: httpx.Client(transport=transport, timeout=10.0)
)
# --- Pipeline tests --------------------------------------------------- #
def test_select_spot_check_visits_returns_in_window_only(loaded_db):
"""select_spot_check_visits honors the DOS window."""
out = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
assert len(out) == 2
# The stale 2025 row was filtered out at the load step.
dos_list = [v.dos for (v, _d, _c) in out]
assert all(date(2026, 1, 1) <= d <= date(2026, 6, 27) for d in dos_list)
def test_select_spot_check_visits_marks_not_in_835(loaded_db):
"""With no SVC rows in DB, visits are NOT_IN_835 (rebill candidates)."""
out = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
dispositions = [d for (_v, d, _c) in out]
assert dispositions == ["NOT_IN_835", "NOT_IN_835"]
def test_select_spot_check_visits_n_truncates(loaded_db):
"""n=1 returns the top 1 by billed_amount desc."""
out = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=1,
)
assert len(out) == 1
assert out[0][0].member_id == "R111111" # $300 > $250
def test_select_spot_check_visits_oa18_excluded(loaded_db):
"""OA-18-only SVC matches are excluded as DENIED_DUPLICATE_NOISE."""
# Insert a batch + remittance + SVC so the Jan-15 visit has a hit.
loaded_db.execute(text("""
INSERT INTO batches (
id, kind, input_filename, parsed_at
) VALUES (
'B-TEST-OA18', '835', 'tp-test-20260115.835',
'2026-01-20 00:00:00'
)
"""))
loaded_db.execute(text("""
INSERT INTO remittances (
id, batch_id, payer_claim_control_number,
total_charge, total_paid, adjustment_amount,
status_code, received_at, is_reversal
) VALUES (
'R-TEST-OA18', 'B-TEST-OA18', 'CLM-TEST-1',
300.00, 0.00, 0.00,
'DENIED', '2026-01-20 00:00:00', 0
)
"""))
loaded_db.execute(text("""
INSERT INTO service_line_payments (
remittance_id, line_number, procedure_qualifier,
procedure_code, service_date, charge, payment
) VALUES (
'R-TEST-OA18', 1, 'HC',
'S5150', '2026-01-15', 300.00, 0.00
)
"""))
loaded_db.commit()
out = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
# R222222 (no SVC) → NOT_IN_835
# R111111 (SVC hit with no CAS yet → DENIED)
dispositions = sorted(d for (_v, d, _c) in out)
assert dispositions == ["DENIED", "NOT_IN_835"]
# Now add OA-18 CAS — R111111 flips to DENIED_DUPLICATE_NOISE.
loaded_db.execute(text("""
INSERT INTO cas_adjustments (
remittance_id, group_code, reason_code, amount
) VALUES ('R-TEST-OA18', 'OA', '18', 300.00)
"""))
loaded_db.commit()
out = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
# R111111 now excluded (DENIED_DUPLICATE_NOISE); only R222222 remains.
assert len(out) == 1
assert out[0][0].member_id == "R222222"
def test_write_spot_check_files_emits_required_segments(tmp_path, loaded_db):
"""Every emitted file passes the plan's literal grep."""
out_dir = tmp_path / "spot-check-837Ps"
visits = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
written = write_spot_check_files(visits, out_dir)
assert len(written) == 2
plan_grep = re.compile(
r"NM1\*41|NM1\*40|NM1\*QC|^CLM\*[A-Z]|SV1\*HC:|DTP\*472",
re.MULTILINE,
)
for sc in written:
content = sc.path.read_text()
hits = plan_grep.findall(content)
# 6 distinct segment classes.
assert len(set(hits)) == 6, (
f"{sc.path.name}: missing one of the 6 required segment "
f"classes; got {set(hits)}"
)
# CLM01 must start with uppercase R.
clm_line = next(
(l for l in content.splitlines() if l.startswith("CLM*")), None
)
assert clm_line and clm_line.startswith("CLM*R"), clm_line
def test_write_spot_check_files_segment_per_line(tmp_path, loaded_db):
"""Files are written segment-per-line (matches prodfiles canonical shape)."""
out_dir = tmp_path / "spot-check-837Ps"
visits = select_spot_check_visits(
loaded_db,
window_start=date(2026, 1, 1),
window_end=date(2026, 6, 27),
n=10,
)
written = write_spot_check_files(visits, out_dir)
# Known segment names that may appear in a well-formed 837P.
valid_segment_prefixes = {
"ISA", "GS", "ST", "BHT", "NM1", "PER", "HL", "PRV", "N3", "N4",
"REF", "SBR", "DMG", "PAT", "CLM", "HI", "LX", "SV1", "SV2",
"SV3", "SV4", "SV5", "SV6", "SV7", "DTP", "SE", "GE", "IEA",
}
for sc in written:
content = sc.path.read_text()
# No tildes in the file body (only segment terminator is newline).
assert "~" not in content, f"{sc.path.name} still contains ~"
# Every line begins with a known segment name.
for line in content.splitlines():
prefix = line.split("*", 1)[0]
assert prefix in valid_segment_prefixes, (
f"unexpected segment prefix in {sc.path.name}: {line!r}"
)
# --- Validate tests --------------------------------------------------- #
def test_is_edifabric_pass_true_on_success():
assert is_edifabric_pass({"Status": "success", "Details": []}) is True
def test_is_edifabric_pass_true_on_warning():
assert is_edifabric_pass({"Status": "warning", "Details": []}) is True
def test_is_edifabric_pass_false_on_error():
assert is_edifabric_pass({"Status": "error", "Details": []}) is False
def test_is_edifabric_pass_false_on_unknown():
assert is_edifabric_pass({"Status": "fubar"}) is False
def test_validate_spot_check_files_passes_on_success_status(tmp_path):
"""validate_spot_check_files returns passed=True on Status='success'."""
f = tmp_path / "test.x12"
f.write_bytes(b"ISA*00*~ST*837*~SE*1*~IEA*1*~")
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(200, json=[_X12_INTERCHANGE])
if request.url.path.endswith("/validate"):
return httpx.Response(
200, json={"Status": "success", "Details": [], "LastIndex": 4},
)
raise AssertionError(f"unexpected path: {request.url.path}")
_mock_secrets(_TEST_KEY)
_install(handler)
out = validate_spot_check_files([f])
assert len(out) == 1
assert out[0]["passed"] is True
assert out[0]["result"]["Status"] == "success"
assert out[0]["transport_error"] is None
def test_validate_spot_check_files_fails_on_error_status(tmp_path):
"""validate_spot_check_files returns passed=False on Status='error'."""
f = tmp_path / "test.x12"
f.write_bytes(b"ISA*00*~ST*837*~SE*1*~IEA*1*~")
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(200, json=[_X12_INTERCHANGE])
if request.url.path.endswith("/validate"):
return httpx.Response(200, json={
"Status": "error",
"Details": [{
"SegmentId": "CLM", "Message": "missing elements",
"Status": "error",
}],
"LastIndex": 2,
})
raise AssertionError(f"unexpected path: {request.url.path}")
_mock_secrets(_TEST_KEY)
_install(handler)
out = validate_spot_check_files([f])
assert out[0]["passed"] is False
assert "missing elements" in out[0]["issue"]
def test_validate_spot_check_files_raises_on_403_quota(tmp_path):
"""On 403 quota, raises EdifabricError — NO fallback to pyx12.
This is the structural fix: the previous scratch driver caught
this error and fell back to a local validator, which silently
marked passed=True without ever talking to Edifabric. The new
module deliberately has no fallback.
"""
f = tmp_path / "test.x12"
f.write_bytes(b"ISA*00*~ST*837*~SE*1*~IEA*1*~")
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/read"):
return httpx.Response(
403,
json={"statusCode": 403,
"message": "Out of call volume quota."},
)
raise AssertionError(f"unexpected path: {request.url.path}")
_mock_secrets(_TEST_KEY)
_install(handler)
with pytest.raises(edifabric.EdifabricError) as exc_info:
validate_spot_check_files([f])
assert exc_info.value.status_code == 403