merge: SP37 canonical submit-batch flow into main
12-commit feature branch delivering:
* Migration 0020: add batches.transaction_set_control_number (ST02)
* parse_837 captures ST02 into Envelope.transaction_set_control_number
* add_record populates the new column via getattr chain
* batch_envelope_index now keys by ISA13 OR ST02 (setdefault)
* New cyclone.submission package with submit_file helper
(DB-first / upload-second invariant; paramiko direct factory
because SftpClient wrapper lacks stat())
* New cyclone submit-batch CLI + POST /api/submit-batch endpoint
(thin wrappers over submit_file, byte-for-byte walker parity)
* 31 new tests across 5 files (8 + 11 + 4 + 3 + 5)
* Live-verified end-to-end: 999 with AK2*837*991102977~ now
resolves to claim_id=CLM001 via the new join key (was orphan)
End-to-end orphan reduction going forward: every batch ingested via
the canonical submit path captures ST02; 999 acks referencing that
ST02 resolve via the updated batch_envelope_index.
Spec: docs/superpowers/specs/2026-07-07-cyclone-submit-batch-canonical-flow-design.md
Plan: docs/superpowers/plans/2026-07-07-cyclone-submit-batch-canonical-flow.md
Tracker: /tmp/refactor-cyclone.md
Reviewed-by: pr-reviewer subagent (PASS, all 8 checklist items)
Live-tested-by: end-to-end 999 POST against real dev DB
(/tmp/sp37-task7-live-evidence.txt)
Merge-shape: --no-ff, single atomic commit (per cyclone-spec skill)
This commit is contained in:
@@ -125,7 +125,7 @@ Frontend triplet for any live page: `use<X>(params)` (initial fetch) + `useTailS
|
|||||||
|
|
||||||
## Backend at a glance
|
## 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`.
|
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`.
|
||||||
|
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ from cyclone.api_routers import (
|
|||||||
providers,
|
providers,
|
||||||
reconciliation,
|
reconciliation,
|
||||||
remittances,
|
remittances,
|
||||||
|
submission,
|
||||||
ta1_acks,
|
ta1_acks,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -49,6 +50,7 @@ routers: list[APIRouter] = [
|
|||||||
providers.router, # gated
|
providers.router, # gated
|
||||||
reconciliation.router, # gated
|
reconciliation.router, # gated
|
||||||
remittances.router, # gated
|
remittances.router, # gated
|
||||||
|
submission.router, # gated
|
||||||
ta1_acks.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/acks"): WRITE_ROLES,
|
||||||
("POST", "/api/batches"): WRITE_ROLES, # /export-837 regenerates X12 from DB rows
|
("POST", "/api/batches"): WRITE_ROLES, # /export-837 regenerates X12 from DB rows
|
||||||
("POST", "/api/eligibility"): WRITE_ROLES,
|
("POST", "/api/eligibility"): WRITE_ROLES,
|
||||||
|
("POST", "/api/submit-batch"): WRITE_ROLES, # SP37: canonical outbound path (mirrors CLI)
|
||||||
|
|
||||||
# CSV export — read-only.
|
# CSV export — read-only.
|
||||||
("GET", "/api/export.csv"): ALL_ROLES,
|
("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()
|
@main.group()
|
||||||
def backup() -> None:
|
def backup() -> None:
|
||||||
"""Encrypted DB backup management (SP17)."""
|
"""Encrypted DB backup management (SP17)."""
|
||||||
|
|||||||
@@ -221,6 +221,13 @@ class Batch(Base):
|
|||||||
totals_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
|
totals_json: Mapped[Optional[dict]] = mapped_column(JSONText, nullable=True)
|
||||||
validation_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)
|
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(
|
claims: Mapped[list["Claim"]] = relationship(
|
||||||
back_populates="batch", cascade="all, delete-orphan"
|
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
|
implementation_guide: str | None = None
|
||||||
# SP3 P1 T2: BHT06 transaction type code (was: transaction_set_purpose_code, which is BHT02).
|
# SP3 P1 T2: BHT06 transaction type code (was: transaction_set_purpose_code, which is BHT02).
|
||||||
transaction_type_code: str | None = None
|
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):
|
class BatchSummary(_Base):
|
||||||
|
|||||||
@@ -83,6 +83,12 @@ def _build_envelope(segments: list[list[str]], input_file: str = "") -> tuple[En
|
|||||||
except (IndexError, ValueError) as exc:
|
except (IndexError, ValueError) as exc:
|
||||||
log.warning("Could not parse BHT date: %s", exc)
|
log.warning("Could not parse BHT date: %s", exc)
|
||||||
elif seg[0] == "ST" and envelope is not None:
|
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:
|
if len(seg) > 3:
|
||||||
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
|
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
|
||||||
return envelope, summary
|
return envelope, summary
|
||||||
|
|||||||
@@ -325,7 +325,13 @@ class CycloneStore:
|
|||||||
return _remove_claim_ack(link_id, event_bus=event_bus)
|
return _remove_claim_ack(link_id, event_bus=event_bus)
|
||||||
|
|
||||||
def batch_envelope_index(self):
|
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()
|
return _batch_envelope_index()
|
||||||
|
|
||||||
# -- SP17: encrypted DB backups -------------------------------------
|
# -- SP17: encrypted DB backups -------------------------------------
|
||||||
|
|||||||
@@ -11,8 +11,10 @@ Five methods on top of the new ``ClaimAck`` ORM table:
|
|||||||
ack-orphans lane).
|
ack-orphans lane).
|
||||||
* ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``.
|
* ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``.
|
||||||
* ``batch_envelope_index`` — D10 in-memory map of
|
* ``batch_envelope_index`` — D10 in-memory map of
|
||||||
``Batch.envelope.control_number → batch.id`` (cheap to rebuild;
|
``{Batch.raw_result_json.envelope.control_number (ISA13) OR
|
||||||
re-built once per ingest).
|
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
|
Each mutating method opens its own ``db.SessionLocal()()`` session
|
||||||
so callers don't have to manage session lifecycles. Publishes are
|
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]:
|
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
|
D10 primary join key (spec §D10). Each 837p batch row contributes
|
||||||
every Batch row whose ``raw_result_json`` carries an envelope —
|
up to two entries:
|
||||||
cost is O(N batches), currently ~16, so trivial. Kept as a
|
|
||||||
plain dict so callers can ``index.get(scn)`` to resolve.
|
* ``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:
|
with db.SessionLocal()() as s:
|
||||||
rows = (
|
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")
|
.filter(Batch.kind == "837p")
|
||||||
.all()
|
.all()
|
||||||
)
|
)
|
||||||
for bid, raw in rows:
|
for bid, raw, stcn in rows:
|
||||||
env = (raw or {}).get("envelope") or {}
|
env = (raw or {}).get("envelope") or {}
|
||||||
ctrl = env.get("control_number")
|
isa_cn = env.get("control_number")
|
||||||
if isinstance(ctrl, str) and ctrl:
|
if isinstance(isa_cn, str) and isa_cn:
|
||||||
out[ctrl] = bid
|
# First write wins (setdefault) — if ISA13 and ST02
|
||||||
return out
|
# 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,
|
totals_json=None,
|
||||||
validation_json=None,
|
validation_json=None,
|
||||||
raw_result_json=json.loads(record.result.model_dump_json()),
|
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)
|
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():
|
def test_migration_latest_idempotent_on_fresh_db():
|
||||||
"""Re-running the migration on the same DB must be a no-op (PRAGMA
|
"""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
|
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
|
||||||
providers/payers/clearhouse, SP10's 0008 payer_rejected,
|
providers/payers/clearhouse, SP10's 0008 payer_rejected,
|
||||||
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
|
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
|
claims.matched_remittance_id index, SP27-Task 17's 0017
|
||||||
claim.patient_control_number backfill UPDATE, SP28's 0018
|
claim.patient_control_number backfill UPDATE, SP28's 0018
|
||||||
claim_acks join table, SP32's 0019
|
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:
|
with db.engine().begin() as c:
|
||||||
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
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.
|
# A second run should not raise and should not bump the version.
|
||||||
db_migrate.run(db.engine())
|
db_migrate.run(db.engine())
|
||||||
with db.engine().begin() as c:
|
with db.engine().begin() as c:
|
||||||
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
||||||
assert v2 == 19
|
assert v2 == 20
|
||||||
|
|
||||||
|
|
||||||
def test_add_ack_persists_row():
|
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.
|
SP28 bumped it to 18 with the claim_acks join table.
|
||||||
SP32 bumped it to 19 with rendering_provider_npi +
|
SP32 bumped it to 19 with rendering_provider_npi +
|
||||||
service_provider_npi on claims and remittances.
|
service_provider_npi on claims and remittances.
|
||||||
|
SP37 bumped it to 20 with batch transaction_set_control_number.
|
||||||
"""
|
"""
|
||||||
engine = _fresh_engine(tmp_path)
|
engine = _fresh_engine(tmp_path)
|
||||||
db_migrate.run(engine)
|
db_migrate.run(engine)
|
||||||
v_after_first = _user_version(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)
|
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:
|
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)
|
engine = _fresh_engine(tmp_path)
|
||||||
|
|
||||||
db_migrate.run(engine)
|
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
|
# Two claims in one batch with the same patient_control_number
|
||||||
# must be insertable. If 0015's table recreation re-introduced a
|
# 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
|
||||||
@@ -187,6 +187,52 @@ absent.)
|
|||||||
`processed_inbound_files` are skipped automatically — to re-process,
|
`processed_inbound_files` are skipped automatically — to re-process,
|
||||||
delete the row first (`DELETE FROM processed_inbound_files WHERE name = ...`).
|
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
|
### Note on per-file parse CLIs
|
||||||
|
|
||||||
`parse-837` and `parse-835` exist as CLIs but only emit JSON files to
|
`parse-837` and `parse-835` exist as CLIs but only emit JSON files to
|
||||||
|
|||||||
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