"""Tests for the FastAPI surface in ``cyclone.api`` for the 835 ERA endpoint.""" from __future__ import annotations import json 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_835.txt" UNBALANCED = Path(__file__).parent / "fixtures" / "unbalanced_835.txt" @pytest.fixture def client() -> TestClient: return TestClient(app) # --------------------------------------------------------------------------- # # JSON response path # --------------------------------------------------------------------------- # def test_parse_835_endpoint_returns_json(client: TestClient): text = FIXTURE.read_text() resp = client.post( "/api/parse-835", files={"file": ("co_medicaid_835.txt", text, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, resp.text body = resp.json() assert "envelope" in body assert "financial_info" in body assert "trace" in body assert "payer" in body assert "payee" in body assert "claims" in body and len(body["claims"]) == 2 assert "summary" in body assert body["summary"]["total_claims"] == 2 assert body["summary"]["passed"] == 2 assert body["summary"]["failed"] == 0 # --------------------------------------------------------------------------- # # NDJSON streaming path # --------------------------------------------------------------------------- # def test_parse_835_endpoint_streams_ndjson(client: TestClient): text = FIXTURE.read_text() resp = client.post( "/api/parse-835", files={"file": ("co_medicaid_835.txt", text, "text/plain")}, ) assert resp.status_code == 200, resp.text assert resp.headers["content-type"].startswith("application/x-ndjson") lines = list(resp.iter_lines()) # 1 envelope + 1 financial_info + 1 trace + 1 payer + 1 payee + 2 claims + 1 summary = 8 assert len(lines) == 8 parsed = [json.loads(line) for line in lines] assert parsed[0]["type"] == "envelope" assert parsed[1]["type"] == "financial_info" assert parsed[2]["type"] == "trace" assert parsed[3]["type"] == "payer" assert parsed[4]["type"] == "payee" assert parsed[5]["type"] == "claim_payment" assert parsed[6]["type"] == "claim_payment" assert parsed[7]["type"] == "summary" # When include_raw_segments defaults to True, each claim carries raw segments. for obj in parsed[5:7]: assert "raw_segments" in obj["data"] assert isinstance(obj["data"]["raw_segments"], list) # Summary numbers match the JSON path. assert parsed[7]["data"]["total_claims"] == 2 assert parsed[7]["data"]["passed"] == 2 # The streaming summary carries the server-side batch id so the # frontend can call /api/batches/{id}/export-837 without a separate # GET /api/batches round-trip (matches the parallel fix on # /api/parse-837). assert parsed[7]["type"] == "summary" assert isinstance(parsed[7]["data"]["batch_id"], str) assert len(parsed[7]["data"]["batch_id"]) == 32 # uuid4().hex def test_parse_835_endpoint_streams_ndjson_without_raw_segments(client: TestClient): text = FIXTURE.read_text() resp = client.post( "/api/parse-835?include_raw_segments=false", files={"file": ("co_medicaid_835.txt", text, "text/plain")}, ) assert resp.status_code == 200 assert resp.headers["content-type"].startswith("application/x-ndjson") claims = [ json.loads(line) for line in resp.iter_lines() if json.loads(line)["type"] == "claim_payment" ] assert len(claims) == 2 for c in claims: assert c["data"]["raw_segments"] == [] # --------------------------------------------------------------------------- # # Validation / error paths # --------------------------------------------------------------------------- # def test_parse_835_endpoint_rejects_missing_file(client: TestClient): # FastAPI's `File(...)` (no default) → 422 Unprocessable Entity. resp = client.post("/api/parse-835") assert resp.status_code == 422 def test_parse_835_endpoint_handles_payer_query_param(client: TestClient): text = FIXTURE.read_text() for payer in ("co_medicaid_835", "generic_835"): resp = client.post( f"/api/parse-835?payer={payer}", files={"file": ("co_medicaid_835.txt", text, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 200, (payer, resp.text) body = resp.json() assert body["summary"]["total_claims"] == 2 assert body["summary"]["passed"] == 2 def test_parse_835_endpoint_rejects_unknown_payer(client: TestClient): text = FIXTURE.read_text() resp = client.post( "/api/parse-835?payer=does_not_exist", files={"file": ("co_medicaid_835.txt", text, "text/plain")}, ) assert resp.status_code == 400 def test_parse_835_endpoint_unbalanced_returns_422(client: TestClient): text = UNBALANCED.read_text() resp = client.post( "/api/parse-835", files={"file": ("unbalanced_835.txt", text, "text/plain")}, headers={"Accept": "application/json"}, ) assert resp.status_code == 422, resp.text body = resp.json() assert body["summary"]["failed"] == 1 # The R835_BAL_BPR_vs_CLP04 rule should have fired. assert any( issue["rule"] == "R835_BAL_BPR_vs_CLP04" for issue in body["validation"]["errors"] ) # --------------------------------------------------------------------------- # # CORS # --------------------------------------------------------------------------- # def test_cors_headers_present_for_835(client: TestClient): resp = client.options( "/api/parse-835", headers={ "Origin": "http://localhost:5173", "Access-Control-Request-Method": "POST", "Access-Control-Request-Headers": "content-type", }, ) assert resp.headers.get("access-control-allow-origin") == "http://localhost:5173" assert "POST" in resp.headers.get("access-control-allow-methods", "").upper() # --------------------------------------------------------------------------- # # Production 835 round-trip (parser → API → store → DB) # --------------------------------------------------------------------------- # # Path to the production 835 ERA files from CO Medicaid. 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_835_DIR = Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "835fromco" # (filename, expected_control_number, expected_total_claims, expected_paid). # Derived from /tmp/parse_prod835.py on 2026-06-20 against the 5 files in # PRODFILE_835_DIR. All 5 are CO Medicaid Information Only (BPR01="I") ERAs # that use a "split payment" pattern: each file carries 3 BPR segments # whose BPR02 amounts sum to the total paid. ``expected_paid`` is the sum # across all BPRs in the file (the parser accumulates them; the validator # emits R835_MULTI_BPR as a warning). EXPECTED_PRODFILES_835: list[tuple[str, str, int, str]] = [ ("tp11525703-835_M019110219-20260525001606050-1of1.x12", "200010701", 735, "58445.24"), ("tp11525703-835_M019200601-20260601003507042-1of1.x12", "200000831", 671, "50263.37"), ("tp11525703-835_M019311719-20260608002507036-1of1.x12", "200007522", 352, "52978.74"), ("tp11525703-835_M019414762-20260615005516914-1of1.x12", "200023767", 349, "17913.97"), ("tp11525703-835_M019506114-20260619075017291-1of1.x12", "200003981", 1267, "87722.18"), ] @pytest.mark.skipif( not PRODFILE_835_DIR.is_dir(), reason=f"production 835 files not present at {PRODFILE_835_DIR} (gitignored)", ) def test_prodfile_round_trip_persists_separately(client: TestClient): """All 5 production 835 ERA files from CO Medicaid 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,374 remittances, 11,163 service payments, 7,510 CAS adjustments, 5 distinct control numbers, $8,231.63 total paid. No 837s are loaded, so all remittances are unmatched by design. The T10 reconciliation summary returned with each 835 must reflect that. """ from decimal import Decimal assert len(global_store.list()) == 0 # 1. POST every file. Each must return 200 with the expected envelope, # financial_info, payer XV, and per-claim CLP count. The 835 path # also runs batch-level validation (R835_BAL_BPR_vs_CLP04, etc.) # which must pass on real data. for filename, expected_ctrl, expected_claims, expected_paid in EXPECTED_PRODFILES_835: path = PRODFILE_835_DIR / filename with open(path, "rb") as f: resp = client.post( "/api/parse-835", files={"file": (filename, f, "application/octet-stream")}, 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["payer"]["id"] == "7912900843", filename # CO Medicaid health plan ID assert body["payee"]["npi"] == "1467507269", filename assert body["summary"]["total_claims"] == expected_claims, filename assert body["summary"]["passed"] == expected_claims, filename assert body["summary"]["failed"] == 0, filename # BPR02 (total paid) is a Decimal-as-string in the JSON envelope. # For CO Medicaid split-payment files, the parser sums the three # BPR segments into one paid_amount. assert Decimal(body["financial_info"]["paid_amount"]) == Decimal(expected_paid), filename # All 5 prod files carry multiple BPR segments; the validator must # flag that as a warning (R835_MULTI_BPR) but pass the batch. multi_bpr_warnings = [ issue for issue in body["validation"]["warnings"] if issue["rule"] == "R835_MULTI_BPR" ] assert multi_bpr_warnings, f"{filename}: expected R835_MULTI_BPR warning" assert body["validation"]["errors"] == [], filename # T10 reconciliation summary is part of the 835 response. No 837s # loaded, so every remit in this batch must be unmatched. assert "reconciliation" in body, filename rec = body["reconciliation"] assert rec["matched"] == 0, filename assert rec["unmatched_claims"] == 0, filename # ``unmatched_remittances`` is global (across all batches), so # we just assert it's a non-negative int — the precise total is # covered by the store-level assertion below. assert rec["unmatched_remittances"] >= 0, filename assert rec["skipped"] == 0, filename # 2. Five batches landed in the store. One per file. batches = global_store.list() assert len(batches) == 5 assert {b.kind for b in batches} == {"835"} assert {b.input_filename for b in batches} == {f for f, _, _, _ in EXPECTED_PRODFILES_835} # 3. Each persisted batch carries the full ParseResult: envelope, # financial_info, payer, payee, and a claims list with the right # size. Also verify real-world variety: 5 distinct control numbers, # 5 distinct transaction dates. by_filename = {b.input_filename: b for b in batches} total_clps = 0 total_svcs = 0 total_cas = 0 unique_pcns: set[str] = set() distinct_ctrl: set[str] = set() distinct_dates: set[str] = set() for filename, expected_ctrl, expected_claims, _ in EXPECTED_PRODFILES_835: 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 for cp in rec.result.claims: assert cp.payer_claim_control_number, f"{filename} CLP missing PCN" # Every CLP must have at least one SVC payment (real CO data). assert cp.service_payments, f"{filename} CLP {cp.payer_claim_control_number} has no SVCs" total_clps += len(rec.result.claims) total_svcs += sum(len(c.service_payments) for c in rec.result.claims) total_cas += sum( len(sp.adjustments) for c in rec.result.claims for sp in c.service_payments ) unique_pcns.update(c.payer_claim_control_number for c in rec.result.claims) distinct_ctrl.add(rec.result.envelope.control_number) distinct_dates.add(str(rec.result.envelope.transaction_date)) assert total_clps == 3374 assert total_svcs == 11163 assert total_cas == 7510 assert len(distinct_ctrl) == 5 assert len(distinct_dates) == 5 # 4. Read path: every batch is retrievable by id and round-trips with # the same remit 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. Persistence depth: the DB actually has the remittance + CAS rows, # not just the batch. ``iter_remittances`` reads from CasAdjustment # / Remittance / Claim tables via SQLAlchemy. # # The store's add() path dedupes remittances by PCN: a second 835 # referencing an already-stored PCN is skipped (the original # Remittance row wins). Real CO Medicaid data has overlapping PCNs # across the 5 prod files, so the persisted row count equals the # count of distinct PCNs across all batches — not the sum of CLPs. from cyclone.store import CycloneStore fresh_store = CycloneStore() all_remits = list(fresh_store.iter_remittances(limit=10000)) assert len(all_remits) == len(unique_pcns) # No duplicate PCNs survived the dedup; sanity check on persistence. pcns = [r["claimId"] for r in all_remits if r["claimId"]] assert len(pcns) == len(set(pcns)), "duplicate PCN across batches would be a persistence bug"