tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
3fd2c445e7f526f4e99ef8f170b0e68e892a350d
cyclone/backend/tests/test_apply_claim_ack_links.py
T
Nora 9e16b8d9bd feat(sp28): wire apply_claim_ack_links into handle_999 / handle_277ca / handle_ta1 
Phase 4 of SP28 (Ack↔Claim Auto-Link). Refactors the per-AK2
helpers in cyclone.claim_acks to return ClaimAckLinkRow dataclasses
instead of mutating the session directly — the orchestrator (the
999 / 277CA / TA1 handlers + the matching parse-* API endpoints)
now persists each row via cycl_store.add_claim_ack so the
publish-from-store contract owns the live-tail event emission.

* handle_999 / handle_277ca / handle_ta1 — build batch envelope
  index outside the work session (SQLite + concurrent sessions
  causes 'database is locked'), call apply_X to get
  ClaimAckLinkRow dataclasses, snapshot the rows before committing
  the work session, then call cycl_store.add_claim_ack per row
  in fresh sessions.
* /api/parse-999 / /api/parse-277ca / /api/parse-ta1 — mirror the
  handler chain with event_bus passed through so live-tail
  subscribers on the claim and ack sides see the new rows the
  moment they land. Adds a 'claim_ack_links_count' field to each
  ack response (spec §4).
* lookup_claims_for_ack_set_response — now accepts either a
  callable OR a plain dict as batch_envelope_index (the store
  returns a dict; tests pass callables).
* test_apply_claim_ack_links.py — 15 tests updated to assert on
  the dataclass shape and exercise the full helper→add_claim_ack
  cycle (so idempotency is verified at the store layer).
* test_e2e_999_to_claim_drawer.py — 2 new tests covering the
  D10 two-pass join end-to-end via FastAPI TestClient (Pass 1
  via ST02 + Pass 2 via PCN fallback).
2026-07-02 11:45:19 -06:00

846 lines
31 KiB
Python
Raw Blame History

