Tyler
6a65c2b750
Three spec-bug fixes from T10 implementation: - Match.claim_id: drop UNIQUE constraint (T4 schema error) — reversals add a 2nd row per claim (audit trail). Add explicit non-unique index ix_matches_claim_id to preserve query performance. Mirrored in ORM and migration 0001_initial.sql. - test_run_orphan_remit_leaves_claim_unmatched: claim PCN-A does not match remit PCN-NEW, so unmatched_claims must be 1, not 0. - test_run_reversal_flips_paid_to_reversed: claim service_date_from was 9 days before reversal remit service_date (outside default 7-day window) so no match occurred; changed to 5 days apart so match() picks the claim. - test_match_unique_per_claim: in test_db_models.py asserted the now-removed UNIQUE behavior; renamed and inverted to assert the audit-trail design (two Match rows per claim allowed).
361 lines
13 KiB
Python
361 lines
13 KiB
Python
"""Tests for the pure reconciliation functions.
|
|
|
|
These tests use lightweight stand-in objects (namedtuples) rather than
|
|
SQLAlchemy ORM instances, so the reconcile module has zero DB dependency.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
from typing import Optional
|
|
|
|
import pytest
|
|
|
|
from cyclone import db, reconcile
|
|
from cyclone.db import (
|
|
ActivityEvent,
|
|
Batch,
|
|
Claim,
|
|
ClaimState,
|
|
Match,
|
|
Remittance,
|
|
)
|
|
from cyclone.reconcile import Match as ReconcileMatch
|
|
|
|
|
|
# --- Stand-in types -----------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FakeClaim:
|
|
id: str
|
|
patient_control_number: str
|
|
service_date_from: Optional[date]
|
|
state: str = "submitted"
|
|
matched_remittance_id: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class FakeRemit:
|
|
id: str
|
|
payer_claim_control_number: str
|
|
service_date: Optional[date]
|
|
is_reversal: bool = False
|
|
|
|
|
|
# --- match() ------------------------------------------------------------------
|
|
|
|
|
|
def test_match_same_pcn_within_window_matches():
|
|
claims = [FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1))]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", date(2026, 6, 2))]
|
|
matches = reconcile.match(claims, remits, window_days=7)
|
|
assert len(matches) == 1
|
|
assert matches[0].claim.id == "CLM-1"
|
|
assert matches[0].remittance.id == "CLP-1"
|
|
assert matches[0].strategy == "auto"
|
|
|
|
|
|
def test_match_dates_outside_window_unmatched():
|
|
claims = [FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1))]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", date(2026, 7, 1))] # 30d apart
|
|
matches = reconcile.match(claims, remits, window_days=7)
|
|
assert matches == []
|
|
|
|
|
|
def test_match_different_pcn_unmatched():
|
|
claims = [FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1))]
|
|
remits = [FakeRemit("CLP-1", "PCN-B", date(2026, 6, 1))]
|
|
matches = reconcile.match(claims, remits)
|
|
assert matches == []
|
|
|
|
|
|
def test_match_multiple_claims_same_pcn_picks_closest_date():
|
|
claims = [
|
|
FakeClaim("CLM-1", "PCN-A", date(2026, 5, 1)),
|
|
FakeClaim("CLM-2", "PCN-A", date(2026, 6, 1)), # closer
|
|
]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", date(2026, 6, 2))]
|
|
matches = reconcile.match(claims, remits)
|
|
assert len(matches) == 1
|
|
assert matches[0].claim.id == "CLM-2"
|
|
|
|
|
|
def test_match_multiple_remits_same_pcn_each_gets_own_claim():
|
|
claims = [
|
|
FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1)),
|
|
FakeClaim("CLM-2", "PCN-A", date(2026, 6, 15)),
|
|
]
|
|
remits = [
|
|
FakeRemit("CLP-1", "PCN-A", date(2026, 6, 2)),
|
|
FakeRemit("CLP-2", "PCN-A", date(2026, 6, 16)),
|
|
]
|
|
matches = reconcile.match(claims, remits)
|
|
assert len(matches) == 2
|
|
claim_ids = {m.claim.id for m in matches}
|
|
assert claim_ids == {"CLM-1", "CLM-2"}
|
|
|
|
|
|
def test_match_skips_already_matched_claims():
|
|
claims = [FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1), matched_remittance_id="OLD")]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", date(2026, 6, 1))]
|
|
matches = reconcile.match(claims, remits)
|
|
assert matches == []
|
|
|
|
|
|
def test_match_falls_back_to_first_claim_when_no_service_date():
|
|
"""When the remit has no service_date, match the first claim for that PCN."""
|
|
claims = [
|
|
FakeClaim("CLM-1", "PCN-A", None),
|
|
FakeClaim("CLM-2", "PCN-A", None),
|
|
]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", None)]
|
|
matches = reconcile.match(claims, remits)
|
|
assert len(matches) == 1
|
|
assert matches[0].claim.id == "CLM-1"
|
|
|
|
|
|
def test_match_reversal_flag_propagates():
|
|
claims = [FakeClaim("CLM-1", "PCN-A", date(2026, 6, 1))]
|
|
remits = [FakeRemit("CLP-1", "PCN-A", date(2026, 6, 1), is_reversal=True)]
|
|
matches = reconcile.match(claims, remits)
|
|
assert matches[0].is_reversal is True
|
|
|
|
|
|
# ApplyIntent + apply_payment / apply_reversal / split_unmatched ----------------
|
|
|
|
|
|
def test_apply_payment_full_amount_returns_paid():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.SUBMITTED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None)
|
|
intent = reconcile.apply_payment(claim, remit, charge=Decimal("100"), paid=Decimal("100"))
|
|
assert intent.new_state == ClaimState.PAID
|
|
assert intent.activity_kind == "reconcile"
|
|
assert intent.skipped is False
|
|
|
|
|
|
def test_apply_payment_partial_amount_returns_partial():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.SUBMITTED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None)
|
|
intent = reconcile.apply_payment(claim, remit, charge=Decimal("100"), paid=Decimal("60"))
|
|
assert intent.new_state == ClaimState.PARTIAL
|
|
|
|
|
|
def test_apply_payment_zero_paid_returns_received():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.SUBMITTED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None)
|
|
intent = reconcile.apply_payment(claim, remit, charge=Decimal("100"), paid=Decimal("0"))
|
|
assert intent.new_state == ClaimState.RECEIVED
|
|
|
|
|
|
def test_apply_payment_status_4_returns_denied():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.SUBMITTED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None)
|
|
intent = reconcile.apply_payment(claim, remit, charge=Decimal("100"), paid=Decimal("0"),
|
|
status_code="4")
|
|
assert intent.new_state == ClaimState.DENIED
|
|
|
|
|
|
def test_apply_payment_already_paid_is_noop():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.PAID.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None)
|
|
intent = reconcile.apply_payment(claim, remit, charge=Decimal("100"), paid=Decimal("100"))
|
|
assert intent.skipped is True
|
|
assert intent.activity_kind == "invalid_state"
|
|
assert intent.new_state is None
|
|
|
|
|
|
def test_apply_reversal_of_paid_returns_reversed_with_prior_state():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.PAID.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None, is_reversal=True)
|
|
intent = reconcile.apply_reversal(claim, remit)
|
|
assert intent.new_state == ClaimState.REVERSED
|
|
assert intent.prior_claim_state == ClaimState.PAID
|
|
assert intent.activity_kind == "reversal"
|
|
assert intent.skipped is False
|
|
|
|
|
|
def test_apply_reversal_of_partial_works_same_way():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.PARTIAL.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None, is_reversal=True)
|
|
intent = reconcile.apply_reversal(claim, remit)
|
|
assert intent.new_state == ClaimState.REVERSED
|
|
assert intent.prior_claim_state == ClaimState.PARTIAL
|
|
|
|
|
|
def test_apply_reversal_of_denied_is_noop():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.DENIED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None, is_reversal=True)
|
|
intent = reconcile.apply_reversal(claim, remit)
|
|
assert intent.skipped is True
|
|
assert intent.activity_kind == "reversal_skipped"
|
|
|
|
|
|
def test_apply_reversal_of_already_reversed_is_noop():
|
|
claim = FakeClaim("CLM-1", "PCN-A", None, state=ClaimState.REVERSED.value)
|
|
remit = FakeRemit("CLP-1", "PCN-A", None, is_reversal=True)
|
|
intent = reconcile.apply_reversal(claim, remit)
|
|
assert intent.skipped is True
|
|
|
|
|
|
def test_split_unmatched_partitions_by_match_set():
|
|
claims = [FakeClaim("CLM-1", "A", None), FakeClaim("CLM-2", "B", None)]
|
|
remits = [FakeRemit("CLP-1", "A", None), FakeRemit("CLP-2", "C", None)]
|
|
matches = [ReconcileMatch(claim=claims[0], remittance=remits[0], strategy="auto", is_reversal=False)]
|
|
unmatched_claims, unmatched_remits = reconcile.split_unmatched(claims, remits, matches)
|
|
assert [c.id for c in unmatched_claims] == ["CLM-2"]
|
|
assert [r.id for r in unmatched_remits] == ["CLP-2"]
|
|
|
|
|
|
# --- reconcile.run() integration tests ----------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def fixture_835(tmp_path, monkeypatch):
|
|
"""Set up a clean DB and return an empty helper for building remits."""
|
|
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/reconcile.db")
|
|
db._reset_for_tests()
|
|
db.init_db()
|
|
yield
|
|
db._reset_for_tests()
|
|
|
|
|
|
def _make_batch(s, batch_id="b1", kind="835"):
|
|
s.add(Batch(
|
|
id=batch_id, kind=kind, input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
s.flush()
|
|
|
|
|
|
def _make_claim(s, claim_id="CLM-1", pcn="PCN-A", charge="100.00",
|
|
service_date=None, state=ClaimState.SUBMITTED):
|
|
s.add(Claim(
|
|
id=claim_id, batch_id="b1", patient_control_number=pcn,
|
|
service_date_from=service_date, charge_amount=Decimal(charge),
|
|
state=state,
|
|
))
|
|
s.flush()
|
|
|
|
|
|
def _make_remit(s, remit_id="CLP-1", pcn="PCN-A", status="1",
|
|
charge="100.00", paid="100.00", service_date=None,
|
|
is_reversal=False):
|
|
rid = remit_id + ":b1xxxxxx"
|
|
s.add(Remittance(
|
|
id=rid, batch_id="b1", payer_claim_control_number=pcn,
|
|
status_code=status, total_charge=Decimal(charge), total_paid=Decimal(paid),
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
service_date=service_date, is_reversal=is_reversal,
|
|
))
|
|
s.flush()
|
|
return rid
|
|
|
|
|
|
def test_run_matches_and_updates_state(fixture_835):
|
|
with db.SessionLocal()() as s:
|
|
_make_batch(s)
|
|
_make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 1))
|
|
_make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00", date(2026, 6, 1))
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
from cyclone import reconcile
|
|
result = reconcile.run(s, "b1")
|
|
s.commit()
|
|
|
|
assert result.matched == 1
|
|
assert result.unmatched_remittances == 0
|
|
assert result.unmatched_claims == 0
|
|
|
|
with db.SessionLocal()() as s:
|
|
claim = s.get(Claim, "CLM-1")
|
|
assert claim.state == ClaimState.PAID
|
|
assert claim.matched_remittance_id == "CLP-1:b1xxxxxx"
|
|
|
|
|
|
def test_run_orphan_remit_leaves_claim_unmatched(fixture_835):
|
|
with db.SessionLocal()() as s:
|
|
_make_batch(s)
|
|
_make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 1))
|
|
_make_remit(s, "CLP-2", "PCN-NEW", "1", "50.00", "50.00", date(2026, 6, 5))
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
from cyclone import reconcile
|
|
result = reconcile.run(s, "b1")
|
|
s.commit()
|
|
|
|
assert result.matched == 0
|
|
assert result.unmatched_remittances == 1
|
|
# Spec-bug fix (T10): claim PCN-A does not match remit PCN-NEW, so the
|
|
# claim stays in the unmatched bucket. Original spec asserted 0.
|
|
assert result.unmatched_claims == 1
|
|
|
|
|
|
def test_run_reversal_flips_paid_to_reversed(fixture_835):
|
|
with db.SessionLocal()() as s:
|
|
_make_batch(s)
|
|
# Spec-bug fix (T10): dates are 5 days apart so match() picks this claim
|
|
# (default window_days=7). Original spec used 2026-06-01 vs 2026-06-10
|
|
# (9 days, outside window) which produced no match.
|
|
_make_claim(s, "CLM-1", "PCN-A", "100.00", date(2026, 6, 5),
|
|
state=ClaimState.PAID)
|
|
# Match row already exists from prior reconcile.
|
|
s.add(Match(
|
|
claim_id="CLM-1", remittance_id="CLP-OLD:b1xxxxxx",
|
|
strategy="auto", matched_at=datetime.now(timezone.utc),
|
|
is_reversal=False,
|
|
))
|
|
s.flush()
|
|
_make_remit(s, "CLP-REV", "PCN-A", "22", "100.00", "-100.00",
|
|
date(2026, 6, 10), is_reversal=True)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
from cyclone import reconcile
|
|
result = reconcile.run(s, "b1")
|
|
s.commit()
|
|
|
|
assert result.matched == 1
|
|
with db.SessionLocal()() as s:
|
|
claim = s.get(Claim, "CLM-1")
|
|
assert claim.state == ClaimState.REVERSED
|
|
# Match rows: original + reversal.
|
|
matches = s.query(Match).filter(Match.claim_id == "CLM-1").all()
|
|
assert len(matches) == 2
|
|
reversal_match = [m for m in matches if m.is_reversal][0]
|
|
assert reversal_match.prior_claim_state == ClaimState.PAID
|
|
|
|
|
|
def test_run_failed_reconcile_writes_activity_event(fixture_835):
|
|
"""If reconcile crashes, the batch + remittances stay; activity event records failure."""
|
|
with db.SessionLocal()() as s:
|
|
_make_batch(s)
|
|
_make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00")
|
|
s.commit()
|
|
|
|
# Monkeypatch reconcile.match to raise.
|
|
from cyclone import reconcile
|
|
real_match = reconcile.match
|
|
def boom(*a, **kw): raise RuntimeError("synthetic fault")
|
|
reconcile.match = boom
|
|
try:
|
|
with db.SessionLocal()() as s:
|
|
try:
|
|
reconcile.run(s, "b1")
|
|
s.commit()
|
|
except RuntimeError:
|
|
pass # caller is responsible for catch; reconcile.run itself shouldn't raise
|
|
finally:
|
|
reconcile.match = real_match
|
|
|
|
# Batch + remit still present.
|
|
with db.SessionLocal()() as s:
|
|
b = s.get(Batch, "b1")
|
|
assert b is not None
|
|
r = s.query(Remittance).first()
|
|
assert r is not None
|