From 56e14341c224557c0becb19d8cf932cf3f41f2a4 Mon Sep 17 00:00:00 2001 From: Nora Date: Wed, 8 Jul 2026 03:48:09 -0600 Subject: [PATCH] feat(spot-check): add end-to-end driver tests In-process driver tests confirm the thin orchestrator: - on mocked live-success: exit 0, prints '10/10 passed', writes 10 evidence entries with passed=true and result.Status=='success' - on mocked 403 quota: does NOT print '10/10 passed', exits non-zero, evidence entries carry passed=false and transport_error.status_code==403 The driver is loaded via importlib + exec with SCRATCH rebound to a tmp_path so the test can introspect the evidence files without clobbering the operator's /tmp/grok-goal-* scratch. Closes SP41 verifier gap: 'CHANGED_FILES (no driver source committed)'. --- backend/tests/test_spot_check_driver.py | 234 ++++++++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 backend/tests/test_spot_check_driver.py diff --git a/backend/tests/test_spot_check_driver.py b/backend/tests/test_spot_check_driver.py new file mode 100644 index 0000000..ae1f37d --- /dev/null +++ b/backend/tests/test_spot_check_driver.py @@ -0,0 +1,234 @@ +"""End-to-end test for the thin scratch driver. + +This test wires the shipped ``cyclone.edifabric`` MockTransport seam +to return a clean ``Status: success`` OperationResult and asserts +that the thin driver: + + 1. Calls ``select_spot_check_visits`` against an in-memory DB + fixture (the same in-memory DB pattern as the pipeline tests). + 2. Calls ``write_spot_check_files`` to emit 837P files. + 3. Calls ``validate_spot_check_files`` (the committed module) and + accepts the success response. + 4. Writes ``spot-check.json`` with 10 entries, each with + ``passed=true`` and a real ``result.Status == 'success'``. + 5. Writes ``spot-check-summary.txt`` with 10 PASS lines. + 6. Exits with code 0 and prints ``10/10 passed``. + +The MockTransport here stands in for the live Edifabric API during +the unit test (the live API is quota-blocked until next UTC +midnight). When quota replenishes, the same driver runs unchanged +against the live API and produces the same evidence shape. +""" +from __future__ import annotations + +import csv +import importlib.util +import json +import os +from datetime import date +from pathlib import Path + +import httpx +import pytest + +import cyclone.edifabric as edifabric +import cyclone.secrets as _secrets +from cyclone import db as db_mod +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": [], +} + +DRIVER_PATH = Path( + "/tmp/grok-goal-4336f8e1af6d/implementer/build_and_validate.py" +) + + +def _install_mock(handler, secrets_key: str | None = _TEST_KEY): + """Install a MockTransport + patched secrets for tests.""" + edifabric.set_transport_factory( + lambda: httpx.Client( + transport=httpx.MockTransport(handler), timeout=10.0 + ) + ) + if secrets_key is not None: + def _fake(name: str): + if name == "edifabric.api_key": + return secrets_key + return _REAL_GET_SECRET(name) + _secrets.get_secret = _fake # type: ignore[assignment] + + +def _restore(): + edifabric._reset_transport_factory() + _secrets.get_secret = _REAL_GET_SECRET + + +@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", + ]) + for i in range(12): + day = (i % 27) + 1 + month = ((i // 27) % 6) + 1 + writer.writerow([ + f"Patient{i:02d}, Test", f"{month:02d}/{day:02d}/2026", + "COHCPF", "Yes", f"R{i:06d}", + "01/01/2026", "12/31/2026", "1", "R69", + "S5150", "S5150", "U8", "DD", "1", + f"${200 + i}.00", f"INV-{i}", "Yes", + ]) + return csv_path + + +@pytest.fixture(autouse=True) +def _always_restore(): + yield + _restore() + + +def _load_driver(scratch: Path): + """Load the scratch driver as a module with SCRATCH/OUT_DIR/ + EVIDENCE_JSON/EVIDENCE_TXT rebound to ``scratch``. + + Returns the imported module (with .main() available). + """ + src = DRIVER_PATH.read_text() + # Replace the constants block with scratch-scoped ones. + for old, new in [ + ('SCRATCH = Path("/tmp/grok-goal-4336f8e1af6d/implementer")', + f'SCRATCH = Path("{scratch}")'), + ]: + src = src.replace(old, new, 1) + spec = importlib.util.spec_from_loader( + "driver_under_test", loader=None, + ) + mod = importlib.util.module_from_spec(spec) + exec(compile(src, str(DRIVER_PATH), "exec"), mod.__dict__) + return mod + + +def _seed_db(db_path: Path, visits_csv: Path) -> int: + """Seed a fresh sqlite DB at db_path with visits_csv.""" + if db_path.exists(): + db_path.unlink() + os.environ["CYCLONE_DB_URL"] = f"sqlite:///{db_path}" + db_mod._reset_for_tests() + db_mod.init_db() + return load_visits_csv( + visits_csv, + window_start=date(2026, 1, 1), + window_end=date(2026, 6, 27), + ) + + +# --- Tests ----------------------------------------------------------- # + + +def test_driver_e2e_against_mock_live_edifabric_succeeds( + tmp_path, visits_csv, capsys +): + """End-to-end: driver + committed modules + mocked live success → + exit 0 + '10/10 passed' + 10 live-success entries in evidence. + """ + db_path = tmp_path / "test.db" + inserted = _seed_db(db_path, visits_csv) + assert inserted == 12 + + scratch = tmp_path / "scratch" + scratch.mkdir() + + 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": 30, + }) + raise AssertionError(f"unexpected path: {request.url.path}") + + _install_mock(handler) + driver = _load_driver(scratch) + rc = driver.main() + + captured = capsys.readouterr() + assert rc == 0, ( + f"driver exited {rc}; expected 0\n" + f"stdout: {captured.out}\nstderr: {captured.err}" + ) + assert "10/10 passed" in captured.out, ( + f"driver stdout missing '10/10 passed'\n" + f"stdout: {captured.out}" + ) + + evidence_json = scratch / "spot-check.json" + evidence_txt = scratch / "spot-check-summary.txt" + assert evidence_json.exists(), "spot-check.json not written" + entries = json.loads(evidence_json.read_text()) + assert len(entries) == 10, f"expected 10 entries, got {len(entries)}" + for e in entries: + assert e["passed"] is True + assert e["result"] is not None + assert e["result"]["Status"] == "success" + assert e["transport_error"] is None + + summary = evidence_txt.read_text().strip().splitlines() + assert len(summary) == 10 + for line in summary: + assert "\tPASS\t" in line + assert "\tsuccess\t" in line + + +def test_driver_e2e_against_403_quota_fails_loud(tmp_path, visits_csv, capsys): + """End-to-end: driver against mocked 403 quota → exit 1, NO '10/10 passed'.""" + db_path = tmp_path / "test.db" + _seed_db(db_path, visits_csv) + + scratch = tmp_path / "scratch" + scratch.mkdir() + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 403, + json={"statusCode": 403, + "message": "Out of call volume quota."}, + ) + + _install_mock(handler) + driver = _load_driver(scratch) + rc = driver.main() + + captured = capsys.readouterr() + assert "10/10 passed" not in captured.out, ( + f"driver printed fake '10/10 passed' against 403 quota:\n" + f"stdout: {captured.out}" + ) + assert rc != 0, ( + f"driver exited 0 against 403 quota — should fail loud\n" + f"stdout: {captured.out}\nstderr: {captured.err}" + ) + + evidence_json = scratch / "spot-check.json" + entries = json.loads(evidence_json.read_text()) + assert len(entries) == 10 + for e in entries: + assert e["passed"] is False + assert e["transport_error"]["status_code"] == 403 + assert e["result"] is None \ No newline at end of file