cyclone/backend/tests/test_scheduler.py

"""SP16 — Inbound MFT polling scheduler tests.

We test the Scheduler class with a fake ``SftpClient`` factory that
returns files we drop on disk (the SFTP stub already does this; we
just need to control which files appear between ticks). The handlers
themselves (999/835/277CA/TA1) are exercised through real parsers
using the fixtures in ``tests/fixtures/``.
"""
from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable

import pytest

from cyclone import db, scheduler as sched_mod
from cyclone.db import ProcessedInboundFile
from cyclone.providers import SftpBlock
from cyclone.scheduler import (
    HANDLERS,
    ROUTED_FILE_TYPES,
    Scheduler,
    STATUS_ERROR,
    STATUS_OK,
    STATUS_SKIPPED,
    TickResult,
)


# ----- fixtures -----------------------------------------------------------


@pytest.fixture
def sftp_block(tmp_path):
    staging = tmp_path / "staging"
    inbound_dir = staging / "FromHPE"
    inbound_dir.mkdir(parents=True)
    return SftpBlock(
        host="mft.example.com",
        port=22,
        username="test",
        paths={
            "outbound": "/ToHPE",
            "inbound": "/FromHPE",
        },
        stub=True,
        staging_dir=str(staging),
        poll_seconds=60,
        auth={"method": "keychain", "secret_ref": "test.password"},
    )


@pytest.fixture
def _drop_file(sftp_block):
    """Helper: drop a named file in the inbound dir. Returns the path."""
    inbound_dir = Path(sftp_block.staging_dir) / "FromHPE"

    def _drop(name: str, body: bytes) -> Path:
        inbound_dir.mkdir(parents=True, exist_ok=True)
        p = inbound_dir / name
        p.write_bytes(body)
        return p

    return _drop


def _make_scheduler(sftp_block, tmp_path) -> Scheduler:
    """Build a Scheduler wired to the real (stub) SftpClient."""
    sched = Scheduler(
        sftp_block,
        poll_interval_seconds=60,
        sftp_block_name="test-block",
        # Use the real SftpClient — it reads from the stub staging dir.
        sftp_client_factory=None,
    )
    return sched


def _load_999_text() -> str:
    return (Path(__file__).parent / "fixtures" / "minimal_999.txt").read_text()


def _load_835_text() -> str:
    return (Path(__file__).parent / "fixtures" / "minimal_835.txt").read_text()


def _load_277ca_text() -> str:
    return (Path(__file__).parent / "fixtures" / "minimal_277ca.txt").read_text()


def _load_ta1_text() -> str:
    return (Path(__file__).parent / "fixtures" / "minimal_ta1.txt").read_text()


# ----- tests --------------------------------------------------------------


class TestSchedulerStatus:
    def test_not_running_by_default(self, sftp_block):
        sched = _make_scheduler(sftp_block, Path("/tmp"))
        st = sched.status()
        assert st.running is False
        assert st.poll_count == 0
        assert st.last_poll_at is None

    def test_running_after_start(self, sftp_block):
        async def _go():
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            await sched.start()
            try:
                assert sched.is_running() is True
            finally:
                await sched.stop()
        asyncio.run(_go())


class TestTickOnEmptyInbox:
    def test_tick_with_no_files_records_zero(self, sftp_block):
        async def _go():
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert isinstance(result, TickResult)
            assert result.files_seen == 0
            assert result.files_processed == 0
            assert result.files_skipped == 0
            assert result.files_errored == 0
            assert result.finished_at is not None
            assert result.errors == []
        asyncio.run(_go())


