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
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
"""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")
|