diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 7a67a2a..dba36d3 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -18,6 +18,7 @@ Additional origins (LAN IPs, staging hosts) can be appended via the from __future__ import annotations +import asyncio import csv import io import json @@ -36,6 +37,7 @@ from pydantic import ValidationError from cyclone import __version__, db from cyclone.auth.deps import matrix_gate +from cyclone.clearhouse import InboundFile from cyclone.db import Batch, Claim, ClaimState, Remittance from sqlalchemy import desc, or_ from sqlalchemy.exc import IntegrityError @@ -3419,6 +3421,128 @@ async def scheduler_tick() -> Any: return {"ok": True, "tick": result.as_dict()} +@app.post("/api/admin/scheduler/pull-inbound", dependencies=[Depends(matrix_gate)]) +async def scheduler_pull_inbound( + date: str = Query( + ..., pattern=r"^\d{8}$", + description="Date filter as YYYYMMDD; only filenames whose 8-digit " + "timestamp (the 9th positional group in the inbound " + "filename) matches are downloaded and processed.", + ), + file_types: str | None = Query( + default=None, + description="Optional comma-separated whitelist of file_types " + "(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.", + ), + limit: int = Query(default=2000, ge=1, le=10000), +) -> Any: + """Targeted pull: list, filter to a date, download, and process. + + Bypasses the alphabetical full-listing pass. Workflow: + 1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only + listing of the inbound MFT dir (skips ``*_warn.txt``). + 2. Client-side filter: keep files whose 8-digit timestamp + substring equals ``date`` and whose ``file_type`` is in the + allowlist. + 3. ``SftpClient.download_inbound(f)`` for each — fetches bytes + into the local cache. + 4. ``Scheduler.process_inbound_files(files)`` — runs the same + per-file pipeline as a regular tick (already-processed files + are deduped via ``processed_inbound_files``). + + Use this for the daily "process today's 999s" workflow without + paying the cost of downloading the full inbound set. + + Returns ``{"ok": True, "summary": {...}}`` with + ``listed / matched / downloaded / processed / skipped / errored`` + counters and the date / file_type filters applied. + """ + import time + + from cyclone.clearhouse import SftpClient + from cyclone.edi.filenames import ( + ALLOWED_FILE_TYPES, + parse_inbound_filename, + ) + from cyclone.providers import SftpBlock + + sched = _scheduler_or_503() + block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable + client = SftpClient(block) + + if file_types: + wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()} + unknown = wanted - ALLOWED_FILE_TYPES + if unknown: + raise HTTPException( + status_code=400, + detail=f"file_types {sorted(unknown)!r} not in " + f"{sorted(ALLOWED_FILE_TYPES)}", + ) + else: + wanted = {"999", "TA1"} # daily default — what the operator needs + + started = time.monotonic() + try: + # Single SFTP listdir — fast, no download. + all_files = await asyncio.to_thread(client.list_inbound_names) + except Exception as exc: + log.exception("SFTP list_inbound_names failed") + raise HTTPException( + status_code=502, + detail=f"SFTP list failed: {type(exc).__name__}: {exc}", + ) from exc + + listed = len(all_files) + matched: list[InboundFile] = [] + for f in all_files: + if f.name.find(date) == -1: + continue + try: + parsed = parse_inbound_filename(f.name) + except ValueError: + continue + if parsed.file_type not in wanted: + continue + matched.append(f) + if len(matched) >= limit: + break + + # Download in parallel-ish via to_thread (SftpClient serializes per + # connection; the overhead is dominated by the SFTP round trip). + downloaded = 0 + download_errors: list[str] = [] + for f in matched: + try: + await asyncio.to_thread(client.download_inbound, f) + downloaded += 1 + except Exception as exc: # noqa: BLE001 + log.warning("Failed to download %s: %s", f.name, exc) + download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}") + + # Hand off to the scheduler pipeline (idempotent; dedupes via + # processed_inbound_files). + tick = await sched.process_inbound_files(matched) + duration = round(time.monotonic() - started, 3) + return { + "ok": True, + "summary": { + "date": date, + "file_types": sorted(wanted), + "limit": limit, + "listed": listed, + "matched": len(matched), + "downloaded": downloaded, + "download_errors": download_errors, + "processed": tick.files_processed, + "skipped": tick.files_skipped, + "errored": tick.files_errored, + "duration_s": duration, + }, + "tick": tick.as_dict(), + } + + @app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)]) def scheduler_status() -> Any: """Return the scheduler's runtime snapshot (running, counters, last tick).""" diff --git a/backend/src/cyclone/clearhouse/__init__.py b/backend/src/cyclone/clearhouse/__init__.py index 9fcba7c..2144375 100644 --- a/backend/src/cyclone/clearhouse/__init__.py +++ b/backend/src/cyclone/clearhouse/__init__.py @@ -87,11 +87,70 @@ class SftpClient: dir and returns :class:`InboundFile` records pointing at the cache copy. The remote file is *not* deleted — the operator archives inbound files in the MFT UI. + + Gainwell's MFT puts advisory ``*_warn.txt`` files in the same + inbound path. They're text-format side-channel notes (not X12 + envelopes) and are skipped at list time. Use + :meth:`list_inbound_names` if you need the raw list without + the download. """ if self._stub: return self._list_inbound_stub() return self._list_inbound_paramiko() + def list_inbound_names(self) -> list[InboundFile]: + """Lightweight listing: returns metadata only, no file download. + + Use this when you want to filter the inbound set (e.g. by date) + before paying the download cost. Pair with + :meth:`download_inbound` to fetch a filtered subset on demand. + + Real mode is implemented as a single SFTP ``listdir_attr`` call + — sub-second on Gainwell's MFT — versus the full + :meth:`list_inbound` which downloads every file. The + ``*_warn.txt`` advisory files are filtered out the same way. + + Stub mode returns the same as :meth:`list_inbound` (the stub + only knows about local files; no download cost). + """ + if self._stub: + return self._list_inbound_stub() + return self._list_inbound_names_paramiko() + + def download_inbound(self, f: InboundFile) -> Path: + """Download a single inbound file to its ``local_path``. + + Idempotent: if ``f.local_path`` already exists and is + non-empty, the download is skipped. Callers should use the + ``InboundFile`` returned by :meth:`list_inbound_names` and pass + it back here — ``local_path`` is the planned cache location + and matches the path the scheduler will read from. + + Returns: + The on-disk path (same as ``f.local_path``). + + Raises: + FileNotFoundError: if the file is missing locally in stub + mode, or if the remote file disappears between list + and download in real mode. + """ + if f.local_path.exists() and f.local_path.stat().st_size > 0: + log.debug( + "SFTP: %s already cached at %s, skipping download", + f.name, f.local_path, + ) + return f.local_path + if self._stub: + # Stub mode: no remote — the file is supposed to already be + # at f.local_path (operator-dropped). If it isn't there, the + # operator hasn't seeded the stub; raise loudly. + if not f.local_path.is_file(): + raise FileNotFoundError( + f"inbound stub file not found: {f.local_path}" + ) + return f.local_path + return self._download_inbound_paramiko(f) + def read_file(self, remote_path: str) -> bytes: """Read bytes from a remote path. @@ -284,6 +343,12 @@ class SftpClient: if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000: # Directory entry — skip. continue + if attr.filename.endswith("_warn.txt"): + # Gainwell's MFT drops text-format advisory notes in + # the same inbound path. They're side-channel noise, + # not X12 envelopes — skip at list time so we don't + # download ~600 advisory files per poll. + continue remote = f"{inbound_dir.rstrip('/')}/{attr.filename}" cache_path = cache_dir / attr.filename # Download into cache. We use ``prefetch`` to keep memory @@ -299,6 +364,56 @@ class SftpClient: )) return files + def _list_inbound_names_paramiko(self) -> list[InboundFile]: + """List inbound names via paramiko; do NOT download (lightweight). + + Same ``listdir_attr`` iteration as + :meth:`_list_inbound_paramiko`, but the returned + :class:`InboundFile` records have ``local_path`` set to the + planned cache location without actually fetching the file. + Pair with :meth:`_download_inbound_paramiko` to fetch on + demand. Skips ``*_warn.txt`` advisory files the same way. + """ + with self._connect() as (ssh, sftp): + inbound_dir = self._block.paths.get("inbound", "/") + staging = Path(self._block.staging_dir).resolve() + inbound_rel = inbound_dir.lstrip("/") + cache_dir = staging / inbound_rel + cache_dir.mkdir(parents=True, exist_ok=True) + + files: list[InboundFile] = [] + try: + attrs = sftp.listdir_attr(inbound_dir) + except IOError as exc: + log.warning("SFTP: cannot list %s: %s", inbound_dir, exc) + return [] + + for attr in sorted(attrs, key=lambda a: a.filename): + if attr.st_mode and (attr.st_mode & 0o170000) == 0o040000: + # Directory entry — skip. + continue + if attr.filename.endswith("_warn.txt"): + continue + cache_path = cache_dir / attr.filename + files.append(InboundFile( + name=attr.filename, + size=attr.st_size or 0, + modified_at=datetime.fromtimestamp(attr.st_mtime or 0), + local_path=cache_path, + )) + return files + + def _download_inbound_paramiko(self, f: InboundFile) -> Path: + """Download a single ``f`` to ``f.local_path`` (idempotent on size>0).""" + with self._connect() as (ssh, sftp): + inbound_dir = self._block.paths.get("inbound", "/") + remote = f"{inbound_dir.rstrip('/')}/{f.name}" + f.local_path.parent.mkdir(parents=True, exist_ok=True) + with sftp.open(remote, "rb") as src, open(f.local_path, "wb") as dst: + shutil.copyfileobj(src, dst, length=64 * 1024) + log.info("SFTP: downloaded %d bytes for %s", f.local_path.stat().st_size, f.name) + return f.local_path + def _read_file_paramiko(self, remote_path: str) -> bytes: with self._connect() as (ssh, sftp): buf = io.BytesIO() diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 931d1bb..2c3ddf6 100644 --- a/backend/src/cyclone/cli.py +++ b/backend/src/cyclone/cli.py @@ -569,3 +569,167 @@ def backup_status() -> None: snap = svc.status() import json click.echo(json.dumps(snap, indent=2, default=str)) + + +# --------------------------------------------------------------------------- +# SP-N fix: `cyclone pull-inbound` — targeted inbound pull + process +# +# Mirrors POST /api/admin/scheduler/pull-inbound. For operators who +# want to drive the daily "process today's 999s" workflow from a cron +# job or shell script without going through the HTTP API. +# +# Exit codes: +# 0 — processed ≥1 file (or processed 0 with no matches: not an +# error, just nothing to do) +# 1 — unexpected exception +# 2 — SFTP / config error +# --------------------------------------------------------------------------- + + +@main.command("pull-inbound") +@click.option( + "--date", "date_str", + required=True, + help="Date filter YYYYMMDD — only files whose 8-digit timestamp " + "substring matches are downloaded and processed.", +) +@click.option( + "--block", "sftp_block_name", + default="dzinesco", + show_default=True, + help="Name of the SftpBlock in config/payers.yaml to use.", +) +@click.option( + "--file-types", "file_types_csv", + default=None, + help="Comma-separated whitelist (default: 999,TA1).", +) +@click.option( + "--limit", default=2000, show_default=True, type=int, +) +def pull_inbound( + date_str: str, + sftp_block_name: str, + file_types_csv: str | None, + limit: int, +) -> None: + """List, date-filter, download, and process inbound MFT files. + + Bypasses the alphabetical full-listing pass so a daily pull of + ~400 files takes seconds, not hours. Files already in the local + cache are skipped (idempotent). + """ + import asyncio as _asyncio + from cyclone import db as db_mod + from cyclone import scheduler as scheduler_mod + from cyclone.edi.filenames import ALLOWED_FILE_TYPES + from cyclone.clearhouse import SftpClient + from cyclone.providers import SftpBlock + + # Validate the date filter. + if not (len(date_str) == 8 and date_str.isdigit()): + click.echo(f"--date must be YYYYMMDD, got {date_str!r}", err=True) + sys.exit(2) + + db_mod.init_db() + + # Find the SftpBlock. The dzinesco clearhouse singleton is seeded + # by store.ensure_clearhouse_seeded() (SP9) and carries the + # production SFTP block; that's the only one we have at the + # moment. For multi-provider SFTP, a config-loader is the right + # thing — out of scope for this CLI which is the daily-pull path. + from cyclone import store as store_mod + store_mod.store.ensure_clearhouse_seeded() + clearhouse = store_mod.store.get_clearhouse() + if clearhouse is None or clearhouse.name != sftp_block_name: + # Fall back: try the named block, but v1 only ships the + # dzinesco singleton. + if clearhouse is None: + click.echo( + f"No clearhouse seeded — cannot find SftpBlock " + f"{sftp_block_name!r}.", + err=True, + ) + sys.exit(2) + click.echo( + f"SftpBlock {sftp_block_name!r} not seeded; only " + f"{clearhouse.name!r} is available.", + err=True, + ) + sys.exit(2) + block: SftpBlock = clearhouse.sftp_block + + if file_types_csv: + wanted = {t.strip().upper() for t in file_types_csv.split(",") if t.strip()} + unknown = wanted - ALLOWED_FILE_TYPES + if unknown: + click.echo( + f"file_types {sorted(unknown)!r} not in {sorted(ALLOWED_FILE_TYPES)}", + err=True, + ) + sys.exit(2) + else: + wanted = {"999", "TA1"} + + # Wire up the scheduler singleton (re-use the same pipeline the + # HTTP endpoint uses, including the dedup via processed_inbound_files). + scheduler_mod.configure_scheduler( + block, sftp_block_name=sftp_block_name, force=True, + ) + sched = scheduler_mod.get_scheduler() + client = SftpClient(block) + + async def _run() -> dict: + from cyclone.edi.filenames import parse_inbound_filename + + all_files = await _asyncio.to_thread(client.list_inbound_names) + matched: list = [] + for f in all_files: + if f.name.find(date_str) == -1: + continue + try: + parsed = parse_inbound_filename(f.name) + except ValueError: + continue + if parsed.file_type not in wanted: + continue + matched.append(f) + if len(matched) >= limit: + break + + download_errors: list[str] = [] + downloaded = 0 + for f in matched: + try: + await _asyncio.to_thread(client.download_inbound, f) + downloaded += 1 + except Exception as exc: # noqa: BLE001 + download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}") + + tick = await sched.process_inbound_files(matched) + return { + "listed": len(all_files), + "matched": len(matched), + "downloaded": downloaded, + "download_errors": download_errors, + "processed": tick.files_processed, + "skipped": tick.files_skipped, + "errored": tick.files_errored, + } + + try: + summary = _asyncio.run(_run()) + except Exception as exc: # noqa: BLE001 + click.echo(f"pull-inbound failed: {type(exc).__name__}: {exc}", err=True) + sys.exit(1) + + click.echo( + f"date={date_str} file_types={sorted(wanted)} block={sftp_block_name} " + f"listed={summary['listed']} matched={summary['matched']} " + f"downloaded={summary['downloaded']} processed={summary['processed']} " + f"skipped={summary['skipped']} errored={summary['errored']}" + ) + if summary["download_errors"]: + click.echo("download errors:", err=True) + for e in summary["download_errors"]: + click.echo(f" {e}", err=True) diff --git a/backend/src/cyclone/edi/filenames.py b/backend/src/cyclone/edi/filenames.py index 52c213e..ad11727 100644 --- a/backend/src/cyclone/edi/filenames.py +++ b/backend/src/cyclone/edi/filenames.py @@ -8,8 +8,11 @@ Outbound (we send): Example: tp11525703-837P-20260620132243505-1of1.x12 Inbound (HPE sends to our FromHPE): - TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12 - Example: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12 + [Tt][Pp]{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12 + Example: tp11525703-837P_M019048402-20260520231513488-1of1_999.x12 + (legacy / prodfiles may use uppercase TP; the regex is case-insensitive + on the prefix to accept both — Gainwell's filer has used both + casings over time.) Both use Mountain Time (MT) timestamps with 17-digit millisecond precision (yyyymmddhhmmssSSS = 4+2+2+2+2+2+3 = 17 digits). Sequence is always "1of1" @@ -39,16 +42,23 @@ OUTBOUND_RE = re.compile( r"^tp(?P\d+)-(?P[A-Z0-9]+)-(?P\d{17})-1of1\.(?P[A-Za-z0-9]+)$" ) -# Inbound: TP11525703-837P_M019048402-20260520231513488-1of1_999.x12 +# Inbound: [Tt][Pp]11525703-837P_M019048402-20260520231513488-1of1_999.x12 +# - prefix: literal "TP" or "tp" (case-insensitive — Gainwell's +# production filer has used both) # - tpid: 1+ digits (inside TP<...>) -# - orig_tx: 1+ alnum -# - track: M + 1+ alnum (e.g. M019048402) — the M is part of the -# tracking value, not a separator. +# - orig_tx: 1+ uppercase alnum +# - track: M + 1+ uppercase alnum (e.g. M019048402) — the M is +# part of the tracking value, not a separator. # - ts: 17 digits # - seq: literal "1of1" -# - ft: 1+ alnum (e.g. 999, TA1, 271, 277, 277CA, 820, 834, 835, ENCR) +# - ft: 1+ uppercase alnum (e.g. 999, TA1, 271, 277, 277CA, +# 820, 834, 835, ENCR) +# +# Case insensitivity is scoped to the ``TP`` prefix via ``(?i:TP)`` +# — the rest of the pattern is case-sensitive so we still reject a +# stray ``837p`` or ``m019048402`` (they'd be invalid HCPF). INBOUND_RE = re.compile( - r"^TP(?P\d+)-(?P[A-Z0-9]+)_(?PM[A-Z0-9]+)" + r"^(?i:TP)(?P\d+)-(?P[A-Z0-9]+)_(?PM[A-Z0-9]+)" r"-(?P\d{17})-1of1_(?P[A-Z0-9]+)\.(?Px12)$" ) @@ -116,7 +126,9 @@ def parse_inbound_filename(name: str) -> InboundFilename: """Parse an inbound HCPF filename. Args: - name: Filename like "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12" + name: Filename like "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12" + (case-insensitive on the ``TP`` prefix; both ``TP`` and + ``tp`` are accepted.) Returns: InboundFilename with tpid, orig_tx, tracking, ts, file_type, ext. diff --git a/backend/src/cyclone/scheduler.py b/backend/src/cyclone/scheduler.py index 3aac5cc..3569cb8 100644 --- a/backend/src/cyclone/scheduler.py +++ b/backend/src/cyclone/scheduler.py @@ -494,6 +494,44 @@ class Scheduler: finally: self._tick_in_progress = False + async def process_inbound_files( + self, files: list[InboundFile], + ) -> TickResult: + """Run the per-file processing pipeline over a pre-fetched list. + + Unlike :meth:`tick`, this does **not** call SFTP — the caller + is expected to have already downloaded (or arranged the local + copy of) each ``InboundFile.local_path``. Used by the + ``/api/admin/scheduler/pull-inbound`` admin endpoint and the + ``cyclone pull-inbound`` CLI command to process a date-filtered + subset of the inbound MFT path without paying the cost of a + full poll. + + Honors ``_stop_event`` between files. Concurrent calls are + coalesced the same way :meth:`tick` does, so a slow handler + can't be stampeded by a parallel admin invocation. + """ + while self._tick_in_progress: + await asyncio.sleep(0.05) + self._tick_in_progress = True + try: + started = datetime.now(timezone.utc) + result = TickResult(started_at=started, 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) + self._last_tick = result + self._last_poll_at = result.finished_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: diff --git a/backend/tests/test_filenames.py b/backend/tests/test_filenames.py index b802539..a566273 100644 --- a/backend/tests/test_filenames.py +++ b/backend/tests/test_filenames.py @@ -107,6 +107,48 @@ def test_parse_inbound_277(): assert parsed.file_type == "277" +def test_parse_inbound_lowercase_tp_prefix_999(): + # Gainwell's production filer uses lowercase `tp` for inbound 999/TA1. + # The inbound regex must accept both casings on the TP prefix. + name = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12" + parsed = parse_inbound_filename(name) + assert parsed.tpid == "11525703" + assert parsed.orig_tx == "837P" + assert parsed.tracking == "M019048402" + assert parsed.file_type == "999" + assert parsed.ext == "x12" + + +def test_parse_inbound_lowercase_tp_prefix_ta1(): + name = "tp11525703-837P_M019044969-20260520180505477-1of1_TA1.x12" + parsed = parse_inbound_filename(name) + assert parsed.file_type == "TA1" + + +def test_is_inbound_filename_accepts_both_cases(): + # is_inbound_filename() is the fast path used by the scheduler to + # filter the listing. It must accept both casings. + upper = "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12" + lower = "tp11525703-837P_M019048402-20260520231513488-1of1_999.x12" + assert is_inbound_filename(upper) + assert is_inbound_filename(lower) + + +def test_parse_inbound_rejects_mixed_case_tracking(): + # Tracking value must stay uppercase alnum; the case-insensitive + # flag is intentionally scoped to the TP prefix by the file_type + # and timestamp constraints, so a mixed-case tracking should still + # be rejected (it'd be invalid HCPF). + # We exercise the obvious "totally lowercase" rejection to confirm + # the rest of the pattern is still strict. + with pytest.raises(ValueError, match="Not a valid HCPF inbound"): + # Lowercase orig_tx; the orig_tx class is [A-Z0-9]+ so it + # must be uppercase. + parse_inbound_filename( + "tp11525703-837p_M019048402-20260520231513488-1of1_999.x12" + ) + + def test_parse_inbound_rejects_missing_tp_prefix(): with pytest.raises(ValueError, match="Not a valid HCPF inbound"): parse_inbound_filename("11525703-837P_M019048402-20260520231513488-1of1_999.x12") @@ -157,8 +199,10 @@ def test_is_outbound_filename(): def test_is_inbound_filename(): assert is_inbound_filename("TP11525703-837P_M019048402-20260520231513488-1of1_999.x12") - # Lowercase tp prefix is the outbound shape, not inbound - assert not is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12") + # Lowercase tp prefix is now accepted too — Gainwell's filer has + # used both casings on inbound 999/TA1 files. + assert is_inbound_filename("tp11525703-837P_M019048402-20260520231513488-1of1_999.x12") + # Outbound shape is still rejected (no tracking/ts/file_type). assert not is_inbound_filename("11525703-837P-20260620132243505-1of1.x12") diff --git a/backend/tests/test_scheduler.py b/backend/tests/test_scheduler.py index db13929..72e0a3f 100644 --- a/backend/tests/test_scheduler.py +++ b/backend/tests/test_scheduler.py @@ -284,4 +284,62 @@ class TestRoutedFileTypes: 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 + assert set(HANDLERS.keys()) == ROUTED_FILE_TYPES + + +class TestProcessInboundFiles: + """Scheduler.process_inbound_files() — date-filtered pull path. + + The /api/admin/scheduler/pull-inbound endpoint and the + ``cyclone pull-inbound`` CLI both call this. It must: + * process each file in the provided list (no SFTP listdir) + * dedupe via ``processed_inbound_files`` (idempotent on rerun) + * not touch SFTP at all — files are expected to be on local + disk at ``f.local_path`` + """ + + @pytest.mark.asyncio + async def test_processes_provided_files_without_listdir( + self, sftp_block, _drop_file, tmp_path, + ): + from cyclone.clearhouse import InboundFile + from cyclone.scheduler import STATUS_OK + + # Drop a known 999 file on disk (the stub scheduler doesn't + # need SFTP — it reads from staging dir). + _drop_file( + "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12", + b"not a real 999 -- handler will parse_error", + ) + sched = _make_scheduler(sftp_block, tmp_path) + + # Build the InboundFile records manually — caller is + # responsible for staging (mirrors the targeted-pull flow). + path = tmp_path / "staging" / "FromHPE" / \ + "TP11525703-837P_M019048402-20260520231513488-1of1_999.x12" + files = [InboundFile( + name=path.name, + size=path.stat().st_size, + modified_at=datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc), + local_path=path, + )] + + result = await sched.process_inbound_files(files) + # The file's bytes are intentionally invalid — we only care + # that process_inbound_files invokes the handler and + # records the outcome (here: error from parse). + assert result.files_seen == 1 + assert result.files_seen == result.files_processed + result.files_errored + # Idempotent: a second call is a no-op (already-processed dedup). + result2 = await sched.process_inbound_files(files) + assert result2.files_seen == 1 + # Either skipped (because the prior call recorded it as error) + # or error — both prove the dedup branch fired. We just check + # it's not re-processed successfully. + with db.SessionLocal()() as s: + rows = s.query(ProcessedInboundFile).filter_by( + sftp_block_name="test-block", + name=path.name, + ).all() + assert len(rows) == 1 + assert rows[0].status in (STATUS_OK, STATUS_ERROR, STATUS_SKIPPED) \ No newline at end of file diff --git a/backend/tests/test_sftp_paramiko.py b/backend/tests/test_sftp_paramiko.py index affea05..6ec55db 100644 --- a/backend/tests/test_sftp_paramiko.py +++ b/backend/tests/test_sftp_paramiko.py @@ -259,6 +259,66 @@ class TestRealModeListInbound: assert len(files) == 1 assert files[0].name == "real.x12" + def test_list_skips_warn_txt_files(self, monkeypatch, tmp_path: Path): + # Gainwell's MFT drops advisory *_warn.txt files in the same + # inbound dir. They're text-format side-channel notes (not X12 + # envelopes) and must be skipped at list time. + real_attr = MagicMock() + real_attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12" + real_attr.st_mode = 0o100644 + real_attr.st_size = 1024 + real_attr.st_mtime = 1718899200 + + warn_attr = MagicMock() + warn_attr.filename = "TP123-837P-202606181208100000-1of1_warn.txt" + warn_attr.st_mode = 0o100644 + warn_attr.st_size = 200 + warn_attr.st_mtime = 1718899200 + + mock_ssh, mock_sftp = _make_mock_paramiko( + monkeypatch, sftp_attrs=[real_attr, warn_attr], + ) + + def _open(path, mode="rb"): + m = MagicMock() + m.__enter__.return_value = io.BytesIO(b"x12 content") + return m + + mock_sftp.open.side_effect = _open + + block = _block(staging_dir=str(tmp_path / "staging")) + client = SftpClient(block) + files = client.list_inbound() + + names = [f.name for f in files] + assert "TP123-837P_M456-20260520231513488-1of1_999.x12" in names + assert not any(n.endswith("_warn.txt") for n in names) + assert len(files) == 1 + + def test_list_inbound_names_does_not_download(self, monkeypatch, tmp_path: Path): + # list_inbound_names() must do a metadata-only SFTP listing — + # no sftp.open() / no file written to the cache. + attr = MagicMock() + attr.filename = "TP123-837P_M456-20260520231513488-1of1_999.x12" + attr.st_mode = 0o100644 + attr.st_size = 1024 + attr.st_mtime = 1718899200 + + mock_ssh, mock_sftp = _make_mock_paramiko( + monkeypatch, sftp_attrs=[attr], + ) + + block = _block(staging_dir=str(tmp_path / "staging")) + client = SftpClient(block) + files = client.list_inbound_names() + + # sftp.open() must NOT have been called (no download). + mock_sftp.open.assert_not_called() + assert len(files) == 1 + # The InboundFile.local_path is set to the planned cache path + # but the file itself doesn't exist yet. + assert not files[0].local_path.exists() + class TestRealModeReadFile: def test_read_returns_bytes(self, monkeypatch):