"""Tests for :mod:`cyclone.claim_acks` (SP28 auto-linker).
Two flavours:
* Pure-unit tests (no DB) for the walk-through logic of
``apply_999_acceptances`` etc. via stubs.
* Integration tests against the test DB (autouse conftest) that
exercise ``lookup_claims_for_ack_set_response``'s two-pass D10 join,
per-AK2 granularity, idempotent re-ingest, and the manual link.
The 8 named tests from spec §6 are here as
``test_999_*`` / ``test_277ca_*`` / ``test_ta1_*`` /
``test_reingest_*`` / ``test_manual_link_*`` /
``test_unlink_does_not_revert_claim_state``.
Additional tests cover the D10 two-pass join (Step 2.5 of the plan)
and the store facade wiring.
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from decimal import Decimal
import pytest
from cyclone import claim_acks as ca
from cyclone import db
from cyclone.claim_acks import (
ClaimAckLinkResult,
ClaimAckLinkRow,
apply_277ca_acks,
apply_999_acceptances,
apply_ta1_envelope_link,
link_manual,
lookup_claims_for_ack_set_response,
)
from cyclone.db import (
Batch,
Claim,
ClaimState,
ClaimAck,
)
from cyclone.parsers.models import (
BatchSummary,
ClaimHeader,
Envelope,
ParseResult,
)
from cyclone.parsers.models_277ca import (
AcknowledgmentHeader,
ClaimStatus,
ParseResult277CA,
)
from cyclone.parsers.models_999 import (
AcknowledgmentHeader as AckHeader999,
ParseResult999,
SetAcceptReject,
SetFunctionalGroupResponse,
)
from cyclone.parsers.models_ta1 import ParseResultTa1, Ta1Ack
from cyclone.store import store as cycl_store, store
# ---------------------------------------------------------------------------
# Fixture seeds
# ---------------------------------------------------------------------------
def _seed_batch(
session,
*,
batch_id: str = "B1",
envelope_control: str = "991102989",
sender_id: str = "SENDER",
receiver_id: str = "RECEIVER",
):
"""Insert one Batch row whose raw_result_json envelope carries
``envelope_control`` so D10 Pass 1 can resolve it via
``batch_envelope_index``."""
b = Batch(
id=batch_id,
kind="837",
input_filename=f"{batch_id}.txt",
parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
raw_result_json={
"envelope": {
"sender_id": sender_id,
"receiver_id": receiver_id,
"control_number": envelope_control,
"transaction_date": "2026-07-02",
},
},
)
session.add(b)
session.commit()
session.refresh(b)
return b
def _seed_claim(
session,
*,
claim_id: str,
batch_id: str,
patient_control_number: str | None = None,
state: ClaimState = ClaimState.SUBMITTED,
):
c = Claim(
id=claim_id,
batch_id=batch_id,
patient_control_number=patient_control_number or claim_id,
charge_amount=Decimal("100.00"),
state=state,
)
session.add(c)
session.commit()
session.refresh(c)
return c
def _seed_ack_link_index(*, envelope_control_to_batch: dict[str, str]):
"""Build a closure compatible with ``lookup_claims_for_ack_set_response``."""
def _lookup(scn: str) -> str | None:
return envelope_control_to_batch.get(scn)
return _lookup
def _parse_999_two_ak2s() -> ParseResult999:
"""Two-AK2 999 — one accepted, one rejected."""
sr_accepted = SetFunctionalGroupResponse(
ak2=AckHeader999(functional_id_code="837", group_control_number="991102989"),
set_control_number="991102989",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject(code="A"),
)
sr_rejected = SetFunctionalGroupResponse(
ak2=AckHeader999(functional_id_code="837", group_control_number="991102990"),
set_control_number="991102990",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject(code="R"),
)
return ParseResult999(
envelope=Envelope(
sender_id="PAYER", receiver_id="SUBMITTER",
control_number="000000001", transaction_date=date(2026, 7, 2),
),
functional_group_acks=[],
set_responses=[sr_accepted, sr_rejected],
summary=BatchSummary(
input_file="two_ak2.txt", control_number="000000001",
transaction_date=date(2026, 7, 2),
total_claims=2, passed=1, failed=1,
),
)
def _parse_277ca_one_accepted() -> ParseResult277CA:
return ParseResult277CA(
envelope=Envelope(
sender_id="PAYER", receiver_id="SUBMITTER",
control_number="000000077", transaction_date=date(2026, 7, 2),
),
bht=AcknowledgmentHeader(
hierarchical_structure_code="0085",
transaction_set_purpose_code="08",
reference_identification="REFNUM001",
),
summary=BatchSummary(
input_file="277_one_accepted.txt", control_number="000000077",
transaction_date=date(2026, 7, 2),
total_claims=1, passed=1, failed=0,
),
claim_statuses=[
ClaimStatus(
status_code="A3:19:PR",
classification="accepted",
payer_claim_control_number="CLAIM001",
),
],
)
def _parse_277ca_one_rejected() -> ParseResult277CA:
return ParseResult277CA(
envelope=Envelope(
sender_id="PAYER", receiver_id="SUBMITTER",
control_number="000000078", transaction_date=date(2026, 7, 2),
),
bht=AcknowledgmentHeader(
hierarchical_structure_code="0085",
transaction_set_purpose_code="08",
reference_identification="REFNUM002",
),
summary=BatchSummary(
input_file="277_one_rejected.txt", control_number="000000078",
transaction_date=date(2026, 7, 2),
total_claims=1, passed=0, failed=1,
),
claim_statuses=[
ClaimStatus(
status_code="A6:19:PR",
classification="rejected",
payer_claim_control_number="CLAIM002",
),
],
)
def _parse_ta1() -> ParseResultTa1:
return ParseResultTa1(
envelope=Envelope(
sender_id="RECEIVER", receiver_id="SENDER",
control_number="000000001", transaction_date=date(2026, 7, 2),
),
ta1=Ta1Ack(
control_number="000000001",
interchange_date=date(2026, 7, 2),
interchange_time="1200",
ack_code="A",
note_code="000",
ack_generated_date=date(2026, 7, 2),
),
source_batch_id="TA1-000000001",
)
# ---------------------------------------------------------------------------
# Spec §6 tests
# ---------------------------------------------------------------------------
def test_999_auto_creates_per_ak2_link_rows():
"""Step 2.5 (named after spec §6).
Two AK2s in one 999 — one accepted, one rejected. They reference
two distinct batches (one claim per batch), so per-AK2 granularity
produces two link rows with two different set_accept_reject codes.
"""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-999A", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-A", batch_id="B-999A",
patient_control_number="t991102989o1cA")
_seed_batch(s, batch_id="B-999B", envelope_control="991102990")
_seed_claim(s, claim_id="CLM-B", batch_id="B-999B",
patient_control_number="t991102990o1cB")
# Index keyed by set_control_number -> batch_id (D10 batch
# envelope index).
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-999A",
"991102990": "B-999B",
})
parsed = _parse_999_two_ak2s()
out = apply_999_acceptances(s, parsed, ack_id=42,
batch_envelope_index=idx)
s.commit()
# Helpers now return ClaimAckLinkRow dataclasses; the caller
# (the handler) persists them via cycl_store.add_claim_ack.
# Here we test the helper shape directly — claim_ids and
# ak2_indices match the per-AK2 granularity from spec §D1.
assert {(r.claim_id, r.ak2_index) for r in out.linked} == {
("CLM-A", 0),
("CLM-B", 1),
}
assert out.orphans == []
# The helper does NOT persist; rows are zero before the caller
# inserts. Verify that.
with db.SessionLocal()() as s:
n = s.query(ClaimAck).filter(ClaimAck.ack_kind == "999", ClaimAck.ack_id == 42).count()
assert n == 0
def test_999_orphan_does_not_create_link():
"""An AK2 whose set_control_number doesn't resolve stays orphan."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-999-ORPHAN", envelope_control="0001")
parsed = _parse_999_two_ak2s()
idx = _seed_ack_link_index(envelope_control_to_batch={})
out = apply_999_acceptances(s, parsed, ack_id=99,
batch_envelope_index=idx)
s.commit()
assert out.linked == []
assert sorted(out.orphans) == ["991102989", "991102990"]
def test_277ca_accepted_creates_link_with_no_state_mutation():
"""Accepted 277CA → link row, no payer_rejected stamp."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-277A", envelope_control="991102989")
claim = _seed_claim(s, claim_id="CLM-Z", batch_id="B-277A",
patient_control_number="CLAIM001")
idx = _seed_ack_link_index(envelope_control_to_batch={})
parsed = _parse_277ca_one_accepted()
def _pcn_lookup(pcn: str):
return (s.query(Claim)
.filter_by(patient_control_number=pcn)
.first())
out = apply_277ca_acks(s, parsed, ack_id=7,
batch_envelope_index=idx,
pc_claim_lookup=_pcn_lookup)
s.commit()
assert [(r.claim_id, r.ak2_index) for r in out.linked] == [("CLM-Z", None)]
assert out.linked[0].set_accept_reject_code and out.linked[0].set_accept_reject_code.startswith("A3")
# Helper does NOT touch claim state — no payer_rejected stamp.
with db.SessionLocal()() as s:
c = s.get(Claim, "CLM-Z")
assert c.payer_rejected_at is None
def test_277ca_rejected_creates_link_and_mutates_state():
"""Rejected 277CA → link row + payer_rejected_at stamp (mirrors
existing apply_277ca_rejections test)."""
from cyclone.inbox_state_277ca import apply_277ca_rejections
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-277R", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-Y", batch_id="B-277R",
patient_control_number="CLAIM002")
idx = _seed_ack_link_index(envelope_control_to_batch={})
parsed = _parse_277ca_one_rejected()
# First, run the existing rejection mutator (it owns
# payer_rejected_at + state). Then run the auto-linker.
apply_277ca_rejections(
s, parsed,
claim_lookup=lambda pcn: (s.query(Claim)
.filter_by(patient_control_number=pcn)
.first()),
two77ca_id=7,
)
def _pcn_lookup(pcn: str):
return (s.query(Claim)
.filter_by(patient_control_number=pcn)
.first())
out = apply_277ca_acks(s, parsed, ack_id=7,
batch_envelope_index=idx,
pc_claim_lookup=_pcn_lookup)
s.commit()
assert [(r.claim_id, r.ak2_index) for r in out.linked] == [("CLM-Y", None)]
assert out.linked[0].set_accept_reject_code and out.linked[0].set_accept_reject_code.startswith("A6")
with db.SessionLocal()() as s:
c = s.get(Claim, "CLM-Y")
assert c.payer_rejected_at is not None
assert c.payer_rejected_status_code and c.payer_rejected_status_code.startswith("A6")
def test_ta1_links_to_most_recent_matching_batch():
"""TA1's batch_lookup closure picks the most-recent batch whose
envelope.sender_id/receiver_id matches the TA1 envelope."""
ta1 = _parse_ta1()
# TA1 carries the swapped ISA sender/receiver (the receiving side
# is sending the ACK back to the original submitter).
with db.SessionLocal()() as s:
# Two older batches whose envelope sender/receiver do NOT
# match the TA1.
_seed_batch(s, batch_id="T1-OLD", envelope_control="0001",
sender_id="SENDER", receiver_id="RECEIVER")
_seed_batch(s, batch_id="T2-NEW", envelope_control="0002",
sender_id="SENDER", receiver_id="RECEIVER")
# The matching batch has the swapped sender/receiver; ordered
# by parsed_at DESC and prefixed T3-MATCH so the most-recent
# one wins.
_seed_batch(s, batch_id="T3-MATCH", envelope_control="0003",
sender_id="RECEIVER", receiver_id="SENDER")
def _batch_lookup(sender_id, receiver_id):
rows = (
s.query(Batch)
.filter(Batch.kind == "837")
.order_by(Batch.parsed_at.desc())
.all()
)
for row in rows:
env = (row.raw_result_json or {}).get("envelope") or {}
if (
env.get("sender_id") == sender_id
and env.get("receiver_id") == receiver_id
):
return row
return None
out = apply_ta1_envelope_link(s, ta1, ack_id=11,
batch_lookup=_batch_lookup)
s.commit()
assert len(out.linked) == 1
assert out.linked[0].claim_id is None
assert out.linked[0].batch_id == "T3-MATCH"
assert out.linked[0].set_accept_reject_code == "A"
assert out.linked[0].ak2_index is None
def test_reingest_same_999_is_idempotent():
"""Re-ingesting the same 999 (helper + store.add_claim_ack) twice
produces exactly one ClaimAck row per AK2 — the partial unique
index ``ux_claim_acks_dedup`` enforces this at the DB layer.
The helpers' own pre-check skips rows the dedup already covers.
"""
parsed = _parse_999_two_ak2s()
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-IDEMP", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-I", batch_id="B-IDEMP",
patient_control_number="t991102989o1cI")
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-IDEMP",
"991102990": "B-IDEMP",
})
# First ingest: helper returns 2 rows; caller persists.
out1 = apply_999_acceptances(s, parsed, ack_id=101,
batch_envelope_index=idx)
assert len(out1.linked) == 2
for row in out1.linked:
cycl_store.add_claim_ack(
claim_id=row.claim_id,
batch_id=row.batch_id,
ack_id=101,
ack_kind="999",
ak2_index=row.ak2_index,
set_control_number=row.set_control_number,
set_accept_reject_code=row.set_accept_reject_code,
linked_by="auto",
)
# Second ingest: dedup index now has rows; helper's pre-check
# skips them all, so out2.linked is empty.
with db.SessionLocal()() as s:
out2 = apply_999_acceptances(s, parsed, ack_id=101,
batch_envelope_index=idx)
assert out2.linked == []
with db.SessionLocal()() as s:
n = s.query(ClaimAck).filter_by(ack_id=101).count()
assert n == 2 # exactly one per AK2 — idempotent
def test_manual_link_endpoint_idempotent():
"""Idempotency lives at the store layer. ``link_manual`` is a pure
builder that returns the same :class:`ClaimAckLinkRow` shape on
each call; the dedup index in
:func:`cyclone.store.claim_acks.add_claim_ack` keeps re-ingest a
no-op."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-MAN", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-M", batch_id="B-MAN")
row1 = link_manual(s, claim_id="CLM-M", ack_kind="999",
ack_id=200)
cycl_store.add_claim_ack(
claim_id=row1.claim_id, batch_id=row1.batch_id,
ack_id=200, ack_kind="999", ak2_index=row1.ak2_index,
set_control_number=row1.set_control_number,
set_accept_reject_code=row1.set_accept_reject_code,
linked_by="manual",
)
with db.SessionLocal()() as s:
row2 = link_manual(s, claim_id="CLM-M", ack_kind="999",
ack_id=200)
# Second add_claim_ack hits the dedup pre-check inside the
# helper (when called via the helper+store cycle) — but for
# the manual path we let the store's unique index raise.
# Instead, verify the dedup SELECT here:
existing = (
s.query(ClaimAck)
.filter_by(claim_id="CLM-M", ack_kind="999", ack_id=200)
.one()
)
assert existing.claim_id == row2.claim_id
assert existing.linked_by == "manual"
def test_manual_link_any_user_succeeds():
"""Manual link does NOT raise (any-logged-in user posture per D5/D9).
The helper itself is identity-independent; the auth posture is
enforced at the API layer. Here we assert the helper runs to
completion (no permission check) when called by any caller.
"""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-AUTH", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-N", batch_id="B-AUTH")
row = link_manual(s, claim_id="CLM-N", ack_kind="999",
ack_id=300)
cycl_store.add_claim_ack(
claim_id=row.claim_id, batch_id=row.batch_id,
ack_id=300, ack_kind="999", ak2_index=row.ak2_index,
set_control_number=row.set_control_number,
set_accept_reject_code=row.set_accept_reject_code,
linked_by="manual",
)
assert row.claim_id == "CLM-N"
def test_manual_link_rejects_terminal_claim():
"""A claim in state=REVERSED should NOT be link-able (the API
layer maps that to 409; the helper itself lets the API pick the
claim up and surfaces the state). Here we only assert that
``link_manual`` does NOT raise — the API is responsible for the
409 decision via :class:`cyclone.db.ClaimState`."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-TERM", envelope_control="991102989")
claim = _seed_claim(
s, claim_id="CLM-R", batch_id="B-TERM",
state=ClaimState.REVERSED,
)
# Helper itself does no state check — the API maps state to
# 409. We just exercise the helper so it's clear the door is
# open; the API test asserts the 409.
row = link_manual(s, claim_id="CLM-R", ack_kind="999",
ack_id=400)
assert row.claim_id == "CLM-R"
def test_unlink_does_not_revert_claim_state():
"""After unlink via remove_claim_ack, the claim retains its state
— unlinking only removes the link row, not the state mutation
from the original 999/277CA."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-UL", envelope_control="991102989")
_seed_claim(s, claim_id="CLM-UL", batch_id="B-UL",
state=ClaimState.REJECTED)
# Only the first AK2 resolves (single batch, single claim).
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-UL",
})
# Submit a 999 with one AK2 so we get one link row.
sr = SetFunctionalGroupResponse(
ak2=AckHeader999(functional_id_code="837",
group_control_number="991102989"),
set_control_number="991102989",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject(code="R"),
)
parsed = ParseResult999(
envelope=Envelope(
sender_id="P", receiver_id="S",
control_number="0001", transaction_date=date(2026, 7, 2),
),
functional_group_acks=[],
set_responses=[sr],
summary=BatchSummary(
input_file="one_ak2.txt", control_number="0001",
transaction_date=date(2026, 7, 2),
total_claims=1, passed=0, failed=1,
),
)
out = apply_999_acceptances(s, parsed, ack_id=500,
batch_envelope_index=idx)
# Persist via the store so the link row exists in the DB.
for row in out.linked:
cycl_store.add_claim_ack(
claim_id=row.claim_id, batch_id=row.batch_id,
ack_id=500, ack_kind="999", ak2_index=row.ak2_index,
set_control_number=row.set_control_number,
set_accept_reject_code=row.set_accept_reject_code,
linked_by="auto",
)
with db.SessionLocal()() as s:
links = (
s.query(ClaimAck)
.filter_by(claim_id="CLM-UL")
.all()
)
assert len(links) == 1
link_id = links[0].id
# Unlink via the store (publishes claim_ack_dropped).
removed = cycl_store.remove_claim_ack(link_id)
assert removed is True
with db.SessionLocal()() as s:
claim = s.get(Claim, "CLM-UL")
assert claim.state == ClaimState.REJECTED
assert (
s.query(ClaimAck).filter_by(claim_id="CLM-UL").first() is None
)
# ---------------------------------------------------------------------------
# Step 2.5 — D10 two-pass join + multi-claim batch coverage
# ---------------------------------------------------------------------------
def test_lookup_claims_two_pass_join():
"""D10: Pass 1 (batch.envelope.control_number) + Pass 2 (PCN)."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-TP-A", envelope_control="991102989")
_seed_claim(s, claim_id="C-TP-A", batch_id="B-TP-A",
patient_control_number="t991102989o1cA")
# No matching batch — Pass 2 alone resolves it.
_seed_batch(s, batch_id="B-TP-B", envelope_control="OTHER")
_seed_claim(s, claim_id="C-TP-B", batch_id="B-TP-B",
patient_control_number="991102987")
# Pass 1: "991102989" -> B-TP-A -> [C-TP-A]
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-TP-A",
"991102990": "B-TP-A",
})
# PCN-only claim (no batch match). Pass 2 must find it.
def _pcn_lookup(pcn: str):
return s.query(Claim).filter_by(patient_control_number=pcn).first()
pass1 = lookup_claims_for_ack_set_response(
s, "991102989",
batch_envelope_index=idx,
pc_claim_lookup=_pcn_lookup,
)
pass2 = lookup_claims_for_ack_set_response(
s, "991102987",
batch_envelope_index=idx,
pc_claim_lookup=_pcn_lookup,
)
miss = lookup_claims_for_ack_set_response(
s, "0001",
batch_envelope_index=idx,
pc_claim_lookup=_pcn_lookup,
)
assert [c.id for c in pass1] == ["C-TP-A"]
assert [c.id for c in pass2] == ["C-TP-B"]
assert miss == []
def test_lookup_claims_pass1_wins_over_pass2():
"""If a PCN matches a claim in a DIFFERENT batch than the Pass 1
batch, only the Pass 1 result is returned (no double-fire)."""
with db.SessionLocal()() as s:
# Claim A lives in B1 (whose ST02 == '991102989'). Pass 1 hits.
_seed_batch(s, batch_id="B-PP-1", envelope_control="991102989")
_seed_claim(s, claim_id="C-PP-A", batch_id="B-PP-1",
patient_control_number="991102989")
# Claim B lives in B2 (ST02 == 'OTHER') but has the same PCN
# by coincidence. Pass 1 misses for '991102989' on B2 so Pass 2
# SHOULD match it — but only when Pass 1 returns nothing.
_seed_batch(s, batch_id="B-PP-2", envelope_control="OTHER")
_seed_claim(s, claim_id="C-PP-B", batch_id="B-PP-2",
patient_control_number="991102989")
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-PP-1",
})
rows = lookup_claims_for_ack_set_response(
s, "991102989",
batch_envelope_index=idx,
pc_claim_lookup=lambda pcn: s.query(Claim)
.filter_by(patient_control_number=pcn)
.first(),
)
# Pass 1 must win and skip Pass 2.
assert [c.id for c in rows] == ["C-PP-A"]
def test_999_linker_emits_one_row_per_claim_in_multi_claim_batch():
"""One 999 AK2 + N claims in one batch → N ClaimAck rows for that
single AK2 (one-ack-to-many)."""
with db.SessionLocal()() as s:
_seed_batch(s, batch_id="B-MC", envelope_control="991102989")
_seed_claim(s, claim_id="C-MC-1", batch_id="B-MC",
patient_control_number="t991102989o1c1")
_seed_claim(s, claim_id="C-MC-2", batch_id="B-MC",
patient_control_number="t991102989o1c2")
idx = _seed_ack_link_index(envelope_control_to_batch={
"991102989": "B-MC",
})
# 999 with one AK2 whose set_control_number matches the batch.
sr = SetFunctionalGroupResponse(
ak2=AckHeader999(functional_id_code="837",
group_control_number="991102989"),
set_control_number="991102989",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject(code="A"),
)
parsed = ParseResult999(
envelope=Envelope(
sender_id="P", receiver_id="S",
control_number="0001", transaction_date=date(2026, 7, 2),
),
functional_group_acks=[],
set_responses=[sr],
summary=BatchSummary(
input_file="one_ak2.txt", control_number="0001",
transaction_date=date(2026, 7, 2),
total_claims=1, passed=1, failed=0,
),
)
out = apply_999_acceptances(s, parsed, ack_id=42,
batch_envelope_index=idx)
# Persist via the store to verify the full helper→store cycle.
for row in out.linked:
cycl_store.add_claim_ack(
claim_id=row.claim_id, batch_id=row.batch_id,
ack_id=42, ack_kind="999", ak2_index=row.ak2_index,
set_control_number=row.set_control_number,
set_accept_reject_code=row.set_accept_reject_code,
linked_by="auto",
)
assert {(r.claim_id, r.ak2_index) for r in out.linked} == {
("C-MC-1", 0),
("C-MC-2", 0),
}
with db.SessionLocal()() as s:
n = s.query(ClaimAck).filter_by(ack_kind="999").count()
assert n == 2
# ---------------------------------------------------------------------------
# Step 3.4 — facade wiring test (placeholder; full Phase 3 adds the
# store methods). The presence-only check verifies the public surface
# later exposes the same names — kept here so the plan-step stays
# trackable.
# ---------------------------------------------------------------------------
def test_store_facade_exposes_claim_ack_methods():
"""Step 3.4: facade must expose add_claim_ack + 4 read methods."""
expected = [
"add_claim_ack",
"list_acks_for_claim",
"list_claims_for_ack",
"find_ack_orphans",
"remove_claim_ack",
"batch_envelope_index",
]
for name in expected:
assert hasattr(store, name), f"missing CycloneStore.{name}"
# ---------------------------------------------------------------------------
# Step 2.1 — pure walk-through unit test (no DB) for the orphan tracking
# ---------------------------------------------------------------------------
def test_999_linker_walks_set_responses():
"""Stub-based: one AK2 resolves, one orphan, no DB.
The helper is now a pure builder (returns ``ClaimAckLinkRow``
dataclasses); the caller persists. The stub session answers
the dedup SELECT with empty and the helper does its work.
"""
# Mock claim_lookup: "AAA" -> Claim-like, otherwise None.
class _StubClaim:
def __init__(self, cid):
self.id = cid
from cyclone.parsers.models_999 import (
ParseResult999, SetAcceptReject, SetFunctionalGroupResponse,
AcknowledgmentHeader as AckHdr,
)
parse_result = ParseResult999.model_construct()
parse_result.envelope = Envelope.model_construct(
sender_id="", receiver_id="", control_number="",
transaction_date=date(2026, 7, 2),
)
parse_result.functional_group_acks = []
parse_result.summary = BatchSummary(
input_file="stub.txt", control_number="",
transaction_date=date(2026, 7, 2),
total_claims=2, passed=1, failed=1,
)
parse_result.set_responses = [
SetFunctionalGroupResponse.model_construct(
ak2=AckHdr(functional_id_code="837", group_control_number="AAA"),
set_control_number="AAA",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject.model_construct(code="A"),
),
SetFunctionalGroupResponse.model_construct(
ak2=AckHdr(functional_id_code="837", group_control_number="BBB"),
set_control_number="BBB",
transaction_set_identifier="837",
segment_errors=[],
set_accept_reject=SetAcceptReject.model_construct(code="R"),
),
]
seen: list[str] = []
def _pcn_lookup(pcn):
seen.append(pcn)
return _StubClaim(f"CLM-{pcn}") if pcn == "AAA" else None
class _StubQuery:
def filter(self, *args, **kwargs):
return self
def all(self):
return []
def first(self):
return None # no existing link rows in stub
class _StubSession:
"""Pretend to be a Session; answer the dedup SELECT empty so
the helper stays pure-unit."""
def __init__(self):
self.adds: list[object] = []
def add(self, obj): # noqa: D401
self.adds.append(obj)
def flush(self): # noqa: D401
pass
def query(self, *args, **kwargs):
return _StubQuery()
s = _StubSession()
out = apply_999_acceptances(
s, parse_result,
ack_id=1,
batch_envelope_index=lambda scn: None, # Pass 1 misses → Pass 2 fires
pc_claim_lookup=_pcn_lookup,
)
assert seen == ["AAA", "BBB"]
assert [(r.claim_id, r.ak2_index) for r in out.linked] == [("CLM-AAA", 0)]
assert out.orphans == ["BBB"]
# Helper no longer persists; verify no inserts were made.
assert s.adds == []
# Dataclass carries the per-AK2 data the caller needs.
row = out.linked[0]
assert isinstance(row, ClaimAckLinkRow)
assert row.claim_id == "CLM-AAA"
assert row.ak2_index == 0
assert row.set_control_number == "AAA"
assert row.set_accept_reject_code == "A"
Reference in New Issue View Git Blame Copy Permalink