Files
cyclone/backend/tests/test_providers_seed.py
T
Nora dd7da18279 fix(sftp): swap inbound/outbound paths — FromHPE is inbound, ToHPE is outbound
The dzinesco SP9 seed had `paths.inbound` and `paths.outbound` mapped to
the wrong Gainwell MFT directories:

  - paths.outbound  was  FromHPE/   (HPE sends files FROM here TO us — inbound)
  - paths.inbound   was  ToHPE/     (we send files TO here — outbound)

So `/api/clearhouse/submit` was writing 837P claims to FromHPE (where
HPE puts acks/835s) and the SP16 scheduler was polling ToHPE (where our
claims go). 999 / TA1 / 835 files in the real FromHPE inbox were
unreachable.

Semantics per operator (2026-06-24):
  - FromHPE = HPE/Gainwell → us  = 999, TA1, 835  (inbound)
  - ToHPE   = us → HPE/Gainwell = 837P claims    (outbound)

**Runtime code (the actual fix):**
  - backend/src/cyclone/store.py            — SP9 seed paths flipped
  - backend/src/cyclone/edi/filenames.py    — docstring corrected

**Test fixtures + assertions (would otherwise fail on the new seed):**
  - backend/tests/test_clearhouse_api.py
  - backend/tests/test_providers_seed.py
  - backend/tests/test_sftp_stub.py        — incl. inbound dir paths
  - backend/tests/test_sftp_paramiko.py
  - backend/tests/test_store_update_clearhouse.py
  - backend/tests/test_api_clearhouse_patch.py
  - backend/tests/test_scheduler.py
  - backend/tests/test_api_scheduler.py

**Docs (text-only — keeps the codebase self-consistent):**
  - README.md
  - docs/reference/co-medicaid.md
  - docs/superpowers/specs/2026-06-20-cyclone-multi-payer-npi-sftp-design.md

**Operator action required after merge:** the existing clearhouse row in
`~/.local/share/cyclone/cyclone.db` was seeded with the old wrong
paths. Easiest recovery:
  1. `rm ~/.local/share/cyclone/cyclone.db` and let `ensure_clearhouse_seeded` re-run on next boot, OR
  2. PATCH /api/clearhouse with the new `paths` block (the SP25
     reconfigure hook picks it up live, including by the running scheduler).

**Verification:**
  - 82 tests in the affected files pass (test_clearhouse_api, test_providers_seed,
    test_sftp_stub, test_sftp_paramiko, test_store_update_clearhouse,
    test_api_clearhouse_patch, test_scheduler, test_api_scheduler, test_filenames).
  - Full backend suite: 1029 pass + 36 pre-existing order-dependent flakes
    unrelated to this change (verified by running the same tests in isolation).
2026-06-24 22:05:55 -06:00

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 "ToHPE" in ch.sftp_block.paths["outbound"]
assert "FromHPE" 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"}