"""Tests for POST /api/batches/{batch_id}/export-837. Reads Claim.raw_json for each requested claim_id and returns a ZIP of regenerated X12 837 files. No DB state mutation. Mirrors the X-Cyclone-Serialize-Errors convention from /api/inbox/rejected/resubmit?download=true. """ import io import json import zipfile from pathlib import Path from fastapi.testclient import TestClient from cyclone.api import app def _seed_batch(client: TestClient, filename: str = "claim.txt") -> dict: """Parse a single 837P file and return a dict with ``batch_id`` and ``claims``. The parse-837 happy-path response does not currently surface ``batch_id`` at the top level (it's a server-side UUID, surfaced in 409 errors but not in 200s). We look it up from the DB — the Batch row is already persisted by the time the parse endpoint returns. """ from cyclone import db from cyclone.db import Batch fixture = Path("tests/fixtures/co_medicaid_837p.txt").read_text() r = client.post( "/api/parse-837", files={"file": (filename, io.BytesIO(fixture.encode()), "text/plain")}, headers={"Accept": "application/json"}, ) assert r.status_code == 200, r.text body = r.json() with db.SessionLocal()() as s: most_recent = s.query(Batch).order_by(Batch.parsed_at.desc()).first() assert most_recent is not None, "expected at least one Batch row after parse" batch_id = most_recent.id return {"batch_id": batch_id, "claims": body["claims"]} def _claim_ids_from_seed(seeded: dict) -> list[str]: return [c["claim_id"] for c in seeded["claims"]] # --------------------------------------------------------------------------- # Happy path # --------------------------------------------------------------------------- def test_happy_path_returns_zip_with_one_x12_per_claim(): with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] claim_ids = _claim_ids_from_seed(seeded) r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": claim_ids}, ) assert r.status_code == 200, r.text assert r.headers["content-type"].startswith("application/zip") assert "attachment" in r.headers["content-disposition"] assert ( f"batch-{batch_id}-{len(claim_ids)}-claims.zip" in r.headers["content-disposition"] ) # Per-claim failures header absent on full success. assert "x-cyclone-serialize-errors" not in {k.lower() for k in r.headers.keys()} with zipfile.ZipFile(io.BytesIO(r.content)) as zf: names = zf.namelist() assert len(names) == len(claim_ids) # Each entry follows the HCPF outbound naming template # "{tpid}-837P-{yyyymmddhhmmssSSS}-1of1.x12" (the seeded # clearhouse TPID is "11525703"). Every name must be unique. from cyclone.edi.filenames import is_outbound_filename seen = set() for name in names: assert is_outbound_filename(name), ( f"expected HCPF outbound filename, got {name!r}" ) assert name not in seen, f"duplicate filename: {name}" seen.add(name) with zf.open(name) as f: first_line = f.readline().decode("ascii", errors="replace") assert first_line.startswith("ISA*"), f"{name} didn't start with ISA" def test_each_x12_in_zip_uses_clearhouse_submit_and_payer_receiver(): """Regression: regenerated 837s used to emit 'CYCLONE' / 'RECEIVER' placeholders. The export endpoint must thread the clearhouse submitter (dzinesco's TPID 11525703) and payer receiver (COMEDASSISTPROG) through to the serializer.""" with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] claim_ids = _claim_ids_from_seed(seeded) r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": claim_ids}, ) assert r.status_code == 200, r.text with zipfile.ZipFile(io.BytesIO(r.content)) as zf: for name in zf.namelist(): with zf.open(name) as f: text = f.read().decode("ascii") # Submitter block must use the clearhouse TPID + name, not # the 'CYCLONE' placeholder. assert "CYCLONE" not in text, f"{name} still emits CYCLONE placeholder" assert "11525703" in text, f"{name} missing clearhouse TPID" assert "Dzinesco" in text, f"{name} missing clearhouse name" # Receiver block must use the CO_TXIX payer config # (COMEDASSISTPROG), not the 'RECEIVER' placeholder. assert "RECEIVER" not in text, f"{name} still emits RECEIVER placeholder" assert "COMEDASSISTPROG" in text, f"{name} missing receiver id" # Loop 1000A requires PER — must be present, not omitted. assert "PER*IC*" in text, f"{name} missing required PER segment" # SBR09 should be 'MC' (Medicaid claim filing indicator), # not the member id. sbr_line = next(seg for seg in text.split("~") if seg.startswith("SBR*")) sbr09 = sbr_line.rstrip("~").split("*")[9] assert sbr09 == "MC", f"{name} SBR09 expected 'MC', got {sbr09!r}" def test_each_x12_in_zip_round_trips_through_parser(): """Fidelity check: each .x12 must parse back to a ClaimOutput deep-equal to the source row's raw_json (modulo recomputed validation).""" from cyclone.parsers.parse_837 import parse from cyclone.parsers.payer import PayerConfig with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] claim_ids = _claim_ids_from_seed(seeded) r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": claim_ids}, ) assert r.status_code == 200, r.text by_id = {c["claim_id"]: c for c in seeded["claims"]} with zipfile.ZipFile(io.BytesIO(r.content)) as zf: for name in zf.namelist(): with zf.open(name) as f: text = f.read().decode("ascii") result = parse(text, PayerConfig(name="CO_MEDICAID")) assert result.claims, f"{name} didn't parse back to any claims" # Claim ids must round-trip (proves the serializer didn't drop # or rewrite the id). assert result.claims[0].claim.claim_id in by_id, ( f"{name} parsed to an unknown claim_id" ) # --------------------------------------------------------------------------- # Partial-failure surface # --------------------------------------------------------------------------- def test_partial_failure_one_claim_with_no_raw_json_returns_zip_and_errors_header(): """If one claim has raw_json=None, the ZIP still returns for the others and the failure is surfaced via X-Cyclone-Serialize-Errors.""" from cyclone import db from cyclone.edi.filenames import is_outbound_filename with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] claim_ids = _claim_ids_from_seed(seeded) assert len(claim_ids) >= 2, "fixture must produce at least 2 claims" # Wipe raw_json on one claim to simulate a corrupted row. with db.SessionLocal()() as s: from cyclone.db import Claim target = s.get(Claim, claim_ids[0]) target.raw_json = None s.commit() r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": claim_ids}, ) assert r.status_code == 200, r.text # Filename uses SUCCESS count, not requested count. expected_success = len(claim_ids) - 1 assert ( f"batch-{batch_id}-{expected_success}-claims.zip" in r.headers["content-disposition"] ) err_header = r.headers.get("x-cyclone-serialize-errors") assert err_header, "expected X-Cyclone-Serialize-Errors header" errs = json.loads(err_header) assert len(errs) == 1 assert errs[0]["claim_id"] == claim_ids[0] assert "raw_json" in errs[0]["reason"].lower() with zipfile.ZipFile(io.BytesIO(r.content)) as zf: names = zf.namelist() assert len(names) == expected_success # Every entry follows HCPF outbound template; none is the # 'claim-CLM-X.x12' placeholder from the old endpoint. for name in names: assert is_outbound_filename(name), f"non-HCPF name: {name!r}" def test_all_claims_fail_to_serialize_returns_422(): from cyclone import db with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] claim_ids = _claim_ids_from_seed(seeded) assert len(claim_ids) >= 1 with db.SessionLocal()() as s: from cyclone.db import Claim for cid in claim_ids: c = s.get(Claim, cid) c.raw_json = None s.commit() r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": claim_ids}, ) assert r.status_code == 422, r.text body = r.json() # The 422 body surfaces the failure list so the UI can show details # without parsing a header. errs = body.get("detail", {}).get("serialize_errors") or body.get("serialize_errors") assert errs is not None assert len(errs) == len(claim_ids) for entry in errs: assert entry["claim_id"] in claim_ids # --------------------------------------------------------------------------- # Error cases # --------------------------------------------------------------------------- def test_empty_claim_ids_returns_400(): with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": []}, ) assert r.status_code == 400, r.text def test_missing_claim_ids_key_returns_400(): with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] r = client.post( f"/api/batches/{batch_id}/export-837", json={}, ) assert r.status_code == 400, r.text def test_unknown_batch_id_returns_404(): with TestClient(app) as client: r = client.post( "/api/batches/BATCH-DOES-NOT-EXIST/export-837", json={"claim_ids": ["CLM-1"]}, ) assert r.status_code == 404, r.text body = r.json() assert "BATCH-DOES-NOT-EXIST" in (body.get("detail") or "") def test_unknown_claim_id_silently_omitted(): """A claim_id that doesn't exist is omitted from the ZIP and surfaced in the errors header — same convention as the resubmit endpoint.""" with TestClient(app) as client: seeded = _seed_batch(client) batch_id = seeded["batch_id"] real_ids = _claim_ids_from_seed(seeded) assert real_ids, "fixture must produce at least 1 claim" requested = real_ids + ["CLM-GHOST"] r = client.post( f"/api/batches/{batch_id}/export-837", json={"claim_ids": requested}, ) assert r.status_code == 200, r.text # Filename uses success count = len(real_ids), not len(requested). assert ( f"batch-{batch_id}-{len(real_ids)}-claims.zip" in r.headers["content-disposition"] ) err_header = r.headers.get("x-cyclone-serialize-errors") assert err_header errs = json.loads(err_header) assert any(e["claim_id"] == "CLM-GHOST" for e in errs) with zipfile.ZipFile(io.BytesIO(r.content)) as zf: names = zf.namelist() # HCPF outbound filenames, one per real claim. We don't pin the # exact ts (it depends on the wall clock at test time), but the # count and the HCPF template must match. from cyclone.edi.filenames import is_outbound_filename assert len(names) == len(real_ids) for name in names: assert is_outbound_filename(name), f"non-HCPF name: {name!r}"