Compare commits
14 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| aff3a13016 | |||
| 7cb0278be0 | |||
| 4c85166734 | |||
| f1dc06ec3b | |||
| dc990cfde3 | |||
| ec065a9082 | |||
| 5d4e06b863 | |||
| 298200fe71 | |||
| 7ad190c267 | |||
| 3a58f561d4 | |||
| 0067ecc5da | |||
| 698f8660bf | |||
| 178898a8e1 | |||
| 710104f491 |
@@ -18,3 +18,23 @@ VITE_API_BASE_URL=
|
||||
# Optional. Set to 1 to disable auth entirely (DEV ONLY). When set,
|
||||
# the backend auto-grants admin access without checking credentials.
|
||||
# CYCLONE_AUTH_DISABLED=0
|
||||
|
||||
# ─── Gainwell / HCPF SFTP credentials (operator convention) ─────────────
|
||||
# These mirror what /home/tyler/EDI/scripts/upload_claims.sh exports and
|
||||
# what the operator's SFTP client uses. Cyclone's secrets module looks up
|
||||
# CYCLONE_SFTP_PASSWORD (or _FILE), not GAINWELL_SFTP_PASS — to bridge:
|
||||
#
|
||||
# export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
|
||||
#
|
||||
# or for the daemon, set CYCLONE_SFTP_PASSWORD_FILE to a 0600 file.
|
||||
GAINWELL_SFTP_USER=colorado-fts\coxix_prod_11525703
|
||||
GAINWELL_SFTP_HOST=mft.gainwelltechnologies.com
|
||||
GAINWELL_SFTP_PASS=
|
||||
GAINWELL_REMOTE_DIR=/CO XIX/PROD/coxix_prod_11525703/ToHPE
|
||||
|
||||
# CYCLONE_SFTP_PASSWORD= # mirror of GAINWELL_SFTP_PASS
|
||||
# CYCLONE_SFTP_PASSWORD_FILE= # path to a 0600 file containing it
|
||||
|
||||
# CYCLONE_SCHEDULER_AUTOSTART= # 1/true/yes to start the MFT poll loop on API launch
|
||||
# CYCLONE_SCHEDULER_POLL_SECONDS=60 # poll interval when autostart is on
|
||||
# CYCLONE_TAIL_HEARTBEAT_S=15 # NDJSON stream heartbeat interval
|
||||
+10
@@ -20,6 +20,16 @@ build/
|
||||
*.swp
|
||||
*.swo
|
||||
|
||||
# macOS extraction residue (AppleDouble / __MACOSX from unzip on macOS).
|
||||
# These are paired with every regular file in an extracted zip — never data.
|
||||
__MACOSX/
|
||||
._*
|
||||
|
||||
# Operator drop zone (untracked working dir; contents are HCPF-delivered
|
||||
# inbound files, never source). Use `mkdir -p ingest && touch ingest/.gitkeep`
|
||||
# if you want the directory itself in the repo.
|
||||
ingest/
|
||||
|
||||
# Local config
|
||||
.env
|
||||
.env.local
|
||||
|
||||
@@ -125,7 +125,7 @@ Frontend triplet for any live page: `use<X>(params)` (initial fetch) + `useTailS
|
||||
|
||||
## Backend at a glance
|
||||
|
||||
`backend/src/cyclone/` is a single namespace. The two largest files are `api.py` (~3,548 LOC, the only large file) and `store.py` (~2,423 LOC, the `CycloneStore` facade — SP21 is in flight to split it into a `cyclone/store/` subpackage; the public API stays unchanged). Subpackages: `api_routers/` (acks, admin, health, ta1_acks), `clearhouse/` (Clearhouse + SftpClient), `edi/` (filenames), `parsers/` (X12 transaction parsers + models + validators + serializers), `workflow/` (placeholder for future sub-project 6).
|
||||
`backend/src/cyclone/` is a single namespace. The two largest files are `api.py` (~3,548 LOC, the only large file) and `store.py` (~2,423 LOC, the `CycloneStore` facade — SP21 is in flight to split it into a `cyclone/store/` subpackage; the public API stays unchanged). Subpackages: `api_routers/` (acks, admin, health, ta1_acks), `clearhouse/` (Clearhouse + SftpClient), `edi/` (filenames), `parsers/` (X12 transaction parsers + models + validators + serializers), `submission/` (SP37 — canonical `submit_file` helper shared by `cyclone submit-batch` CLI + `POST /api/submit-batch` HTTP endpoint; owns parse → DB-write → SFTP-upload → audit per file), `workflow/` (placeholder for future sub-project 6).
|
||||
|
||||
The store is the only read/write surface for the database; every mutating endpoint goes through it. All persistence flows through SQLAlchemy sessions via `db.SessionLocal()()`. SQLAlchemy ORM models live in `db.py`; 12 SQL migrations under `migrations/` (0001_initial through 0012_backups) are walked in order by `db_migrate.py`.
|
||||
|
||||
@@ -137,6 +137,34 @@ Config: `config/payers.yaml` is the on-disk source for providers / payers / clea
|
||||
|
||||
Secrets live in the macOS Keychain (via `keyring` + `cyclone.secrets`): SQLCipher key (service `cyclone`, account `cyclone.db.key`), SFTP password, backup passphrase. No secrets on disk in plaintext.
|
||||
|
||||
## Production SFTP posture (this box: manual mode)
|
||||
|
||||
This host runs in **manual SFTP mode** against Gainwell's MOVEit Transfer MFT at `mft.gainwelltechnologies.com`. The seeded `dzinesco` clearhouse keeps `sftp_block.stub: true`; the operator moves files to/from Gainwell with their SFTP client and drops inbound files into `ingest/` for cyclone to ingest. Do NOT flip `stub` to `false` from this host — see "Auth caveat" below.
|
||||
|
||||
### Env var convention
|
||||
|
||||
The operator's shell exports the Gainwell creds as `GAINWELL_SFTP_USER`, `GAINWELL_SFTP_PASS`, `GAINWELL_SFTP_HOST`, `GAINWELL_REMOTE_DIR`. Cyclone's `secrets.get_secret()` looks up `CYCLONE_SFTP_PASSWORD` (or `CYCLONE_SFTP_PASSWORD_FILE`) per `secrets._ENV_NAME_FOR`. The two are NOT the same name — bridge them per-call or in your shell rc:
|
||||
|
||||
```bash
|
||||
export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
|
||||
```
|
||||
|
||||
### Inbound drop zone
|
||||
|
||||
`/home/tyler/dev/cyclone/ingest/` is the operator-maintained staging dir for inbound MFT files (999 acks, TA1, 835 remittances, 277CA claim acks). Files here are NOT auto-watched; processing happens via the procedure in `docs/RUNBOOK.md` § "Manual SFTP mode". At time of writing this dir holds ~1500 unprocessed 999 acks + 1 TA1 + 1 835 from early July 2026 — the local ingest flow clears them.
|
||||
|
||||
### Auth caveat (do not retry)
|
||||
|
||||
Paramiko reaches `MOVEit Transfer SFTP` cleanly (SSH kex completes, MOVEit offers password auth) but `AuthenticationException: Authentication failed` is returned despite the operator confirming the credentials are known-good from another host. Most likely cause: MOVEit Transfer's IP-based access control — this host's public IP `103.14.26.95` is not whitelisted. Repeated auth attempts risk account lockout. Do NOT keep guessing. Resolve by either: (a) whitelisting this IP with Gainwell's MFT admin, or (b) staying in manual mode and moving files via the operator's SFTP client.
|
||||
|
||||
### Inbound ingestion paths
|
||||
|
||||
The canonical way to write 999/TA1/277CA/835 rows into the DB is `Scheduler.process_inbound_files`, exposed as the CLI `cyclone pull-inbound --date YYYYMMDD` and the HTTP `POST /api/admin/scheduler/pull-inbound`. **There is no `parse-999` / `parse-ta1` / `parse-277ca` CLI command** (CLAUDE.md's "CLI" section above is aspirational on those three). The `parse-837` and `parse-835` CLIs exist but only emit JSON files to `--output-dir`; they do NOT write to the DB. For DB writes, use `pull-inbound` (which dedupes via `processed_inbound_files`).
|
||||
|
||||
### Daemon hot-reload
|
||||
|
||||
`python -m cyclone serve` runs as root (started by `tini`) and keeps the SFTP block in memory. Flipping `stub` directly in the DB (`store.update_clearhouse(...)`) does NOT auto-reload the daemon's view — either restart the daemon or use `PATCH /api/clearhouse` (which calls `scheduler.reconfigure_scheduler` to hot-reload).
|
||||
|
||||
## Frontend at a glance
|
||||
|
||||
`src/` is React 18 + TypeScript + Vite. Routes register in `src/App.tsx` (11 pages, all under a `<Layout>` route wrapper). Pages are pure renderers — every page pairs with a `use<X>` data hook in `src/hooks/` and renders a `<PageHeader>` + a table/list/KPI grid. Drawers (`ClaimDrawer/`, `RemitDrawer/`, plus the new `ProviderDrawer/` and `AckDrawer/`) are mounted by the page and their open/close state is mirrored to the URL via `useDrawerUrlState` so deep-links round-trip. Drill-stack navigation is provided by `<DrillStackProvider>` in `src/components/drill/`.
|
||||
|
||||
@@ -28,6 +28,7 @@ from cyclone.api_routers import (
|
||||
providers,
|
||||
reconciliation,
|
||||
remittances,
|
||||
submission,
|
||||
ta1_acks,
|
||||
)
|
||||
|
||||
@@ -49,6 +50,7 @@ routers: list[APIRouter] = [
|
||||
providers.router, # gated
|
||||
reconciliation.router, # gated
|
||||
remittances.router, # gated
|
||||
submission.router, # gated
|
||||
ta1_acks.router, # gated
|
||||
]
|
||||
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
"""SP37 Task 6: HTTP endpoint for the canonical submit-batch flow.
|
||||
|
||||
Thin wrapper around ``cyclone.submission.submit_file`` — same logic as
|
||||
the ``cyclone submit-batch`` CLI (SP37 Task 5), just framed as JSON in
|
||||
/ JSON out and gated by ``matrix_gate``. The walker pattern, ``._*``
|
||||
AppleDouble skip, ``limit`` semantics, and per-file outcomes all match
|
||||
the CLI byte-for-byte so a batch run via the CLI and the same batch
|
||||
run via this endpoint produce identical DB + SFTP state.
|
||||
|
||||
The endpoint deliberately does NOT inject an ``sftp_client_factory``:
|
||||
``submit_file`` defaults to its paramiko-based factory so SKIPPED is
|
||||
reachable in production (the ``SftpClient`` wrapper has no ``stat()``).
|
||||
Tests monkey-patch ``cyclone.api_routers.submission.submit_file``
|
||||
itself; that avoids the paramiko factory entirely without touching
|
||||
the helper's contract.
|
||||
|
||||
Status code contract (per Task 6 spec §4):
|
||||
- 200: completed run. Per-file failures live in the JSON body.
|
||||
- 401: not authenticated (matrix_gate).
|
||||
- 404: no clearhouse seeded (config-level "missing" → 4xx, not 5xx).
|
||||
- 409: clearhouse SFTP block is in stub mode (refuses to upload).
|
||||
- 422: ``ingest_dir`` missing on disk OR Pydantic body validation
|
||||
failed (missing fields, wrong types).
|
||||
- 5xx: truly unexpected exceptions propagate (do not swallow).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from cyclone.auth.deps import matrix_gate
|
||||
from cyclone.store import store as cycl_store
|
||||
from cyclone.submission.core import submit_file
|
||||
from cyclone.submission.result import SubmitOutcome, SubmitResult
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# No `prefix=` here — every other gated router in this package declares
|
||||
# the full path in the decorator (clearhouse.py uses "/api/clearhouse",
|
||||
# parse.py uses "/api/parse-837", etc.). The decorator sets the full URL.
|
||||
router = APIRouter(
|
||||
tags=["submission"],
|
||||
dependencies=[Depends(matrix_gate)],
|
||||
)
|
||||
|
||||
|
||||
class SubmitBatchRequest(BaseModel):
|
||||
"""Body schema for ``POST /api/submit-batch``.
|
||||
|
||||
``ingest_dir`` has no default so Pydantic raises 422 when it's
|
||||
missing — better UX than letting the walker crash on a missing
|
||||
path. ``validate_files`` and ``actor`` default so a minimal client
|
||||
can skip them. ``limit`` truncates the file list after the walker
|
||||
collects it (mirrors the CLI's post-collection ``if i > limit:
|
||||
break`` semantics, but applied as a slice since the HTTP body
|
||||
model is type-checked up-front).
|
||||
|
||||
The ``validate_files`` field is aliased to ``validate`` in the JSON
|
||||
body to mirror the CLI's ``--validate`` flag and avoid the
|
||||
hardcoded Pydantic warning about ``validate`` shadowing
|
||||
``BaseModel.validate``. ``populate_by_name=True`` lets tests
|
||||
construct the model with either key.
|
||||
"""
|
||||
model_config = ConfigDict(populate_by_name=True, protected_namespaces=())
|
||||
|
||||
ingest_dir: str
|
||||
validate_files: bool = Field(default=True, alias="validate")
|
||||
actor: str = "api-submit-batch"
|
||||
limit: int | None = None
|
||||
|
||||
|
||||
@router.post("/api/submit-batch")
|
||||
def submit_batch(body: SubmitBatchRequest):
|
||||
"""Submit every ``batch-*-claims/*.x12`` under ``ingest_dir``.
|
||||
|
||||
Walks ``ingest_dir`` for any directory matching ``batch-*-claims``,
|
||||
collects each one's ``*.x12`` files (sorted, with ``._*``
|
||||
AppleDouble files skipped), truncates to ``limit`` if set, then
|
||||
calls :func:`cyclone.submission.submit_file` per file with the
|
||||
seeded clearhouse's ``sftp_block`` and ``actor`` from the body.
|
||||
|
||||
Returns counts (``submitted`` / ``skipped`` / ``failed``) plus a
|
||||
per-file ``results`` array. Per-file failures NEVER change the
|
||||
HTTP status code — the response is 200 whenever the run itself
|
||||
completed.
|
||||
"""
|
||||
|
||||
# 1. Config-level guards. Order matters: a missing clearhouse is a
|
||||
# 404 (config-level "missing"), but if it IS present and in stub
|
||||
# mode the operator's request is a 409 (configured-but-wrong).
|
||||
clearhouse = cycl_store.get_clearhouse()
|
||||
if clearhouse is None:
|
||||
raise HTTPException(
|
||||
status_code=404, detail="no clearhouse seeded",
|
||||
)
|
||||
sftp_block = clearhouse.sftp_block
|
||||
if sftp_block.stub:
|
||||
raise HTTPException(
|
||||
status_code=409, detail="clearhouse SFTP block is in stub mode",
|
||||
)
|
||||
|
||||
# 2. Resolve + validate the ingest dir. Use ``resolve()`` so a
|
||||
# symlink-relative path still produces a stable error message.
|
||||
root = Path(body.ingest_dir).resolve()
|
||||
if not root.exists():
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail=f"ingest_dir does not exist: {root}",
|
||||
)
|
||||
|
||||
# 3. Walker — must match the CLI EXACTLY. Same sort, same ``._*``
|
||||
# AppleDouble skip. Any drift here is a quiet split between the
|
||||
# two surfaces and silently produces different batch outcomes.
|
||||
files: list[Path] = []
|
||||
for batch_dir in sorted(root.glob("batch-*-claims")):
|
||||
files.extend(sorted(
|
||||
p for p in batch_dir.glob("*.x12")
|
||||
if not p.name.startswith("._")
|
||||
))
|
||||
|
||||
# 4. ``limit`` truncates after collection. The CLI uses an inline
|
||||
# ``if i > limit: break``; we slice instead because the HTTP
|
||||
# body model validates ``limit`` up-front (Pydantic-level int
|
||||
# check) and slicing keeps the walker branchless.
|
||||
if body.limit is not None:
|
||||
files = files[: body.limit]
|
||||
|
||||
if not files:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail=f"no batch-*-claims/*.x12 files found under {root}",
|
||||
)
|
||||
|
||||
# 5. Per-file submit. Wrap the helper call in try/except so an
|
||||
# unexpected exception in submit_file surfaces as a per-file
|
||||
# failure (outcome="unexpected") instead of crashing the whole
|
||||
# run. The helper's own SubmitOutcome enum covers every typed
|
||||
# failure path; an uncaught exception here is a true
|
||||
# surprise (bug or service outage mid-loop).
|
||||
results: list[dict] = []
|
||||
submitted = skipped = failed = 0
|
||||
for src in files:
|
||||
try:
|
||||
r = submit_file(
|
||||
src,
|
||||
sftp_block=sftp_block,
|
||||
actor=body.actor,
|
||||
validate=body.validate_files,
|
||||
# No ``sftp_client_factory`` — submit_file's default
|
||||
# paramiko factory opens the real MFT. Tests
|
||||
# monkey-patch submit_file itself instead of wiring a
|
||||
# factory here.
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception(
|
||||
"submit-batch unexpected error on %s", src.name,
|
||||
)
|
||||
r = SubmitResult(
|
||||
file=src.name,
|
||||
outcome=SubmitOutcome.SFTP_FAILED,
|
||||
error=f"{exc.__class__.__name__}: {exc}",
|
||||
)
|
||||
|
||||
if r.outcome == SubmitOutcome.SUBMITTED:
|
||||
submitted += 1
|
||||
elif r.outcome == SubmitOutcome.SKIPPED:
|
||||
skipped += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
results.append({
|
||||
"file": r.file,
|
||||
"outcome": r.outcome.value,
|
||||
"batch_id": r.batch_id,
|
||||
"error": r.error,
|
||||
})
|
||||
|
||||
return {
|
||||
"submitted": submitted,
|
||||
"skipped": skipped,
|
||||
"failed": failed,
|
||||
"results": results,
|
||||
}
|
||||
@@ -86,6 +86,7 @@ PERMISSIONS: dict[tuple[str, str], set[Role]] = {
|
||||
("POST", "/api/acks"): WRITE_ROLES,
|
||||
("POST", "/api/batches"): WRITE_ROLES, # /export-837 regenerates X12 from DB rows
|
||||
("POST", "/api/eligibility"): WRITE_ROLES,
|
||||
("POST", "/api/submit-batch"): WRITE_ROLES, # SP37: canonical outbound path (mirrors CLI)
|
||||
|
||||
# CSV export — read-only.
|
||||
("GET", "/api/export.csv"): ALL_ROLES,
|
||||
|
||||
@@ -768,6 +768,99 @@ def resubmit_rejected_claims(
|
||||
)
|
||||
|
||||
|
||||
@main.command("submit-batch")
|
||||
@click.option("--ingest-dir", default="ingest",
|
||||
show_default=True,
|
||||
help="Root dir holding batch-*-claims/ subfolders of .x12 files.")
|
||||
@click.option("--actor", default="cli-submit-batch",
|
||||
show_default=True,
|
||||
help="Audit-log actor tag for the clearhouse.submitted events.")
|
||||
@click.option("--validate/--no-validate", default=True,
|
||||
help="Parse each file via parse_837 before upload.")
|
||||
@click.option("--limit", type=int, default=None,
|
||||
help="Stop after N files (smoke-tests).")
|
||||
def submit_batch(
|
||||
ingest_dir: str,
|
||||
actor: str,
|
||||
validate: bool,
|
||||
limit: int | None,
|
||||
) -> None:
|
||||
"""Parse → DB-write → SFTP-upload each batch-*-claims/*.x12 (SP37).
|
||||
|
||||
Canonical outbound path. Mirrors ``resubmit-rejected-claims`` but
|
||||
routes through ``cyclone.submission.submit_file``, which writes to
|
||||
the DB before uploading. Use this as the preferred outbound path;
|
||||
``resubmit-rejected-claims`` remains for one-off cases where you
|
||||
don't want a DB row (dzinesco-generated fixes not ready for
|
||||
canonical tracking).
|
||||
|
||||
Exit codes: 0 = run completed (even with per-file failures), 1 =
|
||||
unexpected exception, 2 = config-level failure (no clearhouse,
|
||||
SFTP block in stub mode, ingest-dir missing).
|
||||
"""
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.submission.core import submit_file
|
||||
from cyclone.submission.result import SubmitOutcome
|
||||
from cyclone.store import store as cycl_store
|
||||
|
||||
db_mod.init_db()
|
||||
cycl_store.ensure_clearhouse_seeded()
|
||||
clearhouse = cycl_store.get_clearhouse()
|
||||
if clearhouse is None:
|
||||
click.echo("No clearhouse seeded; cannot resolve SFTP block.", err=True)
|
||||
sys.exit(2)
|
||||
sftp_block = clearhouse.sftp_block
|
||||
if sftp_block.stub:
|
||||
click.echo("Clearhouse SFTP block is in stub mode; refusing to upload.", err=True)
|
||||
sys.exit(2)
|
||||
|
||||
root = Path(ingest_dir).resolve()
|
||||
if not root.exists():
|
||||
click.echo(f"--ingest-dir does not exist: {root}", err=True)
|
||||
sys.exit(2)
|
||||
|
||||
files: list[Path] = []
|
||||
for batch_dir in sorted(root.glob("batch-*-claims")):
|
||||
files.extend(sorted(p for p in batch_dir.glob("*.x12")
|
||||
if not p.name.startswith("._")))
|
||||
if not files:
|
||||
click.echo(f"No .x12 files under {root}", err=True)
|
||||
sys.exit(2)
|
||||
|
||||
click.echo(f"found {len(files)} files under {root}")
|
||||
|
||||
submitted = 0
|
||||
skipped = 0
|
||||
failed = 0
|
||||
for i, src in enumerate(files, 1):
|
||||
if limit is not None and i > limit:
|
||||
break
|
||||
try:
|
||||
result = submit_file(
|
||||
src, sftp_block=sftp_block, actor=actor, validate=validate,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
click.echo(f"UNEXPECTED {src.name}: {exc.__class__.__name__}: {exc}", err=True)
|
||||
failed += 1
|
||||
continue
|
||||
if result.outcome == SubmitOutcome.SUBMITTED:
|
||||
submitted += 1
|
||||
click.echo(f"submitted {src.name} batch={result.batch_id}")
|
||||
elif result.outcome == SubmitOutcome.SKIPPED:
|
||||
skipped += 1
|
||||
click.echo(f"skipped {src.name} (already uploaded)")
|
||||
else:
|
||||
failed += 1
|
||||
click.echo(
|
||||
f"{result.outcome.value:<10} {src.name} {result.error or ''}",
|
||||
err=True,
|
||||
)
|
||||
|
||||
click.echo(f"\nsummary: submitted={submitted} skipped={skipped} failed={failed}")
|
||||
# Per-file failures don't bump exit code; details in stdout.
|
||||
# (Per cyclone-cli convention.)
|
||||
|
||||
|
||||
@main.group()
|
||||
def backup() -> None:
|
||||
"""Encrypted DB backup management (SP17)."""
|
||||
|
||||
@@ -221,6 +221,13 @@ class Batch(Base):
|
||||
totals_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
|
||||
validation_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
|
||||
raw_result_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
|
||||
# SP37 Task 2: source 837's ST02 (transaction set control number).
|
||||
# Populated from ``Envelope.transaction_set_control_number`` by
|
||||
# ``store.write.add_record`` for 837P batches; NULL for 835 batches
|
||||
# (the column is an 837P-specific join key for 999 AK2 resolution).
|
||||
# Migration 0020 adds the column additively; no backfill required for
|
||||
# pre-existing rows that lack the value.
|
||||
transaction_set_control_number: Mapped[Optional[str]] = mapped_column(String(32), nullable=True)
|
||||
|
||||
claims: Mapped[list["Claim"]] = relationship(
|
||||
back_populates="batch", cascade="all, delete-orphan"
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
-- version: 20
|
||||
-- SP37: Batch.transaction_set_control_number = parsed 837's ST02.
|
||||
--
|
||||
-- Today's 999 ack join (claim_acks.batch_envelope_index, Pass 1) matches
|
||||
-- on ``Batch.envelope.control_number == 999's set_control_number``. That
|
||||
-- never resolves in production because 999's set_control_number (AK201)
|
||||
-- echoes the source 837's ST02 (transaction set control number), not the
|
||||
-- ISA13 (interchange control number) that Envelope.control_number stores.
|
||||
-- Result: every AK2 set-response against a dzinesco-generated 837 turns
|
||||
-- into an orphan.
|
||||
--
|
||||
-- SP37 fixes this by adding a column populated from the parsed 837's
|
||||
-- ST02 on every ``add_record`` write, then updating Pass 1 to match on
|
||||
-- it (Task 2). This migration is the additive part: nullable, no
|
||||
-- default, backfills from ``raw_result_json.envelope.transaction_set_control_number``
|
||||
-- for any pre-existing batch rows that already carry the value.
|
||||
--
|
||||
-- No new index (column is a primary join key, not a range query; the
|
||||
-- existing batches table is small enough for a full scan during the
|
||||
-- 999 join — see SP37 §"Migration 0013").
|
||||
|
||||
ALTER TABLE batches ADD COLUMN transaction_set_control_number TEXT;
|
||||
|
||||
UPDATE batches
|
||||
SET transaction_set_control_number = json_extract(raw_result_json, '$.envelope.transaction_set_control_number')
|
||||
WHERE raw_result_json IS NOT NULL
|
||||
AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') IS NOT NULL;
|
||||
@@ -120,6 +120,13 @@ class Envelope(_Base):
|
||||
implementation_guide: str | None = None
|
||||
# SP3 P1 T2: BHT06 transaction type code (was: transaction_set_purpose_code, which is BHT02).
|
||||
transaction_type_code: str | None = None
|
||||
# SP37 Task 2: X12 ST02 (transaction set control number). Distinct
|
||||
# from ``control_number`` above, which is the ISA13 interchange
|
||||
# control number. 999 acks echo ST02 back as AK201, so this is the
|
||||
# join key that lets ``add_record``'s batch row round-trip back to
|
||||
# its source 837. Populated only by the 837P parser today; other
|
||||
# parsers share this class but leave the field None.
|
||||
transaction_set_control_number: str | None = None
|
||||
|
||||
|
||||
class BatchSummary(_Base):
|
||||
|
||||
@@ -83,6 +83,12 @@ def _build_envelope(segments: list[list[str]], input_file: str = "") -> tuple[En
|
||||
except (IndexError, ValueError) as exc:
|
||||
log.warning("Could not parse BHT date: %s", exc)
|
||||
elif seg[0] == "ST" and envelope is not None:
|
||||
# SP37 Task 2: capture ST02 (transaction set control number).
|
||||
# 999 acks echo this back as AK201, so this is what makes the
|
||||
# batch row joinable once the 999 ingests. Distinct from ISA13
|
||||
# (which is already on ``control_number``).
|
||||
if len(seg) > 2:
|
||||
envelope = envelope.model_copy(update={"transaction_set_control_number": seg[2].strip()})
|
||||
if len(seg) > 3:
|
||||
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
|
||||
return envelope, summary
|
||||
|
||||
@@ -325,7 +325,13 @@ class CycloneStore:
|
||||
return _remove_claim_ack(link_id, event_bus=event_bus)
|
||||
|
||||
def batch_envelope_index(self):
|
||||
"""Return a {envelope.control_number: batch.id} map for D10 Pass 1."""
|
||||
"""Return a {key: batch.id} map populated from two columns (D10 Pass 1).
|
||||
|
||||
Each 837p batch contributes both ``Batch.raw_result_json.envelope.control_number``
|
||||
(ISA13) and ``Batch.transaction_set_control_number`` (ST02); 835 batches
|
||||
are excluded. Single ``.get(set_control_number)`` lookup resolves
|
||||
either key — SP37 closes the 999 AK201 → source-batch gap.
|
||||
"""
|
||||
return _batch_envelope_index()
|
||||
|
||||
# -- SP17: encrypted DB backups -------------------------------------
|
||||
|
||||
@@ -11,8 +11,10 @@ Five methods on top of the new ``ClaimAck`` ORM table:
|
||||
ack-orphans lane).
|
||||
* ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``.
|
||||
* ``batch_envelope_index`` — D10 in-memory map of
|
||||
``Batch.envelope.control_number → batch.id`` (cheap to rebuild;
|
||||
re-built once per ingest).
|
||||
``{Batch.raw_result_json.envelope.control_number (ISA13) OR
|
||||
Batch.transaction_set_control_number (ST02)} → batch.id``
|
||||
(SP37: populated from two columns; cheap to rebuild, re-built
|
||||
once per ingest).
|
||||
|
||||
Each mutating method opens its own ``db.SessionLocal()()`` session
|
||||
so callers don't have to manage session lifecycles. Publishes are
|
||||
@@ -67,26 +69,43 @@ def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> Non
|
||||
|
||||
|
||||
def batch_envelope_index() -> dict[str, str]:
|
||||
"""Build a ``{envelope.control_number: batch.id}`` map.
|
||||
"""Build a ``{key: batch.id}`` map populated from two columns.
|
||||
|
||||
D10 primary join key (spec §D10). Built once per ingest from
|
||||
every Batch row whose ``raw_result_json`` carries an envelope —
|
||||
cost is O(N batches), currently ~16, so trivial. Kept as a
|
||||
plain dict so callers can ``index.get(scn)`` to resolve.
|
||||
D10 primary join key (spec §D10). Each 837p batch row contributes
|
||||
up to two entries:
|
||||
|
||||
* ``Batch.raw_result_json.envelope.control_number`` (ISA13)
|
||||
* ``Batch.transaction_set_control_number`` (ST02, new in SP37)
|
||||
|
||||
Either key resolves to the same ``batch.id``; callers do a single
|
||||
``idx.get(set_control_number)`` lookup. Pass 2 (PCN) is unchanged.
|
||||
|
||||
Built once per ingest from every Batch row whose kind is "837p";
|
||||
cost is O(N batches), currently small, so trivial. 835 batches
|
||||
are excluded — the column is an 837P-specific join key (see
|
||||
``Batch.transaction_set_control_number`` docstring).
|
||||
"""
|
||||
out: dict[str, str] = {}
|
||||
idx: dict[str, str] = {}
|
||||
with db.SessionLocal()() as s:
|
||||
rows = (
|
||||
s.query(Batch.id, Batch.raw_result_json)
|
||||
s.query(
|
||||
Batch.id,
|
||||
Batch.raw_result_json,
|
||||
Batch.transaction_set_control_number,
|
||||
)
|
||||
.filter(Batch.kind == "837p")
|
||||
.all()
|
||||
)
|
||||
for bid, raw in rows:
|
||||
for bid, raw, stcn in rows:
|
||||
env = (raw or {}).get("envelope") or {}
|
||||
ctrl = env.get("control_number")
|
||||
if isinstance(ctrl, str) and ctrl:
|
||||
out[ctrl] = bid
|
||||
return out
|
||||
isa_cn = env.get("control_number")
|
||||
if isinstance(isa_cn, str) and isa_cn:
|
||||
# First write wins (setdefault) — if ISA13 and ST02
|
||||
# ever collide they map to the same batch anyway.
|
||||
idx.setdefault(isa_cn, bid)
|
||||
if isinstance(stcn, str) and stcn:
|
||||
idx.setdefault(stcn, bid)
|
||||
return idx
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -77,6 +77,17 @@ def add_record(record: BatchRecord, *, event_bus=None) -> None:
|
||||
totals_json=None,
|
||||
validation_json=None,
|
||||
raw_result_json=json.loads(record.result.model_dump_json()),
|
||||
# SP37 Task 2: mirror the parsed 837's ST02 onto the batch
|
||||
# row so 999 AK201 set_control_numbers can resolve back via
|
||||
# Pass 1. The ``getattr`` chain handles the 835 path: the
|
||||
# shared ``Envelope`` class is used by both 837P and 835
|
||||
# parsers, but only ``parse_837`` populates this field — for
|
||||
# 835 records it stays ``None`` and the column is NULL.
|
||||
transaction_set_control_number=getattr(
|
||||
getattr(record.result, "envelope", None),
|
||||
"transaction_set_control_number",
|
||||
None,
|
||||
),
|
||||
)
|
||||
s.add(batch_row)
|
||||
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
"""SP37 Task 4: canonical 837P submission flow.
|
||||
|
||||
The single public helper is ``submit_file`` — it owns parse → DB
|
||||
write → SFTP upload per file. CLI (``cyclone submit-batch``) and HTTP
|
||||
(``POST /api/submit-batch``) are thin wrappers; both call this helper
|
||||
so the ordering, idempotency, and audit shape are identical.
|
||||
|
||||
DB-first, upload-second (per spec decision §2): if the DB write fails,
|
||||
no SFTP call is made. If the SFTP call fails after a successful DB
|
||||
write, the row exists but the file never landed — re-running is safe
|
||||
(idempotent DB write via add_record's s.get check; idempotent SFTP
|
||||
upload via stat-then-put).
|
||||
"""
|
||||
from .core import submit_file
|
||||
from .result import SubmitOutcome, SubmitResult
|
||||
|
||||
__all__ = ["submit_file", "SubmitOutcome", "SubmitResult"]
|
||||
@@ -0,0 +1,225 @@
|
||||
"""SP37 Task 4: parse → DB write → SFTP upload, per file.
|
||||
|
||||
Mirrors ``resubmit_rejected_claims`` but writes to the DB before
|
||||
uploading. Same idempotency check (``stat().st_size == local_size``)
|
||||
skips already-uploaded files without re-emitting audit events.
|
||||
|
||||
DB-first, upload-second (per spec decision §2): if the DB write fails,
|
||||
no SFTP call is made. If the SFTP call fails after a successful DB
|
||||
write, the row exists but the file never landed — re-running is safe
|
||||
(idempotent DB write via add_record's s.get check; idempotent SFTP
|
||||
upload via stat-then-put).
|
||||
|
||||
The default SFTP factory uses paramiko directly (not the
|
||||
``SftpClient`` wrapper) because the wrapper exposes ``write_file``,
|
||||
``list_inbound``, and ``read_file`` but no ``stat()`` — which makes
|
||||
the SKIPPED outcome unreachable in production.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.audit_log import AuditEvent, append_event
|
||||
from cyclone.parsers.parse_837 import parse as parse_837_text
|
||||
from cyclone.parsers.payer import PayerConfig
|
||||
from cyclone.providers import SftpBlock
|
||||
from cyclone.store import store as cycl_store
|
||||
from cyclone.store.records import BatchRecord837
|
||||
|
||||
from .result import SubmitOutcome, SubmitResult
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# The companion-guide payer id we enforce for CO Medicaid submits.
|
||||
# Same value the legacy ``resubmit-rejected-claims`` CLI gates on.
|
||||
EXPECTED_PAYER_ID = "CO_TXIX"
|
||||
|
||||
|
||||
def _default_sftp_factory(sftp_block: SftpBlock) -> Any:
|
||||
"""Open a paramiko SFTP session to the real MFT. Mirrors
|
||||
resubmit_rejected_claims._open_session (cli.py:620-640).
|
||||
Caller is responsible for closing the session.
|
||||
|
||||
The returned ``sftp`` is a ``paramiko.SFTPClient`` — it exposes
|
||||
``stat(remote_path)`` (used for the idempotency check) and
|
||||
``put(local_path, remote_path)`` (used for the upload). The
|
||||
underlying SSH handle is stashed on ``sftp._cyclone_ssh`` so the
|
||||
caller can close it cleanly via ``getattr(sftp, "_cyclone_ssh",
|
||||
None).close()`` after the upload finishes.
|
||||
|
||||
Raises:
|
||||
RuntimeError: if the SFTP block is in stub mode (the CLI/HTTP
|
||||
layer should already guard against this, but we re-check
|
||||
here because paramiko will try to connect even when
|
||||
``stub=True``).
|
||||
"""
|
||||
if sftp_block.stub:
|
||||
# Same posture as resubmit_rejected_claims in cli.py:592-595:
|
||||
# the operator refuses to upload in stub mode, and the helper
|
||||
# surfaces that as SFTP_FAILED with an explicit error.
|
||||
raise RuntimeError("SFTP block is in stub mode")
|
||||
|
||||
from paramiko import AutoAddPolicy, SSHClient
|
||||
|
||||
from cyclone.secrets import get_secret
|
||||
|
||||
pw = get_secret(sftp_block.auth.get("password_keychain_account", ""))
|
||||
ssh = SSHClient()
|
||||
ssh.set_missing_host_key_policy(AutoAddPolicy())
|
||||
ssh.connect(
|
||||
sftp_block.host, port=sftp_block.port,
|
||||
username=sftp_block.username, password=pw,
|
||||
timeout=15, banner_timeout=15, auth_timeout=15,
|
||||
)
|
||||
sftp = ssh.open_sftp()
|
||||
# Attach ssh to sftp so the caller can close it cleanly.
|
||||
sftp._cyclone_ssh = ssh
|
||||
return sftp
|
||||
|
||||
|
||||
def submit_file(
|
||||
path: Path,
|
||||
*,
|
||||
sftp_block: SftpBlock,
|
||||
actor: str,
|
||||
validate: bool = True,
|
||||
sftp_client_factory: Callable[[SftpBlock], Any] | None = None,
|
||||
) -> SubmitResult:
|
||||
"""Submit one 837P file: parse → DB write → SFTP upload.
|
||||
|
||||
Args:
|
||||
path: local 837P file.
|
||||
sftp_block: the clearhouse SftpBlock (for paths + auth).
|
||||
actor: audit-log actor tag (e.g. "api-submit-batch").
|
||||
validate: parse the file before upload (default True). Catches
|
||||
bad byte-fixes early.
|
||||
sftp_client_factory: optional callable that returns an
|
||||
SFTP-client-compatible object (must expose ``stat()`` and
|
||||
``write_file()``). Defaults to :func:`_default_sftp_factory`
|
||||
which opens a real paramiko session. Tests inject a fake.
|
||||
|
||||
Returns:
|
||||
SubmitResult with file, outcome, batch_id (when written),
|
||||
error (when failed).
|
||||
"""
|
||||
file_label = path.name
|
||||
try:
|
||||
content = path.read_bytes()
|
||||
except OSError as exc:
|
||||
return SubmitResult(file_label, SubmitOutcome.PARSE_FAILED, error=str(exc))
|
||||
|
||||
# 1. Validate via parse (optional but recommended).
|
||||
parsed: Any = None
|
||||
if validate:
|
||||
try:
|
||||
parsed = parse_837_text(content.decode(), PayerConfig.co_medicaid())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("submit_file %s: parse failed: %s", file_label, exc)
|
||||
return SubmitResult(file_label, SubmitOutcome.PARSE_FAILED, error=str(exc))
|
||||
mismatch = next(
|
||||
(c for c in parsed.claims if c.payer.id != EXPECTED_PAYER_ID),
|
||||
None,
|
||||
)
|
||||
if mismatch is not None:
|
||||
return SubmitResult(
|
||||
file_label,
|
||||
SubmitOutcome.PAYER_MISMATCH,
|
||||
error=f"payer.id={mismatch.payer.id!r} (expected {EXPECTED_PAYER_ID!r})",
|
||||
)
|
||||
|
||||
# 2. DB write — DB-first, upload-second invariant.
|
||||
# ``parsed`` is required to construct a BatchRecord837 (it embeds the
|
||||
# full ParseResult). validate=False therefore isn't a real path —
|
||||
# the CLI/HTTP always pass validate=True — so reject it loudly
|
||||
# rather than silently building an empty row.
|
||||
if parsed is None:
|
||||
return SubmitResult(
|
||||
file_label,
|
||||
SubmitOutcome.PARSE_FAILED,
|
||||
error="validate=False is not supported; must parse to construct a BatchRecord",
|
||||
)
|
||||
|
||||
# Use the same uuid4().hex id the existing /api/parse-837 path uses.
|
||||
batch_id = uuid.uuid4().hex
|
||||
record = BatchRecord837(
|
||||
id=batch_id,
|
||||
input_filename=file_label,
|
||||
parsed_at=datetime.now(timezone.utc),
|
||||
result=parsed,
|
||||
)
|
||||
|
||||
try:
|
||||
# ``cycl_store.add`` is the public facade method (per the
|
||||
# CycloneStore class in store/__init__.py:158). It delegates
|
||||
# to ``store.write.add_record``, which is the underlying
|
||||
# SQLAlchemy write path.
|
||||
cycl_store.add(record)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("submit_file %s: DB write failed: %s", file_label, exc)
|
||||
return SubmitResult(file_label, SubmitOutcome.DB_FAILED, error=str(exc))
|
||||
|
||||
# 3. SFTP upload. ``_default_sftp_factory`` opens a paramiko
|
||||
# session directly because the SftpClient wrapper has no ``stat()``
|
||||
# method — that omission made the SKIPPED outcome unreachable in
|
||||
# production. Mirror cli.py:620-650.
|
||||
remote_path = f"{sftp_block.paths['outbound']}/{file_label}"
|
||||
factory = sftp_client_factory or _default_sftp_factory
|
||||
sftp = factory(sftp_block)
|
||||
local_size = len(content)
|
||||
try:
|
||||
try:
|
||||
stat = sftp.stat(remote_path)
|
||||
if stat.st_size == local_size:
|
||||
# Already on remote at the right size — re-run is safe;
|
||||
# do NOT emit a duplicate audit event.
|
||||
return SubmitResult(file_label, SubmitOutcome.SKIPPED, batch_id=batch_id)
|
||||
except (IOError, OSError):
|
||||
pass # not on remote yet
|
||||
sftp.write_file(remote_path, content)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning("submit_file %s: SFTP failed: %s", file_label, exc)
|
||||
return SubmitResult(
|
||||
file_label, SubmitOutcome.SFTP_FAILED,
|
||||
batch_id=batch_id, error=str(exc),
|
||||
)
|
||||
finally:
|
||||
# Close the paramiko SSH handle the helper stashed on the
|
||||
# SFTP client (paramiko does not auto-close on GC and a leak
|
||||
# here costs a slot in MOVEit's per-IP session table).
|
||||
ssh_handle = getattr(sftp, "_cyclone_ssh", None)
|
||||
if ssh_handle is not None:
|
||||
try:
|
||||
ssh_handle.close()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
|
||||
# 4. Audit event — same shape the legacy resubmit_rejected_claims CLI
|
||||
# uses (event_type="clearhouse.submitted", entity_type="claim_file",
|
||||
# entity_id=filename, payload has remote_path + source + size).
|
||||
# Best-effort: an audit failure must not roll back the upload.
|
||||
try:
|
||||
with db_mod.SessionLocal()() as session:
|
||||
append_event(session, AuditEvent(
|
||||
event_type="clearhouse.submitted",
|
||||
entity_type="claim_file",
|
||||
entity_id=file_label,
|
||||
payload={
|
||||
"remote_path": remote_path,
|
||||
"source": "submit-batch",
|
||||
"size": local_size,
|
||||
"batch_id": batch_id,
|
||||
},
|
||||
actor=actor,
|
||||
))
|
||||
session.commit()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning(
|
||||
"submit_file %s: audit event failed: %s", file_label, exc,
|
||||
)
|
||||
|
||||
return SubmitResult(file_label, SubmitOutcome.SUBMITTED, batch_id=batch_id)
|
||||
@@ -0,0 +1,40 @@
|
||||
"""SP37 Task 4: typed result for submit_file.
|
||||
|
||||
The CLI and HTTP layer both consume SubmitResult; keeping it in one
|
||||
place means both surfaces agree on the response shape.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class SubmitOutcome(str, Enum):
|
||||
"""Outcome of a single submit_file invocation.
|
||||
|
||||
Audit-event invariant: ``SUBMITTED`` emits a ``clearhouse.submitted``
|
||||
audit event; ``SKIPPED`` does NOT (re-running on an already-uploaded
|
||||
file must not flood the audit log with duplicate events). All
|
||||
failure outcomes (``PARSE_FAILED``, ``PAYER_MISMATCH``, ``DB_FAILED``,
|
||||
``SFTP_FAILED``) emit no event either — failures should be observable
|
||||
via the helper's return value, not the audit log.
|
||||
"""
|
||||
SUBMITTED = "submitted"
|
||||
SKIPPED = "skipped"
|
||||
PARSE_FAILED = "parse_failed"
|
||||
PAYER_MISMATCH = "payer_mismatch"
|
||||
DB_FAILED = "db_failed"
|
||||
SFTP_FAILED = "sftp_failed"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SubmitResult:
|
||||
"""Return value of ``submit_file``: filename, outcome, and optional
|
||||
``batch_id`` (populated when the DB write succeeded) and ``error``
|
||||
(populated on any failure outcome; ``None`` on ``SUBMITTED`` /
|
||||
``SKIPPED``).
|
||||
"""
|
||||
file: str
|
||||
outcome: SubmitOutcome
|
||||
batch_id: str | None = None
|
||||
error: str | None = None
|
||||
@@ -0,0 +1,30 @@
|
||||
ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*^*00501*991102977*1*P*:~
|
||||
GS*HC*11525703*COMEDASSISTPROG*20260611*081417*991102977*X*005010X222A1~
|
||||
ST*837*991102977*005010X222A1~
|
||||
BHT*0019*00*ref-001*20260611*081417*CH~
|
||||
NM1*41*2*Test Submitter*****46*11525703~
|
||||
PER*IC*Test Contact*EM*test@example.com~
|
||||
NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~
|
||||
HL*1**20*1~
|
||||
PRV*BI*PXC*251E00000X~
|
||||
NM1*85*2*Test Provider Inc*****XX*1993999998~
|
||||
N3*123 Test St~
|
||||
N4*Denver*CO*80202~
|
||||
REF*EI*123456789~
|
||||
HL*2*1*22*0~
|
||||
SBR*P*18*******MC~
|
||||
NM1*IL*1*Doe*John****MI*ABC123~
|
||||
N3*456 Member St~
|
||||
N4*Denver*CO*80203~
|
||||
DMG*D8*19800101*M~
|
||||
NM1*PR*2*CO_TXIX*****PI*CO_TXIX~
|
||||
CLM*CLM001*100.00***12:B:1*Y*A*Y*Y~
|
||||
REF*G1*PA123~
|
||||
HI*ABK:Z00~
|
||||
LX*1~
|
||||
SV1*HC:99213*100.00*UN*1***1~
|
||||
DTP*472*D8*20260611~
|
||||
REF*6R*REF001~
|
||||
SE*26*991102977~
|
||||
GE*1*991102977~
|
||||
IEA*1*991102977~
|
||||
@@ -51,7 +51,7 @@ 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 19 after
|
||||
user_version already at the latest version — currently 20 after
|
||||
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
|
||||
providers/payers/clearhouse, SP10's 0008 payer_rejected,
|
||||
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
|
||||
@@ -61,15 +61,16 @@ def test_migration_latest_idempotent_on_fresh_db():
|
||||
claims.matched_remittance_id index, SP27-Task 17's 0017
|
||||
claim.patient_control_number backfill UPDATE, SP28's 0018
|
||||
claim_acks join table, SP32's 0019
|
||||
rendering_provider_npi + service_provider_npi)."""
|
||||
rendering_provider_npi + service_provider_npi, and SP37's 0020
|
||||
transaction_set_control_number)."""
|
||||
with db.engine().begin() as c:
|
||||
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||
assert v1 == 19
|
||||
assert v1 == 20
|
||||
# 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 == 19
|
||||
assert v2 == 20
|
||||
|
||||
|
||||
def test_add_ack_persists_row():
|
||||
|
||||
@@ -0,0 +1,355 @@
|
||||
"""SP37 Task 6: POST /api/submit-batch endpoint.
|
||||
|
||||
Thin HTTP wrapper around ``cyclone.submission.submit_file``. Mirrors the
|
||||
``cyclone submit-batch`` CLI walker exactly: each ``batch-*-claims/*.x12``
|
||||
under ``ingest_dir`` is submitted in order, ``._*`` AppleDouble files are
|
||||
skipped, ``limit`` truncates after collection.
|
||||
|
||||
Tests monkey-patch ``cyclone.api_routers.submission.submit_file`` so we
|
||||
never touch the real paramiko factory or a live MFT.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from cyclone.submission.result import SubmitOutcome, SubmitResult
|
||||
|
||||
_FIXTURE = Path(__file__).parent / "fixtures" / "submit-batch" / "single-claim.x12"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path) -> TestClient:
|
||||
"""Standard TestClient; conftest.py autouse handles DB + auth gate.
|
||||
|
||||
Does NOT touch env vars (per the plan-vs-reality correction in
|
||||
Task 6's instructions). Seeds the clearhouse then flips
|
||||
``sftp_block.stub`` to ``False`` so the file-walking tests reach
|
||||
the walker (otherwise the default-seeded stub=True trips the 409
|
||||
guard before any per-file work runs). The ``stub_mode`` test
|
||||
re-seeds stub=True explicitly so its 409 still fires.
|
||||
"""
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.store import store as cycl_store
|
||||
|
||||
db_mod._reset_for_tests()
|
||||
db_mod.init_db()
|
||||
cycl_store.ensure_clearhouse_seeded()
|
||||
ch = cycl_store.get_clearhouse()
|
||||
# Flip stub=False so the walker actually runs (SFTP calls are
|
||||
# intercepted via the submit_file monkey-patch in each test).
|
||||
cycl_store.update_clearhouse(
|
||||
ch.model_copy(update={"sftp_block": ch.sftp_block.model_copy(update={"stub": False})}),
|
||||
)
|
||||
|
||||
from cyclone.api import app
|
||||
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _stage_batch(tmp_path: Path, n: int) -> Path:
|
||||
"""Copy N copies of the single-claim fixture into ``batch-test-claims/``."""
|
||||
batch_dir = tmp_path / "batch-test-claims"
|
||||
batch_dir.mkdir()
|
||||
payload = _FIXTURE.read_bytes()
|
||||
for i in range(n):
|
||||
(batch_dir / f"claim-{i}.x12").write_bytes(payload)
|
||||
return batch_dir
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Happy path + walker semantics
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_submit_batch_happy_path(client: TestClient, tmp_path, monkeypatch):
|
||||
"""3 valid 837 files → 200 + body shape with submitted=3 (or skipped=3 on re-run).
|
||||
|
||||
Monkey-patches ``cyclone.api_routers.submission.submit_file`` to
|
||||
return a synthetic SUBMITTED SubmitResult per file. Skipped stays
|
||||
possible because SubmitResult's batch_id is what matters and the
|
||||
fake returns it consistently — but a fresh DB should always land
|
||||
in the SUBMITTED branch.
|
||||
"""
|
||||
_stage_batch(tmp_path, 3)
|
||||
|
||||
def _fake_submit(path, **_kw):
|
||||
return SubmitResult(
|
||||
file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"cyclone.api_routers.submission.submit_file", _fake_submit,
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
# Tolerate either counter — counter totals are deterministic with a
|
||||
# fresh DB, but the assertion below focuses on the structural shape
|
||||
# the operator cares about: every file is accounted for.
|
||||
assert body["submitted"] + body["skipped"] + body["failed"] == 3
|
||||
assert isinstance(body["results"], list)
|
||||
assert len(body["results"]) == 3
|
||||
for row in body["results"]:
|
||||
assert "file" in row
|
||||
assert "outcome" in row
|
||||
assert "batch_id" in row
|
||||
assert "error" in row
|
||||
|
||||
|
||||
def test_submit_batch_skips_apple_double_files(client: TestClient, tmp_path, monkeypatch):
|
||||
"""``._foo.x12`` AppleDouble files must be skipped (mirrors the CLI walker)."""
|
||||
batch_dir = tmp_path / "batch-test-claims"
|
||||
batch_dir.mkdir()
|
||||
(batch_dir / "real.x12").write_bytes(_FIXTURE.read_bytes())
|
||||
(batch_dir / "._real.x12").write_bytes(b"\x00\x01\x02") # macOS AppleDouble noise
|
||||
|
||||
def _fake_submit(path, **_kw):
|
||||
return SubmitResult(
|
||||
file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"cyclone.api_routers.submission.submit_file", _fake_submit,
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
# Only real.x12 was submitted (one call); AppleDouble was skipped.
|
||||
assert len(body["results"]) == 1
|
||||
assert body["results"][0]["file"] == "real.x12"
|
||||
|
||||
|
||||
def test_submit_batch_limit_truncates(client: TestClient, tmp_path, monkeypatch):
|
||||
"""``limit=1`` against 3 files → body has exactly 1 result."""
|
||||
_stage_batch(tmp_path, 3)
|
||||
|
||||
submitted_files: list[str] = []
|
||||
|
||||
def _fake_submit(path, **_kw):
|
||||
submitted_files.append(path.name)
|
||||
return SubmitResult(
|
||||
file=path.name, outcome=SubmitOutcome.SUBMITTED, batch_id="b1",
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"cyclone.api_routers.submission.submit_file", _fake_submit,
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False, "limit": 1},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert len(body["results"]) == 1
|
||||
assert body["submitted"] + body["skipped"] + body["failed"] == 1
|
||||
assert len(submitted_files) == 1
|
||||
|
||||
|
||||
def test_submit_batch_empty_ingest_dir_returns_422(client: TestClient, tmp_path):
|
||||
"""No batch-*-claims dir under ingest_dir → 422 (matches the CLI's exit 2 path)."""
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 422, resp.text
|
||||
|
||||
|
||||
def test_submit_batch_per_file_failure_keeps_status_200(
|
||||
client: TestClient, tmp_path, monkeypatch,
|
||||
):
|
||||
"""A SUBMITTED + SFTP_FAILED mix → 200 with failed=1, submitted=1.
|
||||
|
||||
Per-file failures live in the JSON body, NOT in the status code.
|
||||
The endpoint always returns 200 on a completed run.
|
||||
"""
|
||||
_stage_batch(tmp_path, 2)
|
||||
|
||||
outcomes = iter([
|
||||
SubmitResult(file="claim-0.x12", outcome=SubmitOutcome.SUBMITTED, batch_id="b1"),
|
||||
SubmitResult(
|
||||
file="claim-1.x12", outcome=SubmitOutcome.SFTP_FAILED,
|
||||
batch_id="b2", error="sftp down",
|
||||
),
|
||||
])
|
||||
|
||||
def _fake_submit(path, **_kw):
|
||||
return next(outcomes)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"cyclone.api_routers.submission.submit_file", _fake_submit,
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["submitted"] == 1
|
||||
assert body["failed"] == 1
|
||||
assert body["skipped"] == 0
|
||||
assert len(body["results"]) == 2
|
||||
# The failure row carries the error string so the operator can see
|
||||
# what went wrong without a separate API call.
|
||||
assert body["results"][1]["outcome"] == "sftp_failed"
|
||||
assert body["results"][1]["error"] == "sftp down"
|
||||
|
||||
|
||||
def test_submit_batch_unexpected_exception_recorded_in_results(
|
||||
client: TestClient, tmp_path, monkeypatch,
|
||||
):
|
||||
"""If submit_file raises, the file lands in results[] as a failed row.
|
||||
|
||||
The router catches the exception, builds a SubmitResult with
|
||||
``outcome=SubmitOutcome.SFTP_FAILED`` (per Task 6's instruction
|
||||
that "any enum value works as a marker; choose whichever is least
|
||||
misleading"), and counts it as ``failed``. The full error string
|
||||
(including exception class name) is in ``error`` so the operator
|
||||
can tell it was a real exception, not a typed failure path.
|
||||
"""
|
||||
_stage_batch(tmp_path, 1)
|
||||
|
||||
def _explode(path, **_kw):
|
||||
raise RuntimeError("kaboom")
|
||||
|
||||
monkeypatch.setattr(
|
||||
"cyclone.api_routers.submission.submit_file", _explode,
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["failed"] == 1
|
||||
# The marker is the SFTP_FAILED enum value (per Task 6 design
|
||||
# choice). The full exception class+message is in ``error`` so the
|
||||
# operator can distinguish a real exception from a typed failure.
|
||||
assert body["results"][0]["outcome"] == "sftp_failed"
|
||||
assert "kaboom" in body["results"][0]["error"]
|
||||
assert "RuntimeError" in body["results"][0]["error"]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Request validation
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_submit_batch_missing_ingest_dir_returns_422(client: TestClient):
|
||||
"""Body without ``ingest_dir`` → 422 from Pydantic (NOT your manual check)."""
|
||||
resp = client.post("/api/submit-batch", json={"validate": False})
|
||||
assert resp.status_code == 422, resp.text
|
||||
|
||||
|
||||
def test_submit_batch_ingest_dir_missing_on_disk_returns_422(
|
||||
client: TestClient, tmp_path,
|
||||
):
|
||||
"""Body has a path that doesn't exist on disk → 422."""
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={
|
||||
"ingest_dir": str(tmp_path / "does-not-exist"),
|
||||
"validate": False,
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 422, resp.text
|
||||
assert "does not exist" in resp.text or "does-not-exist" in resp.text
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Clearhouse posture (404 / 409)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_submit_batch_no_clearhouse_returns_404(client: TestClient, tmp_path, monkeypatch):
|
||||
"""Delete the clearhouse row → endpoint must refuse with 404, not 500.
|
||||
|
||||
404 per the Task 6 status code contract: a missing clearhouse is a
|
||||
config-level "missing" (4xx), not an unexpected exception.
|
||||
"""
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.db import ClearhouseORM
|
||||
|
||||
# Wipe the seeded clearhouse row.
|
||||
with db_mod.SessionLocal()() as s:
|
||||
row = s.get(ClearhouseORM, 1)
|
||||
if row is not None:
|
||||
s.delete(row)
|
||||
s.commit()
|
||||
|
||||
_stage_batch(tmp_path, 1)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 404, resp.text
|
||||
assert "clearhouse" in resp.text.lower()
|
||||
|
||||
|
||||
def test_submit_batch_stub_mode_returns_409(client: TestClient, tmp_path, monkeypatch):
|
||||
"""When the seeded clearhouse's ``sftp_block.stub`` is True → 409.
|
||||
|
||||
The default client fixture flips stub=False so file-walking tests
|
||||
reach the walker. This test flips it back to True so the
|
||||
stub-mode 409 guard fires before any per-file work runs.
|
||||
"""
|
||||
from cyclone.store import store as cycl_store
|
||||
|
||||
ch = cycl_store.get_clearhouse()
|
||||
assert ch is not None
|
||||
# Re-flip stub=True (the default client fixture set it to False).
|
||||
cycl_store.update_clearhouse(
|
||||
ch.model_copy(update={
|
||||
"sftp_block": ch.sftp_block.model_copy(update={"stub": True}),
|
||||
}),
|
||||
)
|
||||
|
||||
_stage_batch(tmp_path, 1)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 409, resp.text
|
||||
assert "stub" in resp.text.lower()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Auth gate
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_submit_batch_auth_gate(client: TestClient, tmp_path, monkeypatch):
|
||||
"""No cookie + AUTH_DISABLED=False → 401 from matrix_gate.
|
||||
|
||||
The conftest autouse fixture sets ``AUTH_DISABLED = True`` at
|
||||
fixture setup and resets to False at teardown. Inside the test we
|
||||
flip the module attribute so the gate fires — but to avoid leaking
|
||||
the flip into a sibling test we use monkeypatch.setattr (which
|
||||
restores at test teardown, before conftest's finally block runs).
|
||||
"""
|
||||
import cyclone.auth.deps as auth_deps
|
||||
|
||||
monkeypatch.setattr(auth_deps, "AUTH_DISABLED", False)
|
||||
|
||||
_stage_batch(tmp_path, 1)
|
||||
|
||||
resp = client.post(
|
||||
"/api/submit-batch",
|
||||
json={"ingest_dir": str(tmp_path), "validate": False},
|
||||
)
|
||||
assert resp.status_code == 401, resp.text
|
||||
@@ -0,0 +1,126 @@
|
||||
"""SP37 Task 3: batch_envelope_index populates from BOTH columns.
|
||||
|
||||
The dict returned by ``batch_envelope_index`` should resolve lookups
|
||||
by EITHER ``Batch.raw_result_json.envelope.control_number`` (ISA13,
|
||||
preserved) OR ``Batch.transaction_set_control_number`` (ST02, new in
|
||||
SP37 Task 2). One dict, both keys — a single ``.get(set_control_number)``
|
||||
call hits whichever matches, so the D10 Pass 1 join in
|
||||
:func:`cyclone.claim_acks.lookup_claims_for_ack_set_response` resolves
|
||||
999 AK201 (which echoes the source 837's ST02) back to the right batch
|
||||
even when ST02 != ISA13.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.db import Batch
|
||||
from cyclone.store.claim_acks import batch_envelope_index
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _db(tmp_path, monkeypatch):
|
||||
"""Per-test DB; conftest's autouse already wires one too, but we
|
||||
pin the URL again so this module is self-contained if anyone ever
|
||||
lifts it out of the conftest tree."""
|
||||
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
|
||||
db_mod._reset_for_tests()
|
||||
db_mod.init_db()
|
||||
yield
|
||||
|
||||
|
||||
def _make_batch(
|
||||
s,
|
||||
*,
|
||||
id: str,
|
||||
icn: str,
|
||||
stcn: str | None = None,
|
||||
raw_json: dict | None = None,
|
||||
) -> Batch:
|
||||
"""Insert one Batch row whose envelope + ST02 mirror what
|
||||
``store.write.add_record`` writes for an 837P batch."""
|
||||
row = Batch(
|
||||
id=id,
|
||||
kind="837p",
|
||||
input_filename=id + ".x12",
|
||||
parsed_at=datetime.now(timezone.utc),
|
||||
raw_result_json=raw_json or {"envelope": {"control_number": icn}},
|
||||
transaction_set_control_number=stcn,
|
||||
)
|
||||
s.add(row)
|
||||
s.commit()
|
||||
s.refresh(row)
|
||||
return row
|
||||
|
||||
|
||||
def test_index_resolves_by_envelope_control_number():
|
||||
"""Pre-SP37 path: ISA13 (envelope.control_number) still resolves."""
|
||||
with db_mod.SessionLocal()() as s:
|
||||
_make_batch(s, id="b1", icn="ISA000001")
|
||||
idx = batch_envelope_index()
|
||||
assert idx.get("ISA000001") == "b1"
|
||||
|
||||
|
||||
def test_index_resolves_by_transaction_set_control_number():
|
||||
"""SP37 path: ST02 (transaction_set_control_number) resolves too.
|
||||
|
||||
The whole point of Task 3 — Gainwell batches have ST02 != ISA13,
|
||||
so 999 AK201 echoes ST02 and must hit this branch.
|
||||
"""
|
||||
with db_mod.SessionLocal()() as s:
|
||||
_make_batch(
|
||||
s, id="b2", icn="ISA000002", stcn="ST000002",
|
||||
raw_json={"envelope": {"control_number": "ISA000002"}},
|
||||
)
|
||||
idx = batch_envelope_index()
|
||||
assert idx.get("ST000002") == "b2"
|
||||
# Backward compat: ISA still resolves.
|
||||
assert idx.get("ISA000002") == "b2"
|
||||
|
||||
|
||||
def test_index_handles_row_with_no_transaction_set_control_number():
|
||||
"""Pre-migration rows (stcn NULL) should still resolve by ISA.
|
||||
|
||||
Spec says: 'Pre-migration rows (stcn NULL) should still resolve by ISA.'
|
||||
Production row ``b1`` (the only one with claims) has ST02 NULL because
|
||||
the migration's backfill only ran over rows whose raw_result_json
|
||||
envelope had the ST02 key — and that row was written before the
|
||||
parser populated it.
|
||||
"""
|
||||
with db_mod.SessionLocal()() as s:
|
||||
_make_batch(s, id="b3", icn="ISA000003") # no stcn
|
||||
idx = batch_envelope_index()
|
||||
assert idx.get("ISA000003") == "b3"
|
||||
assert idx.get("ST000003") is None
|
||||
|
||||
|
||||
def test_index_ignores_non_837_batches():
|
||||
"""835 batches don't contribute to the 999 join index.
|
||||
|
||||
The D10 two-pass join is specifically for 837 → 999 linkage; 835
|
||||
remittances live on the response side of the loop and have no ST02
|
||||
join key. Filtering by ``kind == "837p"`` keeps the index scoped
|
||||
correctly and avoids an 835 row shadowing an 837 ST02 by accident.
|
||||
"""
|
||||
with db_mod.SessionLocal()() as s:
|
||||
# Add an 837 row first so we can prove the 835 doesn't leak in.
|
||||
_make_batch(s, id="b-837", icn="ISA837", stcn="ST837")
|
||||
# And an 835 row whose ISA13 / ST02 would otherwise pollute.
|
||||
s.add(Batch(
|
||||
id="b-835",
|
||||
kind="835",
|
||||
input_filename="b-835.x12",
|
||||
parsed_at=datetime.now(timezone.utc),
|
||||
raw_result_json={"envelope": {"control_number": "ISA999"}},
|
||||
transaction_set_control_number="ST999",
|
||||
))
|
||||
s.commit()
|
||||
idx = batch_envelope_index()
|
||||
# 837 row resolves both ways.
|
||||
assert idx.get("ISA837") == "b-837"
|
||||
assert idx.get("ST837") == "b-837"
|
||||
# 835 row contributes nothing.
|
||||
assert "ISA999" not in idx
|
||||
assert "ST999" not in idx
|
||||
@@ -0,0 +1,103 @@
|
||||
"""SP37 Task 2: add_record populates Batch.transaction_set_control_number.
|
||||
|
||||
The column should mirror the source 837 envelope's
|
||||
``transaction_set_control_number`` (ST02) so the join-key update in
|
||||
Task 3 can resolve 999 ``set_control_number`` (AK201) values back to
|
||||
the right batch. The 835 path leaves the column NULL because the
|
||||
field is an 837P-specific join key (835 remittances don't need to
|
||||
resolve back to a source 837 — they're the response side of the loop).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.db import Batch
|
||||
from cyclone.parsers.parse_837 import parse as parse_837_text
|
||||
from cyclone.parsers.payer import PayerConfig
|
||||
from cyclone.store import store as cycl_store
|
||||
from cyclone.store.records import BatchRecord837, BatchRecord835
|
||||
|
||||
FIXTURE_837 = Path(__file__).parent / "fixtures" / "minimal_837p.txt"
|
||||
FIXTURE_835 = Path(__file__).parent / "fixtures" / "minimal_835.txt"
|
||||
|
||||
|
||||
def test_add_record_populates_transaction_set_control_number_for_837():
|
||||
"""parse_837 → add_record writes the parsed ST02 onto the Batch row.
|
||||
|
||||
The fixture's ST02 (``991102977``) must round-trip into the new
|
||||
``batches.transaction_set_control_number`` column. This is the
|
||||
join key Task 3's 999-ingest Pass 1 update relies on.
|
||||
"""
|
||||
text = FIXTURE_837.read_text()
|
||||
parsed = parse_837_text(text, PayerConfig.co_medicaid())
|
||||
|
||||
record = BatchRecord837(
|
||||
id="b-txn-cn-1",
|
||||
input_filename="minimal_837p.txt",
|
||||
parsed_at=datetime.now(timezone.utc),
|
||||
result=parsed,
|
||||
)
|
||||
cycl_store.add(record)
|
||||
|
||||
with db_mod.SessionLocal()() as s:
|
||||
row = s.get(Batch, "b-txn-cn-1")
|
||||
assert row is not None
|
||||
assert row.transaction_set_control_number == parsed.envelope.transaction_set_control_number
|
||||
# Sanity: the fixture's ST02 must be the parsed-and-stored value.
|
||||
assert row.transaction_set_control_number == "991102977"
|
||||
|
||||
|
||||
def test_envelope_model_exposes_transaction_set_control_number():
|
||||
"""The Envelope Pydantic model carries the new field with a None default.
|
||||
|
||||
Guards against a regression where someone removes the field from
|
||||
the model — the rest of the chain (parser + ORM + write path)
|
||||
silently degrades to None if the model loses the attribute.
|
||||
"""
|
||||
from cyclone.parsers.models import Envelope
|
||||
|
||||
env = Envelope(
|
||||
sender_id="S",
|
||||
receiver_id="R",
|
||||
control_number="000000001",
|
||||
transaction_date=date(2026, 1, 1),
|
||||
)
|
||||
assert env.transaction_set_control_number is None
|
||||
|
||||
env2 = env.model_copy(update={"transaction_set_control_number": "0001"})
|
||||
assert env2.transaction_set_control_number == "0001"
|
||||
|
||||
|
||||
def test_add_record_leaves_835_column_null():
|
||||
"""835 batches don't carry an ST02 join key; column stays NULL.
|
||||
|
||||
The shared ``Envelope`` class is used by both 837P and 835 parsers,
|
||||
but only ``parse_837`` populates ``transaction_set_control_number``.
|
||||
For 835 records the field is None and ``add_record`` writes NULL
|
||||
to the column — verified end-to-end via a real 835 ingest.
|
||||
"""
|
||||
text = FIXTURE_835.read_text()
|
||||
from cyclone.parsers.parse_835 import parse as parse_835_text
|
||||
parsed835 = parse_835_text(text, payer_config=None) # type: ignore[arg-type]
|
||||
|
||||
# Sanity: the 835 envelope also has the field (shared model class),
|
||||
# but the parser doesn't populate it — so it must be None.
|
||||
assert parsed835.envelope.transaction_set_control_number is None
|
||||
|
||||
record = BatchRecord835(
|
||||
id="b-txn-cn-835-1",
|
||||
input_filename="minimal_835.txt",
|
||||
parsed_at=datetime.now(timezone.utc),
|
||||
result=parsed835,
|
||||
)
|
||||
cycl_store.add(record)
|
||||
|
||||
with db_mod.SessionLocal()() as s:
|
||||
row = s.get(Batch, "b-txn-cn-835-1")
|
||||
assert row is not None
|
||||
assert row.transaction_set_control_number is None
|
||||
# And the kind/kind round-trip is correct (sanity check that
|
||||
# we did exercise the 835 path, not the 837 path).
|
||||
assert row.kind == "835"
|
||||
@@ -126,14 +126,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
|
||||
SP28 bumped it to 18 with the claim_acks join table.
|
||||
SP32 bumped it to 19 with rendering_provider_npi +
|
||||
service_provider_npi on claims and remittances.
|
||||
SP37 bumped it to 20 with batch transaction_set_control_number.
|
||||
"""
|
||||
engine = _fresh_engine(tmp_path)
|
||||
db_migrate.run(engine)
|
||||
v_after_first = _user_version(engine)
|
||||
assert v_after_first == 19, f"expected head=19, got {v_after_first}"
|
||||
assert v_after_first == 20, f"expected head=20, got {v_after_first}"
|
||||
|
||||
db_migrate.run(engine)
|
||||
assert _user_version(engine) == 19, "second run should not bump version"
|
||||
assert _user_version(engine) == 20, "second run should not bump version"
|
||||
|
||||
|
||||
def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@@ -159,7 +160,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
|
||||
engine = _fresh_engine(tmp_path)
|
||||
|
||||
db_migrate.run(engine)
|
||||
assert _user_version(engine) == 19, f"expected head=19, got {_user_version(engine)}"
|
||||
assert _user_version(engine) == 20, f"expected head=20, got {_user_version(engine)}"
|
||||
|
||||
# Two claims in one batch with the same patient_control_number
|
||||
# must be insertable. If 0015's table recreation re-introduced a
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Tests for migration 0020_add_batch_txn_set_control_number.sql.
|
||||
|
||||
SP37 adds a nullable ``batches.transaction_set_control_number`` column
|
||||
populated from the parsed 837's ST02 (transaction set control number)
|
||||
on every write. The 999 ack join (Pass 1) needs to resolve by ST02,
|
||||
not ISA13, so this column is the join key. This migration is purely
|
||||
additive: nullable, no default, backfills from
|
||||
``raw_result_json.envelope.transaction_set_control_number`` where the
|
||||
source JSON already carries it.
|
||||
|
||||
For the backfill-shape tests we point ``db_migrate.MIGRATIONS_DIR``
|
||||
at the real migrations directory, apply all migrations once to bring
|
||||
the fresh DB up to v20, insert representative rows, then replay the
|
||||
exact UPDATE statement the migration uses. Replaying the UPDATE
|
||||
proves the SQL works as intended even though the migration itself
|
||||
already ran over an empty table.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import sqlalchemy as sa
|
||||
|
||||
from cyclone import db_migrate
|
||||
|
||||
|
||||
# The backfill UPDATE the migration executes (extracted so the test can
|
||||
# replay it against rows that didn't exist when init_db ran).
|
||||
BACKFILL_SQL = (
|
||||
"UPDATE batches "
|
||||
"SET transaction_set_control_number = "
|
||||
"json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
|
||||
"WHERE raw_result_json IS NOT NULL "
|
||||
"AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') "
|
||||
"IS NOT NULL"
|
||||
)
|
||||
|
||||
|
||||
def _fresh_engine(path: Path) -> sa.Engine:
|
||||
return sa.create_engine(f"sqlite:///{path}", future=True)
|
||||
|
||||
|
||||
def _table_info(engine: sa.Engine, table: str) -> list[tuple]:
|
||||
"""Return PRAGMA table_info rows for ``table`` as plain tuples."""
|
||||
with engine.connect() as conn:
|
||||
return list(conn.exec_driver_sql(f"PRAGMA table_info({table});").tuples())
|
||||
|
||||
|
||||
def _real_migrations_dir() -> Path:
|
||||
return Path(__file__).parent.parent / "src" / "cyclone" / "migrations"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||
"""Yield an engine at v20 against which every real migration has run.
|
||||
|
||||
Points ``db_migrate.MIGRATIONS_DIR`` at the real migrations
|
||||
directory so the test exercises the actual 0020 file, then runs
|
||||
the migration runner on a per-test fresh DB.
|
||||
"""
|
||||
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", _real_migrations_dir())
|
||||
engine = _fresh_engine(tmp_path / "mig0020.db")
|
||||
db_migrate.run(engine)
|
||||
|
||||
# Confirm head is 20 (every migration applied).
|
||||
with engine.connect() as conn:
|
||||
v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||
assert v == 20, f"expected migration head=20, got {v}"
|
||||
|
||||
yield engine
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def test_migration_0020_creates_column(migrated_engine) -> None:
|
||||
"""Migration 0020 adds ``transaction_set_control_number`` to ``batches``."""
|
||||
cols = _table_info(migrated_engine, "batches")
|
||||
col_names = {row[1] for row in cols}
|
||||
assert "transaction_set_control_number" in col_names, (
|
||||
"batches.transaction_set_control_number missing — migration 0020 did not run. "
|
||||
f"Existing columns: {sorted(col_names)}"
|
||||
)
|
||||
|
||||
|
||||
def test_migration_0020_column_is_nullable(migrated_engine) -> None:
|
||||
"""The new column is nullable (no DEFAULT, no NOT NULL) so existing
|
||||
batches that don't yet carry the ST02 stay valid."""
|
||||
cols = {row[1]: row for row in _table_info(migrated_engine, "batches")}
|
||||
row = cols["transaction_set_control_number"]
|
||||
# PRAGMA table_info tuples: (cid, name, type, notnull, dflt_value, pk)
|
||||
assert row[3] == 0, f"notnull flag must be 0 (nullable), got {row[3]}"
|
||||
assert row[4] is None, f"dflt_value must be NULL, got {row[4]!r}"
|
||||
assert "TEXT" in (row[2] or "").upper(), f"expected TEXT column, got {row[2]!r}"
|
||||
|
||||
|
||||
def test_migration_0020_backfills_when_key_present(
|
||||
migrated_engine: sa.Engine,
|
||||
) -> None:
|
||||
"""Replay the migration's backfill UPDATE against a row whose
|
||||
``raw_result_json`` carries the key — must populate the new column.
|
||||
|
||||
Replay (rather than waiting for ``db_migrate.run()`` to do it) is
|
||||
the only way to test the SQL against a row that didn't exist when
|
||||
init_db() ran; the migration's UPDATE naturally runs only over
|
||||
pre-existing rows.
|
||||
"""
|
||||
st02_value = "ST0001"
|
||||
raw_json = {"envelope": {"transaction_set_control_number": st02_value}}
|
||||
|
||||
with migrated_engine.begin() as conn:
|
||||
conn.exec_driver_sql(
|
||||
"INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
|
||||
"VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)",
|
||||
(json.dumps(raw_json),),
|
||||
)
|
||||
conn.exec_driver_sql(BACKFILL_SQL)
|
||||
|
||||
with migrated_engine.connect() as conn:
|
||||
row = conn.exec_driver_sql(
|
||||
"SELECT transaction_set_control_number FROM batches WHERE id='B-ST02-1'"
|
||||
).first()
|
||||
assert row is not None
|
||||
assert row[0] == st02_value, (
|
||||
f"backfill failed: expected {st02_value!r}, got {row[0]!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_migration_0020_backfill_conditional_on_key_present(
|
||||
migrated_engine: sa.Engine,
|
||||
) -> None:
|
||||
"""Replay the UPDATE against a row whose ``raw_result_json`` does NOT
|
||||
carry the key — must stay NULL. Proves the UPDATE is conditional
|
||||
(the ``json_extract(...) IS NOT NULL`` guard) rather than
|
||||
unconditionally overwriting with NULL."""
|
||||
raw_json = {"envelope": {"control_number": "ISA0001"}} # no txn-set key
|
||||
with migrated_engine.begin() as conn:
|
||||
conn.exec_driver_sql(
|
||||
"INSERT INTO batches (id, kind, input_filename, parsed_at, raw_result_json) "
|
||||
"VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)",
|
||||
(json.dumps(raw_json),),
|
||||
)
|
||||
conn.exec_driver_sql(BACKFILL_SQL)
|
||||
|
||||
with migrated_engine.connect() as conn:
|
||||
row = conn.exec_driver_sql(
|
||||
"SELECT transaction_set_control_number FROM batches WHERE id='B-NOST-1'"
|
||||
).first()
|
||||
assert row is not None
|
||||
assert row[0] is None, (
|
||||
f"backfill must not overwrite when key is absent; got {row[0]!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_migration_0020_backfill_handles_null_raw_result_json(
|
||||
migrated_engine: sa.Engine,
|
||||
) -> None:
|
||||
"""A Batch row with ``raw_result_json IS NULL`` (the prior SP's
|
||||
unparsed state) must not crash the backfill and must stay NULL."""
|
||||
with migrated_engine.begin() as conn:
|
||||
conn.exec_driver_sql(
|
||||
"INSERT INTO batches (id, kind, input_filename, parsed_at) "
|
||||
"VALUES ('B-NULL-1', '837p', 'mig0020-null.txt', '2026-07-07 00:00:00')"
|
||||
)
|
||||
conn.exec_driver_sql(BACKFILL_SQL)
|
||||
|
||||
with migrated_engine.connect() as conn:
|
||||
row = conn.exec_driver_sql(
|
||||
"SELECT transaction_set_control_number FROM batches WHERE id='B-NULL-1'"
|
||||
).first()
|
||||
assert row is not None
|
||||
assert row[0] is None
|
||||
@@ -0,0 +1,208 @@
|
||||
"""SP37 Task 4: submit_file owns parse → DB write → SFTP upload.
|
||||
|
||||
Tests use a fake SFTP client to avoid hitting the real clearhouse.
|
||||
The DB path is real (autouse ``_auto_init_db`` fixture in conftest.py)
|
||||
so the join-key wiring is exercised end-to-end.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from cyclone import db as db_mod
|
||||
from cyclone.submission.core import submit_file
|
||||
from cyclone.submission.result import SubmitOutcome
|
||||
|
||||
|
||||
_FIXTURE = Path(__file__).parent / "fixtures" / "submit-batch" / "single-claim.x12"
|
||||
|
||||
|
||||
class _FakeSftp:
|
||||
"""Implements just enough of SftpClient for submit_file's idempotency check.
|
||||
|
||||
After every successful ``write_file``, the file is added to
|
||||
``existing_files`` keyed by remote_path with the byte size — so a
|
||||
subsequent ``stat`` returns the right size and the idempotency
|
||||
check in ``submit_file`` short-circuits to ``SKIPPED``. Without
|
||||
this bookkeeping, every re-run would look like the file isn't on
|
||||
the remote and would re-upload.
|
||||
"""
|
||||
|
||||
def __init__(self, existing_files: dict[str, int] | None = None):
|
||||
self.existing_files = existing_files or {}
|
||||
self.put_calls: list[tuple[str, bytes]] = []
|
||||
|
||||
def stat(self, path: str):
|
||||
if path not in self.existing_files:
|
||||
raise IOError(f"no such file: {path}")
|
||||
m = MagicMock()
|
||||
m.st_size = self.existing_files[path]
|
||||
return m
|
||||
|
||||
def write_file(self, remote_path: str, content: bytes):
|
||||
self.put_calls.append((remote_path, content))
|
||||
# Mirror the upload so a subsequent stat() sees it.
|
||||
self.existing_files[remote_path] = len(content)
|
||||
|
||||
|
||||
def test_submit_file_happy_path():
|
||||
"""parse → DB write → SFTP upload, all steps succeed."""
|
||||
fake = _FakeSftp()
|
||||
sftp_block = MagicMock()
|
||||
sftp_block.stub = False
|
||||
sftp_block.paths = {"outbound": "/ToHPE"}
|
||||
|
||||
result = submit_file(
|
||||
_FIXTURE,
|
||||
sftp_block=sftp_block,
|
||||
actor="test",
|
||||
validate=True,
|
||||
sftp_client_factory=lambda _block: fake,
|
||||
)
|
||||
|
||||
assert result.outcome == SubmitOutcome.SUBMITTED
|
||||
assert result.batch_id is not None
|
||||
assert len(fake.put_calls) == 1
|
||||
# DB row landed.
|
||||
with db_mod.SessionLocal()() as s:
|
||||
assert s.get(db_mod.Batch, result.batch_id) is not None
|
||||
|
||||
|
||||
def test_submit_file_skipped_when_remote_size_matches():
|
||||
"""Re-running submit_file on an already-uploaded file is SKIPPED.
|
||||
|
||||
The ``SKIPPED`` outcome is reached via the SFTP ``stat()`` check
|
||||
(not via the DB) — same-size remote file ⇒ idempotency short-circuit.
|
||||
Re-runs therefore don't emit a duplicate ``clearhouse.submitted``
|
||||
audit event.
|
||||
"""
|
||||
fake = _FakeSftp()
|
||||
sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
|
||||
factory = lambda _b: fake
|
||||
|
||||
r1 = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
|
||||
r2 = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
|
||||
|
||||
# First call SUBMITTED, second call SKIPPED (size match on stat).
|
||||
assert r1.outcome == SubmitOutcome.SUBMITTED
|
||||
assert r2.outcome == SubmitOutcome.SKIPPED
|
||||
|
||||
|
||||
def test_submit_file_parse_fail(tmp_path):
|
||||
"""Bad 837 → PARSE_FAILED, no DB write, no SFTP call."""
|
||||
bad = tmp_path / "bad.x12"
|
||||
bad.write_bytes(b"not x12")
|
||||
sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
|
||||
fake = _FakeSftp()
|
||||
factory = lambda _b: fake
|
||||
|
||||
result = submit_file(bad, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
|
||||
assert result.outcome == SubmitOutcome.PARSE_FAILED
|
||||
assert len(fake.put_calls) == 0
|
||||
|
||||
|
||||
def test_submit_file_db_fail(monkeypatch):
|
||||
"""If add raises, no SFTP call is made (DB-first invariant)."""
|
||||
from cyclone.store import store as cycl_store
|
||||
monkeypatch.setattr(cycl_store, "add", MagicMock(side_effect=RuntimeError("db down")))
|
||||
|
||||
sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
|
||||
fake = _FakeSftp()
|
||||
factory = lambda _b: fake
|
||||
|
||||
result = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
|
||||
assert result.outcome == SubmitOutcome.DB_FAILED
|
||||
assert len(fake.put_calls) == 0
|
||||
|
||||
|
||||
def test_submit_file_sftp_fail():
|
||||
"""SFTP failure after DB write → SFTP_FAILED, DB row still present."""
|
||||
|
||||
class _Boom:
|
||||
def stat(self, _p):
|
||||
raise IOError("nope")
|
||||
def write_file(self, _p, _c):
|
||||
raise RuntimeError("sftp down")
|
||||
|
||||
sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
|
||||
factory = lambda _b: _Boom()
|
||||
|
||||
result = submit_file(_FIXTURE, sftp_block=sftp_block, actor="t", sftp_client_factory=factory)
|
||||
assert result.outcome == SubmitOutcome.SFTP_FAILED
|
||||
# DB row was written before the SFTP call.
|
||||
assert result.batch_id is not None
|
||||
with db_mod.SessionLocal()() as s:
|
||||
assert s.get(db_mod.Batch, result.batch_id) is not None
|
||||
|
||||
|
||||
def test_submit_file_payer_mismatch():
|
||||
"""PAYER_MISMATCH → no DB write, no SFTP call.
|
||||
|
||||
We monkeypatch ``parse_837_text`` to return a single claim whose
|
||||
payer.id is ``OTHER_PAYER`` (the helper gates on the first
|
||||
claim's payer — same posture as the legacy ``resubmit_rejected_claims``
|
||||
CLI at cli.py:657-668).
|
||||
"""
|
||||
from unittest.mock import patch
|
||||
|
||||
bad_claim = MagicMock()
|
||||
bad_claim.payer.id = "OTHER_PAYER"
|
||||
parsed_with_bad_payer = MagicMock()
|
||||
parsed_with_bad_payer.claims = [bad_claim]
|
||||
parsed_with_bad_payer.parsed_at = None
|
||||
parsed_with_bad_payer.envelope.transaction_set_control_number = "ST123"
|
||||
|
||||
fake = _FakeSftp()
|
||||
sftp_block = MagicMock(stub=False, paths={"outbound": "/ToHPE"})
|
||||
|
||||
with patch("cyclone.submission.core.parse_837_text", return_value=parsed_with_bad_payer):
|
||||
result = submit_file(
|
||||
_FIXTURE,
|
||||
sftp_block=sftp_block,
|
||||
actor="t",
|
||||
validate=True,
|
||||
sftp_client_factory=lambda _b: fake,
|
||||
)
|
||||
assert result.outcome == SubmitOutcome.PAYER_MISMATCH
|
||||
assert "OTHER_PAYER" in (result.error or "")
|
||||
assert len(fake.put_calls) == 0
|
||||
|
||||
|
||||
def test_submit_file_uses_default_paramiko_factory(monkeypatch):
|
||||
"""When no factory is supplied, helper should use paramiko directly
|
||||
(not SftpClient wrapper, which lacks stat()).
|
||||
"""
|
||||
fake_paramiko_sftp = _FakeSftp()
|
||||
monkeypatch.setattr(
|
||||
"cyclone.submission.core._default_sftp_factory",
|
||||
lambda _block: fake_paramiko_sftp,
|
||||
)
|
||||
|
||||
sftp_block = MagicMock()
|
||||
sftp_block.stub = False
|
||||
sftp_block.paths = {"outbound": "/ToHPE"}
|
||||
|
||||
result = submit_file(
|
||||
_FIXTURE,
|
||||
sftp_block=sftp_block,
|
||||
actor="t",
|
||||
validate=True,
|
||||
# NO sftp_client_factory — should use the paramiko default
|
||||
)
|
||||
assert result.outcome == SubmitOutcome.SUBMITTED
|
||||
assert len(fake_paramiko_sftp.put_calls) == 1
|
||||
|
||||
|
||||
def test_submit_batch_cli_help():
|
||||
"""`cyclone submit-batch --help` exits 0 and shows the flags."""
|
||||
import subprocess
|
||||
import sys
|
||||
result = subprocess.run(
|
||||
[sys.executable, "-m", "cyclone.cli", "submit-batch", "--help"],
|
||||
capture_output=True, text=True,
|
||||
cwd=str(Path(__file__).parent.parent),
|
||||
)
|
||||
assert result.returncode == 0
|
||||
assert "--ingest-dir" in result.stdout
|
||||
+153
@@ -138,3 +138,156 @@ started without the secret mount), the scheduler surfaces a
|
||||
`RuntimeError` at the next tick that names the env var and the
|
||||
missing path — this is intentional, so a silent fall-through doesn't
|
||||
mask a real misconfiguration.
|
||||
|
||||
## Manual SFTP mode (this box's current posture)
|
||||
|
||||
Use this when the host's IP isn't whitelisted with Gainwell's MFT, when
|
||||
you don't want a daemon polling every minute, or when you prefer to
|
||||
drag-drop files with FileZilla / WinSCP. The seeded `dzinesco`
|
||||
clearhouse ships in this mode (`sftp_block.stub: true`); no daemon
|
||||
changes required.
|
||||
|
||||
**Posture.** `SftpClient` reads/writes to local staging instead of
|
||||
`mft.gainwelltechnologies.com`:
|
||||
|
||||
- Outbound (`write_file`): `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/ToHPE/`
|
||||
- Inbound (`list_inbound`): `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
|
||||
|
||||
(Paths are relative to the backend cwd; create the directories if
|
||||
absent.)
|
||||
|
||||
### Daily flow (manual mode)
|
||||
|
||||
1. **Pull inbound from Gainwell** with your SFTP client:
|
||||
`/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
|
||||
2. **Stage locally** in `./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/`
|
||||
(or, equivalently, drop them in `/home/tyler/dev/cyclone/ingest/`
|
||||
and copy in).
|
||||
3. **Process** with `pull-inbound` (writes acks/835s to DB, dedupes
|
||||
via `processed_inbound_files`):
|
||||
|
||||
```bash
|
||||
cd /home/tyler/dev/cyclone/backend
|
||||
mkdir -p "./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE"
|
||||
cp /home/tyler/dev/cyclone/ingest/*.x12 \
|
||||
"./var/sftp/staging/CO XIX/PROD/coxix_prod_11525703/FromHPE/"
|
||||
.venv/bin/python -m cyclone pull-inbound --date 20260701
|
||||
.venv/bin/python -m cyclone pull-inbound --date 20260702
|
||||
.venv/bin/python -m cyclone pull-inbound --date 20260703
|
||||
```
|
||||
|
||||
4. **Submit outbound** with `POST /api/clearhouse/submit` (writes
|
||||
serialized 837P files into the stub staging dir; you drag-drop
|
||||
them to your SFTP client's `/CO XIX/PROD/coxix_prod_11525703/ToHPE/`).
|
||||
|
||||
### Backfill a backlog
|
||||
|
||||
`ingest/` often holds days of unprocessed acks. The copy + per-date
|
||||
`pull-inbound` flow above clears it. Files already in
|
||||
`processed_inbound_files` are skipped automatically — to re-process,
|
||||
delete the row first (`DELETE FROM processed_inbound_files WHERE name = ...`).
|
||||
|
||||
### Submitting claims (canonical — SP37)
|
||||
|
||||
For 837P files generated upstream (dzinesco) that you want cyclone to
|
||||
**track in the DB** before uploading, use the canonical submit path.
|
||||
This is the preferred outbound path going forward — it captures the
|
||||
batch's `transaction_set_control_number` (the ST02 control number that
|
||||
999 acks reference in AK201) so future 999 ack links resolve instead
|
||||
of becoming orphans.
|
||||
|
||||
```bash
|
||||
cd /home/tyler/dev/cyclone/backend
|
||||
|
||||
# 1. Lay out files in batch-*-claims subdirs under your ingest dir:
|
||||
# ingest/batch-2026-07-08-claims/claim-001.x12
|
||||
# ingest/batch-2026-07-08-claims/claim-002.x12
|
||||
# ...
|
||||
|
||||
# 2. CLI — walks ingest/, parses, writes to DB, then SFTP-uploads.
|
||||
.venv/bin/python -m cyclone submit-batch \
|
||||
--ingest-dir /home/tyler/dev/cyclone/ingest \
|
||||
--actor cli-submit-batch
|
||||
|
||||
# Or via HTTP (auth-gated by matrix_gate):
|
||||
curl -s -X POST http://localhost:8000/api/submit-batch \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{"ingest_dir": "/home/tyler/dev/cyclone/ingest", "actor": "api-submit-batch"}'
|
||||
```
|
||||
|
||||
Both surfaces share `cyclone.submission.submit_file` for the
|
||||
parse → DB-write → SFTP-upload chain (DB-first, upload-second
|
||||
invariant). The walker pattern is identical: `batch-*-claims/*.x12`,
|
||||
sorted, with `._*` AppleDouble files skipped.
|
||||
|
||||
**When to use `submit-batch` vs `resubmit-rejected-claims`:**
|
||||
- `submit-batch` — canonical path for fresh 837s from dzinesco (or any
|
||||
source) that should be tracked in the DB before upload. Default choice.
|
||||
- `resubmit-rejected-claims` — one-off path for cases where you do NOT
|
||||
want a DB row (e.g., dzinesco-generated fixes not ready for canonical
|
||||
tracking). Legacy, retained for backward compat.
|
||||
|
||||
**Status codes / exit codes:**
|
||||
- HTTP 200 on completed runs (per-file failures live in the JSON body);
|
||||
401 unauthenticated; 404 no clearhouse; 409 stub mode; 422 validation.
|
||||
- CLI exit 0 on completed runs (per-file failures counted, not bumped);
|
||||
2 on config-level failures (no clearhouse / stub mode / missing dir).
|
||||
|
||||
### Note on per-file parse CLIs
|
||||
|
||||
`parse-837` and `parse-835` exist as CLIs but only emit JSON files to
|
||||
`--output-dir`; they do NOT write to the DB. There is no
|
||||
`parse-999` / `parse-ta1` / `parse-277ca` CLI in this version of
|
||||
cyclone — the canonical 999/TA1/277CA/835 ingestion path is
|
||||
`pull-inbound` → `Scheduler.process_inbound_files`. For inspection of
|
||||
a single file without DB writes, use the Python API:
|
||||
|
||||
```python
|
||||
from cyclone.parsers.parse_999 import parse as parse_999
|
||||
result = parse_999(open("path.999.x12").read())
|
||||
```
|
||||
|
||||
### Switching from manual → real (and back)
|
||||
|
||||
When this host's IP is whitelisted with Gainwell's MFT admin, the
|
||||
SP25 procedure above flips the block. The CLI one-liner:
|
||||
|
||||
```bash
|
||||
export CYCLONE_SFTP_PASSWORD="$GAINWELL_SFTP_PASS"
|
||||
.venv/bin/python -c "
|
||||
from cyclone import db, store
|
||||
from cyclone.providers import Clearhouse, SftpBlock
|
||||
db.init_db()
|
||||
ch = store.store.get_clearhouse()
|
||||
sb = ch.sftp_block.model_dump()
|
||||
sb['stub'] = False
|
||||
sb['auth'] = {'password_keychain_account': 'sftp.gainwell.password'}
|
||||
new = Clearhouse(
|
||||
id=1, name=ch.name, tpid=ch.tpid,
|
||||
submitter_id_qual=ch.submitter_id_qual,
|
||||
submitter_name=ch.submitter_name,
|
||||
submitter_contact_name=ch.submitter_contact_name,
|
||||
submitter_contact_email=ch.submitter_contact_email,
|
||||
filename_block=ch.filename_block,
|
||||
sftp_block=SftpBlock.model_validate(sb, strict=True),
|
||||
updated_at=ch.updated_at,
|
||||
)
|
||||
print('updated:', store.store.update_clearhouse(new).sftp_block.stub)
|
||||
"
|
||||
```
|
||||
|
||||
Hot-reload via `PATCH /api/clearhouse` (preferred over direct DB
|
||||
write — it also calls `scheduler.reconfigure_scheduler` so the running
|
||||
daemon picks up the new block without a restart):
|
||||
|
||||
```bash
|
||||
curl -s -b cookies.txt http://127.0.0.1:8000/api/clearhouse > /tmp/ch.json
|
||||
jq '.sftp_block.stub = false
|
||||
| .sftp_block.auth = {"password_keychain_account": "sftp.gainwell.password"}' \
|
||||
/tmp/ch.json > /tmp/ch-patched.json
|
||||
curl -X PATCH http://127.0.0.1:8000/api/clearhouse \
|
||||
-H 'Content-Type: application/json' -b cookies.txt \
|
||||
--data @/tmp/ch-patched.json
|
||||
```
|
||||
|
||||
To revert, set `stub: true` and `auth: {"method": "keychain", "secret_ref": "sftp.gainwell.password"}`.
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,217 @@
|
||||
# Sub-project 37 — Canonical `submit-batch` flow: Design Spec
|
||||
|
||||
**Date:** 2026-07-07
|
||||
**Status:** Draft, awaiting user sign-off
|
||||
**Branch:** `sp37-submit-batch-canonical-flow` (off main, post-SP36 merge)
|
||||
**Aesthetic direction:** No UI changes. Pure backend structural + data-model addition — the public HTTP surface gains one new endpoint; existing CLIs and endpoints are preserved unchanged.
|
||||
|
||||
---
|
||||
|
||||
## 1. Scope
|
||||
|
||||
Today, cyclone's `claims` table is essentially empty in production (`clm-1` is the only row as of 2026-07-07). The reason is structural: dzinesco generates the 837P files (in `ingest/corrected/batch-*-claims/*.x12`), cyclone's `resubmit-rejected-claims` CLI only validates and re-uploads them to Gainwell's MFT — it never writes the claims to the DB. When 999 acknowledgments come back referencing those batches, every AK2 set-response becomes an orphan because there is no `Batch` row to join against.
|
||||
|
||||
The 999→claim join (the D10 two-pass join in `claim_acks.lookup_claims_for_ack_set_response`) has two paths:
|
||||
|
||||
- **Pass 1 (primary):** `Batch.envelope.control_number == 999's set_control_number`. This currently fails because the 999's `set_control_number` (AK201) echoes the source 837's **ST02** (transaction set control number), not the **ISA13** (interchange control number). The parsed `Envelope` model stores ISA13 in `control_number`, so the keys never collide even when the DB row exists.
|
||||
- **Pass 2 (fallback):** `Claim.patient_control_number == 999's set_control_number`. This also fails because the `_claim_837_row` ORM builder sets `patient_control_number = claim.claim_id` (CLM01), and CLM01 ≠ ST02 in production batches.
|
||||
|
||||
The fix is twofold: (a) make cyclone write a `Batch` + `Claim` rows at submission time so there is something to join against, and (b) make the join key actually match — by storing the ST02 explicitly on `Batch` and using it in Pass 1.
|
||||
|
||||
**In scope:**
|
||||
|
||||
- A new top-level CLI `cyclone submit-batch` that does parse → DB write → SFTP upload per file, in that order.
|
||||
- A new HTTP endpoint `POST /api/submit-batch` with the same semantics.
|
||||
- A shared helper module `cyclone.submission.submit_file` so the CLI and HTTP path share one source of truth.
|
||||
- A new nullable column `Batch.transaction_set_control_number`, populated from the parsed 837's ST02 on every `add_record` write.
|
||||
- An Alembic migration `0013_add_batch_txn_set_control_number.py` (additive, nullable, backfills from `raw_result_json.envelope.transaction_set_control_number` where present).
|
||||
- An update to `claim_acks.batch_envelope_index` so Pass 1 also matches on `transaction_set_control_number` (Pass 2 unchanged).
|
||||
- Per-claim `claim_submitted` activity events and per-file `clearhouse.submitted` audit events (same shape `resubmit-rejected-claims` already emits).
|
||||
|
||||
**Out of scope (explicit):**
|
||||
|
||||
- Deprecating `parse-837` CLI or `/api/parse-837` endpoint. They remain for one-off testing and dry-run validation.
|
||||
- Deprecating `resubmit-rejected-claims`. It remains the "validate-then-upload without DB write" path for the dzinesco-generated files that aren't ready for canonical tracking.
|
||||
- UI changes. The Claims page already renders the fields a new tracked claim needs; the existing `claim_written` live-tail event fires automatically from `add_record`.
|
||||
- Bulk backfill of the existing `ingest/corrected/batch-*-claims/*.x12` backlog. A follow-up SP will walk that directory once `submit-batch` is proven.
|
||||
- Any change to the `claims` row schema beyond the new `Batch` column.
|
||||
- SFTP auth work. This box stays in manual mode (post-2026-07-07 incident); `submit-batch` works in stub mode for local testing and against a whitelisted IP for production.
|
||||
|
||||
## 2. Decisions (locked during brainstorming)
|
||||
|
||||
These were each a multiple-choice question with operator sign-off on 2026-07-07.
|
||||
|
||||
1. **Where does the DB write happen?** New canonical `submit` flow. `submit-batch` parses → writes to DB → uploads, replacing neither `parse-837` nor `resubmit-rejected-claims` but superseding them as the preferred path.
|
||||
2. **DB vs upload ordering?** DB-first, upload-second. If DB write fails, the file never goes on the wire. If DB write succeeds but upload fails, re-running is safe (idempotent DB write, idempotent SFTP put).
|
||||
3. **Join key for 999 acks?** Add `Batch.transaction_set_control_number` column. Update Pass 1 to match on it. Don't repurpose `envelope.control_number` (preserves backward compat with existing joins) and don't force CLM01 == ST02 (Gainwell's existing 999s use different values).
|
||||
4. **Deprecation posture?** Add `submit-batch`; preserve `parse-837` and `resubmit-rejected-claims`. Document `submit-batch` as preferred in CLAUDE.md + RUNBOOK. Future SP can deprecate if usage shifts.
|
||||
|
||||
## 3. Architecture
|
||||
|
||||
The new flow is a single, small public surface — one CLI command, one HTTP endpoint, one shared helper. The helper owns the parse → write → upload sequence and is the only place that knows about the order.
|
||||
|
||||
### Module layout
|
||||
|
||||
A new package `cyclone.submission` (under `backend/src/cyclone/submission/`) holds the shared helper:
|
||||
|
||||
- `__init__.py` — re-exports `submit_file`, `SubmitResult`, `SubmitOutcome`
|
||||
- `core.py` — `submit_file(path: Path, *, sftp_block, actor: str, event_bus=None, validate: bool = True) -> SubmitResult`. Owns its own DB session for the parse-write step, returns a result dataclass the caller inspects.
|
||||
- `cli.py` — Click handler for `cyclone submit-batch`. Walks `--ingest-dir`, calls `submit_file` per file, aggregates counts, exits 0/1/2 per cyclone-cli convention.
|
||||
- `api.py` — FastAPI handler for `POST /api/submit-batch`. Mounted in `api_routers/submission.py`. Calls `submit_file` per file. Auth-gated by the existing `matrix_gate`.
|
||||
|
||||
### `submit_file` algorithm
|
||||
|
||||
The walker globs `<ingest_dir>/batch-*-claims/*.x12` (mirrors `resubmit-rejected-claims` exactly — same pattern, same `_`-prefix skip for AppleDouble residue). For each matched file:
|
||||
|
||||
1. Read bytes; if `validate=True`, run `parse_837_text(bytes, payer_cfg)`. On parse failure, record `{path, "parse_failed", exc}` and return without writing or uploading.
|
||||
2. Run a payer-id check (must be `CO_TXIX`); on mismatch, record and return.
|
||||
3. Call `store.add_record(record, event_bus=...)`. On DB failure, record and return (no SFTP call).
|
||||
4. Build the remote path `{outbound_root}/{filename}`. Open an `SftpClient` (the seeded singleton). On SFTP connect failure, record and return — the DB row is already written; the re-run will skip it via idempotency.
|
||||
5. `sftp_client.stat(remote_path)` — if `st_size == local_size`, mark as `skipped` (already uploaded; do not re-emit audit). Otherwise `sftp_client.write_file(remote_path, bytes)`.
|
||||
6. On successful upload, emit one `clearhouse.submitted` audit event with `source_file`, `batch_id`, `actor`.
|
||||
|
||||
### Idempotency
|
||||
|
||||
Two layers, both per-file:
|
||||
|
||||
- **DB layer:** `store.add_record` already does a `s.get(Claim, claim.claim_id)` check before insert. A re-parse of the same file logs "claim already exists; skipping" and continues. Re-running `submit-batch` on an already-tracked file is a no-op on the DB.
|
||||
- **SFTP layer:** the `sftp.stat().st_size == local_size` check before `write_file` short-circuits already-uploaded files.
|
||||
|
||||
### Failure modes summary
|
||||
|
||||
- Parse fails → no DB write, no SFTP put, file counted as `failed`.
|
||||
- Payer-id mismatch → same as parse fail.
|
||||
- DB write fails → no SFTP put, file counted as `failed`. The DB session rolls back; no orphan rows.
|
||||
- SFTP connect fails → DB row already committed; file counted as `failed`. Re-run is safe.
|
||||
- SFTP stat succeeds + size match → counted as `skipped`; no audit event.
|
||||
- SFTP write fails → file counted as `failed`. DB row already committed. Re-run is safe.
|
||||
- Crash between DB write and SFTP put → DB row exists, file never landed. Re-run will see the size mismatch (remote doesn't have it) and upload.
|
||||
|
||||
## 4. Data flow & error handling
|
||||
|
||||
### `POST /api/submit-batch`
|
||||
|
||||
Request body (JSON):
|
||||
- `ingest_dir: string` — root directory holding `batch-*-claims/*.x12` subfolders (mirrors the CLI flag).
|
||||
- `validate: bool = true` — pass-through to `submit_file`.
|
||||
- `actor: string = "api-submit-batch"` — audit actor tag.
|
||||
- `limit: int | null = null` — stop after N files (smoke test).
|
||||
|
||||
Response body (200):
|
||||
- `submitted: int` — files uploaded this run.
|
||||
- `skipped: int` — files already on remote (size match).
|
||||
- `failed: int` — files that hit parse / payer / DB / SFTP errors.
|
||||
- `results: list<{file: string, outcome: "submitted"|"skipped"|"parse_failed"|"payer_mismatch"|"db_failed"|"sftp_failed", batch_id: string|null, error: string|null}>` — per-file detail.
|
||||
|
||||
Status codes:
|
||||
- `200` — run completed (may have failures; details in the response body).
|
||||
- `401` — no auth cookie (matrix_gate).
|
||||
- `422` — body validation failure.
|
||||
- `500` — unexpected exception (caught by the existing `@app.exception_handler(Exception)`).
|
||||
|
||||
### CLI exit codes (per `cyclone-cli` convention)
|
||||
|
||||
- `0` — run completed, even with some per-file failures (the per-file detail is in stdout).
|
||||
- `1` — unexpected exception (uncaught error in the walker or the helper itself).
|
||||
- `2` — config-level failure (no clearhouse seeded, SFTP block in stub mode for non-stubbed invocation, ingest-dir doesn't exist).
|
||||
|
||||
### Audit + live-tail events
|
||||
|
||||
- `claim_submitted` (one per Claim row) — emitted by `add_record` already; no new code path.
|
||||
- `clearhouse.submitted` (one per successful upload) — emitted by `submit_file` after the SFTP put succeeds. Same actor tag and payload shape as the current `resubmit-rejected-claims` path uses.
|
||||
|
||||
### Migration `0013_add_batch_txn_set_control_number.py`
|
||||
|
||||
- `op.add_column('batches', sa.Column('transaction_set_control_number', sa.String(32), nullable=True))`.
|
||||
- Backfill: `UPDATE batches SET transaction_set_control_number = json_extract(raw_result_json, '$.envelope.transaction_set_control_number') WHERE raw_result_json IS NOT NULL AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') IS NOT NULL`.
|
||||
- Reversible: drop column.
|
||||
- The down migration is `op.drop_column('batches', 'transaction_set_control_number')`.
|
||||
|
||||
### Join-key update in `claim_acks.batch_envelope_index`
|
||||
|
||||
The existing function reads every `Batch` row whose `raw_result_json` has an envelope and returns `dict[str, str]` keyed by `envelope.control_number`. The updated version returns the **same shape** — a single `dict[str, str]` — but **populated from two columns**:
|
||||
|
||||
- Each row contributes one entry keyed by `envelope.control_number` (preserved).
|
||||
- Each row also contributes one entry keyed by `transaction_set_control_number` (new), when that column is non-NULL.
|
||||
|
||||
If both keys resolve to the same value (Gainwell batches where the ST02 happens to equal the ISA13 — unusual but legal), the second insert is a no-op. If they differ, both keys point to the same `batch.id` — a single `idx.get(set_control_number)` lookup will match either one. The dict may grow by up to `len(batches)` new entries; callers that previously assumed `len(index) == len(batches)` are unaffected because that invariant was already loose (multiple ST02s can share one ISA13).
|
||||
|
||||
`lookup_claims_for_ack_set_response` Pass 1 is unchanged at the call site — one `idx.get(set_control_number)` lookup that succeeds whether the source 837's ST02 matches a `transaction_set_control_number` row or an `envelope.control_number` row. Pass 2 (PCN) fires as today if both miss.
|
||||
|
||||
## 5. Testing approach
|
||||
|
||||
Per `cyclone-tests` skill: pytest under `backend/tests/`, two flavors — `test_api_*` (FastAPI integration) and `test_*` (pure unit). Frontend tests out of scope (no UI changes).
|
||||
|
||||
### Unit tests (`backend/tests/test_submission.py`)
|
||||
|
||||
- `submit_file happy path` — mock SFTP, real DB; assert one Batch + N Claim rows + one audit event.
|
||||
- `submit_file idempotency on DB` — call `submit_file` twice with the same file; assert second call returns `skipped` (DB no-op).
|
||||
- `submit_file idempotency on SFTP` — mock SFTP `stat` to return matching size; assert `skipped`, no second `write_file` call, no second audit event.
|
||||
- `submit_file parse fail` — pass a bad 837; assert no DB write, no SFTP call, `parse_failed` outcome.
|
||||
- `submit_file payer mismatch` — pass a valid 837 with wrong payer_id; assert no DB write, no SFTP call, `payer_mismatch` outcome.
|
||||
- `submit_file DB fail` — monkey-patch `add_record` to raise; assert no SFTP call, `db_failed` outcome.
|
||||
- `submit_file SFTP fail` — mock SFTP `write_file` to raise; assert `sftp_failed` outcome, DB row still present.
|
||||
- `submit_file re-run after partial failure` — submit a file that fails SFTP, then submit again with SFTP working; assert second run uploads and emits audit.
|
||||
|
||||
### Integration tests (`backend/tests/test_api_submit_batch.py`)
|
||||
|
||||
- `POST /api/submit-batch` happy path — 3 valid 837 fixtures in `ingest_dir`; assert 200 + body shape.
|
||||
- `POST /api/submit-batch` auth gate — no cookie → 401 (matrix_gate).
|
||||
- `POST /api/submit-batch` body validation — missing `ingest_dir` → 422.
|
||||
|
||||
### Migration tests (`backend/tests/test_migration_0013.py`)
|
||||
|
||||
- Up + down on a fresh DB; column exists + is nullable; backfill populates the column from raw JSON.
|
||||
|
||||
### Join-key update tests (`backend/tests/test_claim_acks_index.py`)
|
||||
|
||||
- `batch_envelope_index` returns both maps; Pass 1 finds a claim by ST02 even when ISA13 differs; Pass 1 finds by ISA13 (backward compat); Pass 2 fallback still works.
|
||||
|
||||
### Live test (after merge)
|
||||
|
||||
Walk an actual `ingest/batch-*-claims/` directory via the CLI; verify:
|
||||
- `claims` table has new rows with the expected `patient_control_number` (CLM01).
|
||||
- `batches` table has new rows with `transaction_set_control_number` matching the parsed ST02.
|
||||
- Live-tail `claim_written` events fire on the API.
|
||||
|
||||
Then drop a synthetic 999 (with the matching AK2 set_control_number) into `ingest/`, run `pull-inbound`, verify the `ClaimAck` link row is created (not an orphan).
|
||||
|
||||
## 6. Threat model (post-SP24 alignment)
|
||||
|
||||
No new attack surface. The new endpoint and CLI:
|
||||
|
||||
- Are gated by the existing `matrix_gate` (HTTP) and `CYCLONE_AUTH_DISABLED` env var.
|
||||
- Use the same SFTP client and credential paths as `resubmit-rejected-claims` today.
|
||||
- Write to the same store facade; no new SQL surface.
|
||||
- Emit the same audit events with the same payload shape.
|
||||
|
||||
The `transaction_set_control_number` column is operator-visible (it's in the raw `batches` row), but no auth boundary depends on it being absent. PII exposure is unchanged (the column carries the same value the 999 ack would have echoed back — already public-to-Gainwell).
|
||||
|
||||
## 7. Risks & mitigations
|
||||
|
||||
- **Risk:** The new `submission` package is the third "command runner" in the codebase (alongside `cli.py`'s standalone commands and `scheduler.py`). **Mitigation:** Keep `submit_file` as a pure helper with no FastAPI/Click dependencies; the API and CLI handlers are thin wrappers. Document the layering in `submission/__init__.py`'s module docstring.
|
||||
- **Risk:** `resubmit-rejected-claims` and `submit-batch` will diverge over time (one writes to DB, one doesn't). **Mitigation:** Both call the same `submit_file` helper if `submit-batch`'s `--write-to-db` flag is set to true (default). Document the divergence point in RUNBOOK.
|
||||
- **Risk:** Migration backfill could be slow on a large DB. **Mitigation:** `UPDATE ... WHERE json_extract(...) IS NOT NULL` is one pass; on a million-row DB this is sub-second. Index on `transaction_set_control_number` not added (low cardinality per batch; existing `ix_claims_patient_control_number` covers the hot path).
|
||||
- **Risk:** Live-tail event volume doubles when `submit-batch` runs (one `claim_submitted` per claim in addition to the historical `claim_written`). **Mitigation:** This is desired behavior — the operator wants the activity feed to show submissions. Document in the RUNBOOK that submitting a 1000-claim batch will publish ~1000 events.
|
||||
|
||||
## 8. Rollout
|
||||
|
||||
- **Branch:** `sp37-submit-batch-canonical-flow`, off main as of `710104f`.
|
||||
- **Implementation order:**
|
||||
1. Migration `0013` + backfill (additive, no behavior change yet).
|
||||
2. Update `claim_acks.batch_envelope_index` + Pass 1 lookup (no behavior change yet — the new map is built but unused until `submit-batch` writes rows).
|
||||
3. `cyclone/submission/core.py` helper + unit tests.
|
||||
4. CLI handler `cyclone submit-batch` + smoke test.
|
||||
5. HTTP handler `POST /api/submit-batch` + integration tests.
|
||||
6. Live test with one of the existing `ingest/batch-*-claims/*.x12` directories (will be in stub mode on this host).
|
||||
- **Live test cadence:** per the user's standing directive — live-test after each significant step, run autoreview, commit.
|
||||
- **Merge shape:** single atomic merge commit per SP-N convention. No squash. No rebase.
|
||||
- **Post-merge:** RUNBOOK update for the canonical flow; CLAUDE.md update to mark `submit-batch` as preferred.
|
||||
|
||||
## 9. Open questions
|
||||
|
||||
None blocking the spec. Tracked for follow-up SPs:
|
||||
|
||||
- **Backfill command** — once `submit-batch` is proven, a `cyclone backfill-claims` CLI that walks the existing `ingest/corrected/batch-*-claims/*.x12` backlog (without uploading — they're already on Gainwell's side) and writes the DB rows. This is what would un-orphan the 1491+ orphan 999 acks currently in the DB.
|
||||
- **Deprecation timeline for `resubmit-rejected-claims`** — once `submit-batch` is the default, consider folding `resubmit-rejected-claims` into a `--no-write-to-db` flag on `submit-batch` to consolidate the surface.
|
||||
- **Cross-link to 277CA acks** — same `transaction_set_control_number` join key will resolve 277CA orphans if/when those arrive. The migration and index update cover both 999 and 277CA since both use Pass 1.
|
||||
Reference in New Issue
Block a user