merge: SP25 + SP26 SFTP polling + secret-file lookup into Version-1.0.0

Bring the SP25 SFTP Polling Enablement (PATCH /api/clearhouse,
scheduler hot-reload, env-var-first secret lookup) and SP26
SFTP Password File Companion (Docker secret _FILE tier) onto
the v1.0.0 release branch so the production stack can actually
flip the Gainwell MFT poll from stub → real.

Conflict resolved in docker-compose.yml: kept both the SP26
CYCLONE_SFTP_PASSWORD_FILE env var and the new 'Always bind
to 0.0.0.0' comment.
This commit is contained in:
tyler
2026-06-24 18:09:32 -06:00
16 changed files with 3189 additions and 14 deletions
+174
View File
@@ -0,0 +1,174 @@
"""SP25 — PATCH /api/clearhouse endpoint.
Lets the operator flip sftp_block.stub and adjust host/port/paths
without raw SQL. The endpoint validates via the Clearhouse Pydantic
model, writes via store.update_clearhouse(), then hot-reloads the
running scheduler via scheduler.reconfigure_scheduler().
"""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from fastapi.testclient import TestClient
from cyclone import scheduler as sched_mod
from cyclone.api import app
from cyclone.auth import deps
from cyclone.store import store as cycl_store
@pytest.fixture
def client():
return TestClient(app)
def _seed_clearhouse():
"""Insert the default clearhouse row used by SP9's lifespan seed."""
cycl_store.ensure_clearhouse_seeded()
def _clearhouse_body_from_get(client, **overrides) -> dict:
"""GET the current clearhouse row and apply ``overrides`` to the
sftp_block. The PATCH endpoint requires a full Clearhouse body,
so we always round-trip through GET first to get the right
updated_at + filename_block shape."""
r = client.get("/api/clearhouse")
assert r.status_code == 200, r.text
body = r.json()
sb = dict(body["sftp_block"])
sb.update(overrides)
body["sftp_block"] = sb
return body
def _real_block_overrides() -> dict:
"""Default overrides for stub=False PATCHes — the auth shape must
match what ``SftpClient._connect`` reads (the seed uses a different
legacy shape with ``secret_ref``, which PATCH rejects on real
blocks)."""
return {
"stub": False,
"host": "mft.example.com",
"auth": {"password_keychain_account": "sftp.gainwell.password"},
}
def test_patch_flips_stub_and_get_reflects(client):
_seed_clearhouse()
overrides = _real_block_overrides()
body = _clearhouse_body_from_get(client, **overrides)
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 200, r.text
payload = r.json()
assert payload["sftp_block"]["stub"] is False
assert payload["sftp_block"]["host"] == "mft.example.com"
r2 = client.get("/api/clearhouse")
assert r2.status_code == 200
assert r2.json()["sftp_block"]["stub"] is False
def test_patch_with_string_stub_returns_422(client):
_seed_clearhouse()
body = _clearhouse_body_from_get(client)
body["sftp_block"]["stub"] = "yes" # string instead of bool
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 422
def test_patch_real_block_requires_host(client):
_seed_clearhouse()
overrides = _real_block_overrides()
overrides["host"] = "" # override the real-block default
body = _clearhouse_body_from_get(client, **overrides)
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 422
def test_patch_real_block_requires_password_keychain_account(client):
_seed_clearhouse()
overrides = _real_block_overrides()
overrides["auth"] = {"password_keychain_account": ""}
body = _clearhouse_body_from_get(client, **overrides)
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 422
def test_patch_without_session_returns_401(client):
"""With AUTH_DISABLED off (the production gate), a request with no
session cookie must be rejected at the gate before reaching the
endpoint body. Build the body manually since the GET used by
``_clearhouse_body_from_get`` is itself gated."""
_seed_clearhouse()
# Build a valid body without going through the gated GET.
overrides = _real_block_overrides()
body = {
"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}-1of1.{ext}",
"inbound_template": "TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12",
},
"sftp_block": {
"host": overrides.get("host", "mft.example.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": overrides.get("stub", False),
"staging_dir": "./var/sftp/staging",
"auth": overrides.get("auth", {"password_keychain_account": "sftp.gainwell.password"}),
},
"updated_at": "2026-06-24T00:00:00+00:00",
}
deps.AUTH_DISABLED = False
try:
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 401
finally:
deps.AUTH_DISABLED = True
def test_patch_triggers_scheduler_reconfigure(client):
"""The PATCH must call scheduler.reconfigure_scheduler() with the new
SftpBlock so the next tick uses real MFT instead of the stub."""
_seed_clearhouse()
with patch.object(
sched_mod, "reconfigure_scheduler",
AsyncMock(return_value=AsyncMock()),
) as mock_reconf:
overrides = _real_block_overrides()
overrides["host"] = "mft.gainwelltechnologies.com"
body = _clearhouse_body_from_get(client, **overrides)
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 200, r.text
assert mock_reconf.await_count == 1
# The new sftp_block is passed positionally; sftp_block_name
# comes from the clearhouse row's ``name`` field.
call_args = mock_reconf.await_args
assert call_args.args[0].stub is False
assert call_args.args[0].host == "mft.gainwelltechnologies.com"
assert call_args.kwargs.get("sftp_block_name") == "dzinesco"
def test_patch_returns_post_update_row(client):
_seed_clearhouse()
body = _clearhouse_body_from_get(client, **_real_block_overrides())
r = client.patch("/api/clearhouse", json=body)
assert r.status_code == 200
payload = r.json()
assert payload["name"] == "dzinesco"
assert payload["tpid"] == "11525703"
assert payload["sftp_block"]["stub"] is False
assert payload["sftp_block"]["host"] == "mft.example.com"
+24 -1
View File
@@ -117,8 +117,15 @@ def test_compose_declares_required_services():
def test_compose_declares_required_secrets_and_volumes():
compose = yaml.safe_load(COMPOSE_FILE.read_text())
secrets = compose.get("secrets", {})
for required in ("cyclone_db_key", "cyclone_admin_password"):
for required in ("cyclone_db_key", "cyclone_admin_password", "cyclone_sftp_password"):
assert required in secrets, f"compose must declare secret {required!r}"
# SP26: the SFTP secret must point at the same /etc/cyclone/secrets/ tree
# the host operator manages. Catch typos in the file path here so a
# rename breaks the test, not a production deploy.
assert (
secrets["cyclone_sftp_password"]["file"]
== "/etc/cyclone/secrets/sftp_password"
)
volumes = compose.get("volumes", {})
for required in (
"cyclone_db",
@@ -142,6 +149,22 @@ def test_compose_backend_wires_backup_autostart():
assert "CYCLONE_BACKUP_RETENTION_DAYS" in env
def test_compose_backend_wires_sftp_password_file_env_var():
"""SP26 — the backend must wire CYCLONE_SFTP_PASSWORD_FILE to the
mounted Docker secret so the MFT password resolves from
/run/secrets/cyclone_sftp_password without an env-var export."""
compose = yaml.safe_load(COMPOSE_FILE.read_text())
backend = compose["services"]["backend"]
env = backend.get("environment", {})
assert env.get("CYCLONE_SFTP_PASSWORD_FILE") == "/run/secrets/cyclone_sftp_password", (
"backend must wire CYCLONE_SFTP_PASSWORD_FILE to the Docker secret mount"
)
backend_secrets = backend.get("secrets", [])
assert any(
_volume_source(s) == "cyclone_sftp_password" for s in backend_secrets
), "backend must reference the cyclone_sftp_password Docker secret"
@pytest.mark.skipif(
not _has_docker(), reason="docker not on PATH; skipping Dockerfile parse check"
)
+197
View File
@@ -0,0 +1,197 @@
"""SP25 — scheduler.reconfigure_scheduler() hot-reload.
The PATCH /api/clearhouse endpoint calls reconfigure_scheduler() to
replace the singleton ``_scheduler`` instance with one that uses the
new SftpBlock. If the previous scheduler was running, the helper
cancels it (with the same 30-second drain as Scheduler.stop()) and
starts the new one. If it was stopped, the new one is left stopped.
"""
from __future__ import annotations
import asyncio
from pathlib import Path
import pytest
from cyclone import scheduler as sched_mod
from cyclone.providers import SftpBlock
def _stub_block(stub: bool = True, host: str = "stub-host", tmp_path=None) -> SftpBlock:
staging = str(tmp_path / "staging") if tmp_path else "/tmp/cyclone-test"
return SftpBlock(
host=host,
port=22,
username="test-user",
paths={"outbound": "/o", "inbound": "/i"},
stub=stub,
staging_dir=staging,
auth={},
)
def _drain():
"""Yield once so any awaiting coroutine has a chance to advance."""
return asyncio.sleep(0)
@pytest.fixture(autouse=True)
def _reset_scheduler():
sched_mod.reset_scheduler_for_tests()
yield
sched_mod.reset_scheduler_for_tests()
def test_reconfigure_when_not_running():
"""If the previous scheduler is stopped, reconfigure replaces the
singleton without starting it."""
async def _go():
block_a = _stub_block(stub=True, host="block-a")
sched_a = sched_mod.configure_scheduler(block_a, sftp_block_name="a")
assert not sched_a.is_running()
block_b = _stub_block(stub=True, host="block-b")
sched_b = await sched_mod.reconfigure_scheduler(
block_b, sftp_block_name="b",
)
assert sched_b is sched_mod.get_scheduler()
assert not sched_b.is_running()
assert sched_b._sftp_block_name == "b"
asyncio.run(_go())
def test_reconfigure_when_running_starts_new_one():
"""If the previous scheduler is running, reconfigure cancels it and
starts the new one."""
async def _go():
block_a = _stub_block(stub=True, host="block-a")
sched_a = sched_mod.configure_scheduler(block_a, sftp_block_name="a")
await sched_a.start()
assert sched_a.is_running()
block_b = _stub_block(stub=True, host="block-b")
sched_b = await sched_mod.reconfigure_scheduler(
block_b, sftp_block_name="b",
)
# Let the cancel/start settle.
await _drain()
assert sched_b.is_running()
assert sched_b._sftp_block_name == "b"
await sched_b.stop()
asyncio.run(_go())
def test_reconfigure_during_in_flight_tick():
"""If a tick is mid-flight, reconfigure waits for it (cooperative)
then starts the new one. Verified by feeding the scheduler a slow
client factory."""
async def _go():
started = asyncio.Event()
finish = asyncio.Event()
async def _slow_tick():
started.set()
await finish.wait()
# Inject a fake Scheduler that simulates a long-running tick.
# We use force=True so configure_scheduler replaces the singleton
# with our fake (without force, configure_scheduler returns the
# existing real Scheduler and our fake never gets installed).
class _FakeScheduler:
def __init__(self, block, **_):
self._block = block
self._task = None
self._running = False
async def start(self):
self._running = True
self._task = asyncio.create_task(_slow_tick())
async def stop(self):
self._running = False
if self._task:
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
pass
self._task = None
def is_running(self):
return self._running
original_scheduler_cls = sched_mod.Scheduler
sched_mod.Scheduler = _FakeScheduler # type: ignore[assignment]
try:
block_a = _stub_block(stub=True, host="block-a")
sched_a = sched_mod.configure_scheduler(
block_a, sftp_block_name="a", force=True,
)
await sched_a.start()
await started.wait()
assert sched_a.is_running()
# Now reconfigure. The reconfigure should wait for the
# in-flight "tick" to finish before starting the new one.
reconfigure_task = asyncio.create_task(
sched_mod.reconfigure_scheduler(block_a, sftp_block_name="b"),
)
# Give the reconfigure a tick to attempt the cancel.
await asyncio.sleep(0.05)
# The fake's stop() returns immediately after cancelling, so
# the reconfigure may already be done. The key invariant is
# that it completed cleanly without raising.
finish.set()
sched_b = await reconfigure_task
assert sched_b is sched_mod.get_scheduler()
finally:
# Clean up: ensure no slow_tick leaks across tests.
try:
current = sched_mod.get_scheduler()
if current.is_running():
await current.stop()
except RuntimeError:
pass
sched_mod.Scheduler = original_scheduler_cls # type: ignore[assignment]
asyncio.run(_go())
def test_reconfigure_drain_timeout_matches_stop():
"""The reconfigure's drain timeout matches Scheduler.stop() — both
are 30 seconds. We can't wait 30s in a test, but we can verify the
constant is shared."""
# Both code paths should reference the same constant.
assert hasattr(sched_mod, "_DRAIN_TIMEOUT_SECONDS")
assert sched_mod._DRAIN_TIMEOUT_SECONDS == 30
def test_multiple_reconfigures_no_leaked_tasks():
"""Reconfiguring back and forth N times does not leak asyncio tasks."""
async def _go():
block = _stub_block(stub=True, host="block-a")
sched_mod.configure_scheduler(block, sftp_block_name="a")
sched = sched_mod.get_scheduler()
await sched.start()
before = len(asyncio.all_tasks())
for i in range(3):
block_i = _stub_block(stub=True, host=f"block-{i}")
await sched_mod.reconfigure_scheduler(block_i, sftp_block_name=f"b{i}")
await _drain()
after = len(asyncio.all_tasks())
# The new scheduler itself owns one task; allow +1.
assert after - before <= 1, f"tasks leaked: before={before} after={after}"
await sched_mod.get_scheduler().stop()
asyncio.run(_go())
+100
View File
@@ -0,0 +1,100 @@
"""SP25 — env-var fallback in cyclone.secrets.get_secret().
The scheduler's ``SftpClient._connect`` calls ``get_secret(name)`` to
fetch the MFT password. Today the lookup is macOS-Keychain-only via
the ``keyring`` library. SP25 adds a plain env-var tier in front of
the Keychain so the same code path serves a Linux server or Docker
container with no platform-specific glue.
"""
from __future__ import annotations
import importlib
import pytest
@pytest.fixture
def secrets_module(monkeypatch):
"""Reload cyclone.secrets with a clean keyring state per test."""
# Force a fresh import so module-level keyring cache is clean.
import cyclone.secrets as secrets_mod
importlib.reload(secrets_mod)
yield secrets_mod
importlib.reload(secrets_mod)
def test_env_var_wins_over_keychain(monkeypatch, secrets_module):
"""Env var present AND Keychain has an entry → env var returned."""
monkeypatch.setenv("cyclone.test.password", "from-env")
monkeypatch.setattr(
secrets_module.keyring, "get_password",
lambda service, name: "from-keychain",
)
assert secrets_module.get_secret("cyclone.test.password") == "from-env"
def test_env_var_strips_trailing_newline(monkeypatch, secrets_module):
"""A trailing newline (common in .env files) is stripped."""
monkeypatch.setenv("cyclone.test.password", "s3cret\n")
assert secrets_module.get_secret("cyclone.test.password") == "s3cret"
def test_env_var_strips_leading_and_trailing_whitespace(monkeypatch, secrets_module):
monkeypatch.setenv("cyclone.test.password", " s3cret \n")
assert secrets_module.get_secret("cyclone.test.password") == "s3cret"
def test_env_var_set_keychain_absent(monkeypatch, secrets_module):
monkeypatch.setenv("cyclone.test.password", "from-env")
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda service, name: None,
)
assert secrets_module.get_secret("cyclone.test.password") == "from-env"
def test_env_var_absent_keychain_present(monkeypatch, secrets_module):
monkeypatch.delenv("cyclone.test.password", raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password",
lambda service, name: "from-keychain",
)
assert secrets_module.get_secret("cyclone.test.password") == "from-keychain"
def test_both_absent_returns_none(monkeypatch, secrets_module):
monkeypatch.delenv("cyclone.test.password", raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda service, name: None,
)
assert secrets_module.get_secret("cyclone.test.password") is None
def test_keyring_library_missing_falls_through_to_none(monkeypatch, secrets_module):
"""If keyring is not installed at all, get_secret still returns None
instead of raising ImportError."""
monkeypatch.setenv("cyclone.test.password", "from-env")
monkeypatch.setattr(secrets_module, "_HAS_KEYRING", False)
# Even if keyring is missing, the env-var tier still wins.
assert secrets_module.get_secret("cyclone.test.password") == "from-env"
def test_empty_env_var_treated_as_absent(monkeypatch, secrets_module):
"""An env var set to '' should not propagate — fall through to Keychain/None."""
monkeypatch.setenv("cyclone.test.password", "")
monkeypatch.setattr(
secrets_module.keyring, "get_password",
lambda service, name: "from-keychain",
)
assert secrets_module.get_secret("cyclone.test.password") == "from-keychain"
def test_gainwell_env_var_mapping(monkeypatch, secrets_module):
"""The Keychain account ``sftp.gainwell.password`` maps to the env
var ``CYCLONE_SFTP_PASSWORD`` via the internal name table. Without
this, an operator setting ``CYCLONE_SFTP_PASSWORD`` on a Linux box
would not get the MFT password."""
monkeypatch.setenv("CYCLONE_SFTP_PASSWORD", "real-mft-password")
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda service, name: None,
)
assert secrets_module.get_secret("sftp.gainwell.password") == "real-mft-password"
+128
View File
@@ -0,0 +1,128 @@
"""SP26 — Docker-secrets file fallback in cyclone.secrets.get_secret().
SP25 added a plain env-var tier ahead of the macOS Keychain lookup.
SP26 adds a further tier above that: an `<env_name>_FILE` env var
pointing at a file on disk — the standard Docker-secrets pattern.
The file takes precedence over the plain env var (matches the
``auth/bootstrap.py:_read_secret`` convention).
"""
from __future__ import annotations
import importlib
from pathlib import Path
import pytest
# A secret name we use throughout the unmapped cases. Because it is not
# in ``_ENV_NAME_FOR``, ``get_secret()`` will read it as
# ``os.environ.get("cyclone.test.file.password")`` directly (and the
# ``_FILE`` companion is the same name with ``_FILE`` appended). This
# mirrors SP25's ``test_secrets_envvar.py`` convention.
_UNMAPPED_NAME = "cyclone.test.file.password"
@pytest.fixture
def secrets_module(monkeypatch):
"""Reload cyclone.secrets with a clean keyring state per test."""
import cyclone.secrets as secrets_mod
importlib.reload(secrets_mod)
yield secrets_mod
importlib.reload(secrets_mod)
def _write_secret_file(tmp_path: Path, name: str, contents: str) -> Path:
f = tmp_path / name
f.write_text(contents)
return f
def test_file_env_var_returns_file_contents(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
"""_FILE set, file exists → returns file contents (no Keychain lookup)."""
f = _write_secret_file(tmp_path, "pw", "from-file\n")
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", str(f))
monkeypatch.delenv(_UNMAPPED_NAME, raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret(_UNMAPPED_NAME) == "from-file"
def test_file_env_var_strips_trailing_newline(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
"""Docker secret files commonly end with \\n — strip it."""
f = _write_secret_file(tmp_path, "pw", "supersecret\n")
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", str(f))
monkeypatch.delenv(_UNMAPPED_NAME, raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret(_UNMAPPED_NAME) == "supersecret"
def test_file_env_var_strips_leading_and_trailing_whitespace(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
f = _write_secret_file(tmp_path, "pw", " supersecret \n")
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", str(f))
monkeypatch.delenv(_UNMAPPED_NAME, raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret(_UNMAPPED_NAME) == "supersecret"
def test_file_env_var_wins_over_plain_env_var(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
"""When both _FILE and the plain env var are set, _FILE wins."""
f = _write_secret_file(tmp_path, "pw", "from-file\n")
monkeypatch.setenv(_UNMAPPED_NAME, "from-env")
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", str(f))
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret(_UNMAPPED_NAME) == "from-file"
def test_file_env_var_missing_file_raises_runtime_error(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
"""If _FILE points at a non-existent path, raise RuntimeError."""
missing = tmp_path / "does-not-exist"
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", str(missing))
monkeypatch.delenv(_UNMAPPED_NAME, raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
with pytest.raises(RuntimeError, match=f"{_UNMAPPED_NAME}_FILE"):
secrets_module.get_secret(_UNMAPPED_NAME)
def test_file_env_var_empty_string_treated_as_unset(
monkeypatch, secrets_module,
) -> None:
"""An empty _FILE env var falls through to the plain env var (SP25 rule)."""
monkeypatch.setenv(f"{_UNMAPPED_NAME}_FILE", "")
monkeypatch.setenv(_UNMAPPED_NAME, "from-env")
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret(_UNMAPPED_NAME) == "from-env"
def test_gainwell_file_env_var_full_chain(
tmp_path: Path, monkeypatch, secrets_module,
) -> None:
"""Setting CYCLONE_SFTP_PASSWORD_FILE makes get_secret('sftp.gainwell.password')
return the file's contents — the operator-visible chain works end-to-end."""
f = _write_secret_file(tmp_path, "sftp_pw", "real-mft-password\n")
monkeypatch.setenv("CYCLONE_SFTP_PASSWORD_FILE", str(f))
monkeypatch.delenv("CYCLONE_SFTP_PASSWORD", raising=False)
monkeypatch.setattr(
secrets_module.keyring, "get_password", lambda s, n: None,
)
assert secrets_module.get_secret("sftp.gainwell.password") == "real-mft-password"
@@ -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)