"""SP27 Task 11: matched-pair invariant. The matched pair is a denormalized FK pair — ``Claim.matched_remittance_id`` and ``Remittance.claim_id`` — that MUST agree in both directions at every moment. Three writers can mutate it: 1. ``CycloneStore.manual_match`` — operator override. SP27 Task 11 pins it writes BOTH sides in one transaction. 2. ``CycloneStore.manual_unmatch`` — operator reversal. SP27 Task 11 pins it clears BOTH sides in one transaction. 3. ``reconcile.run`` (inside ``CycloneStore.add`` post Task 10) — auto- match. Already pinned by the existing reconcile tests. The drift check (``check_matched_pair_drift``) catches pre-existing mismatches at startup and logs them at WARNING. Non-blocking. These tests pin the invariants directly. They are explicitly redundant with the integration tests in ``test_store_reconcile.py`` — the goal here is a focused regression pin so a future refactor that touches ``manual_match`` / ``manual_unmatch`` (e.g. to defer the symmetric write to a post-commit pass) doesn't silently break the invariant. """ from __future__ import annotations from datetime import date, datetime, timezone from decimal import Decimal import pytest from cyclone import db from cyclone.parsers.models import ( BatchSummary, BillingProvider, ClaimHeader, ClaimOutput, Envelope, Payer, ParseResult, Subscriber, ValidationReport, ) from cyclone.parsers.models_835 import ( BatchSummary as BatchSummary835, ClaimAdjustment, ClaimPayment, Envelope as Envelope835, FinancialInfo, ParseResult835, Payer835, Payee835, ReassociationTrace, ServicePayment, ) from cyclone.store import ( CycloneStore, check_matched_pair_drift, ) @pytest.fixture(autouse=True) def _setup(tmp_path, monkeypatch): monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db") db._reset_for_tests() db.init_db() yield db._reset_for_tests() def _build_claim(claim_id, pcn): """Build a single 837 claim + persist it through CycloneStore.add.""" from cyclone.store import BatchRecord837 co = ClaimOutput( claim_id=claim_id, control_number="0001", transaction_date=date(2026, 6, 19), billing_provider=BillingProvider(name="Test", npi="1234567890"), subscriber=Subscriber( first_name="Jane", last_name="Doe", member_id=pcn, ), payer=Payer(name="Test Payer", id="P1"), claim=ClaimHeader( claim_id=claim_id, total_charge=Decimal("100"), frequency_code="1", place_of_service="11", ), diagnoses=[], service_lines=[], validation=ValidationReport(passed=True, errors=[], warnings=[]), raw_segments=[], ) pr837 = ParseResult( envelope=Envelope( sender_id="S", receiver_id="R", control_number="0001", transaction_date=date(2026, 6, 19), ), claims=[co], summary=BatchSummary( input_file="c.txt", control_number="0001", transaction_date=date(2026, 6, 19), total_claims=1, passed=1, failed=0, ), ) CycloneStore().add(BatchRecord837( id="b-837", kind="837p", input_filename="c.txt", parsed_at=date(2026, 6, 19).isoformat() + "T12:00:00+00:00", result=pr837, )) def _build_remit(remit_id, pcn): """Build a single 835 remit + persist it through CycloneStore.add.""" from datetime import datetime, timezone from cyclone.store import BatchRecord835 cp = ClaimPayment( payer_claim_control_number=pcn, status_code="1", status_label="Primary", total_charge=Decimal("124.00"), total_paid=Decimal("100.00"), service_payments=[ ServicePayment( line_number=1, procedure_qualifier="HC", procedure_code="99213", charge=Decimal("124.00"), payment=Decimal("100.00"), adjustments=[ClaimAdjustment( group_code="CO", reason_code="45", amount=Decimal("24.00"), )], ), ], ) pr835 = ParseResult835( envelope=Envelope835( sender_id="S", receiver_id="R", control_number="0001", transaction_date=date(2026, 6, 19), ), financial_info=FinancialInfo( handling_code="C", paid_amount=Decimal("0"), credit_debit_flag="C", payment_method=None, ), trace=ReassociationTrace( trace_type_code="1", trace_number="0001", originating_company_id="S", ), payer=Payer835(name="X", id="SKCO0"), payee=Payee835(name="Y", npi="1234567890"), claims=[cp], summary=BatchSummary835( input_file="era.txt", control_number="0001", transaction_date=date(2026, 6, 19), total_claims=1, passed=1, failed=0, ), ) CycloneStore().add(BatchRecord835( id="b-835", kind="835", input_filename="era.txt", parsed_at=datetime(2026, 6, 19, 12, 5, tzinfo=timezone.utc), result=pr835, )) # --------------------------------------------------------------------------- # manual_match — writes BOTH sides in one transaction # --------------------------------------------------------------------------- def test_manual_match_writes_both_sides_in_one_transaction(): """After ``manual_match`` returns, ``Claim.matched_remittance_id`` AND ``Remittance.claim_id`` both point at their pair. Without the symmetric write, a later ``list_unmatched`` filter (which keys off ``Remittance.claim_id IS NULL``) would surface an already-matched remit as "orphan". Pre-T11 the symmetric write was correct but only implicitly tested; this pins it explicitly. Note: claim PCN and remit PCN are DELIBERATELY DIFFERENT so the auto-match inside ``CycloneStore.add`` (Task 10) doesn't pre-pair them and ``manual_match`` becomes a no-op no-throw. """ from cyclone.db import Claim, Remittance _build_claim("CLM-1", pcn="CLM-PATIENT-PCN") _build_remit("CLP-1", pcn="CLP-1") CycloneStore().manual_match("CLM-1", "CLP-1") with db.SessionLocal()() as s: claim = s.get(Claim, "CLM-1") # Remit PK is payer_claim_control_number (see Remittance ORM). from sqlalchemy import select remit = s.execute( select(Remittance).where( Remittance.payer_claim_control_number == "CLP-1" ) ).scalar_one() assert claim.matched_remittance_id == remit.id, ( f"claim.matched_remittance_id={claim.matched_remittance_id!r} " f"!= remit.id={remit.id!r}" ) assert remit.claim_id == claim.id, ( f"remit.claim_id={remit.claim_id!r} != claim.id={claim.id!r}" ) # --------------------------------------------------------------------------- # manual_unmatch — clears BOTH sides in one transaction # --------------------------------------------------------------------------- def test_manual_unmatch_clears_both_sides(): """After ``manual_unmatch`` returns, BOTH ``Claim.matched_remittance_id`` AND ``Remittance.claim_id`` return to NULL. Without the symmetric clear, an operator-initiated unmatch would leave the orphaned remit pointing at the claim while the claim is correctly cleared — ``list_unmatched`` would then NOT surface the pair (remit side is non-NULL), even though the operator intended to put it back in the unmatched bucket. PCN asymmetry prevents auto-match from pre-pairing — see ``test_manual_match_writes_both_sides_in_one_transaction``. """ from sqlalchemy import select from cyclone.db import Claim, Remittance _build_claim("CLM-1", pcn="CLM-PATIENT-PCN") _build_remit("CLP-1", pcn="CLP-1") store = CycloneStore() store.manual_match("CLM-1", "CLP-1") store.manual_unmatch("CLM-1") with db.SessionLocal()() as s: claim = s.get(Claim, "CLM-1") remit = s.execute( select(Remittance).where( Remittance.payer_claim_control_number == "CLP-1" ) ).scalar_one() assert claim.matched_remittance_id is None assert remit.claim_id is None def test_manual_match_rollback_clears_symmetric_writes(monkeypatch): """If anything between the two writes raises, the session's rollback MUST clear BOTH writes — neither side set. Pins the SP27 Task 11 invariant "writes BOTH sides in one transaction" at the rollback level. Without this pin, a future refactor that pushes the symmetric write to a post-commit handler (or splits ``manual_match`` across two sessions) would still pass the happy-path test but leave one side set on a failed match — the very class of drift Task 11 was added to surface. """ from sqlalchemy import select from cyclone.db import Claim, Remittance from cyclone import reconcile _build_claim("CLM-ROLL", pcn="CLM-PATIENT-PCN") _build_remit("CLP-ROLL", pcn="CLP-ROLL") # Force the second-loop line-reconcile pass to raise. By the time # it runs, the claim + remit writes are staged in the session # but not committed — the raise must trigger a rollback. def boom(session, claim, remittance): raise RuntimeError("simulated post-write fault") monkeypatch.setattr(reconcile, "_reconcile_pair", boom) with pytest.raises(RuntimeError, match="simulated post-write fault"): CycloneStore().manual_match("CLM-ROLL", "CLP-ROLL") with db.SessionLocal()() as s: claim = s.get(Claim, "CLM-ROLL") remit = s.execute( select(Remittance).where( Remittance.payer_claim_control_number == "CLP-ROLL" ) ).scalar_one() # Neither side set — the session rolled back the partial # state machine. claim.matched_remittance_id is the seeded # NULL (no manual match); remit.claim_id is seeded NULL. assert claim.matched_remittance_id is None assert remit.claim_id is None # --------------------------------------------------------------------------- # Drift check — startup audit, non-blocking # --------------------------------------------------------------------------- def test_drift_check_returns_zero_for_clean_db(): """A clean DB (no manual_match calls, no remits with claim_id) reports zero drift and does not log any warning.""" _build_claim("CLM-1", pcn="PCN-A") # No remits paired — no drift expected. count = check_matched_pair_drift() assert count == 0 def test_drift_check_logs_warning_for_mismatched_pair(caplog): """Inject drift directly via SQL (bypassing store so we can build a state the writers can never produce), call the check, assert a WARNING was logged with the offending ids. The drift check is read-only and non-blocking — operators can clean up manually. Logging at WARNING makes it visible in the standard boot log without paging anyone. This case is the inverse direction: a REMIT points at a claim whose ``matched_remittance_id`` doesn't point back. Pairs with ``test_drift_check_logs_warning_for_case_a_mismatch`` to cover both SQL paths through the check. """ from cyclone.db import Claim, ClaimState, Remittance from sqlalchemy import select # Build the inverse state: a remit whose claim_id points at a # claim that does NOT have matched_remittance_id back. Only Case # B fires; Case A (claim → remit) is clean by construction. with db.SessionLocal()() as s: s.add(Remittance( id="orphan-remit", batch_id="b1", payer_claim_control_number="PCN-B", status_code="1", total_charge=Decimal("100"), total_paid=Decimal("100"), received_at=datetime.now(timezone.utc), service_date=None, is_reversal=False, claim_id="orphan-claim", # ← only this side set )) s.add(Claim( id="orphan-claim", batch_id="b1", patient_control_number="PCN-B-CLAIM", service_date_from=None, charge_amount=Decimal("100"), state=ClaimState.SUBMITTED, matched_remittance_id=None, # ← back-pointer missing )) s.commit() with caplog.at_level("WARNING", logger="cyclone.store"): count = check_matched_pair_drift() assert count == 1, "exactly one drift was injected" # The warning log should mention both ids so the operator can grep. assert any( "orphan-claim" in record.message and "orphan-remit" in record.message for record in caplog.records ), "drift warning should mention the offending claim + remit ids" def test_drift_check_logs_warning_for_case_a_mismatch(caplog): """Pins the OTHER direction: claim → remit drift where the claim's matched_remittance_id points at a remit whose claim_id doesn't point back. Without this test (paired with the inverse test above) a SQL typo in Case A or Case B would silently pass the suite. """ from cyclone.db import Claim, ClaimState, Remittance with db.SessionLocal()() as s: # Remit exists but its claim_id is NULL (claim side is the # only one pointing). s.add(Remittance( id="dangle-remit", batch_id="b1", payer_claim_control_number="PCN-A", status_code="1", total_charge=Decimal("100"), total_paid=Decimal("100"), received_at=datetime.now(timezone.utc), service_date=None, is_reversal=False, claim_id=None, )) s.add(Claim( id="dangle-claim", batch_id="b1", patient_control_number="PCN-A", service_date_from=None, charge_amount=Decimal("100"), state=ClaimState.SUBMITTED, matched_remittance_id="dangle-remit", )) s.commit() with caplog.at_level("WARNING", logger="cyclone.store"): count = check_matched_pair_drift() assert count == 1 assert any( "dangle-claim" in record.message and "dangle-remit" in record.message for record in caplog.records ) def test_drift_check_returns_count_of_mismatched_pairs(): """The return value is the number of mismatched pairs so operators can alert on it (e.g. fail boot if drift > 0 in a future iteration).""" from cyclone.db import Claim, ClaimState, Remittance with db.SessionLocal()() as s: # Two drifted pairs: case A and case B in a single DB. for i, (claim_id, remit_id) in enumerate( [("drift-A-claim", "drift-A-remit"), ("drift-B-claim", "drift-B-remit")] ): s.add(Remittance( id=remit_id, batch_id="b1", payer_claim_control_number=f"PCN-{i}", status_code="1", total_charge=Decimal("100"), total_paid=Decimal("100"), received_at=datetime.now(timezone.utc), service_date=None, is_reversal=False, )) s.add(Claim( id=claim_id, batch_id="b1", patient_control_number=f"PCN-{i}", service_date_from=None, charge_amount=Decimal("100"), state=ClaimState.SUBMITTED, matched_remittance_id=remit_id, )) s.commit() count = check_matched_pair_drift() assert count == 2 def test_drift_check_does_not_raise_on_query_error(monkeypatch): """The drift check is non-blocking at boot — if a DB query fails we want a logged exception, NOT a crashed backend. Wrapped at the api.py call site (Task 11 wiring); this pins that the function itself doesn't try/catch (the call site does). """ from sqlalchemy.exc import OperationalError # Force the SessionLocal to raise on use. def boom(): raise OperationalError("SELECT 1", {}, Exception("synthetic")) monkeypatch.setattr(db, "SessionLocal", lambda: boom) with pytest.raises(OperationalError): check_matched_pair_drift() # The exception propagates — api.py::lifespan wraps it with a # try/except so the boot continues. The pin here is that the # function doesn't swallow by accident.