cyclone/backend/tests/test_scheduler_status_errors.py

"""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