Nora
d5f95b4f3c
Move 'reconcile.run(s, record.id)' inside CycloneStore.add()'s ingest session, before s.commit(). The placeholder adjustment_amount set by _remittance_835_row is overwritten by reconcile's CAS-aggregate pass in the same transaction — readers never see a half-reconciled Remittance row. If reconcile raises, the entire 835 ingest rolls back via the session's __exit__. The previous flow committed batch + remittance rows in session-1, then opened session-2 to run reconcile.fail-soft. Two visible problems closed: a race window where readers fetched the placeholder adjustment_amount, and a half-reconciled state left visible if reconcile crashed. Deviation from the plan (N4 in autoreview): the reconcile call lives in cycl_store.add() rather than handle_835 calling a new reconcile.run_now(batch_id) helper after add(). Same end state, one fewer module surface, handler stays a thin wrapper over the store.
253 lines
10 KiB
Python
253 lines
10 KiB
Python
"""Direct tests for the ``handle_835`` handler (SP27 Task 5 + Task 10).
|
|
|
|
Locks the handler's contract independent of the scheduler lifecycle
|
|
so a regression in the scheduler wiring doesn't hide a regression
|
|
in the handler.
|
|
|
|
Note: the 835 handler is the largest of the four because the schema
|
|
covers per-claim remittances + CAS adjustments + validation. SP27
|
|
Task 10 unified the two-phase ingest path (batch row first, then a
|
|
separate ``reconcile`` pass that overwrote ``adjustment_amount``)
|
|
into one critical section. These tests pin the atomicity invariants:
|
|
after ``handle()`` returns, the persisted ``adjustment_amount`` is
|
|
already the authoritative CAS sum; if reconcile raises, no batch
|
|
rows land.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cyclone import db
|
|
from cyclone.handlers.handle_835 import handle
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
|
|
MINIMAL = Path(__file__).parent / "fixtures" / "minimal_835.txt"
|
|
UNBALANCED = Path(__file__).parent / "fixtures" / "unbalanced_835.txt"
|
|
CO_MEDICAID = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt"
|
|
|
|
|
|
def test_handle_835_happy_persists_batch_and_remittance():
|
|
"""Happy path: minimal_835 has 1 CLP claim. Handler persists a
|
|
BatchRecord + 1 Remittance row. Returns (parse_835, 1)."""
|
|
text = MINIMAL.read_text()
|
|
parser_used, claim_count = handle(text, source_file=MINIMAL.name)
|
|
assert parser_used == "parse_835"
|
|
assert claim_count == 1
|
|
|
|
# Lock persistence: a Batch record exists, plus a Remittance row.
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import Batch, Remittance
|
|
from sqlalchemy import select
|
|
batch_rows = session.execute(
|
|
select(Batch.__table__.c.id, Batch.__table__.c.kind).where(
|
|
Batch.__table__.c.kind == "835"
|
|
)
|
|
).all()
|
|
assert len(batch_rows) >= 1
|
|
batch_id = batch_rows[-1][0] # last inserted
|
|
rem_rows = session.execute(
|
|
select(Remittance.__table__.c.id).where(
|
|
Remittance.__table__.c.batch_id == batch_id
|
|
)
|
|
).all()
|
|
assert len(rem_rows) == 1
|
|
|
|
|
|
def test_handle_835_with_cas_persists_casadjustment_rows():
|
|
"""co_medicaid_835 has more claims + CAS adjustments — exercise
|
|
the CAS-adjustment persistence path. We don't assert a specific
|
|
count (the fixture may grow); we assert that *some*
|
|
CasAdjustment rows exist."""
|
|
text = CO_MEDICAID.read_text()
|
|
parser_used, claim_count = handle(text, source_file=CO_MEDICAID.name)
|
|
assert parser_used == "parse_835"
|
|
assert claim_count >= 1
|
|
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import CasAdjustment, Remittance
|
|
from sqlalchemy import select
|
|
# Lock CAS adjustment persistence: at least one row was
|
|
# written for the remittances we just ingested.
|
|
cas_rows = session.execute(
|
|
select(CasAdjustment.__table__.c.id)
|
|
).all()
|
|
# CAS rows match per-remit-group; at minimum, the minimal
|
|
# happy-path test left one CAS row behind if minimal_835.txt
|
|
# includes a CAS segment. We assert just that the count is
|
|
# observable.
|
|
assert isinstance(cas_rows, list)
|
|
|
|
|
|
def test_handle_835_validation_failure_handler_returns_normally():
|
|
"""A validation-failing 835 persists with failed_count == claim_count
|
|
(per scheduler's inline behavior). The handler does NOT raise on
|
|
validation failure — that's a parser-vs-validator distinction; the
|
|
parser raises CycloneParseError only on bad EDI."""
|
|
text = UNBALANCED.read_text()
|
|
try:
|
|
parser_used, claim_count = handle(text, source_file=UNBALANCED.name)
|
|
except ValueError:
|
|
# Could raise if the parser rejects the unbalanced file.
|
|
return
|
|
assert parser_used == "parse_835"
|
|
# claim_count is whatever was parsed; the important contract is
|
|
# that the call returned with a parser name (not raise without
|
|
# contract), so the scheduler can record the outcome.
|
|
assert isinstance(claim_count, int)
|
|
assert claim_count >= 0
|
|
|
|
|
|
def test_handle_835_raises_on_completely_unparseable_input():
|
|
"""Garbage that the parser can't tokenize raises ValueError
|
|
(wraps CycloneParseError)."""
|
|
bad = "this is not even EDI"
|
|
with pytest.raises((CycloneParseError, ValueError)):
|
|
handle(bad, source_file="bad.835")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SP27 Task 10: atomicity invariants
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_handle_835_adjustment_amount_matches_cas_sum():
|
|
"""After ``handle()`` returns, every persisted Remittance's
|
|
``adjustment_amount`` equals the SUM of its CasAdjustment rows.
|
|
|
|
Pins the Task 10 atomicity fix: ingest + reconcile live in the
|
|
same DB session, so reconcile's CAS-aggregate write happens
|
|
BEFORE commit and the placeholder value never escapes. Before
|
|
Task 10, a reader could observe the placeholder
|
|
(``_remittance_835_row`` sums only the first service line's CAS)
|
|
while the second-phase ``reconcile.run`` was still pending.
|
|
"""
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import func, select
|
|
|
|
from cyclone.db import CasAdjustment, Remittance
|
|
|
|
text = MINIMAL.read_text()
|
|
handle(text, source_file=MINIMAL.name)
|
|
|
|
with db.SessionLocal()() as session:
|
|
remits = session.execute(select(Remittance)).scalars().all()
|
|
assert len(remits) >= 1
|
|
for r in remits:
|
|
cas_sum = session.execute(
|
|
select(func.coalesce(func.sum(CasAdjustment.amount), 0))
|
|
.where(CasAdjustment.remittance_id == r.id)
|
|
).scalar_one()
|
|
assert r.adjustment_amount == Decimal(str(cas_sum)), (
|
|
f"remit {r.id}: adjustment_amount={r.adjustment_amount} "
|
|
f"!= CAS sum={cas_sum}"
|
|
)
|
|
|
|
|
|
def test_handle_835_reconcile_failure_rolls_back_ingest(monkeypatch):
|
|
"""If reconcile raises mid-ingest, the batch + remittance rows
|
|
don't land at all (atomic rollback, SP27 Task 10).
|
|
|
|
Before Task 10, ``store.add`` committed the batch first and then
|
|
ran reconcile in a separate session with fail-soft semantics —
|
|
a reconcile crash left the half-reconciled batch visible. After
|
|
Task 10, ``reconcile.run`` runs inside the ingest session, so
|
|
any reconcile exception rolls the whole transaction back.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
from cyclone import reconcile
|
|
from cyclone.db import Batch, Remittance
|
|
|
|
def boom(*a, **kw):
|
|
raise RuntimeError("simulated reconcile outage")
|
|
|
|
monkeypatch.setattr(reconcile, "match", boom)
|
|
|
|
text = MINIMAL.read_text()
|
|
with pytest.raises(RuntimeError, match="simulated reconcile outage"):
|
|
handle(text, source_file=MINIMAL.name)
|
|
|
|
with db.SessionLocal()() as session:
|
|
# No batch, no remittance — atomic rollback worked.
|
|
batches = session.execute(select(Batch)).scalars().all()
|
|
remits = session.execute(select(Remittance)).scalars().all()
|
|
assert batches == [], (
|
|
f"reconcile raised but {len(batches)} batch rows landed"
|
|
)
|
|
assert remits == [], (
|
|
f"reconcile raised but {len(remits)} remittance rows landed"
|
|
)
|
|
|
|
|
|
def test_handle_835_late_reconcile_failure_rolls_back_match_loop(monkeypatch):
|
|
"""A crash in the SECOND pipeline (``_reconcile_pair``) still
|
|
rolls back the FIRST pipeline's mutations — Claim.state,
|
|
Claim.matched_remittance_id, Remittance.claim_id, the new
|
|
ActivityEvent row, the new Match row.
|
|
|
|
Pins the atomic rollback across the whole reconcile pipeline,
|
|
not just the listing-error path. Without this, a future change
|
|
that splits ``reconcile.run`` into two sessions would quietly
|
|
re-introduce the same race Task 10 closed.
|
|
"""
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
|
|
from cyclone import reconcile
|
|
from cyclone.db import (
|
|
ActivityEvent, Batch, Claim, ClaimState, Match, Remittance,
|
|
)
|
|
|
|
# Pre-seed a Claim that will auto-match the minimal_835 remit
|
|
# (``payer_claim_control_number = "CLM001"``). Without this,
|
|
# ``reconcile.match`` returns no matches and ``_reconcile_pair``
|
|
# is never called — the test would silently become a no-op.
|
|
with db.SessionLocal()() as s:
|
|
s.add(Claim(
|
|
id="CLM001",
|
|
batch_id="seed",
|
|
patient_control_number="CLM001",
|
|
service_date_from=date(2026, 6, 2),
|
|
charge_amount=Decimal("100.00"),
|
|
state=ClaimState.SUBMITTED,
|
|
))
|
|
s.commit()
|
|
|
|
def boom_pair(session, claim, remittance):
|
|
raise RuntimeError("simulated late-stage reconcile outage")
|
|
|
|
monkeypatch.setattr(reconcile, "_reconcile_pair", boom_pair)
|
|
|
|
text = MINIMAL.read_text()
|
|
with pytest.raises(RuntimeError, match="simulated late-stage reconcile outage"):
|
|
handle(text, source_file=MINIMAL.name)
|
|
|
|
with db.SessionLocal()() as session:
|
|
# Zero-state invariants — same as the early-failure test,
|
|
# but also pinning that the match-loop's side effects (Match
|
|
# row, ActivityEvent) didn't slip through the rollback.
|
|
batches = session.execute(select(Batch)).scalars().all()
|
|
remits = session.execute(select(Remittance)).scalars().all()
|
|
match_rows = session.execute(select(Match)).scalars().all()
|
|
activity_rows = session.execute(select(ActivityEvent)).scalars().all()
|
|
assert batches == []
|
|
assert remits == []
|
|
assert match_rows == [], (
|
|
f"_reconcile_pair raised but {len(match_rows)} Match rows landed"
|
|
)
|
|
assert activity_rows == [], (
|
|
f"_reconcile_pair raised but {len(activity_rows)} ActivityEvent rows landed"
|
|
)
|
|
|
|
# And the pre-seeded claim is unchanged — no matched_remittance_id,
|
|
# still in SUBMITTED state. The match-loop's mutations against
|
|
# this row were rolled back too.
|
|
claim = session.get(Claim, "CLM001")
|
|
assert claim.matched_remittance_id is None
|
|
assert claim.state == ClaimState.SUBMITTED
|