Tyler
c62826daea
- 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
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""SP9 — providers/payers/clearhouse seed + CRUD tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from cyclone import db as db_mod
|
|
from cyclone.providers import Provider
|
|
from cyclone.store import store
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_db(tmp_path, monkeypatch):
|
|
"""Use a fresh in-memory SQLite for each test."""
|
|
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
|
|
db_mod._reset_for_tests()
|
|
db_mod.init_db()
|
|
yield
|
|
db_mod._reset_for_tests()
|
|
|
|
|
|
def test_seed_creates_3_providers():
|
|
store.ensure_clearhouse_seeded()
|
|
providers = store.list_providers()
|
|
labels = {p.label for p in providers}
|
|
assert labels == {"Montrose", "Delta", "Salida"}
|
|
npis = {p.npi for p in providers}
|
|
assert npis == {"1881068062", "1851446637", "1467507269"}
|
|
|
|
|
|
def test_seed_uses_consistent_tax_id_and_taxonomy():
|
|
store.ensure_clearhouse_seeded()
|
|
providers = store.list_providers()
|
|
for p in providers:
|
|
assert p.tax_id == "721587149"
|
|
assert p.taxonomy_code == "251E00000X"
|
|
assert p.legal_name == "TOC, Inc."
|
|
assert p.address_line1 == "1100 East Main St"
|
|
assert p.address_line2 == "Suite A"
|
|
assert p.city == "Montrose"
|
|
assert p.state == "CO"
|
|
assert p.zip == "814014063"
|
|
|
|
|
|
def test_seed_creates_clearhouse_singleton():
|
|
store.ensure_clearhouse_seeded()
|
|
ch = store.get_clearhouse()
|
|
assert ch is not None
|
|
assert ch.id == 1
|
|
assert ch.name == "dzinesco"
|
|
assert ch.tpid == "11525703"
|
|
assert ch.submitter_name == "Dzinesco"
|
|
assert ch.sftp_block.host == "mft.gainwelltechnologies.com"
|
|
assert ch.sftp_block.stub is True
|
|
assert "FromHPE" in ch.sftp_block.paths["outbound"]
|
|
assert "ToHPE" in ch.sftp_block.paths["inbound"]
|
|
|
|
|
|
def test_seed_creates_co_txix_payer_with_both_configs():
|
|
store.ensure_clearhouse_seeded()
|
|
payers = store.list_payers()
|
|
assert {p.payer_id for p in payers} == {"CO_TXIX"}
|
|
p837 = store.get_payer_config("CO_TXIX", "837P")
|
|
p835 = store.get_payer_config("CO_TXIX", "835")
|
|
assert p837 is not None and p837["payer_id"] == "CO_TXIX"
|
|
assert p835 is not None and "1811725341" in p835["expected_payer_tax_ids"]
|
|
|
|
|
|
def test_seed_is_idempotent():
|
|
store.ensure_clearhouse_seeded()
|
|
store.ensure_clearhouse_seeded()
|
|
store.ensure_clearhouse_seeded()
|
|
assert len(store.list_providers()) == 3
|
|
assert len(store.list_payers()) == 1
|
|
assert store.get_clearhouse() is not None
|
|
|
|
|
|
def test_get_provider_returns_none_for_unknown_npi():
|
|
store.ensure_clearhouse_seeded()
|
|
assert store.get_provider("9999999999") is None
|
|
|
|
|
|
def test_upsert_provider_creates_then_updates():
|
|
store.ensure_clearhouse_seeded()
|
|
p = Provider(
|
|
npi="1111111111",
|
|
label="Test",
|
|
legal_name="Test, Inc.",
|
|
tax_id="123456789",
|
|
taxonomy_code="207R00000X",
|
|
address_line1="1 Test Way",
|
|
city="Testville",
|
|
state="CO",
|
|
zip="80000",
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
stored = store.upsert_provider(p)
|
|
assert stored.label == "Test"
|
|
p2 = p.model_copy(update={"label": "Updated"})
|
|
stored2 = store.upsert_provider(p2)
|
|
assert stored2.label == "Updated"
|
|
assert len(store.list_providers(is_active=None)) == 4
|
|
|
|
|
|
def test_list_providers_filter_is_active():
|
|
store.ensure_clearhouse_seeded()
|
|
p = Provider(
|
|
npi="2222222222",
|
|
label="Inactive",
|
|
legal_name="X",
|
|
tax_id="1",
|
|
taxonomy_code="X",
|
|
address_line1="X",
|
|
city="X",
|
|
state="CO",
|
|
zip="0",
|
|
is_active=False,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
store.upsert_provider(p)
|
|
active = store.list_providers(is_active=True)
|
|
assert {pp.label for pp in active} == {"Montrose", "Delta", "Salida"}
|
|
all_p = store.list_providers(is_active=None)
|
|
assert {pp.label for pp in all_p} == {"Montrose", "Delta", "Salida", "Inactive"}
|