Tyler
534130ee2b
Migration 0014 changes the PRIMARY KEYs of claims and remittances from single-column id to composite (batch_id, id). Enables the spec'd cross-batch CLM01 / CLP01 collision workflow: same CLM01 in multiple batches is now representable (resubmits), pre-flight dedup 409 path is genuinely exercisable, force-insert can skip pre-existing duplicates. Strategy: PRAGMA defer_foreign_keys = ON + table recreation for every table whose FKs pointed at the old single-column PK. Tables recreated: remittances, claims, matches, cas_adjustments, service_line_payments, line_reconciliations. Each child table gains a batch_id (or remittance_batch_id) column for the composite FK side; INSERT INTO new SELECT FROM old JOINs populate it from the already-recreated parent. Cross-table FKs (remittances.claim_id, claims.matched_remittance_id) cannot be SQL-enforced with composite PKs (SQLite has no ALTER CONSTRAINT). Dropped at SQL level; enforced via application-layer invariants in store.manual_match / manual_unmatch / reconcile.run and the dedup.preflight_* helpers. ORM updates (db.py): - Claim / Remittance: composite PK via explicit PrimaryKeyConstraint in __table_args__ (column order matches SQL: batch_id, id). - New Claim.matched_remittance_batch_id column. - Match / CasAdjustment / ServiceLinePayment / LineReconciliation: added the batch side of their composite FK to the parent table. - SQLAlchemy before_insert events auto-populate the batch side of the composite FK from session.new, then identity_map, then SQL fallback. Production code updates: - store.manual_match / manual_unmatch: also write matched_remittance_batch_id on the claim (was missing). - api.py manual-match endpoint: same fix. - reconcile.run: same fix for auto-matched pairs. Test updates: replaced s.get(Claim, X) with the composite key (batch_id, id) where batch_id is known, or s.query().filter().first() where the test only knows the id. Tests that previously inserted a Match row pointing at a non-existent Remittance now seed the parent Remittance so the new NOT NULL composite FK is satisfied.
256 lines
9.6 KiB
Python
256 lines
9.6 KiB
Python
"""Tests for :func:`cyclone.inbox_state_277ca.apply_277ca_rejections`.
|
|
|
|
SP10 T2. The 277CA's STC A4/A6/A7 codes stamp payer-rejection fields
|
|
on matching claim rows. Distinct from the 999 envelope rejection.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from cyclone import db
|
|
from cyclone.db import Claim, init_db
|
|
from cyclone.inbox_state_277ca import apply_277ca_rejections
|
|
from cyclone.parsers.models_277ca import ClaimStatus, ParseResult277CA
|
|
from cyclone.parsers.parse_277ca import parse_277ca_text
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Fixtures
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _fresh_db():
|
|
"""Each test gets a fresh in-memory DB."""
|
|
init_db()
|
|
yield
|
|
|
|
|
|
def _make_claim(session, *, claim_id: str, pcn: str = "CLAIM001") -> Claim:
|
|
c = Claim(
|
|
id=claim_id,
|
|
batch_id="BATCH-1",
|
|
patient_control_number=pcn,
|
|
charge_amount=100.00,
|
|
)
|
|
session.add(c)
|
|
session.commit()
|
|
session.refresh(c)
|
|
return c
|
|
|
|
|
|
def _make_rejected_status(pcn: str | None = "CLAIM001") -> ClaimStatus:
|
|
return ClaimStatus(
|
|
status_code="A6",
|
|
status_description="19",
|
|
entity_identifier="PR",
|
|
classification="rejected",
|
|
payer_claim_control_number=pcn,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Tests
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestApply277CARejectionsHappyPath:
|
|
def test_rejected_status_stamps_matching_claim(self):
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
claim = _make_claim(s, claim_id="c1")
|
|
pcn = claim.patient_control_number
|
|
result = parse_277ca_text(_minimal_277ca_one_rejected())
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(pcn_q):
|
|
return s.query(Claim).filter_by(patient_control_number=pcn_q).first()
|
|
outcome = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome.matched == ["c1"]
|
|
assert outcome.orphans == []
|
|
with db.SessionLocal()() as s:
|
|
# Migration 0014: composite PK; only one claim with id "c1"
|
|
# exists in this test, so filter-by-id returns it.
|
|
c = s.query(Claim).filter(Claim.id == "c1").first()
|
|
assert c.payer_rejected_at is not None
|
|
assert c.payer_rejected_status_code == "A6"
|
|
assert "A6" in (c.payer_rejected_reason or "")
|
|
assert c.payer_rejected_by_277ca_id == "ACK-1"
|
|
|
|
|
|
class TestApply277CARejectionsOrphans:
|
|
def test_unknown_pcn_becomes_orphan(self):
|
|
from cyclone import db
|
|
# No claim exists — PCN won't match.
|
|
text = _minimal_277ca_one_rejected()
|
|
result = parse_277ca_text(text)
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(_):
|
|
return None
|
|
outcome = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome.matched == []
|
|
assert outcome.orphans == ["CLAIM001"]
|
|
|
|
def test_status_without_ref_1k_becomes_orphan(self):
|
|
"""A rejected STC with no REF*1K cannot match a claim."""
|
|
from cyclone import db
|
|
text = _minimal_277ca_no_ref1k()
|
|
result = parse_277ca_text(text)
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(_):
|
|
return None
|
|
outcome = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome.matched == []
|
|
# The orphan entry uses the status code (since PCN is missing).
|
|
assert outcome.orphans == ["A6"]
|
|
|
|
|
|
class TestApply277CARejectionsIdempotent:
|
|
def test_already_stamped_is_not_overwritten(self):
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
_make_claim(s, claim_id="c1")
|
|
text = _minimal_277ca_one_rejected()
|
|
result = parse_277ca_text(text)
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(_):
|
|
return s.query(Claim).filter_by(patient_control_number="CLAIM001").first()
|
|
outcome1 = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome1.matched == ["c1"]
|
|
# Migration 0014: composite PK — see test_rejected_status_stamps_matching_claim.
|
|
original_reason = s.query(Claim).filter(Claim.id == "c1").first().payer_rejected_reason
|
|
original_at = s.query(Claim).filter(Claim.id == "c1").first().payer_rejected_at
|
|
# Run again with same code.
|
|
outcome2 = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome2.matched == []
|
|
assert outcome2.already_rejected == ["c1"]
|
|
with db.SessionLocal()() as s:
|
|
# Migration 0014: see above — composite PK.
|
|
c = s.query(Claim).filter(Claim.id == "c1").first()
|
|
# Reason and timestamp unchanged.
|
|
assert c.payer_rejected_reason == original_reason
|
|
assert c.payer_rejected_at == original_at
|
|
|
|
|
|
class TestApply277CAOnlyRejectsRejected:
|
|
def test_accepted_status_does_not_stamp(self):
|
|
"""An A3 (accepted) status must NOT trigger a payer_rejected stamp."""
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
_make_claim(s, claim_id="c1")
|
|
text = _minimal_277ca_one_accepted()
|
|
result = parse_277ca_text(text)
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(_):
|
|
return s.query(Claim).filter_by(patient_control_number="CLAIM001").first()
|
|
outcome = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome.matched == []
|
|
with db.SessionLocal()() as s:
|
|
# Migration 0014: composite PK.
|
|
c = s.query(Claim).filter(Claim.id == "c1").first()
|
|
assert c.payer_rejected_at is None
|
|
|
|
|
|
class TestApply277CAMultipleStatuses:
|
|
def test_mixed_batch_only_stamps_rejected(self):
|
|
"""Of three statuses (A3/A6/A8), only the A6 claim gets stamped."""
|
|
from cyclone import db
|
|
with db.SessionLocal()() as s:
|
|
_make_claim(s, claim_id="c1", pcn="CLAIM001")
|
|
_make_claim(s, claim_id="c2", pcn="CLAIM002")
|
|
_make_claim(s, claim_id="c3", pcn="CLAIM003")
|
|
text = (
|
|
"ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
|
|
"GS*HN*A*B*20240620*1200*1*X*005010X214~"
|
|
"ST*277CA*0001*005010X214~"
|
|
"BHT*0085*08*X*20240620*1200*TH~"
|
|
"HL*1**20*1~"
|
|
"HL*2*1*21*1~"
|
|
"HL*3*2*19*1~"
|
|
"HL*4*3*PT~"
|
|
"REF*1K*CLAIM001~"
|
|
"STC*A3:19:PR*20240620*WQ*100.00~"
|
|
"HL*5*3*PT~"
|
|
"REF*1K*CLAIM002~"
|
|
"STC*A6:19:PR*20240620*U*250.00~"
|
|
"HL*6*3*PT~"
|
|
"REF*1K*CLAIM003~"
|
|
"STC*A8:19:PR*20240620*U*175.00~"
|
|
"SE*16*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
result = parse_277ca_text(text)
|
|
with db.SessionLocal()() as s:
|
|
def _lookup(pcn):
|
|
return s.query(Claim).filter_by(patient_control_number=pcn).first()
|
|
outcome = apply_277ca_rejections(s, result, claim_lookup=_lookup, two77ca_id="ACK-1")
|
|
assert outcome.matched == ["c2"]
|
|
with db.SessionLocal()() as s:
|
|
# Migration 0014: composite PK — filter-by-id returns the
|
|
# single claim with that id (no cross-batch duplicates in
|
|
# this test fixture).
|
|
assert s.query(Claim).filter(Claim.id == "c1").first().payer_rejected_at is None
|
|
assert s.query(Claim).filter(Claim.id == "c2").first().payer_rejected_at is not None
|
|
assert s.query(Claim).filter(Claim.id == "c3").first().payer_rejected_at is None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Test fixtures
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _minimal_277ca_one_rejected() -> str:
|
|
return (
|
|
"ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
|
|
"GS*HN*A*B*20240620*1200*1*X*005010X214~"
|
|
"ST*277CA*0001*005010X214~"
|
|
"BHT*0085*08*X*20240620*1200*TH~"
|
|
"HL*1**20*1~"
|
|
"HL*2*1*21*1~"
|
|
"HL*3*2*19*1~"
|
|
"HL*4*3*PT~"
|
|
"REF*1K*CLAIM001~"
|
|
"STC*A6:19:PR*20240620*U*100.00~"
|
|
"SE*9*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
|
|
|
|
def _minimal_277ca_one_accepted() -> str:
|
|
return (
|
|
"ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
|
|
"GS*HN*A*B*20240620*1200*1*X*005010X214~"
|
|
"ST*277CA*0001*005010X214~"
|
|
"BHT*0085*08*X*20240620*1200*TH~"
|
|
"HL*1**20*1~"
|
|
"HL*2*1*21*1~"
|
|
"HL*3*2*19*1~"
|
|
"HL*4*3*PT~"
|
|
"REF*1K*CLAIM001~"
|
|
"STC*A3:19:PR*20240620*WQ*100.00~"
|
|
"SE*9*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
|
|
|
|
def _minimal_277ca_no_ref1k() -> str:
|
|
"""Patient HL with STC A6 but no REF*1K."""
|
|
return (
|
|
"ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
|
|
"GS*HN*A*B*20240620*1200*1*X*005010X214~"
|
|
"ST*277CA*0001*005010X214~"
|
|
"BHT*0085*08*X*20240620*1200*TH~"
|
|
"HL*1**20*1~"
|
|
"HL*2*1*21*1~"
|
|
"HL*3*2*19*1~"
|
|
"HL*4*3*PT~"
|
|
"STC*A6:19:PR*20240620*U*100.00~"
|
|
"SE*8*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|