feat(sp25): store.update_clearhouse() for PATCH endpoint

This commit is contained in:
Nora
2026-06-24 15:30:53 -06:00
parent b2d88d13d3
commit 2d3667738d
2 changed files with 132 additions and 0 deletions
+43
View File
@@ -2302,6 +2302,49 @@ class CycloneStore:
"updated_at": row.updated_at, "updated_at": row.updated_at,
}) })
def update_clearhouse(self, block: Clearhouse) -> Clearhouse:
"""Replace the singleton clearhouse row. SP25.
Used by ``PATCH /api/clearhouse`` to flip ``sftp_block.stub``
and adjust host/port/paths without touching SQLite directly.
Raises ``LookupError`` if the singleton row is missing — the
caller is expected to run the lifespan seed first.
"""
from cyclone.db import ClearhouseORM
with db.SessionLocal()() as s:
row = s.get(ClearhouseORM, 1)
if row is None:
raise LookupError(
"clearhouse singleton row missing; run the lifespan "
"seed (ensure_clearhouse_seeded) before PATCH /api/clearhouse"
)
row.name = block.name
row.tpid = block.tpid
row.submitter_name = block.submitter_name
row.submitter_id_qual = block.submitter_id_qual
row.submitter_contact_name = block.submitter_contact_name
row.submitter_contact_email = block.submitter_contact_email
row.filename_block_json = json.loads(
json.dumps(block.filename_block.model_dump())
)
row.sftp_block_json = json.loads(
json.dumps(block.sftp_block.model_dump())
)
row.updated_at = datetime.now(timezone.utc).isoformat()
s.commit()
return Clearhouse.model_validate({
"id": 1,
"name": row.name,
"tpid": row.tpid,
"submitter_name": row.submitter_name,
"submitter_id_qual": row.submitter_id_qual,
"submitter_contact_name": row.submitter_contact_name,
"submitter_contact_email": row.submitter_contact_email,
"filename_block": dict(row.filename_block_json),
"sftp_block": dict(row.sftp_block_json),
"updated_at": row.updated_at,
})
def ensure_clearhouse_seeded(self) -> None: def ensure_clearhouse_seeded(self) -> None:
"""Insert the default clearhouse singleton + 3 providers + CO_TXIX payer """Insert the default clearhouse singleton + 3 providers + CO_TXIX payer
if they don't exist. Idempotent. Called from the API lifespan.""" if they don't exist. Idempotent. Called from the API lifespan."""
@@ -0,0 +1,89 @@
"""SP25 — store.update_clearhouse() round-trip test.
The PATCH /api/clearhouse endpoint needs a way to replace the singleton
clearhouse row inside one session. This test exercises the method
directly against a fresh in-memory DB.
"""
from __future__ import annotations
import pytest
from cyclone import db
from cyclone.providers import Clearhouse, SftpBlock
from cyclone.store import store as cycl_store
def _seed_clearhouse():
"""Insert the default clearhouse row used by SP9's lifespan seed."""
cycl_store.ensure_clearhouse_seeded()
def test_update_clearhouse_round_trip():
"""Replace all fields on the singleton row; GET returns the new values."""
_seed_clearhouse()
new_block = 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=False,
staging_dir="./var/sftp/staging",
auth={"password_keychain_account": "sftp.gainwell.password"},
)
new_clearhouse = Clearhouse.model_validate({
"id": 1,
"name": "dzinesco",
"tpid": "11525703",
"submitter_name": "Dzinesco",
"submitter_id_qual": "46",
"submitter_contact_name": "Tyler Martinez",
"submitter_contact_email": "tyler@dzinesco.com",
"filename_block": {
"tz": "America/Denver",
"outbound_template": "{tpid}-{tx}-{ts_mt}-1of1.{ext}",
"inbound_template": "TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12",
},
"sftp_block": new_block.model_dump(),
"updated_at": "2026-06-24T00:00:00+00:00",
})
cycl_store.update_clearhouse(new_clearhouse)
fetched = cycl_store.get_clearhouse()
assert fetched is not None
assert fetched.sftp_block.stub is False
assert fetched.sftp_block.host == "mft.gainwelltechnologies.com"
assert fetched.sftp_block.auth == {"password_keychain_account": "sftp.gainwell.password"}
def test_update_clearhouse_missing_row_raises():
"""If the singleton row was never seeded, update raises LookupError."""
# Wipe the DB.
db._reset_for_tests()
db.init_db()
new_block = SftpBlock(
host="mft.example.com", port=22, username="u",
paths={"outbound": "/o", "inbound": "/i"},
stub=True, staging_dir="/tmp", auth={},
)
new_clearhouse = Clearhouse.model_validate({
"id": 1, "name": "x", "tpid": "1",
"submitter_name": "X", "submitter_id_qual": "46",
"submitter_contact_name": "X", "submitter_contact_email": "x@x",
"filename_block": {
"tz": "America/Denver",
"outbound_template": "{tpid}-{tx}-{ts}-1of1.{ext}",
"inbound_template": "TP{tpid}-{tx}_{ts}.x12",
},
"sftp_block": new_block.model_dump(),
"updated_at": "2026-06-24T00:00:00+00:00",
})
with pytest.raises(LookupError):
cycl_store.update_clearhouse(new_clearhouse)