From 23fc85cad2798103d29596a6d30909e359357108 Mon Sep 17 00:00:00 2001 From: Nora Date: Wed, 8 Jul 2026 03:43:57 -0600 Subject: [PATCH] refactor(sp41): split spot-check into committed pipeline + validate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../src/cyclone/rebill/spot_check_pipeline.py | 274 +++++++++++++ .../src/cyclone/rebill/spot_check_validate.py | 126 ++++++ backend/tests/test_spot_check_driver.py | 205 ---------- backend/tests/test_spot_check_pipeline.py | 383 ++++++++++++++++++ 4 files changed, 783 insertions(+), 205 deletions(-) create mode 100644 backend/src/cyclone/rebill/spot_check_pipeline.py create mode 100644 backend/src/cyclone/rebill/spot_check_validate.py delete mode 100644 backend/tests/test_spot_check_driver.py create mode 100644 backend/tests/test_spot_check_pipeline.py diff --git a/backend/src/cyclone/rebill/spot_check_pipeline.py b/backend/src/cyclone/rebill/spot_check_pipeline.py new file mode 100644 index 0000000..80ad0fe --- /dev/null +++ b/backend/src/cyclone/rebill/spot_check_pipeline.py @@ -0,0 +1,274 @@ +"""SP41-goal: spot-check 837P generation pipeline (committed under +``cyclone.rebill``). + +Splits the SP41 spot-check flow into a pure-Python generation step +that consumes the canonical ``visits`` + ``service_line_payments`` ++ ``cas_adjustments`` tables and writes well-formed 837P files to +disk. The companion validation step lives in +:mod:`cyclone.rebill.spot_check_validate` (no fallback path — +quota / transport errors fail loud). + +Public surface +-------------- + +- :func:`select_spot_check_visits` — DB-only: read the in-window + visits, reconcile against the ingested 835 SVC rows, drop + PAID / OA-18-only adjudications, return the top-N candidates by + ``billed_amount`` desc. +- :func:`write_spot_check_files` — produce one well-formed 837P per + visit via the canonical ``cyclone.parsers.serialize_837.serialize_837`` + single-claim path. Files are written segment-per-line to ``out_dir`` + with the HCPF-spec filename via + :func:`cyclone.edi.filenames.build_outbound_filename`. + +The dedup of OA-18 (Exact Duplicate) adjudications reflects the +operator's note that the 835s show multiple rejections because +files were submitted multiple times; visits whose only adjudication +is OA-18 on a duplicate SVC are NOT in-window rebill candidates. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from pathlib import Path + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from cyclone import db as db_mod +from cyclone.db import Visit +from cyclone.edi.filenames import build_outbound_filename +from cyclone.parsers.serialize_837 import serialize_837 +from cyclone.rebill.spot_check import ( + VisitRow, + build_claim_output, +) + +_log = logging.getLogger(__name__) + +# Spot-check envelope constants (must match the HCPF production +# trading-partner profile seeded in config/payers.yaml). +SENDER_ID = "11525703" +RECEIVER_ID = "CO_TXIX" +SUBMITTER_NAME = "Dzinesco" +SUBMITTER_CONTACT_NAME = "Tyler Martinez" +SUBMITTER_CONTACT_EMAIL = "tyler@dzinesco.com" +SUBMITTER_CONTACT_PHONE = "8005550100" +RECEIVER_NAME = "COLORADO MEDICAL ASSISTANCE PROGRAM" +SBR09 = "MC" # Medicaid + + +@dataclass(frozen=True) +class SpotCheckFile: + """One well-formed spot-check 837P file on disk + its source visit.""" + + path: Path + visit: VisitRow + disposition: str # "DENIED" / "PARTIAL" / "NOT_IN_835" + claim_id: str + + +def _visit_row_from_db(row: Visit) -> VisitRow: + """Translate a ``Visit`` ORM row to a flat ``VisitRow``. + + The visits table stores modifiers colon-joined (post-normalize); + split them back into a tuple so :func:`build_claim_output` sees + a list-of-modifiers shape. + """ + mods = tuple( + m.strip() for m in (row.modifiers or "").split(":") if m.strip() + ) + return VisitRow( + dos=row.dos, + member_id=row.member_id, + client_name=row.client_name or "", + procedure_code=row.procedure_code, + modifiers=mods, + billed_amount=Decimal(str(row.billed_amount or 0)), + icd10=row.icd10, + prior_auth=row.prior_auth, + ) + + +def _build_claim_id(visit: VisitRow, idx: int) -> str: + """Spot-check CLM01 — uppercase ``R`` prefix satisfies the plan's + literal grep ``^CLM\\*[A-Z]`` (the lowercase ``r`` would not match). + The shape ``R--`` round-trips back to the visit. + """ + return f"R{visit.dos.isoformat().replace('-', '')}-{visit.member_id}-{idx:02d}" + + +def select_spot_check_visits( + session: Session, + *, + window_start: date, + window_end: date, + n: int = 10, +) -> list[tuple[VisitRow, str, list[str]]]: + """Pick ``n`` spot-check candidates from the canonical visits table. + + Reads ``visits`` for the DOS window and ``service_line_payments`` / + ``cas_adjustments`` for the same window. Reconciles each visit on + ``(member_id, procedure_code, DOS)`` against the SVC rows. + + Returns: + A list of ``(VisitRow, disposition, cas_reasons)`` tuples, top + ``n`` by ``billed_amount`` desc, from the + ``{"DENIED", "PARTIAL", "NOT_IN_835"}`` cohort (PAID and + ``DENIED_DUPLICATE_NOISE`` are excluded). + + Disposition classification: + * ``PAID`` — at least one matching SVC paid ≥ 95% of billed. + * ``PARTIAL`` — at least one SVC paid but < 95%. + * ``DENIED`` — matching SVCs but no payment. + * ``DENIED_DUPLICATE_NOISE`` — pure OA-18 (Exact Duplicate) CAS. + Excluded per the operator's note about prior duplicate + submissions — these are NOT in-window rebill candidates. + * ``NOT_IN_835`` — no matching SVC at all. + """ + visits = session.execute( + select(Visit).where( + Visit.dos >= window_start, Visit.dos <= window_end, + ) + ).scalars().all() + + svc_rows = session.execute( + db_mod.text( + """ + SELECT slp.procedure_code AS procedure, + slp.service_date AS svc_date, + slp.charge AS charge, + slp.payment AS paid, + slp.remittance_id AS remittance_id + FROM service_line_payments slp + JOIN remittances r ON r.id = slp.remittance_id + WHERE slp.service_date BETWEEN :start AND :end + AND r.is_reversal = 0 + """ + ), + {"start": window_start.isoformat(), "end": window_end.isoformat()}, + ).mappings().all() + + cas_by_remit: dict[str, list[str]] = {} + cas_rows = session.execute( + db_mod.text( + """ + SELECT remittance_id, group_code, reason_code + FROM cas_adjustments + WHERE remittance_id IN ( + SELECT id FROM remittances WHERE is_reversal = 0 + ) + """ + ) + ).all() + for cr in cas_rows: + cas_by_remit.setdefault(str(cr.remittance_id), []).append( + f"{cr.group_code}-{cr.reason_code}" + ) + + # SQLite returns DATE columns as ISO strings; normalize to date. + def _to_date(v) -> date: + if isinstance(v, date): + return v + return date.fromisoformat(str(v)[:10]) + + # Best-of-N SVC match by (procedure, DOS). The DB-side 835 parser + # (SP20) does not capture member_id at SVC scope, so we use the + # visits table's member_id directly — the visits table is the + # canonical member-of-record source, and an SVC at (procedure, + # DOS) for the same billed amount within 5% tolerance is + # unambiguous in practice. + by_pdos: dict[tuple[str, date], list] = {} + for s in svc_rows: + d = _to_date(s.svc_date) + by_pdos.setdefault((s.procedure, d), []).append(s) + + out: list[tuple[VisitRow, str, list[str]]] = [] + for v in visits: + vr = _visit_row_from_db(v) + cands = by_pdos.get((vr.procedure_code, vr.dos), []) + if not cands: + out.append((vr, "NOT_IN_835", [])) + continue + total_paid = sum( + (Decimal(str(c.paid or 0)) for c in cands), Decimal("0") + ) + any_paid = any(Decimal(str(c.paid or 0)) > 0 for c in cands) + cas_reasons: list[str] = [] + for c in cands: + cas_reasons.extend(cas_by_remit.get(str(c.remittance_id), [])) + if any_paid and total_paid >= vr.billed_amount * Decimal("0.95"): + out.append((vr, "PAID", cas_reasons)) + elif any_paid: + out.append((vr, "PARTIAL", cas_reasons)) + else: + if cas_reasons and set(cas_reasons) == {"OA-18"}: + out.append((vr, "DENIED_DUPLICATE_NOISE", cas_reasons)) + else: + out.append((vr, "DENIED", cas_reasons)) + + # Keep DENIED / PARTIAL / NOT_IN_835; drop PAID + duplicate-noise. + candidates = [(v, d, c) for (v, d, c) in out + if d in ("DENIED", "PARTIAL", "NOT_IN_835")] + candidates.sort(key=lambda t: t[0].billed_amount, reverse=True) + return candidates[:n] + + +def write_spot_check_files( + visits: list[tuple[VisitRow, str, list[str]]], + out_dir: Path, +) -> list[SpotCheckFile]: + """Write one well-formed 837P file per visit into ``out_dir``. + + Each file is produced via the canonical + :func:`cyclone.parsers.serialize_837.serialize_837` single-claim + path. Output is segment-per-line (X12 spec accepts both forms; + segment-per-line matches the prodfiles canonical shape so the + plan's literal grep works on the file directly). + + Returns: + A list of :class:`SpotCheckFile` records (path + source visit + + disposition + claim_id) — caller passes the ``.path`` to the + validator. + """ + out_dir.mkdir(parents=True, exist_ok=True) + out: list[SpotCheckFile] = [] + for idx, (visit, disp, _cas) in enumerate(visits, start=1): + claim_id = _build_claim_id(visit, idx) + claim = build_claim_output(visit, claim_id=claim_id) + x12_text = serialize_837( + claim, + sender_id=SENDER_ID, + receiver_id=RECEIVER_ID, + submitter_name=SUBMITTER_NAME, + submitter_contact_name=SUBMITTER_CONTACT_NAME, + submitter_contact_phone=SUBMITTER_CONTACT_PHONE, + submitter_contact_email=SUBMITTER_CONTACT_EMAIL, + receiver_name=RECEIVER_NAME, + claim_filing_indicator_code=SBR09, + ) + fname = build_outbound_filename(SENDER_ID, "837P") + # Disambiguate same-millisecond filenames with the claim_id. + stem, dot_ext = fname.rsplit(".", 1) + path = out_dir / f"{stem}-{claim_id}.{dot_ext}" + # Write segment-per-line so the plan's literal grep works + # directly on the file. + path.write_text( + x12_text.replace("~", "\n").rstrip("\n") + "\n", + encoding="ascii", + ) + out.append(SpotCheckFile( + path=path, + visit=visit, + disposition=disp, + claim_id=claim_id, + )) + _log.info( + "wrote spot-check file %s (disposition=%s, dos=%s, member=%s, " + "procedure=%s, amount=%s)", + path.name, disp, visit.dos, visit.member_id, + visit.procedure_code, visit.billed_amount, + ) + return out \ No newline at end of file diff --git a/backend/src/cyclone/rebill/spot_check_validate.py b/backend/src/cyclone/rebill/spot_check_validate.py new file mode 100644 index 0000000..921c07a --- /dev/null +++ b/backend/src/cyclone/rebill/spot_check_validate.py @@ -0,0 +1,126 @@ +"""SP41-goal: spot-check 837P validation against the Edifabric +``/v2/x12/validate`` endpoint (committed under ``cyclone.rebill``). + +Public surface +-------------- + +- :func:`is_edifabric_pass` — inspect an ``OperationResult`` dict and + return ``True`` only for ``Status in {"success", "warning"}``. +- :func:`validate_spot_check_files` — submit every file in ``paths`` + to ``cyclone.edifabric.validate_edi`` (the shipped /v2/x12/validate + client) and return a list of per-file result dicts. + +No fallback path +---------------- + +The original scratch driver fell back to local pyX12 on a 403 +quota response from Edifabric. That fallback was masking the +acceptance-criterion breach: AC5/VP4 require live Edifabric +validation, and a pyx12 substitute does not satisfy the criterion. +This module deliberately has NO fallback. On any non-2xx HTTP +response (``EdifabricError``) or any ``Status == "error"`` +OperationResult, the per-file record carries ``passed: false`` and +the OperationResult body is preserved verbatim so the operator can +diagnose. + +Transport / network failures raise so the caller can decide +whether to exit-loud or retry — the orchestrator (the thin driver) +prints ``10/10 passed`` only when all 10 entries are live Edifabric +passes. +""" +from __future__ import annotations + +import logging +from pathlib import Path + +from cyclone.edifabric import EdifabricError, validate_edi + +_log = logging.getLogger(__name__) + + +def is_edifabric_pass(result: dict) -> bool: + """Return True iff ``result`` is a passing ``OperationResult``. + + Per the EdiNation docs, ``Status`` is one of: + * ``"success"`` — passes. + * ``"warning"`` — passes (advisory notices don't block a + structurally valid envelope; the SP41 contract treats + warning as a pass per the operator's sign-off). + * ``"error"`` — fails; the Details array carries the failing + segments / messages. + * anything else — fail loud so we don't silently approve. + """ + status = (result.get("Status") or "").lower() + return status in ("success", "warning") + + +def validate_spot_check_files(paths: list[Path]) -> list[dict]: + """Submit each file in ``paths`` to Edifabric ``/v2/x12/validate``. + + Returns a list of per-file result dicts, one per input path, in + the same order. Each dict has the shape:: + + { + "file": str, # path to the submitted file + "passed": bool, # True only if Status is success/warning + "issue": str, # human-readable verdict + "result": dict | None, # OperationResult verbatim (None on transport error) + "transport_error": dict | None, # {status_code, body} on EdifabricError + } + + Raises: + EdifabricError: on the first transport failure. The caller + decides whether to retry, exit non-zero, or fall back. + This module never silently substitutes a different + validator. + """ + out: list[dict] = [] + for p in paths: + body = p.read_bytes() + try: + result = validate_edi(body) + except EdifabricError as exc: + out.append({ + "file": str(p), + "passed": False, + "issue": ( + f"EdifabricError {exc.status_code}: {exc.body!r}" + ), + "result": None, + "transport_error": { + "status_code": exc.status_code, + "body": exc.body, + }, + }) + # Re-raise so the caller can't silently absorb the failure + # (the original scratch driver caught + fallback'd which + # is the bug this module fixes). + raise + + if is_edifabric_pass(result): + out.append({ + "file": str(p), + "passed": True, + "issue": f"edifabric {result.get('Status')!r}", + "result": result, + "transport_error": None, + }) + else: + details = result.get("Details") or [] + first_msgs = [] + for d in details[:5]: + if isinstance(d, dict): + first_msgs.append( + f"{d.get('SegmentId', '?')}: {d.get('Message', '?')}" + ) + out.append({ + "file": str(p), + "passed": False, + "issue": ( + f"edifabric {result.get('Status')!r}: " + + ("; ".join(first_msgs) or "no details") + ), + "result": result, + "transport_error": None, + }) + return out \ No newline at end of file diff --git a/backend/tests/test_spot_check_driver.py b/backend/tests/test_spot_check_driver.py deleted file mode 100644 index 0375e30..0000000 --- a/backend/tests/test_spot_check_driver.py +++ /dev/null @@ -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" \ No newline at end of file diff --git a/backend/tests/test_spot_check_pipeline.py b/backend/tests/test_spot_check_pipeline.py new file mode 100644 index 0000000..f0d6509 --- /dev/null +++ b/backend/tests/test_spot_check_pipeline.py @@ -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 \ No newline at end of file