230 lines
8.1 KiB
Python
230 lines
8.1 KiB
Python
"""SP27 Task 9: scheduler status surfaces SFTP failures.
|
|
|
|
The 06/25 silent hang exposed a gap: ``Scheduler.status()`` had no
|
|
signal that polling had stalled. The operator's UI couldn't tell
|
|
"all quiet on the MFT front" from "we've been unable to reach the
|
|
MFT server for 3 hours". This pins the fix — the status dict
|
|
carries ``consecutive_failures``, ``last_error``, ``last_error_at``,
|
|
and ``last_sftp_attempt_at`` so the operator pill can flip to
|
|
destructive after 3 (or however many) consecutive failures.
|
|
|
|
Discriminator under test: ``TickResult.sftp_failed`` is the flag
|
|
``tick()`` keys off — NOT ``TickResult.errors``. Per-file processing
|
|
errors append to ``errors`` but must NOT bump ``consecutive_failures``.
|
|
The test suite pins both halves of the discriminator (SFTP failure
|
|
bumps; per-file failure doesn't).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from cyclone.clearhouse import InboundFile
|
|
from cyclone.providers import SftpBlock
|
|
from cyclone.scheduler import Scheduler
|
|
|
|
|
|
def _block() -> SftpBlock:
|
|
return SftpBlock(
|
|
host="mft.example.com",
|
|
port=22,
|
|
username="user",
|
|
auth={"password_keychain_account": "x"},
|
|
paths={"inbound": "/inbound", "outbound": "/outbound"},
|
|
stub=True,
|
|
)
|
|
|
|
|
|
# ---- pre-state: a fresh scheduler has no errors -------------------------
|
|
|
|
|
|
def test_status_starts_with_zero_failures_and_no_error():
|
|
"""A fresh scheduler must report no errors — pins that the new
|
|
fields default cleanly and don't trip on first read."""
|
|
sched = Scheduler(_block())
|
|
status = sched.status()
|
|
assert status.consecutive_failures == 0
|
|
assert status.last_error is None
|
|
assert status.last_error_at is None
|
|
assert status.last_sftp_attempt_at is None
|
|
|
|
|
|
# ---- a failed SFTP call bumps consecutive_failures -----------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_records_last_error_after_failure():
|
|
"""A failed tick bumps consecutive_failures and records the error.
|
|
|
|
Simulates a broken SFTP client. Pins the post-Task-8 behavior
|
|
where ``_tick_impl`` catches the error and ``tick()`` records
|
|
it on the scheduler's status.
|
|
"""
|
|
class BrokenClient:
|
|
def list_inbound(self):
|
|
raise RuntimeError("simulated outage")
|
|
|
|
# SP27 Task 8: scheduler calls async_list_inbound() (the
|
|
# wait_for-wrapped variant). The broken stub raises before
|
|
# the wait_for wrapper runs — that's a real SFTP failure
|
|
# either way.
|
|
async def async_list_inbound(self):
|
|
raise RuntimeError("simulated outage")
|
|
|
|
sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient())
|
|
|
|
before = datetime.now(timezone.utc)
|
|
await sched.tick()
|
|
after = datetime.now(timezone.utc)
|
|
|
|
status = sched.status()
|
|
assert status.consecutive_failures == 1
|
|
assert status.last_error is not None
|
|
assert "simulated outage" in status.last_error
|
|
# last_error_at is set to roughly the time of the failed tick
|
|
assert status.last_error_at is not None
|
|
assert before <= status.last_error_at <= after
|
|
# last_sftp_attempt_at is also bumped — separate from last_error_at
|
|
# when the failure was an SFTP-side issue (not a parse error).
|
|
assert status.last_sftp_attempt_at is not None
|
|
assert before <= status.last_sftp_attempt_at <= after
|
|
|
|
|
|
# ---- a successful tick clears consecutive_failures -----------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_successful_tick_clears_consecutive_failures():
|
|
"""A tick that succeeds (or skips cleanly) resets the counter.
|
|
|
|
Without this, a transient blip would lock the destructive pill
|
|
on forever. Pins the recovery path.
|
|
"""
|
|
class BrokenClient:
|
|
def list_inbound(self):
|
|
raise RuntimeError("transient")
|
|
|
|
async def async_list_inbound(self):
|
|
raise RuntimeError("transient")
|
|
|
|
class HealthyClient:
|
|
def list_inbound(self):
|
|
return []
|
|
|
|
async def async_list_inbound(self):
|
|
return []
|
|
|
|
# First: two failing ticks to bump the counter
|
|
sched = Scheduler(_block(), sftp_client_factory=lambda b: BrokenClient())
|
|
await sched.tick()
|
|
await sched.tick()
|
|
assert sched.status().consecutive_failures == 2
|
|
|
|
# Then: switch to a healthy client and tick again
|
|
sched._sftp_client_factory = lambda b: HealthyClient()
|
|
await sched.tick()
|
|
status = sched.status()
|
|
assert status.consecutive_failures == 0
|
|
# last_error is preserved (audit trail) but last_error_at is from
|
|
# the last failure, not the success.
|
|
assert status.last_error is not None # the historical error stays
|
|
assert "transient" in status.last_error
|
|
|
|
|
|
# ---- last_sftp_attempt_at moves on every attempt, not just failures ---
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_last_sftp_attempt_at_advances_on_every_tick():
|
|
"""Even a successful tick should bump last_sftp_attempt_at.
|
|
|
|
Without this, the operator can't tell when the scheduler last
|
|
*tried* to reach the MFT — only when it last failed.
|
|
"""
|
|
sched = Scheduler(_block()) # default stub returns []
|
|
before = datetime.now(timezone.utc)
|
|
await sched.tick()
|
|
after = datetime.now(timezone.utc)
|
|
first = sched.status().last_sftp_attempt_at
|
|
assert first is not None
|
|
assert before <= first <= after
|
|
|
|
# Wait a moment, tick again — last_sftp_attempt_at should move forward.
|
|
await asyncio.sleep(0.01)
|
|
await sched.tick()
|
|
second = sched.status().last_sftp_attempt_at
|
|
assert second is not None
|
|
assert second > first
|
|
|
|
|
|
# ---- as_dict surfaces the new fields for the API -----------------------
|
|
|
|
|
|
def test_status_as_dict_includes_new_fields():
|
|
"""The API endpoint (/api/health or /api/admin/scheduler) reads
|
|
status.as_dict() — the new fields must be present and serialized."""
|
|
sched = Scheduler(_block())
|
|
d = sched.status().as_dict()
|
|
assert "consecutive_failures" in d
|
|
assert d["consecutive_failures"] == 0
|
|
assert "last_error" in d
|
|
assert d["last_error"] is None
|
|
assert "last_error_at" in d
|
|
assert d["last_error_at"] is None
|
|
assert "last_sftp_attempt_at" in d
|
|
assert d["last_sftp_attempt_at"] is None
|
|
|
|
|
|
# ---- per-file processing errors must NOT bump consecutive_failures ------
|
|
#
|
|
# This pins the SP27 Task 9 discriminator (``TickResult.sftp_failed``
|
|
# vs. ``TickResult.errors``). Without this test, a future refactor
|
|
# that broadens ``result.errors`` semantics — or rewires `tick()` to
|
|
# key off it again — would silently re-break the discrimination.
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_per_file_error_does_not_bump_consecutive_failures(monkeypatch):
|
|
"""Per-file processing errors must NOT flip the destructive pill.
|
|
|
|
The listing step succeeds but ``_handle_one`` simulates the
|
|
real handler's error path: it appends to ``result.errors``
|
|
(the way ``_handle_one`` does on parser exceptions) but does
|
|
NOT touch ``result.sftp_failed``. The pill must stay clean.
|
|
"""
|
|
|
|
class FilesClient:
|
|
async def async_list_inbound(self):
|
|
return [
|
|
InboundFile(
|
|
name="tp123-999_MTRACKING-20260525001606060-1of1.x12",
|
|
size=100,
|
|
modified_at=datetime.now(timezone.utc),
|
|
# ``local_path`` is None because fake_handle_one
|
|
# never reaches the download step.
|
|
local_path=None,
|
|
)
|
|
]
|
|
|
|
sched = Scheduler(_block(), sftp_client_factory=lambda b: FilesClient())
|
|
|
|
async def fake_handle_one(f, result):
|
|
# Mirrors the real ``_handle_one`` parser-failure path:
|
|
# appends to result.errors, does NOT set sftp_failed.
|
|
result.errors.append(f"{f.name}: ValueError: garbage")
|
|
|
|
monkeypatch.setattr(sched, "_handle_one", fake_handle_one)
|
|
|
|
await sched.tick()
|
|
status = sched.status()
|
|
|
|
# Counter stays clean — per-file error is not an SFTP outage.
|
|
assert status.consecutive_failures == 0
|
|
# ``last_error`` is reserved for SFTP-side failures.
|
|
assert status.last_error is None
|
|
# But ``last_sftp_attempt_at`` moves forward regardless — the
|
|
# scheduler did try, it just failed to ingest the file.
|
|
assert status.last_sftp_attempt_at is not None
|