tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp10): 277CA parser + Payer-Rejected Inbox lane

Browse Source 
- Add cyclone.parsers.models_277ca + parse_277ca (X12 005010X214)
  - Per-Patient HL ClaimStatus with REF*1K (PCN), REF*EJ (tax ID),
    STC category code, amount, service date
  - STC classifier: A1-A3 accepted, A4/A6/A7 rejected, A8/A9 pended,
    P1-P5 paid, anything else unknown
  - Multiple STCs per Patient HL: last wins (canonical pattern for
    'first pended, then paid' acknowledgments)
  - Subscriber-level STCs surface in unscoped_statuses
- Add apply_277ca_rejections — stamps Claim.payer_rejected_* fields on
  matching rows (lookup by patient_control_number, mirrors 999 path)
- New /api/parse-277ca, /api/277ca-acks, /api/277ca-acks/{id} endpoints
- New two77ca_acks table + Two77caAck ORM model
- New Claim columns: payer_rejected_at, _reason, _status_code, _by_277ca_id
- New payer_rejected lane in /api/inbox/lanes (distinct from 999
  envelope rejected lane)
- New PayerConfig277CA block in config/payers.yaml + Pydantic model
- Migration 0008 bumps user_version to 8

Tests: 654 -> 688 (parser 22 + apply 6 + API 8 + config 6 + adjustments).
All 688 backend tests pass; 1 pre-existing skipped test class unaffected.
This commit is contained in:
Tyler
2026-06-20 23:41:01 -06:00
parent ae2d48102e
commit 2c0afbe9c5
22 changed files with 1899 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_apply_277ca_rejections.py
+247
View File
@@ -0,0 +1,247 @@
"""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:
c = s.get(Claim, "c1")
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"]
original_reason = s.get(Claim, "c1").payer_rejected_reason
original_at = s.get(Claim, "c1").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:
c = s.get(Claim, "c1")
# 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:
c = s.get(Claim, "c1")
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:
assert s.get(Claim, "c1").payer_rejected_at is None
assert s.get(Claim, "c2").payer_rejected_at is not None
assert s.get(Claim, "c3").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~"
)