Files
cyclone/backend/tests/test_clearhouse_api.py
T
Tyler c62826daea feat(sp9): multi-payer, multi-NPI, SFTP stub for dzinesco/TOC
- providers table seeded with 3 NPIs (Montrose, Delta, Salida)
- payers + payer_configs (per-tx config) join table
- clearhouse singleton (dzinesco identity, SFTP block, filename block)
- config/payers.yaml loader with Pydantic schema validation
- cyclone/edi/filenames.py: HCPF X12 File Naming Standards helpers
- cyclone/clearhouse/sftp.py: stub that copies to ./var/sftp/staging/...
- cyclone/secrets.py: macOS Keychain wrapper via keyring
- 11 new validation rules R200-R210 (CO MAP + HCPF naming)
- 6 new API endpoints (/api/clearhouse/*, /api/config/*, /api/admin/reload-config)
- migration 0007
- 72 new tests (646 total passing, was 574)

See spec: docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md
2026-06-20 23:07:31 -06:00

104 lines
3.1 KiB
Python

"""SP9 — clearhouse API endpoint tests."""
from __future__ import annotations
import os
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from cyclone import db as db_mod
from cyclone.api import app
@pytest.fixture
def client(tmp_path, monkeypatch):
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
db_mod._reset_for_tests()
db_mod.init_db()
with TestClient(app) as c:
yield c
db_mod._reset_for_tests()
def test_get_clearhouse_seeded(client):
# Lifespan runs ensure_clearhouse_seeded()
r = client.get("/api/clearhouse")
assert r.status_code == 200
body = r.json()
assert body["name"] == "dzinesco"
assert body["tpid"] == "11525703"
assert body["sftp_block"]["stub"] is True
assert "FromHPE" in body["sftp_block"]["paths"]["outbound"]
assert "ToHPE" in body["sftp_block"]["paths"]["inbound"]
def test_list_providers(client):
r = client.get("/api/config/providers")
assert r.status_code == 200, r.text
body = r.json()
assert {p["label"] for p in body} == {"Montrose", "Delta", "Salida"}
assert {p["npi"] for p in body} == {"1881068062", "1851446637", "1467507269"}
def test_get_provider_by_npi(client):
r = client.get("/api/config/providers/1881068062")
assert r.status_code == 200
assert r.json()["label"] == "Montrose"
def test_get_provider_404(client):
r = client.get("/api/config/providers/9999999999")
assert r.status_code == 404
def test_list_payers(client):
r = client.get("/api/config/payers")
assert r.status_code == 200
body = r.json()
assert len(body) == 1
assert body[0]["payer_id"] == "CO_TXIX"
def test_list_payer_configs_for_co_txix(client):
r = client.get("/api/config/payers/CO_TXIX/configs")
assert r.status_code == 200
body = r.json()
# The lifespan loads config/payers.yaml; configs come from the registry
tx_types = {c["transaction_type"] for c in body}
assert "837P" in tx_types or "835" in tx_types
def test_reload_config(client):
r = client.post("/api/admin/reload-config")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert isinstance(body["loaded"], int)
def test_submit_clearhouse_rejects_empty_claim_ids(client):
r = client.post("/api/clearhouse/submit", json={"claim_ids": [], "payer_id": "CO_TXIX"})
assert r.status_code == 400
def test_submit_clearhouse_rejects_missing_payer_id(client):
r = client.post("/api/clearhouse/submit", json={"claim_ids": ["X"]})
assert r.status_code == 400
def test_submit_clearhouse_handles_unknown_claim_id(client):
r = client.post("/api/clearhouse/submit", json={
"claim_ids": ["DOES_NOT_EXIST"],
"payer_id": "CO_TXIX",
})
# 200 with per-claim ok:false entries (graceful degradation)
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["stub"] is True
assert len(body["submitted"]) == 1
assert body["submitted"][0]["ok"] is False
assert "not found" in body["submitted"][0]["error"] or "cannot be re-serialized" in body["submitted"][0]["error"]