class TestTickRoutesFiles:
    def test_999_file_processed(self, sftp_block, _drop_file):
        async def _go():
            _drop_file("TP11525703-837P_M019048402-20260618130000000-1of1_999.x12",
                       _load_999_text().encode("utf-8"))
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert result.files_seen == 1
            assert result.files_processed == 1
            assert result.files_errored == 0

            with db.SessionLocal()() as session:
                rows = (
                    session.query(ProcessedInboundFile)
                    .filter_by(name="TP11525703-837P_M019048402-20260618130000000-1of1_999.x12")
                    .all()
                )
                assert len(rows) == 1
                assert rows[0].status == STATUS_OK
                assert rows[0].parser_used == "parse_999"
        asyncio.run(_go())

    def test_ta1_file_processed(self, sftp_block, _drop_file):
        async def _go():
            _drop_file("TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12",
                       _load_ta1_text().encode("utf-8"))
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert result.files_processed == 1, result.errors

            with db.SessionLocal()() as session:
                row = (
                    session.query(ProcessedInboundFile)
                    .filter_by(name="TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12")
                    .first()
                )
                assert row is not None
                assert row.status == STATUS_OK
                assert row.parser_used == "parse_ta1"
        asyncio.run(_go())

    def test_unknown_file_type_marked_skipped(self, sftp_block, _drop_file):
        async def _go():
            # 270 (eligibility request) is in the HCPF allowed set but
            # NOT in ROUTED_FILE_TYPES — Cyclone doesn't have a 270 parser.
            _drop_file("TP11525703-837P_M019048402-20260618130000000-1of1_270.x12",
                       b"some bytes")
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert result.files_skipped == 1
            assert result.files_processed == 0

            with db.SessionLocal()() as session:
                row = (
                    session.query(ProcessedInboundFile)
                    .filter_by(name="TP11525703-837P_M019048402-20260618130000000-1of1_270.x12")
                    .first()
                )
                assert row is not None
                assert row.status == STATUS_SKIPPED
                assert row.error_message and "270" in row.error_message
        asyncio.run(_go())

    def test_filename_not_matching_hcpf_marked_skipped(self, sftp_block, _drop_file):
        async def _go():
            _drop_file("random.txt", b"garbage")
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert result.files_skipped == 1
        asyncio.run(_go())

    def test_parse_error_marked_error(self, sftp_block, _drop_file):
        async def _go():
            # Valid filename but malformed body — parser raises.
            _drop_file("TP11525703-837P_M019048402-20260618130000000-1of1_999.x12",
                       b"this is not a 999 file at all")
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            result = await sched.tick()
            assert result.files_errored == 1
            assert result.files_processed == 0

            with db.SessionLocal()() as session:
                row = (
                    session.query(ProcessedInboundFile)
                    .filter_by(name="TP11525703-837P_M019048402-20260618130000000-1of1_999.x12")
                    .first()
                )
                assert row.status == STATUS_ERROR
                assert row.error_message
        asyncio.run(_go())


class TestTickIdempotent:
    def test_second_tick_does_not_reprocess(self, sftp_block, _drop_file):
        async def _go():
            _drop_file("TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12",
                       _load_ta1_text().encode("utf-8"))
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            r1 = await sched.tick()
            r2 = await sched.tick()
            assert r1.files_processed == 1
            assert r2.files_seen == 1   # still lists it
            assert r2.files_processed == 0  # but skips — already done
        asyncio.run(_go())


class TestSchedulerStartStop:
    def test_start_then_stop_returns_to_not_running(self, sftp_block):
        async def _go():
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            await sched.start()
            assert sched.is_running()
            await sched.stop()
            assert not sched.is_running()
        asyncio.run(_go())

    def test_double_start_is_idempotent(self, sftp_block):
        async def _go():
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            await sched.start()
            await sched.start()  # no-op
            assert sched.is_running()
            await sched.stop()
        asyncio.run(_go())

    def test_stop_when_not_running_is_safe(self, sftp_block):
        async def _go():
            sched = _make_scheduler(sftp_block, Path("/tmp"))
            await sched.stop()  # no-op
        asyncio.run(_go())


class TestModuleSingleton:
    def test_get_scheduler_raises_if_not_configured(self):
        sched_mod.reset_scheduler_for_tests()
        with pytest.raises(RuntimeError, match="not configured"):
            sched_mod.get_scheduler()

    def test_configure_then_get_returns_same_instance(self, sftp_block):
        sched_mod.reset_scheduler_for_tests()
        s = sched_mod.configure_scheduler(sftp_block, sftp_block_name="t")
        try:
            assert sched_mod.get_scheduler() is s
        finally:
            sched_mod.reset_scheduler_for_tests()


class TestRoutedFileTypes:
    """Frozen-set guards — adding a new routed type without updating
    the dispatch table would silently skip every inbound file of that
    type. These tests are the regression net."""

    def test_handlers_cover_all_routed_types(self):
        assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES