"""Smoke tests: every prodfiles/* file we have parsers for must parse cleanly. These tests are data-tolerant: they discover files via ``glob`` and assert invariants per file rather than hard-coded expected counts. If ops drops more files into ``docs/prodfiles/``, the tests stay green as long as the new files parse cleanly. Scope: - ``docs/prodfiles/claims/`` → 113 single-claim 837P files (output format reference) - ``docs/prodfiles/FromHPE/`` 837P .txt → 4 production 837 submissions to CO Medicaid - ``docs/prodfiles/FromHPE/`` *.999.x12 → ~1012 production transaction-set acks - ``docs/prodfiles/FromHPE/`` *.TA1.x12 → ~352 production interchange acks """ 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 DOCS_PRODFILES = Path(__file__).parent.parent.parent / "docs" / "prodfiles" CLAIMS_DIR = DOCS_PRODFILES / "claims" FROMHPE_DIR = DOCS_PRODFILES / "FromHPE" @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 _post(client: TestClient, endpoint: str, path: Path, content_type: str): with open(path, "rb") as f: return client.post( endpoint, files={"file": (path.name, f, content_type)}, headers={"Accept": "application/json"}, ) # --------------------------------------------------------------------------- # # 837P — claims/ (output format reference) + FromHPE/ (real submissions) # --------------------------------------------------------------------------- # @pytest.mark.skipif( not CLAIMS_DIR.is_dir(), reason=f"docs/prodfiles/claims/ not present at {CLAIMS_DIR}", ) def test_claims_prodfiles_parse_via_837_endpoint(client: TestClient): """Every 837P file in docs/prodfiles/claims/ parses cleanly. These single-claim files represent the output format our writer needs to produce. If any of them fails to parse, our round-trip is broken even before submission to the payer. Note: claims/ PCNs intentionally overlap with the axiscare 837 files loaded by ``test_prodfile_round_trip_persists_separately`` (verified 100% overlap). That means the store.add() dedup logic skips them silently — by design, since they represent the same claim shape. We assert the API response (parse + validate), not persistence count. """ paths = sorted(p for p in CLAIMS_DIR.iterdir() if p.is_file() and p.suffix == ".x12") assert paths, f"no 837P files at {CLAIMS_DIR}" for path in paths: resp = _post(client, "/api/parse-837", path, "text/plain") 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 assert body["summary"]["passed"] == body["summary"]["total_claims"], path.name # Every claim must have an id (CLM01) — that's what makes it # addressable in the API and the DB. for claim in body["claims"]: assert claim.get("claim_id"), f"{path.name}: missing claim_id" # Sender / receiver shape must match the axiscare prod data. assert body["envelope"]["sender_id"] == "11525703", path.name assert body["envelope"]["receiver_id"] == "COMEDASSISTPROG", path.name @pytest.mark.skipif( not FROMHPE_DIR.is_dir(), reason=f"docs/prodfiles/FromHPE/ not present at {FROMHPE_DIR}", ) def test_fromhpe_837_prodfiles_parse(client: TestClient): """The 4 production 837P submissions in FromHPE/ parse cleanly.""" paths = sorted(FROMHPE_DIR.glob("tp11525703-837P-*.txt")) assert paths, f"no 837P .txt files at {FROMHPE_DIR}" for path in paths: resp = _post(client, "/api/parse-837", path, "text/plain") 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 # All FromHPE 837s are sender 11525703 → COMEDASSISTPROG (HPE gateway) assert body["envelope"]["sender_id"] == "11525703", path.name batches = [b for b in global_store.list() if b.kind == "837p"] assert len(batches) == len(paths) # --------------------------------------------------------------------------- # # 999 — FromHPE/ production transaction-set acks # --------------------------------------------------------------------------- # @pytest.mark.skipif( not FROMHPE_DIR.is_dir(), reason=f"docs/prodfiles/FromHPE/ not present at {FROMHPE_DIR}", ) def test_fromhpe_999_prodfiles_parse(client: TestClient): """Every 999 ACK file in FromHPE/ parses and persists to /api/acks. These ~1012 files are real acks Colorado sends back after receiving our 837 submissions. If any fail to parse, our ingest of payer responses is broken. Known data anomaly: Colorado occasionally sends ``AK9*A*1*1*1`` — 1 received, 1 accepted, 1 rejected from a single set, which violates X12 spec (AK904 should be ``received - accepted``). The parser reports what the AK9 literally says; if this becomes a real problem, fix ``_ack_count_summary`` in api.py to clamp ``rejected = max(0, received - accepted)``. """ paths = sorted(FROMHPE_DIR.glob("*999.x12")) assert paths, f"no 999 files at {FROMHPE_DIR}" a_count = 0 # accepted r_count = 0 # rejected (set-level) e_count = 0 # partial / error malformed_ak9 = 0 # AK9 with accepted + rejected > received (CO bug) for path in paths: resp = _post(client, "/api/parse-999", path, "application/octet-stream") assert resp.status_code == 200, f"{path.name}: {resp.text}" body = resp.json() ack = body["ack"] assert ack["ack_code"] in {"A", "E", "R"}, ( f"{path.name}: unexpected ack_code {ack['ack_code']!r}" ) assert ack["received_count"] >= 1, path.name # Colorado's real 999s occasionally ship a malformed AK9 # (e.g. ``AK9*A*1*1*1``: 1 received, 1 accepted, 1 rejected, # which is impossible from a single set). The parser faithfully # reports what the AK9 says; we count those as anomalies but # don't fail on them. See follow-up note in test docstring. if ack["accepted_count"] + ack["rejected_count"] > ack["received_count"]: malformed_ak9 += 1 if ack["ack_code"] == "A": a_count += 1 elif ack["ack_code"] == "R": r_count += 1 else: e_count += 1 # Persisted as acks rows. Use the store directly to count (GET /api/acks # is paginated and we want to assert the total). from cyclone.store import CycloneStore fresh_store = CycloneStore() # acks live in the DB via store.add_ack; read via the same store path. assert _ack_count(fresh_store) == len(paths), ( f"persisted { _ack_count(fresh_store) } acks, expected {len(paths)}" ) # Sanity: every persisted ack round-trips through GET /api/acks/{id}. list_resp = client.get("/api/acks?limit=1000", headers={"Accept": "application/json"}) assert list_resp.status_code == 200, list_resp.text assert list_resp.json()["total"] == len(paths) # Real-data distribution: surface counts so an anomaly (e.g. all R) # is visible in test output even when the test passes. print( f"\n[999 prodfile distribution] A={a_count} R={r_count} E={e_count} " f"malformed_AK9={malformed_ak9} total={len(paths)}" ) # --------------------------------------------------------------------------- # # TA1 — FromHPE/ production interchange acks # --------------------------------------------------------------------------- # @pytest.mark.skipif( not FROMHPE_DIR.is_dir(), reason=f"docs/prodfiles/FromHPE/ not present at {FROMHPE_DIR}", ) def test_fromhpe_ta1_prodfiles_parse(client: TestClient): """Every TA1 file in FromHPE/ parses and persists to /api/ta1-acks. These files are real interchange acks Colorado sends back in response to our 837 submissions. All real production files are R (rejected) — that's the dataset ops gave us to test against. """ paths = sorted(FROMHPE_DIR.glob("*TA1.x12")) assert paths, f"no TA1 files at {FROMHPE_DIR}" a_count = 0 e_count = 0 r_count = 0 for path in paths: resp = _post(client, "/api/parse-ta1", path, "application/octet-stream") assert resp.status_code == 200, f"{path.name}: {resp.text}" body = resp.json() ta1 = body["ta1"] assert ta1["ack_code"] in {"A", "E", "R"}, ( f"{path.name}: unexpected ack_code {ta1['ack_code']!r}" ) assert ta1["control_number"], f"{path.name}: empty TA101" assert ta1["source_batch_id"].startswith("TA1-"), path.name if ta1["ack_code"] == "A": a_count += 1 elif ta1["ack_code"] == "R": r_count += 1 else: e_count += 1 # Verify persistence via GET /api/ta1-acks. The endpoint caps at # limit=1000 but reports total across all rows, so the assertion # works even with more rows than the cap. list_resp = client.get("/api/ta1-acks?limit=1000", headers={"Accept": "application/json"}) assert list_resp.status_code == 200, list_resp.text assert list_resp.json()["total"] == len(paths) # Real-data distribution: surface counts so an anomaly is visible. print( f"\n[TA1 prodfile distribution] A={a_count} R={r_count} E={e_count} " f"total={len(paths)}" ) def _ack_count(store: CycloneStore) -> int: """Count rows in the acks table.""" from cyclone import db with db.SessionLocal()() as s: return s.query(db.Ack).count()