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
parent fbe9940a3f
commit c62826daea
23 changed files with 2869 additions and 6 deletions
@@ -0,0 +1,92 @@
+"""SP9 — SFTP stub tests."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from cyclone.clearhouse import SftpClient
+from cyclone.providers import SftpBlock
+
+
+@pytest.fixture
+def sftp_block(tmp_path):
+    staging = tmp_path / "staging"
+    return SftpBlock(
+        host="mft.gainwelltechnologies.com",
+        port=22,
+        username="colorado-fts\\coxix_prod_11525703",
+        paths={
+            "outbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+            "inbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+        },
+        stub=True,
+        staging_dir=str(staging),
+        poll_seconds=300,
+        auth={"method": "keychain", "secret_ref": "sftp.gainwell.password"},
+    )
+
+
+def test_stub_writes_preserving_remote_path(sftp_block, tmp_path):
+    client = SftpClient(sftp_block)
+    remote = "/CO XIX/PROD/coxix_prod_11525703/FromHPE/11525703-837P-20260620132243505-1of1.x12"
+    target = client.write_file(remote, b"ISA*00*...~IEA*1*1~")
+    assert target.exists()
+    assert target.read_bytes() == b"ISA*00*...~IEA*1*1~"
+    # Confirm the full nested MFT path is preserved under staging
+    rel = target.relative_to(sftp_block.staging_dir)
+    assert str(rel) == "CO XIX/PROD/coxix_prod_11525703/FromHPE/11525703-837P-20260620132243505-1of1.x12"
+
+
+def test_stub_creates_parent_dirs(sftp_block):
+    client = SftpClient(sftp_block)
+    remote = "/deep/nested/path/file.x12"
+    target = client.write_file(remote, b"data")
+    assert target.exists()
+    assert target.parent.is_dir()
+
+
+def test_stub_list_inbound_empty_when_no_local_files(sftp_block):
+    client = SftpClient(sftp_block)
+    assert client.list_inbound() == []
+
+
+def test_stub_list_inbound_returns_local_files(sftp_block):
+    # Simulate operator dropping a file in the inbound staging dir
+    inbound_dir = Path(sftp_block.staging_dir) / "CO XIX/PROD/coxix_prod_11525703/ToHPE"
+    inbound_dir.mkdir(parents=True, exist_ok=True)
+    (inbound_dir / "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12").write_bytes(b"X")
+    client = SftpClient(sftp_block)
+    files = client.list_inbound()
+    assert len(files) == 1
+    assert files[0].name.startswith("TP11525703-837P_M019048402")
+    assert files[0].size == 1
+
+
+def test_stub_get_secret_returns_stub_when_keychain_empty(sftp_block):
+    client = SftpClient(sftp_block)
+    # Keychain is empty on the test box, so we get the stub sentinel
+    secret = client.get_secret("sftp.gainwell.password")
+    assert secret == "<stub-secret>"
+
+
+def test_real_mode_write_raises_not_implemented(sftp_block):
+    block = sftp_block.model_copy(update={"stub": False})
+    client = SftpClient(block)
+    with pytest.raises(NotImplementedError, match="SP13"):
+        client.write_file("/x.x12", b"y")
+
+
+def test_real_mode_list_inbound_raises_not_implemented(sftp_block):
+    block = sftp_block.model_copy(update={"stub": False})
+    client = SftpClient(block)
+    with pytest.raises(NotImplementedError, match="SP13"):
+        client.list_inbound()
+
+
+def test_stub_read_file_raises(sftp_block):
+    client = SftpClient(sftp_block)
+    with pytest.raises(RuntimeError, match="Stub SFTP cannot read"):
+        client.read_file("/x.x12")