feat(sp37): cyclone.submission.submit_file helper
Owns the parse → DB write → SFTP upload sequence in one place. CLI and HTTP thin-call it. DB-first invariant: if add_record raises, no SFTP call is made. Idempotency: add_record dedupes via s.get(Claim, claim_id); SFTP layer dedupes via stat().st_size match.
This commit is contained in:
@@ -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,164 @@
|
|||||||
|
"""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).
|
||||||
|
"""
|
||||||
|
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.clearhouse import SftpClient
|
||||||
|
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
|
||||||
|
|
||||||
|
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 submit_file(
|
||||||
|
path: Path,
|
||||||
|
*,
|
||||||
|
sftp_block: Any,
|
||||||
|
actor: str,
|
||||||
|
validate: bool = True,
|
||||||
|
sftp_client_factory: Callable[[Any], 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
|
||||||
|
SftpClient-compatible object. Defaults to the real
|
||||||
|
``SftpClient(block=sftp_block)``. 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.
|
||||||
|
remote_path = f"{sftp_block.paths['outbound']}/{file_label}"
|
||||||
|
factory = sftp_client_factory or (lambda b: SftpClient(block=b))
|
||||||
|
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),
|
||||||
|
)
|
||||||
|
|
||||||
|
# 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,26 @@
|
|||||||
|
"""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):
|
||||||
|
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:
|
||||||
|
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~
|
||||||
@@ -0,0 +1,140 @@
|
|||||||
|
"""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 fixture) 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.clearhouse import SftpClient
|
||||||
|
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)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def _db(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
|
||||||
|
db_mod._reset_for_tests()
|
||||||
|
db_mod.init_db()
|
||||||
|
yield
|
||||||
|
|
||||||
|
|
||||||
|
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_idempotent_db():
|
||||||
|
"""Re-running submit_file on the same file is a no-op on the DB side."""
|
||||||
|
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
|
||||||
Reference in New Issue
Block a user