"""Successful parses must persist to the store; failed parses must not.""" from __future__ import annotations from pathlib import Path import pytest from fastapi.testclient import TestClient from cyclone.api import app from cyclone.store import store as global_store FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt" FIXTURE_835 = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt" @pytest.fixture(autouse=True) def clear_store(): """Reset the module-level store before and after each test.""" with global_store._lock: global_store._batches.clear() yield with global_store._lock: global_store._batches.clear() @pytest.fixture def client() -> TestClient: return TestClient(app) def test_successful_parse_837_creates_batch(client: TestClient): assert len(global_store.list()) == 0 text = FIXTURE.read_text() resp = client.post( "/api/parse-837", files={"file": ("test.txt", text, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, resp.text assert len(global_store.list()) == 1 rec = global_store.list()[0] assert rec.kind == "837p" assert rec.input_filename == "test.txt" def test_failed_parse_837_does_not_create_batch(client: TestClient): """An empty / garbage file should NOT create a batch.""" assert len(global_store.list()) == 0 resp = client.post( "/api/parse-837", files={"file": ("garbage.txt", "not-edi", "text/plain")}, headers={"Accept": "application/json"}, ) # Either a 4xx from the parser, or a 200 with 0 claims — the contract # is that NO batch is added either way. assert resp.status_code in (200, 400, 422) assert len(global_store.list()) == 0 def test_validation_failed_837_does_not_create_batch(client: TestClient): """A parse that produced claims with validation errors (422) must not be stored. The spec only mandates "do not store on CycloneParseError or unhandled exception"; we choose the stricter interpretation that 422 validation failures are also excluded so the data on disk is always clean. """ # Construct a minimal but technically malformed 837P that the parser # will accept but validation will reject. We inject a wrong NPI # length to trigger the per-claim NPI validation rule. text = FIXTURE.read_text() # Replace the provider NPI (XX*1881068062) with an invalid 8-digit NPI # (the rule is NPI must be 10 digits). This keeps the parse step # green while failing claim validation. text = text.replace("XX*1881068062", "XX*12345678") resp = client.post( "/api/parse-837", files={"file": ("bad.txt", text, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 422, resp.text assert len(global_store.list()) == 0 def test_parse_835_response_includes_reconciliation_summary( client: TestClient, tmp_path: Path ): """A successful 835 parse returns matched/unmatched counts in JSON.""" fixture = FIXTURE_835.read_text() p = tmp_path / "era.txt" p.write_text(fixture) with open(p, "rb") as f: r = client.post( "/api/parse-835", files={"file": ("era.txt", f, "text/plain")}, headers={"Accept": "application/json"}, ) assert r.status_code == 200 body = r.json() assert "reconciliation" in body, f"missing reconciliation: {list(body.keys())}" rec = body["reconciliation"] assert "matched" in rec assert "unmatched_claims" in rec assert "unmatched_remittances" in rec assert "skipped" in rec # --------------------------------------------------------------------------- # # Production 837P round-trip (parser → API → store → DB) # --------------------------------------------------------------------------- # # Path to the production 837P files from axiscare. These are NOT in git # (.gitignore: ``docs/prodfiles/*/``) — the test is skipped if the directory # is missing so it stays green in clean checkouts but exercises real data # when ops has dropped files in for pipeline validation. PRODFILE_DIR = Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare" # (filename, expected_control_number, expected_total_claims). Derived from # /tmp/parse_prodfile.py on 2026-06-20 against the 7 files in PRODFILE_DIR. # The first two files share control_number 991102984 — a deliberate replay. EXPECTED_PRODFILES: list[tuple[str, str, int]] = [ ("tp11525703-837P-20260618151119397-1of1.txt", "991102984", 141), ("tp11525703-837P-20260618153339862-1of1.txt", "991102984", 141), ("tp11525703-837P-20260618153343460-1of1.txt", "991102983", 97), ("tp11525703-837P-20260618153346107-1of1.txt", "991102982", 28), ("tp11525703-837P-20260618153349188-1of1.txt", "991102981", 69), ("tp11525703-837P-20260618153354947-1of1.txt", "991102978", 149), ("tp11525703-837P-20260618153358831-1of1.txt", "991102977", 99), ] @pytest.mark.skipif( not PRODFILE_DIR.is_dir(), reason=f"production files not present at {PRODFILE_DIR} (gitignored)", ) def test_prodfile_round_trip_persists_separately(client: TestClient): """All 7 production 837P files from axiscare must parse, persist as separate batches, and be retrievable by id. Exercises the full FastAPI path (parse → validate → store.add → DB) against real production data — 3 distinct billing-provider NPIs, 2 distinct transaction dates, 6 unique control numbers, 724 claims total. """ assert len(global_store.list()) == 0 # 1. POST every file. Each must return 200 with the expected envelope # and summary — proves the parse + payer-config + validation gates # work on real EDI, not just the synthetic fixture. for filename, expected_ctrl, expected_claims in EXPECTED_PRODFILES: path = PRODFILE_DIR / filename with open(path, "rb") as f: resp = client.post( "/api/parse-837", files={"file": (filename, f, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, f"{filename}: {resp.text}" body = resp.json() assert body["envelope"]["control_number"] == expected_ctrl, filename assert body["envelope"]["sender_id"] == "11525703", filename assert body["summary"]["total_claims"] == expected_claims, filename assert body["summary"]["passed"] == expected_claims, filename assert body["summary"]["failed"] == 0, filename # 2. Seven batches landed in the store (one per file). Confirms the # store.add() path was hit for every upload and that the in-process # state matches what the response said. batches = global_store.list() assert len(batches) == 7 assert {b.kind for b in batches} == {"837p"} assert {b.input_filename for b in batches} == {f for f, _, _ in EXPECTED_PRODFILES} # 3. Each persisted batch carries the full ParseResult: envelope, # summary, claims. Also sanity-check real-world variety: 3 NPIs, # 2 transaction dates (one each for 2026-06-11 vs 2026-06-17). by_filename = {b.input_filename: b for b in batches} total_claims = 0 distinct_npis: set[str] = set() distinct_dates: set[str] = set() for filename, expected_ctrl, expected_claims in EXPECTED_PRODFILES: rec = by_filename[filename] assert rec.result.envelope.control_number == expected_ctrl, filename assert rec.result.summary.total_claims == expected_claims, filename assert len(rec.result.claims) == expected_claims, filename total_claims += rec.result.summary.total_claims for claim in rec.result.claims: assert claim.billing_provider.npi, f"{filename} claim {claim.claim_id} missing NPI" distinct_npis.add(claim.billing_provider.npi) distinct_dates.add(str(claim.transaction_date)) assert total_claims == 724 assert len(distinct_npis) >= 3, f"expected ≥3 NPIs across the batch, got {distinct_npis}" assert len(distinct_dates) >= 2, f"expected ≥2 dates, got {distinct_dates}" # 4. Read path: every batch is retrievable by id and round-trips with # the same claim count that was just written. Covers global_store.get # and the DB-backed lookup that powers /api/batches/{id}. for rec in batches: fetched = global_store.get(rec.id) assert fetched is not None, rec.id assert fetched.id == rec.id assert len(fetched.result.claims) == rec.result.summary.total_claims assert fetched.result.envelope.control_number == rec.result.envelope.control_number # 5. Replay sanity: the first two files share control_number 991102984 # but were persisted as TWO distinct batches (no implicit dedup). # Documents the current behavior; flip the assertion if/when a # duplicate-control-number policy is added. ctrl_991102984 = [b for b in batches if b.result.envelope.control_number == "991102984"] assert len(ctrl_991102984) == 2 assert len({b.id for b in ctrl_991102984}) == 2 # --------------------------------------------------------------------------- # # Cross-pipeline reconciliation (837 + 835 together) # --------------------------------------------------------------------------- # PRODFILE_835_DIR_FOR_XP = ( Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "835fromco" ) @pytest.mark.skipif( not (PRODFILE_DIR.is_dir() and PRODFILE_835_DIR_FOR_XP.is_dir()), reason=f"production 837 or 835 files not present (gitignored): {PRODFILE_DIR}, {PRODFILE_835_DIR_FOR_XP}", ) def test_prodfile_cross_pipeline_reconciles(client: TestClient): """All production 837s + all production 835s end-to-end. Pipeline under test: POST /api/parse-837 (×N) → store.add (claim rows) → DB POST /api/parse-835 (×M) → store.add (remit rows) → reconcile.run → DB Designed to survive variable 837 / 835 file counts and arbitrary PCN overlap. Hard-codes NO match counts. Instead asserts invariants: 1. Every 837 file parses (200, total_claims ≥ 1) and lands as one 837p batch. 2. Every 835 file parses (200, total_claims ≥ 1) and lands as one 835 batch; each carries an R835_MULTI_BPR warning. 3. After all batches: store carries len(837 files) + len(835 files) batches with the expected kind split. 4. After all 835s loaded: store.list_unmatched() returns shapes: unmatched_claims = total_unique_837_pcns - total_matched unmatched_remittances = total_unique_835_pcns - total_matched Both counts are ≥ 0; the matched count is a function of how many 837 PCNs happen to appear in the 835 set — that's data- dependent and not asserted directly. 5. Per-835 reconciliation summaries are well-formed (matched, unmatched_claims, unmatched_remittances, skipped — all ints). Files are discovered via glob so adding more 837 files (the production set grows over time) needs no test edit. """ from cyclone.store import CycloneStore assert len(global_store.list()) == 0 # Discover files dynamically — works for N 837s and M 835s. prodfiles_837 = sorted(p for p in PRODFILE_DIR.iterdir() if p.is_file()) prodfiles_835 = sorted(p for p in PRODFILE_835_DIR_FOR_XP.iterdir() if p.is_file()) assert prodfiles_837, f"no 837 files at {PRODFILE_DIR}" assert prodfiles_835, f"no 835 files at {PRODFILE_835_DIR_FOR_XP}" # 1. Load every 837 first (claims must exist before 835 reconcile runs). unique_837_pcns: set[str] = set() for path in prodfiles_837: with open(path, "rb") as f: resp = client.post( "/api/parse-837", files={"file": (path.name, f, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, f"{path.name}: {resp.text}" body = resp.json() assert body["summary"]["total_claims"] >= 1, path.name assert body["summary"]["failed"] == 0, path.name for claim in body["claims"]: pcn = claim.get("claim_id") or claim.get("patient_control_number") if pcn: unique_837_pcns.add(pcn) # 2. Load every 835. Each triggers reconcile.run against the existing # 837 claims; the per-batch reconciliation summary records how many # of THIS batch's remits matched. unique_835_pcns: set[str] = set() per_835_matched: list[int] = [] per_835_summary_keys = {"matched", "unmatched_claims", "unmatched_remittances", "skipped"} for path in prodfiles_835: with open(path, "rb") as f: resp = client.post( "/api/parse-835", files={"file": (path.name, f, "application/octet-stream")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, f"{path.name}: {resp.text}" body = resp.json() assert body["summary"]["total_claims"] >= 1, path.name assert body["summary"]["failed"] == 0, path.name # CO Medicaid split-payment pattern: validator surfaces the # non-standard data as a warning; the batch still passes. assert any( issue["rule"] == "R835_MULTI_BPR" for issue in body["validation"]["warnings"] ), f"{path.name}: missing R835_MULTI_BPR warning" # Per-claim PCN extraction — UI shape is claim_id (not PCN). for cp in body["claims"]: pcn = cp.get("claim_id") or cp.get("payer_claim_control_number") if pcn: unique_835_pcns.add(pcn) # Reconciliation summary must be well-formed. rec = body["reconciliation"] assert set(rec.keys()) >= per_835_summary_keys, path.name for key in per_835_summary_keys: assert isinstance(rec[key], int), (path.name, key, rec[key]) assert rec["skipped"] == 0, path.name per_835_matched.append(rec["matched"]) # 3. Store shape: one batch per file, split by kind. batches = global_store.list() assert len(batches) == len(prodfiles_837) + len(prodfiles_835) by_kind: dict[str, set[str]] = {"837p": set(), "835": set()} for b in batches: by_kind[b.kind].add(b.input_filename) assert by_kind["837p"] == {p.name for p in prodfiles_837} assert by_kind["835"] == {p.name for p in prodfiles_835} # 4. Reconciliation invariants. matched_count is bounded by the size # of the smaller set; unmatched counts are the disjoint remainder. fresh_store = CycloneStore() unmatched = fresh_store.list_unmatched(kind="both") unmatched_claims = len(unmatched["claims"]) unmatched_remits = len(unmatched["remittances"]) total_matched = sum(per_835_matched) assert 0 <= total_matched <= len(unique_837_pcns), ( f"matched {total_matched} outside [0, {len(unique_837_pcns)}]" ) # Every matched pair consumes one claim and one remit. assert unmatched_claims == len(unique_837_pcns) - total_matched, ( f"unmatched_claims {unmatched_claims} != " f"{len(unique_837_pcns)} - {total_matched}" ) assert unmatched_remits == len(unique_835_pcns) - total_matched, ( f"unmatched_remittances {unmatched_remits} != " f"{len(unique_835_pcns)} - {total_matched}" ) # The two PCN sets are independent sources of truth: total deduped # claim rows + remittance rows in the DB must equal the input totals # adjusted for cross-batch dedup. The DB-level ground truth: claim_rows = sum( len(b.result.claims) for b in batches if b.kind == "837p" ) # Deduped claim rows == unique PCNs (one Claim row per PCN due to # the store.add dedup logic, same as 835 PCNs). assert claim_rows >= len(unique_837_pcns), ( f"claim_rows {claim_rows} < unique_837_pcns {len(unique_837_pcns)}" )