diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index a602897..3837b9e 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -95,6 +95,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: serves requests. No teardown needed for Cyclone's local-only posture. """ from cyclone import db, payers as payer_loader + from cyclone import scheduler as scheduler_mod db.init_db() app.state.event_bus = EventBus() @@ -104,8 +105,40 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: store.ensure_clearhouse_seeded() except Exception as exc: # noqa: BLE001 log.exception("SP9 seed failed: %s", exc) + + # SP16: configure the inbound MFT polling scheduler. The + # dzinesco clearhouse singleton (seeded by SP9) carries the SFTP + # block — multi-provider polling is out of scope for v1. + # Auto-start is opt-in via env var so operators running the API + # for one-off parsing don't accidentally poll the MFT on startup. + try: + clearhouse = store.get_clearhouse() + if clearhouse is not None: + sched = scheduler_mod.configure_scheduler( + clearhouse.sftp_block, + sftp_block_name=clearhouse.name or "default", + ) + import os as _os + if _os.environ.get("CYCLONE_SCHEDULER_AUTOSTART", "").lower() in ( + "1", "true", "yes", + ): + await sched.start() + except Exception as exc: # noqa: BLE001 + # Scheduler setup must never block the API from coming up. + log.exception("SP16 scheduler config failed: %s", exc) + yield + # SP16: stop the scheduler on shutdown so the test harness + # (TestClient context manager) doesn't leak background tasks. + try: + sched = scheduler_mod.get_scheduler() + await sched.stop() + except RuntimeError: + pass # never configured + except Exception as exc: # noqa: BLE001 + log.warning("Scheduler shutdown failed: %s", exc) + # Mirror cli._PAYER_FACTORIES. Kept local so the API doesn't depend on the # Click-based CLI module. @@ -2430,6 +2463,113 @@ def rotate_db_key_endpoint(body: dict | None = None) -> Any: _db_rotate_lock.release() +# --------------------------------------------------------------------------- +# SP16: live MFT polling scheduler (admin) +# +# The scheduler lives in :mod:`cyclone.scheduler` and is configured by +# the lifespan handler. The endpoints below expose start / stop / +# one-shot tick / status / history so an operator (or a cron job) +# can drive the scheduler without touching the DB. +# +# Note: the scheduler is OFF by default. Auto-start is opt-in via +# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints +# are the operator's manual controls. +# --------------------------------------------------------------------------- +from cyclone import scheduler as _scheduler_mod + + +def _scheduler_or_503(): + """Return the configured scheduler or raise 503.""" + try: + return _scheduler_mod.get_scheduler() + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) + + +@app.post("/api/admin/scheduler/start") +async def scheduler_start() -> Any: + """Begin polling the MFT inbound path every poll_interval_seconds.""" + sched = _scheduler_or_503() + await sched.start() + return {"status": sched.status().as_dict()} + + +@app.post("/api/admin/scheduler/stop") +async def scheduler_stop() -> Any: + """Stop polling. Waits up to 30s for the current tick to finish.""" + sched = _scheduler_or_503() + await sched.stop() + return {"status": sched.status().as_dict()} + + +@app.post("/api/admin/scheduler/tick") +async def scheduler_tick() -> Any: + """Run a single poll cycle synchronously and return the result. + + Useful for: forcing a poll without waiting for the next interval; + verifying SFTP connectivity; running a one-shot import from the + CLI (``curl -X POST .../api/admin/scheduler/tick``). + """ + sched = _scheduler_or_503() + result = await sched.tick() + return {"ok": True, "tick": result.as_dict()} + + +@app.get("/api/admin/scheduler/status") +def scheduler_status() -> Any: + """Return the scheduler's runtime snapshot (running, counters, last tick).""" + sched = _scheduler_or_503() + return sched.status().as_dict() + + +@app.get("/api/admin/scheduler/processed-files") +def scheduler_processed_files( + limit: int = Query(default=100, ge=1, le=1000), + status: str | None = Query(default=None), +) -> Any: + """List rows from ``processed_inbound_files``, newest first. + + The operator's "what did the scheduler do?" view. Filters by + ``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``). + Returns ``{"count": N, "files": [...]}`` where ``files[i]`` + matches the ORM row as a JSON dict. + """ + from cyclone.db import ProcessedInboundFile + from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING + + valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING} + if status is not None and status not in valid_statuses: + raise HTTPException( + status_code=400, + detail=f"status must be one of {sorted(valid_statuses)}", + ) + + with db.SessionLocal()() as s: + q = s.query(db.ProcessedInboundFile) + if status is not None: + q = q.filter(db.ProcessedInboundFile.status == status) + rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all() + return { + "count": len(rows), + "files": [ + { + "id": r.id, + "sftp_block_name": r.sftp_block_name, + "name": r.name, + "size": r.size, + "modified_at": r.modified_at.isoformat() if r.modified_at else None, + "file_type": r.file_type, + "processed_at": r.processed_at.isoformat() if r.processed_at else None, + "parser_used": r.parser_used, + "claim_count": r.claim_count, + "status": r.status, + "error_message": r.error_message, + } + for r in rows + ], + } + + @app.get("/api/config/providers/{npi}") def get_configured_provider(npi: str): p = store.get_provider(npi) diff --git a/backend/src/cyclone/clearhouse/__init__.py b/backend/src/cyclone/clearhouse/__init__.py index 55a2add..9fcba7c 100644 --- a/backend/src/cyclone/clearhouse/__init__.py +++ b/backend/src/cyclone/clearhouse/__init__.py @@ -93,13 +93,24 @@ class SftpClient: return self._list_inbound_paramiko() def read_file(self, remote_path: str) -> bytes: - """Read bytes from a remote path. Stub raises in stub mode.""" + """Read bytes from a remote path. + + Stub mode: reads from ``{staging_dir}/{remote_path}``. Used by + the SP16 scheduler so it can exercise the same code path on a + workstation without a real MFT connection. + """ if self._stub: - raise RuntimeError( - "Stub SFTP cannot read remote files. Use the local staging dir." - ) + return self._read_file_stub(remote_path) return self._read_file_paramiko(remote_path) + def _read_file_stub(self, remote_path: str) -> bytes: + """Read bytes from ``{staging_dir}/{remote_path}`` (SP16 stub).""" + staging = Path(self._block.staging_dir).resolve() + target = staging / remote_path.lstrip("/") + if not target.is_file(): + raise FileNotFoundError(f"inbound stub file not found: {target}") + return target.read_bytes() + def get_secret(self, name: str) -> Optional[str]: """Fetch the auth secret from Keychain. Returns the stub secret if absent.""" value = secrets.get_secret(name) diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index 36f7058..1af417e 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -676,6 +676,51 @@ class AuditLog(Base): ) +# --------------------------------------------------------------------------- +# SP16: inbound MFT scheduler +# --------------------------------------------------------------------------- + + +class ProcessedInboundFile(Base): + """One row per inbound MFT file the scheduler has downloaded. + + SP16. Lets the scheduler be idempotent: a re-tick or restart must + not re-parse the same inbound file. The unique index on + (sftp_block_name, name) prevents duplicate inserts and lets the + scheduler fast-skip already-processed files via a SELECT. + + Status values: + * ok - parsed cleanly, results persisted to the store + * error - parser raised; error_message captured + * skipped - file_type not in the scheduler's allowed set + * pending - file was downloaded but a downstream step failed; + the scheduler retries on the next tick + """ + + __tablename__ = "processed_inbound_files" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + sftp_block_name: Mapped[str] = mapped_column(String(128), nullable=False) + name: Mapped[str] = mapped_column(String(256), nullable=False) + size: Mapped[int] = mapped_column(Integer, nullable=False) + modified_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) + file_type: Mapped[Optional[str]] = mapped_column(String(16), nullable=True) + processed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + parser_used: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) + claim_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + status: Mapped[str] = mapped_column(String(16), nullable=False) + error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + + __table_args__ = ( + Index( + "ux_processed_inbound_files_block_name", + "sftp_block_name", "name", unique=True, + ), + Index("ix_processed_inbound_files_processed_at", "processed_at"), + Index("ix_processed_inbound_files_status", "status"), + ) + + # --------------------------------------------------------------------------- # SP9: providers, payers, payer_configs, clearhouse # --------------------------------------------------------------------------- diff --git a/backend/src/cyclone/migrations/0011_processed_inbound_files.sql b/backend/src/cyclone/migrations/0011_processed_inbound_files.sql new file mode 100644 index 0000000..a77288f --- /dev/null +++ b/backend/src/cyclone/migrations/0011_processed_inbound_files.sql @@ -0,0 +1,47 @@ +-- version: 11 +-- SP16: Inbound MFT polling scheduler +-- +-- Tracks every file the background scheduler has downloaded from +-- the Gainwell MFT inbound path so a re-tick (or a restart) does not +-- re-process the same file. Idempotency is required for production: +-- the scheduler polls every N seconds and a slow MFT server may hand +-- us the same file across two polls. +-- +-- We key on (sftp_block_name, name) — the sftp_block_name disambiguates +-- multi-provider installations (SP9+SP-multi-NPI), name is the inbound +-- filename as it appears on the MFT server. +-- +-- Status values: +-- * ok — parsed cleanly, results persisted to the store +-- * error — parser raised; error_message captured for the operator +-- * skipped — file_type not in the scheduler's allowed set +-- * pending — file was downloaded but a downstream step failed +-- (e.g. DB write); the scheduler retries on the next tick +-- +-- claim_count is the number of claims/remittances/acks the parser +-- surfaced. Surfaced on /api/admin/scheduler/status so the operator can +-- see throughput without parsing logs. +-- +-- Compliance: not part of the HIPAA audit chain (SP11). This is +-- operational metadata; an SFTP outage shouldn't pollute the audit log. + +CREATE TABLE processed_inbound_files ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sftp_block_name TEXT NOT NULL, + name TEXT NOT NULL, + size INTEGER NOT NULL, + modified_at TEXT, + file_type TEXT, + processed_at TEXT NOT NULL, + parser_used TEXT, + claim_count INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL, + error_message TEXT +); + +CREATE UNIQUE INDEX ux_processed_inbound_files_block_name + ON processed_inbound_files(sftp_block_name, name); +CREATE INDEX ix_processed_inbound_files_processed_at + ON processed_inbound_files(processed_at DESC); +CREATE INDEX ix_processed_inbound_files_status + ON processed_inbound_files(status); \ No newline at end of file diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py new file mode 100644 index 0000000..0de34d9 --- /dev/null +++ b/backend/src/cyclone/scheduler.py @@ -0,0 +1,713 @@ +"""Background inbound MFT polling scheduler (SP16). + +Turns Cyclone from a manual upload tool into a live clearinghouse: +a long-running asyncio task that periodically polls the Gainwell MFT +inbound path, downloads each new file, and runs it through the +appropriate parser. The operator no longer has to watch for inbound +files and POST them to ``/api/parse-999`` etc. by hand. + +Design constraints +------------------ + +* **Idempotent.** A re-tick (or a process restart) must not re-parse + the same inbound file. We persist a ``processed_inbound_files`` row + per file and skip ones we've already seen. + +* **Crash-safe.** If the parser raises or the DB write fails, the + scheduler logs the error, records an ``error`` row, and moves on. + The next tick continues from the next file. + +* **Bounded blast radius.** A bad file must not stop the scheduler. + Each file is wrapped in try/except so a 999 parser crash doesn't + prevent us from processing the next inbound 835. + +* **Operator-controlled.** The scheduler is OFF by default; the + operator must explicitly start it (``POST /api/admin/scheduler/start`` + or ``CYCLONE_SCHEDULER_AUTOSTART=true``). When it's running, status + is exposed via ``GET /api/admin/scheduler/status``. + +* **No threading.** We use ``asyncio.create_task`` + ``asyncio.sleep`` + rather than APScheduler or threading because the rest of the + codebase is asyncio-native (FastAPI). The whole polling loop runs + in the FastAPI event loop on the main thread. + +Compliance: SP16 is operational metadata only. Inbound file +processing is NOT part of the HIPAA audit chain (SP11) — an SFTP +outage shouldn't pollute the audit log with parser errors. +""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +import traceback +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Optional + +from sqlalchemy.exc import IntegrityError + +from cyclone import db +from cyclone.store import store as cycl_store +from cyclone.audit_log import AuditEvent, append_event +from cyclone.clearhouse import InboundFile, SftpClient +from cyclone.db import ProcessedInboundFile +from cyclone.edi.filenames import parse_inbound_filename +from cyclone.inbox_state import apply_999_rejections +from cyclone.inbox_state_277ca import apply_277ca_rejections +from cyclone.providers import SftpBlock + +log = logging.getLogger(__name__) + + +# Status values for ProcessedInboundFile.status. +STATUS_OK = "ok" +STATUS_ERROR = "error" +STATUS_SKIPPED = "skipped" +STATUS_PENDING = "pending" + + +# File types we know how to route. The HCPF set is broader (270/271/ +# 276/277/278/820/834/ENCR) but Cyclone's parser only covers the +# four below. Files with unknown types are recorded as ``skipped`` +# so the operator can see them in the audit table. +ROUTED_FILE_TYPES = frozenset({"999", "835", "277", "277CA", "TA1"}) + + +@dataclass +class TickResult: + """Outcome of a single scheduler tick (one poll cycle).""" + + started_at: datetime + finished_at: Optional[datetime] = None + files_seen: int = 0 + files_processed: int = 0 + files_skipped: int = 0 + files_errored: int = 0 + errors: list[str] = field(default_factory=list) + + def as_dict(self) -> dict[str, Any]: + return { + "started_at": self.started_at.isoformat(), + "finished_at": ( + self.finished_at.isoformat() if self.finished_at else None + ), + "files_seen": self.files_seen, + "files_processed": self.files_processed, + "files_skipped": self.files_skipped, + "files_errored": self.files_errored, + "errors": list(self.errors), + } + + +@dataclass +class SchedulerStatus: + """Snapshot of the scheduler's runtime state.""" + + running: bool + poll_interval_seconds: int + sftp_block_name: str + last_poll_at: Optional[datetime] + poll_count: int + total_processed: int + total_skipped: int + total_errored: int + last_tick: Optional[TickResult] = None + + def as_dict(self) -> dict[str, Any]: + return { + "running": self.running, + "poll_interval_seconds": self.poll_interval_seconds, + "sftp_block_name": self.sftp_block_name, + "last_poll_at": ( + self.last_poll_at.isoformat() if self.last_poll_at else None + ), + "poll_count": self.poll_count, + "total_processed": self.total_processed, + "total_skipped": self.total_skipped, + "total_errored": self.total_errored, + "last_tick": self.last_tick.as_dict() if self.last_tick else None, + } + + +# --------------------------------------------------------------------------- +# Per-file-type handlers. Each returns (parser_name, claim_count) and +# persists its own DB rows. The scheduler records the outcome. +# --------------------------------------------------------------------------- + + +def _handle_999(text: str, source_file: str) -> tuple[str, int]: + """Parse a 999, apply rejections, persist ack row. Returns (parser, count).""" + from cyclone.parsers.parse_999 import parse_999_text + from cyclone.parsers.exceptions import CycloneParseError + + try: + result = parse_999_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"999 parse error: {exc}") from exc + + received, accepted, rejected, ack_code = _ack_count_summary(result) + icn = result.envelope.control_number + synthetic_id = _ack_synthetic_source_batch_id(icn) + + with db.SessionLocal()() as session: + def _lookup(pcn: str): + return ( + session.query(db.Claim) + .filter_by(patient_control_number=pcn) + .first() + ) + rejection_result = apply_999_rejections( + session, result, claim_lookup=_lookup, + ) + if rejection_result.matched: + for cid in rejection_result.matched: + append_event(session, AuditEvent( + event_type="claim.rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id}, + actor="999-parser-scheduler", + )) + row = cycl_store.add_ack( + source_batch_id=synthetic_id, + accepted_count=accepted, + rejected_count=rejected, + received_count=received, + ack_code=ack_code, + raw_json=json.loads(result.model_dump_json()), + ) + session.commit() + return "parse_999", received + + +def _handle_835(text: str, source_file: str) -> tuple[str, int]: + """Parse an 835, run validation, persist batch + remittances.""" + import uuid + from cyclone.parsers.parse_835 import parse as parse_835 + from cyclone.parsers.exceptions import CycloneParseError + from cyclone.parsers.validator_835 import validate as validate_835 + from cyclone.payers import PAYER_FACTORIES_835 + from cyclone.store import BatchRecord + + config = PAYER_FACTORIES_835["co_medicaid_835"]() + try: + result = parse_835(text, config, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"835 parse error: {exc}") from exc + + # Validation report (mirrors the API endpoint). + report = validate_835(result, config) + n = len(result.claims) + if report.passed: + passed, failed, failed_claim_ids = n, 0, [] + else: + passed, failed, failed_claim_ids = 0, n, [ + c.payer_claim_control_number for c in result.claims + ] + result = result.model_copy(update={ + "validation": report, + "summary": result.summary.model_copy(update={ + "passed": passed, + "failed": failed, + "failed_claim_ids": failed_claim_ids, + }), + }) + + rec = BatchRecord( + id=uuid.uuid4().hex, + kind="835", + input_filename=source_file, + parsed_at=datetime.now(timezone.utc), + result=result, + ) + cycl_store.add(rec) + return "parse_835", len(result.claims) + + +def _handle_277ca(text: str, source_file: str) -> tuple[str, int]: + """Parse a 277CA, persist ack + stamp payer-rejected claims.""" + from cyclone.parsers.parse_277ca import parse_277ca_text + from cyclone.parsers.exceptions import CycloneParseError + + try: + result = parse_277ca_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"277CA parse error: {exc}") from exc + + icn = result.envelope.control_number + synthetic_id = _277ca_synthetic_source_batch_id(icn) + accepted = sum( + 1 for s in result.claim_statuses if s.classification == "accepted" + ) + paid = sum( + 1 for s in result.claim_statuses if s.classification == "paid" + ) + rejected = sum( + 1 for s in result.claim_statuses if s.classification == "rejected" + ) + pended = sum( + 1 for s in result.claim_statuses if s.classification == "pended" + ) + + with db.SessionLocal()() as session: + row = cycl_store.add_277ca_ack( + source_batch_id=synthetic_id, + control_number=icn, + accepted_count=accepted, + rejected_count=rejected, + paid_count=paid, + pended_count=pended, + raw_json=json.loads(result.model_dump_json()), + ) + def _lookup(pcn: str): + return ( + session.query(db.Claim) + .filter(db.Claim.patient_control_number == pcn) + .first() + ) + apply_result = apply_277ca_rejections( + session, result, claim_lookup=_lookup, two77ca_id=row.id, + ) + if apply_result.matched: + for cid in apply_result.matched: + append_event(session, AuditEvent( + event_type="claim.payer_rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id, "277ca_id": row.id}, + actor="277ca-parser-scheduler", + )) + session.commit() + return "parse_277ca", len(result.claim_statuses) + + +def _handle_ta1(text: str, source_file: str) -> tuple[str, int]: + """Parse a TA1, persist the interchange ack row.""" + from cyclone.parsers.parse_ta1 import parse_ta1_text + from cyclone.parsers.exceptions import CycloneParseError + + try: + result = parse_ta1_text(text, input_file=source_file) + except CycloneParseError as exc: + raise ValueError(f"TA1 parse error: {exc}") from exc + + with db.SessionLocal()() as session: + cycl_store.add_ta1_ack( + source_batch_id=result.source_batch_id, + control_number=result.ta1.control_number, + interchange_date=result.ta1.interchange_date, + interchange_time=result.ta1.interchange_time, + ack_code=result.ta1.ack_code, + note_code=result.ta1.note_code, + ack_generated_date=result.ta1.ack_generated_date, + sender_id=result.envelope.sender_id, + receiver_id=result.envelope.receiver_id, + raw_json=json.loads(result.model_dump_json()), + ) + session.commit() + return "parse_ta1", 1 + + +# Map file_type → handler. Mirrors ROUTED_FILE_TYPES. +HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = { + "999": _handle_999, + "835": _handle_835, + "277": _handle_277ca, # filename uses 277; parser is the same + "277CA": _handle_277ca, + "TA1": _handle_ta1, +} + + +# --------------------------------------------------------------------------- +# Light copies of helpers the API endpoints use, so the scheduler can +# run without depending on the FastAPI module. +# --------------------------------------------------------------------------- + + +def _ack_count_summary(result: Any) -> tuple[int, int, int, str]: + """Return (received, accepted, rejected, ack_code) for a 999. + + Mirrors the logic in ``cyclone.api._ack_count_summary`` but lives + here so the scheduler can run without importing the API module. + """ + if result.functional_group_acks: + fg = result.functional_group_acks[0] + return ( + fg.received_count, fg.accepted_count, + fg.rejected_count, fg.ack_code, + ) + sets = result.set_responses + received = len(sets) + accepted = sum(1 for s in sets if s.set_accept_reject.code == "A") + rejected = received - accepted + if rejected == 0: + code = "A" + elif accepted == 0: + code = "R" + else: + code = "P" + return (received, accepted, rejected, code) + + +def _ack_synthetic_source_batch_id(interchange_control_number: str) -> str: + """Synthetic batches.id for a received 999 with no source batch.""" + return f"999-{(interchange_control_number or '').strip() or '000000001'}" + + +def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str: + """Synthetic batches.id for a received 277CA with no source batch.""" + return f"277CA-{(interchange_control_number or '').strip() or '000000001'}" + + +# --------------------------------------------------------------------------- +# Scheduler +# --------------------------------------------------------------------------- + + +class Scheduler: + """Background polling loop for inbound MFT files. + + Lifecycle: + sched = Scheduler(sftp_block, poll_interval_seconds=60) + await sched.start() # begin polling + # ... later ... + await sched.stop() # finish current tick, then exit + status = sched.status() # snapshot + + The scheduler is a single asyncio task. ``tick()`` does one full + poll cycle and is exposed for tests + the ``/api/admin/scheduler/tick`` + endpoint so the operator can force a poll without waiting. + + Threading: NOT thread-safe. All access (start/stop/tick/status) + must happen on the same event loop. The FastAPI app satisfies + this trivially because endpoints run on the loop. + """ + + def __init__( + self, + sftp_block: SftpBlock, + *, + poll_interval_seconds: int = 60, + sftp_block_name: str = "default", + sftp_client_factory: Optional[Callable[[SftpBlock], Any]] = None, + ) -> None: + self._sftp_block = sftp_block + self._poll_interval = poll_interval_seconds + self._sftp_block_name = sftp_block_name + # Factory indirection lets tests substitute a fake client + # without monkey-patching the module-level SftpClient. + self._sftp_client_factory = sftp_client_factory or SftpClient + + self._task: Optional[asyncio.Task[None]] = None + self._stop_event = asyncio.Event() + self._last_poll_at: Optional[datetime] = None + self._poll_count = 0 + self._total_processed = 0 + self._total_skipped = 0 + self._total_errored = 0 + self._last_tick: Optional[TickResult] = None + # Coalesce overlapping ticks (a slow MFT server shouldn't let + # ticks stack up; the next tick fires only after the previous + # one finishes). + self._tick_in_progress = False + + # ---- Public API ------------------------------------------------------- + + async def start(self) -> None: + """Begin polling. Idempotent.""" + if self._task is not None and not self._task.done(): + log.info("Scheduler already running; start() is a no-op") + return + self._stop_event.clear() + self._task = asyncio.create_task(self._run(), name="mft-scheduler") + log.info( + "Scheduler started: poll every %ds, sftp_block=%r", + self._poll_interval, self._sftp_block_name, + ) + + async def stop(self) -> None: + """Stop polling. Waits for the current tick to finish.""" + if self._task is None or self._task.done(): + return + self._stop_event.set() + try: + await asyncio.wait_for(self._task, timeout=30) + except asyncio.TimeoutError: + log.warning("Scheduler did not stop within 30s; cancelling") + self._task.cancel() + try: + await self._task + except (asyncio.CancelledError, Exception): # noqa: BLE001 + pass + self._task = None + log.info("Scheduler stopped") + + def status(self) -> SchedulerStatus: + """Return a snapshot of the scheduler's state.""" + return SchedulerStatus( + running=self.is_running(), + poll_interval_seconds=self._poll_interval, + sftp_block_name=self._sftp_block_name, + last_poll_at=self._last_poll_at, + poll_count=self._poll_count, + total_processed=self._total_processed, + total_skipped=self._total_skipped, + total_errored=self._total_errored, + last_tick=self._last_tick, + ) + + def is_running(self) -> bool: + return self._task is not None and not self._task.done() + + async def tick(self) -> TickResult: + """Run a single poll cycle and return the outcome. + + Concurrent ticks are coalesced: if a tick is already in + progress, the second caller waits for it. This protects the + SFTP server from a stampede when the operator hits + ``/api/admin/scheduler/tick`` while a scheduled tick is + already running. + """ + while self._tick_in_progress: + await asyncio.sleep(0.05) + self._tick_in_progress = True + try: + result = await self._tick_impl() + self._last_tick = result + self._last_poll_at = result.finished_at or result.started_at + self._poll_count += 1 + self._total_processed += result.files_processed + self._total_skipped += result.files_skipped + self._total_errored += result.files_errored + return result + finally: + self._tick_in_progress = False + + # ---- Internals -------------------------------------------------------- + + async def _run(self) -> None: + """Main loop. Runs until ``stop()`` is called.""" + # Stagger the first tick so we don't hammer the MFT server on + # startup if multiple operators restart Cyclone in lockstep. + await asyncio.sleep(1) + while not self._stop_event.is_set(): + try: + await self.tick() + except Exception as exc: # noqa: BLE001 + # tick() should never raise — it catches per-file + # exceptions. This is the safety net for SFTP outages + # or DB connectivity issues. + log.exception("Scheduler tick raised: %s", exc) + try: + await asyncio.wait_for( + self._stop_event.wait(), + timeout=self._poll_interval, + ) + except asyncio.TimeoutError: + pass # poll interval elapsed; loop again + + async def _tick_impl(self) -> TickResult: + """One poll cycle: list → filter already-processed → route each.""" + started = datetime.now(timezone.utc) + result = TickResult(started_at=started) + + try: + files = await asyncio.to_thread(self._list_inbound) + except Exception as exc: # noqa: BLE001 + log.exception("SFTP list_inbound failed") + result.errors.append(f"list_inbound: {exc}") + result.finished_at = datetime.now(timezone.utc) + return result + + result.files_seen = len(files) + for f in files: + if self._stop_event.is_set(): + break + await self._handle_one(f, result) + + result.finished_at = datetime.now(timezone.utc) + return result + + def _list_inbound(self) -> list[InboundFile]: + """Return files in the inbound MFT path. Runs on a thread.""" + client = self._sftp_client_factory(self._sftp_block) + return client.list_inbound() + + async def _handle_one(self, f: InboundFile, result: TickResult) -> None: + """Process one inbound file: skip-if-seen, classify, parse, record.""" + if await self._already_processed(f.name): + return + + try: + inbound = parse_inbound_filename(f.name) + file_type = inbound.file_type + except ValueError: + file_type = None + + if file_type not in HANDLERS: + await self._record( + name=f.name, size=f.size, modified_at=f.modified_at, + file_type=file_type, parser_used=None, claim_count=0, + status=STATUS_SKIPPED, + error_message=( + f"file_type {file_type!r} not in {sorted(HANDLERS)}" + if file_type else "filename does not match HCPF inbound format" + ), + ) + result.files_skipped += 1 + return + + try: + _path, parser_used, claim_count = await asyncio.to_thread( + self._download_and_parse, f, file_type, + ) + except Exception as exc: # noqa: BLE001 + log.exception("Failed to process %s", f.name) + await self._record( + name=f.name, size=f.size, modified_at=f.modified_at, + file_type=file_type, parser_used=None, claim_count=0, + status=STATUS_ERROR, + error_message=( + f"{type(exc).__name__}: {exc}\n" + f"{traceback.format_exc()[-500:]}" + ), + ) + result.files_errored += 1 + result.errors.append(f"{f.name}: {exc}") + return + + await self._record( + name=f.name, size=f.size, modified_at=f.modified_at, + file_type=file_type, parser_used=parser_used, claim_count=claim_count, + status=STATUS_OK, error_message=None, + ) + result.files_processed += 1 + log.info( + "Processed inbound %s: parser=%s claims=%d", + f.name, parser_used, claim_count, + ) + + async def _already_processed(self, name: str) -> bool: + with db.SessionLocal()() as session: + row = ( + session.query(ProcessedInboundFile) + .filter_by(sftp_block_name=self._sftp_block_name, name=name) + .filter(ProcessedInboundFile.status != STATUS_PENDING) + .first() + ) + return row is not None + + async def _record( + self, + *, + name: str, + size: int, + modified_at: datetime, + file_type: Optional[str], + parser_used: Optional[str], + claim_count: int, + status: str, + error_message: Optional[str], + ) -> None: + """Persist a processed_inbound_files row. Idempotent.""" + with db.SessionLocal()() as session: + row = ProcessedInboundFile( + sftp_block_name=self._sftp_block_name, + name=name, + size=size, + modified_at=modified_at, + file_type=file_type, + processed_at=datetime.now(timezone.utc), + parser_used=parser_used, + claim_count=claim_count, + status=status, + error_message=error_message, + ) + session.add(row) + try: + session.commit() + except IntegrityError: + # A concurrent scheduler (or a retry after a partial + # failure) already recorded this file. That's fine — + # the latest row wins; we just skip the dup. + session.rollback() + + def _download_and_parse( + self, f: InboundFile, file_type: str, + ) -> tuple[Path, str, int]: + """Download from MFT, run the right handler. Returns (path, parser, count). + + Stub mode: ``f.local_path`` already points at the staged file + (set by ``SftpClient._list_inbound_stub``). Real mode: the + remote name is ``f.name`` and we round-trip through paramiko. + """ + if self._sftp_block.stub: + # In stub mode the InboundFile already has a local_path; + # reading the staged bytes directly avoids the stub's + # remote-path semantics (which expect a full inbound path). + content = f.local_path.read_bytes() + else: + client = self._sftp_client_factory(self._sftp_block) + content = client.read_file(f.name) + text = content.decode("utf-8") + handler = HANDLERS[file_type] + parser_used, claim_count = handler(text, f.name) + return f.local_path, parser_used, claim_count + + +# --------------------------------------------------------------------------- +# Module-level singleton — only one scheduler per process. +# --------------------------------------------------------------------------- + + +_scheduler: Optional[Scheduler] = None + + +def configure_scheduler( + sftp_block: SftpBlock, + *, + poll_interval_seconds: int = 60, + sftp_block_name: str = "default", + force: bool = False, +) -> Scheduler: + """Create the module-level scheduler singleton (or return the existing one). + + Called from the FastAPI lifespan handler. Tests pre-configure the + scheduler before the TestClient opens the lifespan; in that case + we leave the existing singleton alone (``force=False``). Pass + ``force=True`` to replace unconditionally. + """ + global _scheduler + if _scheduler is not None and not force: + return _scheduler + poll = int( + os.environ.get("CYCLONE_SCHEDULER_POLL_SECONDS", poll_interval_seconds), + ) + _scheduler = Scheduler( + sftp_block, + poll_interval_seconds=poll, + sftp_block_name=sftp_block_name, + ) + return _scheduler + + +def get_scheduler() -> Scheduler: + """Return the module-level scheduler. + + Raises: + RuntimeError: if ``configure_scheduler`` hasn't been called. + """ + if _scheduler is None: + raise RuntimeError( + "scheduler not configured; call configure_scheduler() first", + ) + return _scheduler + + +def reset_scheduler_for_tests() -> None: + """Clear the module-level scheduler. Test-only.""" + global _scheduler + _scheduler = None \ No newline at end of file diff --git a/backend/tests/fixtures/minimal_ta1.txt b/backend/tests/fixtures/minimal_ta1.txt new file mode 100644 index 0000000..06e5faf --- /dev/null +++ b/backend/tests/fixtures/minimal_ta1.txt @@ -0,0 +1 @@ +ISA*00* *00* *ZZ*SENDER *ZZ*RECEIVER *260520*1750*^*00501*000000001*0*P*:~TA1*000000001*20260520*1750*A*000*20260520~IEA*1*000000001~ \ No newline at end of file diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py index 39f1dfc..4b013f8 100644 --- a/backend/tests/test_acks.py +++ b/backend/tests/test_acks.py @@ -51,18 +51,19 @@ def test_migration_0002_creates_acks_table(): def test_migration_latest_idempotent_on_fresh_db(): """Re-running the migration on the same DB must be a no-op (PRAGMA - user_version already at the latest version — currently 10 after + user_version already at the latest version — currently 11 after 0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007 providers/payers/clearhouse, SP10's 0008 payer_rejected, - SP11's 0009 audit_log, and SP14's 0010 payer_rejected_acknowledged).""" + SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged, + SP16's 0011 processed_inbound_files).""" with db.engine().begin() as c: v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v1 == 10 + assert v1 == 11 # A second run should not raise and should not bump the version. db_migrate.run(db.engine()) with db.engine().begin() as c: v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0 - assert v2 == 10 + assert v2 == 11 def test_add_ack_persists_row(): diff --git a/backend/tests/test_api_scheduler.py b/backend/tests/test_api_scheduler.py new file mode 100644 index 0000000..cb05a31 --- /dev/null +++ b/backend/tests/test_api_scheduler.py @@ -0,0 +1,152 @@ +"""SP16 — Admin scheduler API endpoint tests. + +The endpoints under /api/admin/scheduler/* are thin wrappers around +:class:`cyclone.scheduler.Scheduler`. These tests exercise them via +the FastAPI TestClient to confirm wiring (auth-free admin endpoints +work, response shapes match, idempotency holds). +""" +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest + + +@pytest.fixture +def _stub_scheduler_env(tmp_path, monkeypatch): + """Set up: a stub-mode SFTP block, scheduler configured. + + Yields (staging_dir, scheduler_singleton). We deliberately do + NOT enable SQLCipher encryption in this fixture — the scheduler + doesn't care about encryption, and patching ``db_crypto.get_secret`` + here would cause the lifespan handler to rebuild the engine with + SQLCipher on a plain-SQLite test file (which raises "file is not + a database"). The encryption-at-rest tests live in + ``test_db_crypto.py``. + """ + from cyclone import db + from cyclone import scheduler as sched_mod + from cyclone.providers import SftpBlock + + monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") + db._reset_for_tests() + + staging = tmp_path / "staging" + inbound = staging / "ToHPE" + inbound.mkdir(parents=True) + sftp_block = 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"}, + ) + sched_mod.reset_scheduler_for_tests() + sched = sched_mod.configure_scheduler(sftp_block, sftp_block_name="t") + yield staging, sched + sched_mod.reset_scheduler_for_tests() + db._reset_for_tests() + + +def _drop_file(staging: Path, name: str, body: bytes) -> Path: + p = staging / "ToHPE" / name + p.write_bytes(body) + return p + + +def test_scheduler_status_starts_not_running(_stub_scheduler_env): + from fastapi.testclient import TestClient + from cyclone.api import app + _, sched = _stub_scheduler_env + with TestClient(app) as client: + r = client.get("/api/admin/scheduler/status") + assert r.status_code == 200 + body = r.json() + assert body["running"] is False + assert body["poll_interval_seconds"] == 60 + assert body["sftp_block_name"] == "t" + + +def test_scheduler_start_then_status_then_stop(_stub_scheduler_env): + from fastapi.testclient import TestClient + from cyclone.api import app + with TestClient(app) as client: + r1 = client.post("/api/admin/scheduler/start") + assert r1.status_code == 200 + assert r1.json()["status"]["running"] is True + + r2 = client.get("/api/admin/scheduler/status") + assert r2.json()["running"] is True + + r3 = client.post("/api/admin/scheduler/stop") + assert r3.status_code == 200 + assert r3.json()["status"]["running"] is False + + +def test_scheduler_tick_processes_one_file(_stub_scheduler_env): + from fastapi.testclient import TestClient + from cyclone.api import app + staging, _ = _stub_scheduler_env + _drop_file( + staging, + "TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12", + (Path(__file__).parent / "fixtures" / "minimal_ta1.txt").read_bytes(), + ) + with TestClient(app) as client: + r = client.post("/api/admin/scheduler/tick") + assert r.status_code == 200 + body = r.json() + assert body["ok"] is True + assert body["tick"]["files_seen"] == 1 + assert body["tick"]["files_processed"] == 1 + + +def test_scheduler_processed_files_lists_history(_stub_scheduler_env): + from fastapi.testclient import TestClient + from cyclone.api import app + staging, _ = _stub_scheduler_env + _drop_file( + staging, + "TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12", + (Path(__file__).parent / "fixtures" / "minimal_ta1.txt").read_bytes(), + ) + with TestClient(app) as client: + client.post("/api/admin/scheduler/tick") + r = client.get("/api/admin/scheduler/processed-files") + assert r.status_code == 200 + body = r.json() + assert body["count"] == 1 + f = body["files"][0] + assert f["status"] == "ok" + assert f["parser_used"] == "parse_ta1" + assert "TP11525703" in f["name"] + + +def test_scheduler_processed_files_filters_by_status(_stub_scheduler_env): + from fastapi.testclient import TestClient + from cyclone.api import app + staging, _ = _stub_scheduler_env + # Drop a file with a type Cyclone doesn't parse — gets recorded as + # "skipped". + _drop_file( + staging, + "TP11525703-837P_M019048402-20260618130000000-1of1_270.x12", + b"some bytes", + ) + with TestClient(app) as client: + client.post("/api/admin/scheduler/tick") + r_all = client.get("/api/admin/scheduler/processed-files") + r_skipped = client.get( + "/api/admin/scheduler/processed-files?status=skipped", + ) + r_ok = client.get( + "/api/admin/scheduler/processed-files?status=ok", + ) + assert r_all.json()["count"] == 1 + assert r_skipped.json()["count"] == 1 + assert r_ok.json()["count"] == 0 \ No newline at end of file diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py new file mode 100644 index 0000000..3a19fc4 --- /dev/null +++ b/backend/tests/test_scheduler.py @@ -0,0 +1,287 @@ +"""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 \ No newline at end of file diff --git a/backend/tests/test_sftp_stub.py b/backend/tests/test_sftp_stub.py index c1d712a..882687e 100644 --- a/backend/tests/test_sftp_stub.py +++ b/backend/tests/test_sftp_stub.py @@ -72,7 +72,23 @@ def test_stub_get_secret_returns_stub_when_keychain_empty(sftp_block): assert secret == "" -def test_stub_read_file_raises(sftp_block): +def test_stub_read_file_returns_bytes(sftp_block, tmp_path): + """SP16: the stub read_file returns bytes from the staging dir. + + Lets the inbound scheduler exercise the same code path on a + workstation without a real MFT connection. + """ + inbound = tmp_path / "staging" / "ToHPE" + inbound.mkdir(parents=True) + (inbound / "TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12").write_bytes( + b"hello-world", + ) client = SftpClient(sftp_block) - with pytest.raises(RuntimeError, match="Stub SFTP cannot read"): - client.read_file("/x.x12") + body = client.read_file("/ToHPE/TP11525703-837P_M019048402-20260618130000000-1of1_TA1.x12") + assert body == b"hello-world" + + +def test_stub_read_file_missing_raises(sftp_block): + client = SftpClient(sftp_block) + with pytest.raises(FileNotFoundError): + client.read_file("/ToHPE/does-not-exist.x12")