"""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 / "ToHPE" inbound_dir.mkdir(parents=True) return SftpBlock( host="mft.example.com", port=22, username="test", paths={ "outbound": "/FromHPE", "inbound": "/ToHPE", }, 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) / "ToHPE" 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