Tyler
2c0afbe9c5
- 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.
248 lines
9.4 KiB
Python
248 lines
9.4 KiB
Python
"""Tests for the 277CA Claim Acknowledgment parser.
|
|
|
|
SP10 T1. The 277CA is the semantic claim-level ack from the payer —
|
|
the parser must walk the HL hierarchy and capture one ClaimStatus
|
|
per Patient HL, with REF*1K (payer_claim_control_number) and the
|
|
STC category code.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
from cyclone.parsers.models_277ca import (
|
|
ACCEPTED_CODES,
|
|
PAID_CODES,
|
|
PENDED_CODES,
|
|
REJECTED_CODES,
|
|
ParseResult277CA,
|
|
classify_status_code,
|
|
)
|
|
from cyclone.parsers.parse_277ca import parse_277ca_text
|
|
|
|
FIXTURE_DIR = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Status-code classification
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestClassifyStatusCode:
|
|
def test_accepted_codes(self):
|
|
for code in ACCEPTED_CODES:
|
|
assert classify_status_code(code) == "accepted"
|
|
|
|
def test_rejected_codes(self):
|
|
for code in REJECTED_CODES:
|
|
assert classify_status_code(code) == "rejected"
|
|
|
|
def test_pended_codes(self):
|
|
for code in PENDED_CODES:
|
|
assert classify_status_code(code) == "pended"
|
|
|
|
def test_paid_codes(self):
|
|
for code in PAID_CODES:
|
|
assert classify_status_code(code) == "paid"
|
|
|
|
def test_unknown_code_does_not_raise(self):
|
|
"""Unknown codes (e.g. payer-specific) surface as 'unknown', not error."""
|
|
assert classify_status_code("ZZ") == "unknown"
|
|
assert classify_status_code("") == "unknown"
|
|
assert classify_status_code(" ") == "unknown"
|
|
|
|
def test_case_insensitive(self):
|
|
assert classify_status_code("a3") == "accepted"
|
|
assert classify_status_code("A6") == "rejected"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: minimal happy path with mixed statuses
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CAMixed:
|
|
@pytest.fixture
|
|
def parsed(self) -> ParseResult277CA:
|
|
text = (FIXTURE_DIR / "minimal_277ca.txt").read_text()
|
|
return parse_277ca_text(text, input_file="minimal_277ca.txt")
|
|
|
|
def test_envelope_control_number(self, parsed):
|
|
assert parsed.envelope.control_number == "000000123"
|
|
assert parsed.envelope.implementation_guide == "005010X214"
|
|
|
|
def test_bht_captures_reference_and_date(self, parsed):
|
|
assert parsed.bht.hierarchical_structure_code == "0085"
|
|
assert parsed.bht.transaction_set_purpose_code == "08"
|
|
assert parsed.bht.reference_identification == "REFNUM001"
|
|
assert parsed.bht.transaction_set_creation_date == date(2024, 6, 20)
|
|
assert parsed.bht.transaction_type_code == "TH"
|
|
|
|
def test_three_patient_hls_three_statuses(self, parsed):
|
|
assert len(parsed.claim_statuses) == 3
|
|
|
|
def test_first_status_is_accepted(self, parsed):
|
|
s = parsed.claim_statuses[0]
|
|
assert s.payer_claim_control_number == "CLAIM001"
|
|
assert s.billing_provider_tax_id == "721587149"
|
|
assert s.status_code == "A3"
|
|
assert s.classification == "accepted"
|
|
assert s.total_claim_charge_amount == 100.00
|
|
|
|
def test_second_status_is_rejected(self, parsed):
|
|
s = parsed.claim_statuses[1]
|
|
assert s.payer_claim_control_number == "CLAIM002"
|
|
assert s.status_code == "A6"
|
|
assert s.classification == "rejected"
|
|
assert s.total_claim_charge_amount == 250.00
|
|
|
|
def test_third_status_is_pended(self, parsed):
|
|
s = parsed.claim_statuses[2]
|
|
assert s.payer_claim_control_number == "CLAIM003"
|
|
assert s.status_code == "A8"
|
|
assert s.classification == "pended"
|
|
|
|
def test_service_date_parsed_from_dtp_472(self, parsed):
|
|
"""DTP*472 with RD8 fmt should return the start date."""
|
|
s = parsed.claim_statuses[0]
|
|
assert s.service_date == date(2024, 6, 15)
|
|
|
|
def test_summary_counts(self, parsed):
|
|
# accepted=1 (A3), rejected=1 (A6), pended=1 (A8)
|
|
assert parsed.summary.total_claims == 3
|
|
assert parsed.summary.passed == 1 # only "accepted"
|
|
assert parsed.summary.failed == 1 # only "rejected" (pended excluded)
|
|
|
|
def test_round_trips_via_json(self, parsed):
|
|
blob = json.loads(parsed.model_dump_json())
|
|
assert blob["envelope"]["control_number"] == "000000123"
|
|
assert len(blob["claim_statuses"]) == 3
|
|
rebuilt = ParseResult277CA.model_validate(blob)
|
|
assert rebuilt.claim_statuses[1].status_code == "A6"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: ST*277 (instead of ST*277CA) — X12 spec also allows it
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CAAltST:
|
|
def test_st_277_accepted(self):
|
|
text = (FIXTURE_DIR / "minimal_277ca_st277.txt").read_text()
|
|
result = parse_277ca_text(text, input_file="minimal_277ca_st277.txt")
|
|
assert len(result.claim_statuses) == 1
|
|
assert result.claim_statuses[0].classification == "accepted"
|
|
|
|
def test_st_wrong_value_rejected(self):
|
|
"""A non-277 ST segment should raise CycloneParseError."""
|
|
text = (
|
|
"ISA*00* *00* *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
|
|
"GS*HN*A*B*20240620*1200*1*X*005010X214~"
|
|
"ST*835*0001*005010X221A1~"
|
|
"BHT*0085*08*X*20240620*1200*TH~"
|
|
"SE*3*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
with pytest.raises(CycloneParseError, match="ST\\*277"):
|
|
parse_277ca_text(text)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: rejected-only fixture (single A7)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CARejectedOnly:
|
|
def test_single_a7_rejected(self):
|
|
text = (FIXTURE_DIR / "minimal_277ca_rejected_only.txt").read_text()
|
|
result = parse_277ca_text(text, input_file="minimal_277ca_rejected_only.txt")
|
|
assert len(result.claim_statuses) == 1
|
|
s = result.claim_statuses[0]
|
|
assert s.status_code == "A7"
|
|
assert s.classification == "rejected"
|
|
assert s.payer_claim_control_number == "CLAIM099"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: error handling
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CAErrors:
|
|
def test_missing_isa_raises(self):
|
|
"""No ISA envelope → CycloneParseError, never silent fail."""
|
|
with pytest.raises(CycloneParseError, match="ISA"):
|
|
parse_277ca_text("not a valid 277ca\n")
|
|
|
|
def test_empty_input_raises(self):
|
|
with pytest.raises(CycloneParseError):
|
|
parse_277ca_text("")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: multiple STC per patient (last wins)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CAMultipleStcPerPatient:
|
|
def test_last_stc_wins_per_patient(self):
|
|
"""A patient HL with multiple STC segments: the last one wins.
|
|
|
|
This is the canonical pattern for "first pended, then paid"
|
|
acknowledgments — the most recent action is authoritative.
|
|
"""
|
|
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*CLAIM555~"
|
|
"STC*A1:19:PR*20240620*WQ*100.00~"
|
|
"STC*A8:19:PR*20240620*WQ*100.00~"
|
|
"STC*A3:19:PR*20240620*WQ*100.00~"
|
|
"SE*9*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
result = parse_277ca_text(text)
|
|
assert len(result.claim_statuses) == 1
|
|
assert result.claim_statuses[0].status_code == "A3"
|
|
assert result.claim_statuses[0].classification == "accepted"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Parser: subscriber-level STCs surface in unscoped_statuses
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class TestParse277CASubscriberLevelStc:
|
|
def test_subscriber_stc_goes_to_unscoped(self):
|
|
"""Subscriber-level (HL*19) STCs without a Patient child go to unscoped_statuses."""
|
|
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~"
|
|
"STC*A6:19:PR*20240620*U~"
|
|
"SE*7*0001~"
|
|
"GE*1*1~"
|
|
"IEA*1*000000001~"
|
|
)
|
|
result = parse_277ca_text(text)
|
|
assert len(result.claim_statuses) == 0
|
|
assert len(result.unscoped_statuses) == 1
|
|
assert result.unscoped_statuses[0].status_code == "A6"
|
|
assert result.unscoped_statuses[0].classification == "rejected"
|