Nora
142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
"""Direct tests for the ``handle_277ca`` handler (SP27 Task 4).
|
|
|
|
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: ``claim.rejected_after_remit`` audit emission (when a 277CA
|
|
rejects a claim already matched to a remit) is deferred to SP27
|
|
Task 13. For now we lock the existing claim.payer_rejected behavior
|
|
only.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cyclone import db
|
|
from cyclone.audit_log import AuditEvent
|
|
from cyclone.handlers.handle_277ca import handle
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
|
|
ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_277ca.txt"
|
|
REJECTED = Path(__file__).parent / "fixtures" / "minimal_277ca_rejected_only.txt"
|
|
|
|
|
|
def _seed_claim(claim_id: str, pcn: str) -> None:
|
|
with db.SessionLocal()() as s:
|
|
from cyclone.db import Claim
|
|
s.add(Claim(
|
|
id=claim_id, batch_id="BATCH-1",
|
|
patient_control_number=pcn, charge_amount=100.00,
|
|
))
|
|
s.commit()
|
|
|
|
|
|
def test_handle_277ca_persists_ack_with_classification_counts():
|
|
"""Happy path: persist a 277CA ack row with the right counts.
|
|
|
|
The minimal_277ca.txt fixture has 3 claim statuses:
|
|
- CLAIM001 → A3:19 → accepted
|
|
- CLAIM002 → A6:19 → rejected
|
|
- CLAIM003 → A8:19 → pended
|
|
|
|
So accepted=1, rejected=1, pended=1, paid=0.
|
|
"""
|
|
text = ACCEPTED.read_text()
|
|
parser_used, claim_count = handle(text, source_file=ACCEPTED.name)
|
|
assert parser_used == "parse_277ca"
|
|
assert claim_count == 3
|
|
|
|
# Lock persistence: a row was added to two77ca_acks.
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import Two77caAck
|
|
rows = (
|
|
session.query(Two77caAck)
|
|
.filter_by(source_batch_id="277CA-000000123")
|
|
.all()
|
|
)
|
|
assert len(rows) == 1
|
|
row = rows[0]
|
|
assert row.control_number == "000000123"
|
|
assert row.accepted_count == 1
|
|
assert row.rejected_count == 1
|
|
assert row.pended_count == 1
|
|
assert row.paid_count == 0
|
|
|
|
|
|
def test_handle_277ca_accepted_only_no_inbox_change():
|
|
"""A 277CA with only accepted statuses (no rejections) leaves the
|
|
inbox unchanged — no claim.payer_rejected audit row, no
|
|
payer_rejected_at stamp on any claim.
|
|
"""
|
|
text = ACCEPTED.read_text()
|
|
# Seed all 3 claims so the 277CA can match them by PCN.
|
|
_seed_claim("c1", "CLAIM001")
|
|
_seed_claim("c2", "CLAIM002")
|
|
_seed_claim("c3", "CLAIM003")
|
|
|
|
handle(text, source_file=ACCEPTED.name)
|
|
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import Claim
|
|
# CLAIM001 (accepted A3) — no payer_rejected_at
|
|
c1 = session.get(Claim, "c1")
|
|
assert c1.payer_rejected_at is None
|
|
# CLAIM002 (rejected A6) — stamped
|
|
c2 = session.get(Claim, "c2")
|
|
assert c2.payer_rejected_at is not None
|
|
assert c2.payer_rejected_status_code == "A6"
|
|
# CLAIM003 (pended A8) — no payer_rejected_at (A8 is pended, not rejected)
|
|
c3 = session.get(Claim, "c3")
|
|
assert c3.payer_rejected_at is None
|
|
# Activity-event audit row exists for the rejected one (c2)
|
|
from sqlalchemy import select
|
|
events = session.execute(
|
|
select(db.AuditLog.__table__.c.event_type).where(
|
|
db.AuditLog.__table__.c.entity_id == "c2"
|
|
)
|
|
).all()
|
|
event_types = [e[0] for e in events]
|
|
assert "claim.payer_rejected" in event_types
|
|
|
|
|
|
def test_handle_277ca_rejects_only_one_claim():
|
|
"""The rejected_only fixture has one A7:19 for CLAIM099.
|
|
Seed a matching claim and verify only it gets stamped.
|
|
"""
|
|
_seed_claim("c_rejected", "CLAIM099")
|
|
_seed_claim("c_unrelated", "OTHERPCN")
|
|
|
|
text = REJECTED.read_text()
|
|
parser_used, claim_count = handle(text, source_file=REJECTED.name)
|
|
assert parser_used == "parse_277ca"
|
|
assert claim_count == 1 # one claim_status in the fixture
|
|
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import Claim, Two77caAck
|
|
# The matching claim is stamped as rejected.
|
|
c_rej = session.get(Claim, "c_rejected")
|
|
assert c_rej.payer_rejected_at is not None
|
|
assert c_rej.payer_rejected_status_code == "A7"
|
|
# The unrelated claim is untouched.
|
|
c_unr = session.get(Claim, "c_unrelated")
|
|
assert c_unr.payer_rejected_at is None
|
|
# The two77ca_ack row was persisted with the rejected count.
|
|
rows = (
|
|
session.query(Two77caAck)
|
|
.filter_by(source_batch_id="277CA-000000789")
|
|
.all()
|
|
)
|
|
assert len(rows) == 1
|
|
assert rows[0].rejected_count == 1
|
|
assert rows[0].accepted_count == 0
|
|
|
|
|
|
def test_handle_277ca_raises_value_error_on_bad_x12():
|
|
"""Garbage that tokenize may accept but parse_277ca will reject."""
|
|
bad = "ISA*00*bad~ST*277CA*0001~SE*2*0001~IEA*0*0~"
|
|
with pytest.raises((CycloneParseError, ValueError)):
|
|
handle(bad, source_file="bad.277ca")
|