merge: SP16 live MFT polling scheduler into main
This commit is contained in:
Tyler
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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);
|
||||
@@ -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
|
||||
+1
@@ -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~
|
||||
@@ -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():
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -72,7 +72,23 @@ def test_stub_get_secret_returns_stub_when_keychain_empty(sftp_block):
|
||||
assert secret == "<stub-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")
|
||||
|
||||
Reference in New Issue
Block a user