tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9bade2429c91a552939c79ab0dd35c89dffd6b75
cyclone/backend/tests/test_reconcile.py
T

656 lines
26 KiB
Python
Raw Blame History

"""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 == "pcn-exact"
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="pcn-exact", 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="pcn-exact", 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_reconcile_raise_in_session_leaves_prior_commits_alone(fixture_835):
"""A ``reconcile.run`` raise inside an open session does not damage
previously-committed rows.
The test seeds a batch + remit, commits them in one session, then
calls ``reconcile.run`` in a fresh session with ``match`` monkey-
patched to raise. The pre-existing rows must survive — they're on
disk from the prior commit, separate from the rolling-back
session. This pins the unit-level invariant that ``reconcile.run``
itself never commits and never silently mutates rows outside the
session it was given; it is the *caller's* responsibility (now
``CycloneStore.add`` per SP27 Task 10) to control the transaction
boundary.
"""
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
# --- SP31: content-keys fallback matcher (pure helper) -----------------------
def _make_remit_for_keys(remit_id: str, pcn: str, charge, npi: str):
"""Tiny shim with the three fields _content_keys_match reads."""
from cyclone.reconcile import _RemitLike # protocol not enforced at runtime
return type("R", (), {
"id": remit_id,
"payer_claim_control_number": pcn,
"total_charge_amount": charge,
"rendering_provider_npi": npi,
})()
def _make_claim_for_keys(claim_id: str, pcn: str, charge, npi: str):
return type("C", (), {
"id": claim_id,
"patient_control_number": pcn,
"total_charge": charge,
"rendering_provider_npi": npi,
})()
def test_content_keys_pcn_plus_charge_matches():
"""PCN + charge match (NPI mismatched) → True."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "")
c = _make_claim_for_keys("c1", "ABC", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c) is True
def test_content_keys_pcn_plus_npi_matches():
"""PCN + NPI match (charge mismatched) → True."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "ABC", Decimal("200.00"), "1111111111")
assert _content_keys_match(r, c) is True
def test_content_keys_charge_plus_npi_matches():
"""Charge + NPI match (PCN mismatched) → True."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "XYZ", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c) is True
def test_content_keys_all_three_match():
"""All 3 match → True."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "ABC", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c) is True
def test_content_keys_only_pcn_does_not_match():
"""Only PCN matches (charge + NPI both wrong) → False."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "ABC", Decimal("200.00"), "2222222222")
assert _content_keys_match(r, c) is False
def test_content_keys_only_charge_does_not_match():
"""Only charge matches (PCN + NPI both wrong) → False."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "XYZ", Decimal("100.00"), "2222222222")
assert _content_keys_match(r, c) is False
def test_content_keys_none_match():
"""All 3 fields disagree → False."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "1111111111")
c = _make_claim_for_keys("c1", "XYZ", Decimal("200.00"), "2222222222")
assert _content_keys_match(r, c) is False
def test_content_keys_charge_tolerance_is_one_cent():
"""Charge differs by exactly $0.005 (rounding) → True ($0.01 tolerance)."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.005"), "1111111111")
c = _make_claim_for_keys("c1", "ABC", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c) is True
def test_content_keys_charge_differs_by_two_cents_no_match():
"""Charge differs by $0.02 → False (outside tolerance)."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
# PCN and NPI deliberately differ so the only potential match is charge.
# Charge is outside the $0.01 tolerance → 0 of 3 match → False.
r = _make_remit_for_keys("r1", "ABC", Decimal("100.02"), "1111111111")
c = _make_claim_for_keys("c1", "XYZ", Decimal("100.00"), "2222222222")
assert _content_keys_match(r, c) is False
def test_content_keys_empty_npi_does_not_count():
"""Empty NPI on remit counts as not-matched (falls back to PCN+charge rule)."""
from cyclone.reconcile import _content_keys_match
from decimal import Decimal
r = _make_remit_for_keys("r1", "ABC", Decimal("100.00"), "")
c = _make_claim_for_keys("c1", "ABC", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c) is True # PCN + charge still match
c2 = _make_claim_for_keys("c2", "XYZ", Decimal("100.00"), "1111111111")
assert _content_keys_match(r, c2) is False # only charge, no PCN
# --- SP31: content-keys fallback matcher (DB helper) ------------------------
def test_score_fallback_unique_match_returns_claim_id(db_session, make_claim, make_remit):
"""One claim in pool matches 2-of-3 → returns that claim."""
from decimal import Decimal
from datetime import date, timedelta
from cyclone.reconcile import _score_fallback_candidates
c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 1))
c2 = make_claim(patient_control_number="XYZ", total_charge=Decimal("999.99"),
rendering_provider_npi="2222222222",
service_date_from=date(2026, 6, 1))
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=date(2026, 6, 5)) # ±30-day window
assert _score_fallback_candidates(db_session, r) == c1.id
def test_score_fallback_zero_matches_returns_none(db_session, make_claim, make_remit):
"""No claim in pool matches 2-of-3 → returns None."""
from decimal import Decimal
from datetime import date
from cyclone.reconcile import _score_fallback_candidates
make_claim(patient_control_number="XYZ", total_charge=Decimal("999.99"),
rendering_provider_npi="2222222222",
service_date_from=date(2026, 6, 1))
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=date(2026, 6, 5))
assert _score_fallback_candidates(db_session, r) is None
def test_score_fallback_two_matches_returns_none(db_session, make_claim, make_remit):
"""Two claims in pool both match 2-of-3 → returns None (ambiguous)."""
from decimal import Decimal
from datetime import date
from cyclone.reconcile import _score_fallback_candidates
# Both claims have same PCN and same charge — both would match PCN+charge.
make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 1))
make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 2))
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=date(2026, 6, 5))
assert _score_fallback_candidates(db_session, r) is None
def test_score_fallback_skips_already_matched(db_session, make_claim, make_remit):
"""Claim with matched_remittance_id set → excluded from pool."""
from decimal import Decimal
from datetime import date
from cyclone.reconcile import _score_fallback_candidates
c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 1),
matched_remittance_id="some-other-remit-id")
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=date(2026, 6, 5))
assert _score_fallback_candidates(db_session, r) is None # only candidate was excluded
def test_score_fallback_skips_terminal_state(db_session, make_claim, make_remit):
"""Claim in PAID/DENIED/REJECTED/REVERSED/RECONCILED → excluded from pool."""
from decimal import Decimal
from datetime import date
from cyclone.db import ClaimState
from cyclone.reconcile import _score_fallback_candidates
make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 1),
state=ClaimState.PAID)
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=date(2026, 6, 5))
assert _score_fallback_candidates(db_session, r) is None
def test_score_fallback_widened_window_catches_late_remit(db_session, make_claim, make_remit):
"""Remit 25 days after claim → still in pool (30-day window)."""
from decimal import Decimal
from datetime import date, timedelta
from cyclone.reconcile import _score_fallback_candidates
claim_date = date(2026, 6, 1)
remit_date = claim_date + timedelta(days=25)
c1 = make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=claim_date)
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=remit_date)
assert _score_fallback_candidates(db_session, r) == c1.id
def test_score_fallback_outside_window_excluded(db_session, make_claim, make_remit):
"""Remit 60 days after claim → excluded (outside 30-day window)."""
from decimal import Decimal
from datetime import date, timedelta
from cyclone.reconcile import _score_fallback_candidates
claim_date = date(2026, 6, 1)
remit_date = claim_date + timedelta(days=60)
make_claim(patient_control_number="ABC", total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=claim_date)
r = make_remit(payer_claim_control_number="ABC",
total_charge_amount=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date=remit_date)
assert _score_fallback_candidates(db_session, r) is None
# --- SP31 Task 3 fix: Remittance NPI lives in raw_json ----------------------
def test_content_keys_npi_from_remit_raw_json(db_session, make_claim):
"""NPI on the Remit is read from ``raw_json`` when the attribute is absent.
The 835 parser stores rendering_provider_npi on ``Remittance.raw_json``
(the ORM has no dedicated column) — reconcile.run() hands a raw
``Remittance`` instance to ``_content_keys_match``, so the helper must
fall through ``getattr(...rendering_provider_npi...)`` to the raw_json
read. This test pins that path with a real ORM row (no transient
``setattr`` shadowing the production layout).
Configuration: PCN mismatched, charge matches, NPI matches via raw_json
→ 2 of 3 keys agree → True.
"""
from decimal import Decimal
from datetime import date, datetime, timezone
from cyclone.db import Remittance
from cyclone.reconcile import _content_keys_match
claim = make_claim(
patient_control_number="ABC",
total_charge=Decimal("100.00"),
rendering_provider_npi="1111111111",
service_date_from=date(2026, 6, 1),
)
# Real Remittance ORM instance — NPI lives ONLY in raw_json, not as a
# transient attribute. (The 835 parser populates raw_json that way;
# the make_remit factory's transient ``rendering_provider_npi`` setattr
# would shadow the production read.)
remit = Remittance(
id="real-remit-1",
batch_id="test-batch",
payer_claim_control_number="OTHER-PCN", # PCN: won't match "ABC"
status_code="1",
total_charge=Decimal("100.00"), # charge: matches
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
raw_json={"rendering_provider_npi": "1111111111"}, # NPI: matches
)
db_session.add(remit)
db_session.flush()
assert _content_keys_match(remit, claim) is True # charge + NPI = 2 of 3
Reference in New Issue View Git Blame Copy Permalink