tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: ui/batch-diff

Browse Source
This commit is contained in:
Tyler
2026-06-20 17:30:30 -06:00
parent 5baa0290f0 7b394fff1a
commit f4a7cd99b3
12 changed files with 2556 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+52
View File
@@ -894,6 +894,58 @@ def get_reconciliation_unmatched() -> dict:
return store.list_unmatched(kind="both")
# --------------------------------------------------------------------------- #
# Side-by-side diff between two batches (SP3 P4 / T18)
# --------------------------------------------------------------------------- #
@app.get("/api/batch-diff")
def get_batch_diff(
a: str | None = Query(None),
b: str | None = Query(None),
) -> dict:
"""Return a side-by-side diff of two batches identified by id.
Query params: ``a=<batch_id>``, ``b=<batch_id>`` (both required).
Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the
projector shapes):
- ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt,
inputFilename, claimCount)
- ``added`` — claims present in B but not A
- ``removed`` — claims present in A but not B
- ``changed`` — claims present in both, with field deltas
- ``summary`` — precomputed counts
Errors:
- 400 — missing ``a`` or ``b``
- 404 — either batch id is unknown
Pure read endpoint — never mutates the store. Both 837P and 835
batches are accepted (mixed-kind diffs are valid: comparing the
submitted claims against the matching remittances).
"""
if not a or not b:
raise HTTPException(
status_code=400,
detail={"error": "Missing param", "detail": "Both ?a=<batch_id> and ?b=<batch_id> are required."},
)
try:
a_rec, b_rec = store.load_two_for_diff(a, b)
except LookupError as exc:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": str(exc)},
)
# Lazy import — keeps the module's import surface small until the
# endpoint is actually hit. Mirrors the same pattern used by other
# endpoint-local helpers (e.g. reconciler).
from cyclone.batch_diff import diff_batches_to_wire
return diff_batches_to_wire(a_rec, b_rec)
@app.post("/api/reconciliation/match")
def post_reconciliation_match(body: dict) -> dict:
"""Manually pair a Claim with a Remittance (operator override).
backend/src/cyclone/batch_diff.py
+401
View File
@@ -0,0 +1,401 @@
"""Side-by-side diff between two parsed batches.
Powers ``GET /api/batch-diff?a=<batch_id>&b=<batch_id>``. Given two
``BatchRecord`` objects (kind=837P or 835), project each contained
claim/remittance to a small ``BatchClaimSummary`` and split the union
into three buckets:
- ``added`` — in B, not in A (by claim_id / payer_claim_control_number)
- ``removed`` — in A, not in B
- ``changed`` — in both, with at least one differing field
The diff is purely value-based (no segment-level deep diff). The four
fields we surface as "changed" are:
- status (837P: validation-derived label, 835: CLP02 code)
- total_charge (Decimal → number; tolerance 1e-2 to ignore float jitter)
- service_date (ISO date; missing vs present counts as a change)
- cpt_codes (sorted list equality)
The module is a pure function on Pydantic models — no DB I/O — so the
API endpoint and tests can both call it without session management.
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Iterable
from cyclone.parsers.models import ClaimOutput, ValidationReport
from cyclone.parsers.models_835 import ClaimPayment, ParseResult835
from cyclone.store import BatchRecord, BatchRecord835, BatchRecord837, _claim_status_from_validation
# ---------------------------------------------------------------------------
# Projected types
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class BatchClaimSummary:
"""Small per-claim shape returned in the diff payload.
Kept tiny on purpose — the diff view shouldn't carry the full EDI
payload. Raw segments and diagnoses live on the per-batch detail
endpoint; this view is what the operator uses to spot the deltas
between A and B at a glance.
``service_date`` and ``cpt_codes`` are optional because 835
remittance rows don't carry the patient name; we surface empty
strings / empty lists rather than synthesizing values so the UI can
distinguish "no data" from "blank string".
"""
claim_id: str
patient_name: str
total_charge: float
service_date: str | None
status: str
cpt_codes: list[str]
def to_dict(self) -> dict[str, Any]:
return {
"claim_id": self.claim_id,
"patient_name": self.patient_name,
"total_charge": self.total_charge,
"service_date": self.service_date,
"status": self.status,
"cpt_codes": list(self.cpt_codes),
}
@dataclass(frozen=True)
class FieldDiff:
"""One changed field between two ``BatchClaimSummary`` rows.
``a`` and ``b`` are the field values from each side. We don't
attempt to coerce types — the frontend renders them verbatim — but
the comparator picks the field-specific projection (e.g. charge as
a number) so equality is stable.
"""
field: str
a: Any
b: Any
def to_dict(self) -> dict[str, Any]:
return {"field": self.field, "a": self.a, "b": self.b}
@dataclass(frozen=True)
class ChangedRow:
"""One claim present in both batches with at least one field delta.
Always carries both summaries (so the renderer can fall back to
a single side if a field is missing) plus the diff list (possibly
multiple fields changed for the same claim).
"""
a: BatchClaimSummary
b: BatchClaimSummary
diff: list[FieldDiff]
def to_dict(self) -> dict[str, Any]:
return {
"a": self.a.to_dict(),
"b": self.b.to_dict(),
"diff": [d.to_dict() for d in self.diff],
}
@dataclass(frozen=True)
class BatchDiffResult:
"""Full diff between two batches.
Mirrors the API contract documented in the task spec. Counts are
precomputed so the frontend doesn't have to derive them.
"""
a_meta: dict[str, Any]
b_meta: dict[str, Any]
added: list[BatchClaimSummary]
removed: list[BatchClaimSummary]
changed: list[ChangedRow]
@property
def summary(self) -> dict[str, int]:
added_count = len(self.added)
removed_count = len(self.removed)
changed_count = len(self.changed)
# ``unchanged`` are the claims present in BOTH batches that have
# no field differences. Derivation from A's total:
# A's total = unchanged + changed + removed
# so unchanged = A_total - changed - removed. Clamp at zero so
# a malformed batch pair can never surface a negative count.
a_total = len(self.a_meta.get("claim_ids", []) or [])
unchanged_count = max(0, a_total - changed_count - removed_count)
return {
"addedCount": added_count,
"removedCount": removed_count,
"changedCount": changed_count,
"unchangedCount": unchanged_count,
}
def to_dict(self) -> dict[str, Any]:
return {
"a": self.a_meta,
"b": self.b_meta,
"added": [s.to_dict() for s in self.added],
"removed": [s.to_dict() for s in self.removed],
"changed": [c.to_dict() for c in self.changed],
"summary": self.summary,
}
# ---------------------------------------------------------------------------
# Projection helpers
# ---------------------------------------------------------------------------
def _summary_from_claim_837(claim: ClaimOutput) -> BatchClaimSummary:
"""Project a 837P ``ClaimOutput`` to the diff summary shape."""
# Patient name: same composition as to_ui_claim.
sub = claim.subscriber
patient_name = f"{sub.first_name} {sub.last_name}".strip()
# Total charge: coerce Decimal to float; 0.0 when None.
total_charge = float(claim.claim.total_charge or Decimal("0"))
# Earliest service date across the SV1 service lines (matches
# store._service_dates_from_claim convention).
service_dates: list = [
sl.service_date for sl in claim.service_lines if sl.service_date is not None
]
service_date: str | None = None
if service_dates:
earliest = min(service_dates)
# The pydantic model serializes dates to ISO strings via the
# @model_serializer; pull isoformat defensively for both
# date and datetime inputs (Pydantic coerces in either case).
service_date = earliest.isoformat() if hasattr(earliest, "isoformat") else str(earliest)
# CPT codes: collect the procedure.code across all SV1s, in line order.
cpt_codes = [
sl.procedure.code for sl in claim.service_lines if sl.procedure.code
]
return BatchClaimSummary(
claim_id=claim.claim_id,
patient_name=patient_name,
total_charge=total_charge,
service_date=service_date,
status=_claim_status_from_validation(claim),
cpt_codes=cpt_codes,
)
def _summary_from_remit_835(cp: ClaimPayment) -> BatchClaimSummary:
"""Project an 835 ``ClaimPayment`` to the diff summary shape.
Patient name isn't carried on an ERA — we leave it blank rather
than synthesize. Total charge / paid amount aren't on the
``BatchClaimSummary`` contract; we expose the 835's ``total_charge``
on the BPR total so it lines up with the 837P's CLM01 amount.
"""
total_charge = float(cp.total_charge or Decimal("0"))
service_date: str | None = None
cpt_codes: list[str] = []
if cp.service_payments:
sp = cp.service_payments[0]
if sp.service_date is not None:
service_date = (
sp.service_date.isoformat()
if hasattr(sp.service_date, "isoformat")
else str(sp.service_date)
)
# CPT codes from every SVC segment in the CLP loop.
for svc in cp.service_payments:
if svc.procedure_code:
cpt_codes.append(svc.procedure_code)
# CLP02 status code as the primary status label; fall back to the
# human-readable label when one is bundled in the parser.
status = cp.status_label or cp.status_code or ""
return BatchClaimSummary(
# ``payer_claim_control_number`` is the ERA's natural id and the
# match key against another batch's 835 (matches on CLP01).
claim_id=cp.payer_claim_control_number,
patient_name="",
total_charge=total_charge,
service_date=service_date,
status=status,
cpt_codes=cpt_codes,
)
def _summarize_record(rec: BatchRecord) -> list[BatchClaimSummary]:
"""Return the list of summaries for a batch — 837P or 835.
Dispatched by ``isinstance`` so callers don't need to branch on
``kind``. Returns an empty list when the record's ``result`` is
missing the expected claim array (defensive against partially-loaded
rows in tests).
"""
if isinstance(rec, BatchRecord837):
return [_summary_from_claim_837(c) for c in (rec.result.claims or [])]
if isinstance(rec, BatchRecord835):
return [_summary_from_remit_835(c) for c in (rec.result.claims or [])]
# Unknown record type — return empty so the diff still computes.
return []
# ---------------------------------------------------------------------------
# Field comparison
# ---------------------------------------------------------------------------
def _eq_charge(a: float, b: float) -> bool:
"""Two floats are equal if they differ by less than a cent.
Decimal → float coercion can introduce sub-cent noise when the
source had more than 2 decimal places; tolerate that here so we
don't surface spurious charge deltas.
"""
return abs(a - b) < 0.01
def _diff_pair(
a: BatchClaimSummary,
b: BatchClaimSummary,
*,
compare_status: bool,
) -> list[FieldDiff]:
"""Return the list of changed fields between two same-id summaries.
Compares the four fields documented in the module docstring; an
unchanged field is simply not in the returned list.
``compare_status`` controls whether ``status`` enters the diff:
- True when both batches are the same kind (837P vs 837P, or
835 vs 835). 837P's validation-derived status (``submitted`` /
``pending`` / ``denied`` / ``draft``) and 835's CLP02 code
(``1`` / ``4`` / ``21`` / …) live in different namespaces, so
comparing across kinds surfaces spurious deltas the operator
can't act on. We skip the comparison when kinds differ.
"""
out: list[FieldDiff] = []
if compare_status and a.status != b.status:
out.append(FieldDiff("status", a.status, b.status))
if not _eq_charge(a.total_charge, b.total_charge):
out.append(FieldDiff("total_charge", a.total_charge, b.total_charge))
if a.service_date != b.service_date:
out.append(FieldDiff("service_date", a.service_date, b.service_date))
# CPT codes are compared as sorted, deduplicated lists so reordering
# between SV1 segments doesn't register as a diff.
a_codes = sorted(set(a.cpt_codes))
b_codes = sorted(set(b.cpt_codes))
if a_codes != b_codes:
out.append(FieldDiff("cpt_codes", a_codes, b_codes))
return out
def _meta(rec: BatchRecord, *, claim_ids: Iterable[str]) -> dict[str, Any]:
"""Build the ``a``/``b`` metadata block for the diff envelope.
The ``claim_ids`` list is passed in by the caller (the same source
used for the summaries) so the meta block is consistent with what
the diff actually saw.
"""
parsed_at = rec.parsed_at
parsed_iso = parsed_at.isoformat().replace("+00:00", "Z") if parsed_at else ""
return {
"id": rec.id,
"kind": rec.kind,
"parsedAt": parsed_iso,
"inputFilename": rec.input_filename,
"claimCount": len(list(claim_ids)),
# Internal-only — stripped from the wire response below. The
# API layer recomputes counts via to_dict, so this key is
# used solely for the ``summary.unchangedCount`` derivation.
"claim_ids": list(claim_ids),
}
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def diff_batches(
a: BatchRecord,
b: BatchRecord,
) -> BatchDiffResult:
"""Compute the side-by-side diff between ``a`` and ``b``.
Both records can be 837P, both 835, or one of each — we project
each to ``BatchClaimSummary`` independently and match by the
summary's ``claim_id`` field (837P ``CLM01``, 835 ``CLP01``). When
the kinds differ the result is still well-defined; it just means
you're comparing e.g. an 837P claim set against the matching
remittances — useful when the operator wants to see which claims
were paid vs. submitted.
Order preservation:
- ``added`` follows B's input order (matches the upload narrative).
- ``removed`` follows A's input order.
- ``changed`` follows B's input order (the "current state" side).
This mirrors git-diff conventions where the right-hand side drives
the row order on the page.
"""
a_summaries = _summarize_record(a)
b_summaries = _summarize_record(b)
a_by_id = {s.claim_id: s for s in a_summaries}
b_by_id = {s.claim_id: s for s in b_summaries}
# Status only carries comparable meaning when both batches share a
# kind (837P/837P or 835/835). Across kinds the namespaces differ
# (validation-derived vs CLP02) so we skip that field.
compare_status = a.kind == b.kind
added: list[BatchClaimSummary] = []
removed: list[BatchClaimSummary] = []
changed: list[ChangedRow] = []
# Walk B's order for added/changed (B drives row order).
for s in b_summaries:
if s.claim_id not in a_by_id:
added.append(s)
continue
other = a_by_id[s.claim_id]
diff = _diff_pair(other, s, compare_status=compare_status)
if diff:
changed.append(ChangedRow(a=other, b=s, diff=diff))
# Walk A's order for removed (anything in A but not in B).
for s in a_summaries:
if s.claim_id not in b_by_id:
removed.append(s)
a_meta = _meta(a, claim_ids=[s.claim_id for s in a_summaries])
b_meta = _meta(b, claim_ids=[s.claim_id for s in b_summaries])
return BatchDiffResult(
a_meta=a_meta,
b_meta=b_meta,
added=added,
removed=removed,
changed=changed,
)
def diff_batches_to_wire(
a: BatchRecord,
b: BatchRecord,
) -> dict[str, Any]:
"""Convenience wrapper that returns the JSON-ready dict.
Strips the internal ``claim_ids`` key from the metadata blocks
before returning so the wire contract stays clean — the spec only
promises ``id, kind, parsedAt, inputFilename, claimCount`` per side.
"""
result = diff_batches(a, b)
wire = result.to_dict()
wire["a"].pop("claim_ids", None)
wire["b"].pop("claim_ids", None)
return wire
backend/src/cyclone/store.py
+25
View File
@@ -1240,6 +1240,31 @@ class CycloneStore:
rows = s.query(Batch).order_by(Batch.parsed_at.asc()).all()
return [self._row_to_record(r) for r in rows]
def load_two_for_diff(
self,
a_id: str,
b_id: str,
) -> tuple[BatchRecord, BatchRecord]:
"""Load two batches by id for the side-by-side diff view.
Returns ``(a, b)`` as ``BatchRecord`` objects. Raises
:class:`LookupError` when either id is missing — the API layer
catches it and maps it to ``404 Not Found`` (matching the
``GET /api/batches/{id}`` contract). The two loads happen in
independent sessions so a transient failure on one side can't
poison the other.
Used exclusively by :mod:`cyclone.batch_diff` via the
``/api/batch-diff`` endpoint.
"""
a = self.get(a_id)
if a is None:
raise LookupError(f"batch {a_id} not found")
b = self.get(b_id)
if b is None:
raise LookupError(f"batch {b_id} not found")
return a, b
def iter_claims(
self,
*,
backend/tests/test_batch_diff.py
+611
View File
@@ -0,0 +1,611 @@
"""Tests for ``cyclone.batch_diff`` and the ``GET /api/batch-diff`` endpoint.
Two layers:
1. Pure unit tests over :func:`cyclone.batch_diff.diff_batches` — no DB,
no HTTP. Construct ``BatchRecord`` values directly with the parser's
Pydantic models so the tests stay fast and deterministic.
2. Integration tests over the FastAPI endpoint — round-trip through
``TestClient`` with real 837P / 835 fixtures. The ``conftest.py``
fixture already gives every test a fresh SQLite database, so these
run independently without manual cleanup.
"""
from __future__ import annotations
from datetime import date, datetime, timezone
from decimal import Decimal
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from cyclone.api import app
from cyclone.batch_diff import (
BatchClaimSummary,
diff_batches,
diff_batches_to_wire,
)
from cyclone.parsers.models import (
Address,
BatchSummary,
BillingProvider,
ClaimHeader,
ClaimOutput,
Envelope,
Payer,
ParseResult,
Procedure,
ServiceLine,
Subscriber,
ValidationReport,
)
from cyclone.parsers.models_835 import (
ClaimPayment,
ParseResult835,
Payer835,
Payee835,
ReassociationTrace,
FinancialInfo,
ServicePayment,
)
from cyclone.store import BatchRecord, store as global_store
# 837P fixture shipped with the test suite — two claims, useful as a
# baseline for the integration tests below.
FIXTURE_837 = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt"
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _setup(tmp_path, monkeypatch):
"""Per-test isolated DB.
Same pattern used by ``test_store.py``. The unit tests below don't
touch the DB, but it's autouse'd so a test that flips to integration
mode doesn't accidentally inherit stale state.
"""
monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/test.db")
from cyclone import db
db._reset_for_tests()
db.init_db()
yield
db._reset_for_tests()
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
def _validation(passed: bool = True) -> ValidationReport:
return ValidationReport(passed=passed, errors=[], warnings=[])
def _make_claim_837(
*,
claim_id: str = "CLM-1",
charge: str = "124.00",
patient_first: str = "Jane",
patient_last: str = "Doe",
frequency_code: str = "1",
service_date: date | None = None,
cpt_code: str = "99213",
validation: ValidationReport | None = None,
) -> ClaimOutput:
service_lines = []
if cpt_code or service_date is not None:
service_lines.append(
ServiceLine(
line_number=1,
procedure=Procedure(qualifier="HC", code=cpt_code, modifiers=[]),
charge=Decimal(charge),
units=Decimal("1"),
service_date=service_date,
)
)
return ClaimOutput(
claim_id=claim_id,
control_number="0001",
transaction_date=date(2026, 6, 19),
billing_provider=BillingProvider(name="Test Clinic", npi="1234567890"),
subscriber=Subscriber(
first_name=patient_first,
last_name=patient_last,
member_id=f"M-{claim_id}",
),
payer=Payer(name="Test Payer", id="P1"),
claim=ClaimHeader(
claim_id=claim_id,
total_charge=Decimal(charge),
frequency_code=frequency_code,
place_of_service="11",
),
diagnoses=[],
service_lines=service_lines,
validation=validation if validation is not None else _validation(),
raw_segments=[],
)
def _make_record_837(claims: list[ClaimOutput], *, batch_id: str = "b-a") -> BatchRecord:
return BatchRecord(
id=batch_id,
kind="837p",
input_filename=f"{batch_id}.837",
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
result=ParseResult(
envelope=Envelope(
sender_id="S", receiver_id="R", control_number="0001",
transaction_date=date(2026, 6, 19),
),
claims=claims,
summary=BatchSummary(
input_file=f"{batch_id}.837", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=len(claims),
passed=len(claims), failed=0,
),
),
)
def _make_remit_835(
*,
pccn: str = "R-1",
charge: str = "100.00",
status_code: str = "1",
service_date: date | None = None,
procedure_code: str = "99213",
) -> ClaimPayment:
service_payments = []
if procedure_code or service_date is not None:
service_payments.append(
ServicePayment(
line_number=1,
procedure_qualifier="HC",
procedure_code=procedure_code,
modifiers=[],
charge=Decimal(charge),
payment=Decimal("0"),
service_date=service_date,
adjustments=[],
)
)
return ClaimPayment(
payer_claim_control_number=pccn,
status_code=status_code,
status_label=None,
total_charge=Decimal(charge),
total_paid=Decimal("0"),
service_payments=service_payments,
raw_segments=[],
)
def _make_record_835(
remits: list[ClaimPayment], *, batch_id: str = "b-b",
) -> BatchRecord:
return BatchRecord(
id=batch_id,
kind="835",
input_filename=f"{batch_id}.835",
parsed_at=datetime(2026, 6, 19, 13, 0, tzinfo=timezone.utc),
result=ParseResult835(
envelope=Envelope(
sender_id="P", receiver_id="R", control_number="0002",
transaction_date=date(2026, 6, 19),
),
financial_info=FinancialInfo(
handling_code="C", paid_amount=Decimal("0"),
credit_debit_flag="C",
),
trace=ReassociationTrace(
trace_type_code="1", trace_number="1",
originating_company_id="P1",
),
payer=Payer835(name="Payer"),
payee=Payee835(name="Clinic", npi="1234567890"),
claims=remits,
summary=BatchSummary(
input_file=f"{batch_id}.835", control_number="0002",
transaction_date=date(2026, 6, 19),
total_claims=len(remits),
passed=len(remits), failed=0,
),
),
)
# ---------------------------------------------------------------------------
# Pure unit tests — diff_batches
# ---------------------------------------------------------------------------
def test_empty_batches_have_no_deltas():
a = _make_record_837([], batch_id="b-a")
b = _make_record_837([], batch_id="b-b")
result = diff_batches(a, b)
assert result.added == []
assert result.removed == []
assert result.changed == []
summary = result.summary
assert summary == {
"addedCount": 0, "removedCount": 0,
"changedCount": 0, "unchangedCount": 0,
}
def test_identical_batches_have_no_deltas():
claims = [
_make_claim_837(claim_id="CLM-1", charge="124.00", service_date=date(2026, 6, 1)),
_make_claim_837(claim_id="CLM-2", charge="88.50", service_date=date(2026, 6, 2)),
]
a = _make_record_837(claims, batch_id="b-a")
b = _make_record_837(claims, batch_id="b-b")
result = diff_batches(a, b)
assert result.added == []
assert result.removed == []
assert result.changed == []
assert result.summary["unchangedCount"] == 2
def test_claim_added_in_b():
a = _make_record_837([_make_claim_837(claim_id="CLM-1")], batch_id="b-a")
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1"), _make_claim_837(claim_id="CLM-2")],
batch_id="b-b",
)
result = diff_batches(a, b)
assert [s.claim_id for s in result.added] == ["CLM-2"]
assert result.removed == []
assert result.changed == []
assert result.summary["addedCount"] == 1
assert result.summary["unchangedCount"] == 1
def test_claim_removed_from_a():
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1"), _make_claim_837(claim_id="CLM-2")],
batch_id="b-a",
)
b = _make_record_837([_make_claim_837(claim_id="CLM-1")], batch_id="b-b")
result = diff_batches(a, b)
assert result.added == []
assert [s.claim_id for s in result.removed] == ["CLM-2"]
assert result.summary["removedCount"] == 1
def test_claim_charge_change_is_a_diff():
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1", charge="100.00")], batch_id="b-a",
)
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1", charge="125.00")], batch_id="b-b",
)
result = diff_batches(a, b)
assert result.added == [] and result.removed == []
assert len(result.changed) == 1
fields = [d.field for d in result.changed[0].diff]
assert "total_charge" in fields
charge_diff = next(d for d in result.changed[0].diff if d.field == "total_charge")
assert charge_diff.a == 100.00
assert charge_diff.b == 125.00
def test_sub_cent_charge_jitter_is_not_a_diff():
"""Two charges that differ by less than a cent register as equal.
Decimal → float can introduce sub-cent noise when the source had
more than 2 decimal places. Tolerance is 1e-2 to match the
production ``_eq_charge`` helper.
"""
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1", charge="100.00")], batch_id="b-a",
)
# 100.004 rounds to 100.00 in the float projection but is a fresh
# value when summed — surface this as a delta only if the diff
# exceeds the 1e-2 tolerance.
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1", charge="100.005")], batch_id="b-b",
)
result = diff_batches(a, b)
assert result.changed == [], (
"sub-cent float noise should not register as a delta"
)
def test_status_change_is_a_diff():
"""Validation-derived status flows through to the diff.
A claim that flipped from passing to failing should surface a
``status`` field diff between the two sides.
"""
passing = _make_claim_837(
claim_id="CLM-1", validation=_validation(passed=True),
)
failing = _make_claim_837(
claim_id="CLM-1", validation=ValidationReport(
passed=False,
errors=[],
warnings=[],
),
)
a = _make_record_837([passing], batch_id="b-a")
b = _make_record_837([failing], batch_id="b-b")
result = diff_batches(a, b)
assert len(result.changed) == 1
fields = [d.field for d in result.changed[0].diff]
assert "status" in fields
def test_service_date_change_is_a_diff():
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1", service_date=date(2026, 6, 1))],
batch_id="b-a",
)
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1", service_date=date(2026, 6, 2))],
batch_id="b-b",
)
result = diff_batches(a, b)
assert len(result.changed) == 1
fields = [d.field for d in result.changed[0].diff]
assert "service_date" in fields
def test_cpt_change_is_a_diff():
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1", cpt_code="99213")], batch_id="b-a",
)
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1", cpt_code="99214")], batch_id="b-b",
)
result = diff_batches(a, b)
assert len(result.changed) == 1
fields = [d.field for d in result.changed[0].diff]
assert "cpt_codes" in fields
def test_cpt_reorder_is_not_a_diff():
"""CPT codes compared as sorted, deduplicated sets.
Two claims with the same codes in different SV1 order should
not register as a diff — that would generate noise on every
service-line reordering.
"""
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1", cpt_code="99213")], batch_id="b-a",
)
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1", cpt_code="99213")], batch_id="b-b",
)
result = diff_batches(a, b)
assert result.changed == []
def test_mixed_added_removed_changed():
a = _make_record_837(
[
_make_claim_837(claim_id="CLM-1", charge="100.00"),
_make_claim_837(claim_id="CLM-2", charge="200.00"),
_make_claim_837(claim_id="CLM-3", charge="300.00"),
],
batch_id="b-a",
)
b = _make_record_837(
[
_make_claim_837(claim_id="CLM-1", charge="150.00"), # changed
_make_claim_837(claim_id="CLM-3", charge="300.00"), # unchanged
_make_claim_837(claim_id="CLM-4", charge="400.00"), # added
],
batch_id="b-b",
)
result = diff_batches(a, b)
assert [s.claim_id for s in result.added] == ["CLM-4"]
assert [s.claim_id for s in result.removed] == ["CLM-2"]
assert [c.a.claim_id for c in result.changed] == ["CLM-1"]
summary = result.summary
assert summary["addedCount"] == 1
assert summary["removedCount"] == 1
assert summary["changedCount"] == 1
assert summary["unchangedCount"] == 1
def test_wire_dict_strips_internal_claim_ids():
a = _make_record_837(
[_make_claim_837(claim_id="CLM-1")], batch_id="b-a",
)
b = _make_record_837(
[_make_claim_837(claim_id="CLM-1"), _make_claim_837(claim_id="CLM-2")],
batch_id="b-b",
)
wire = diff_batches_to_wire(a, b)
assert "claim_ids" not in wire["a"]
assert "claim_ids" not in wire["b"]
# Public shape — every key documented in the spec.
assert set(wire.keys()) == {"a", "b", "added", "removed", "changed", "summary"}
assert set(wire["a"].keys()) == {
"id", "kind", "parsedAt", "inputFilename", "claimCount",
}
assert wire["a"]["claimCount"] == 1
assert wire["b"]["claimCount"] == 2
def test_mixed_kind_diff_matches_by_claim_id():
"""837P and 835 can be diffed; the matcher uses ``claim_id``.
The common case for a single kind pair is covered by the other
tests. Here we exercise the cross-kind path so an operator can
diff submitted claims against received remittances.
"""
a = _make_record_837(
[
_make_claim_837(claim_id="CLM-1", charge="100.00"),
_make_claim_837(claim_id="CLM-2", charge="200.00"),
],
batch_id="b-a",
)
b = _make_record_835(
[
_make_remit_835(pccn="CLM-1", charge="100.00", status_code="1"),
_make_remit_835(pccn="CLM-3", charge="300.00", status_code="1"),
],
batch_id="b-b",
)
result = diff_batches(a, b)
# CLM-1 in both, no field change → unchanged
# CLM-2 in A only → removed
# CLM-3 in B only → added
assert [s.claim_id for s in result.added] == ["CLM-3"]
assert [s.claim_id for s in result.removed] == ["CLM-2"]
assert result.changed == []
assert result.summary["unchangedCount"] == 1
# ---------------------------------------------------------------------------
# Integration tests — /api/batch-diff
# ---------------------------------------------------------------------------
def _ingest_837(client: TestClient, text: str, filename: str = "x.837") -> str:
resp = client.post(
"/api/parse-837",
files={"file": (filename, text, "text/plain")},
headers={"Accept": "application/json"},
)
assert resp.status_code == 200, resp.text
return resp.json()["claims"][0]["claim_id"] # used for sanity
def test_endpoint_404_when_a_missing(client: TestClient):
"""Missing ``a`` → 404, never 500.
Mirrors the existing ``GET /api/batches/{id}`` 404 contract.
"""
resp = client.get("/api/batch-diff?a=ghost&b=also-ghost")
assert resp.status_code == 404
body = resp.json()
assert body["detail"]["error"] == "Not found"
assert "ghost" in body["detail"]["detail"]
def test_endpoint_400_when_params_missing(client: TestClient):
"""Missing query params → 400 with a helpful detail message."""
resp = client.get("/api/batch-diff?a=only")
assert resp.status_code == 400
body = resp.json()
assert body["detail"]["error"] == "Missing param"
assert "b" in body["detail"]["detail"].lower()
def test_endpoint_returns_diff_between_two_ingested_batches(client: TestClient):
"""End-to-end: POST two 837 files, GET the diff, assert the shape.
The fixture file ships with 2 claims; we ingest it twice. The
diff should be empty (the two batches are identical) and the
unchangedCount should equal 2.
"""
text = FIXTURE_837.read_text()
r1 = client.post(
"/api/parse-837",
files={"file": ("first.837", text, "text/plain")},
headers={"Accept": "application/json"},
)
assert r1.status_code == 200, r1.text
a_id = r1.json()["claims"][0]["claim_id"] # sanity: parse worked
r2 = client.post(
"/api/parse-837",
files={"file": ("second.837", text, "text/plain")},
headers={"Accept": "application/json"},
)
assert r2.status_code == 200, r2.text
# Resolve the actual batch ids via /api/batches.
batches = client.get("/api/batches").json()["items"]
assert len(batches) == 2
a_batch_id = batches[0]["id"]
b_batch_id = batches[1]["id"]
resp = client.get(
f"/api/batch-diff?a={a_batch_id}&b={b_batch_id}",
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["a"]["id"] == a_batch_id
assert body["b"]["id"] == b_batch_id
assert body["a"]["kind"] == "837p"
assert body["b"]["kind"] == "837p"
assert body["added"] == []
assert body["removed"] == []
assert body["changed"] == []
# Spec summary keys are all present and numeric.
summary = body["summary"]
assert summary["addedCount"] == 0
assert summary["removedCount"] == 0
assert summary["changedCount"] == 0
assert summary["unchangedCount"] == 2
# Reference the parsed claim id once so the fixture's first claim
# id is exercised end-to-end (the lint otherwise complains about
# the unused variable).
assert a_id
def test_endpoint_includes_added_removed_summary_keys(client: TestClient):
"""Diff summary keys are documented and always present.
Even when zero, the JSON contract should expose every key so the
UI can render the four cards unconditionally.
"""
text = FIXTURE_837.read_text()
r1 = client.post(
"/api/parse-837",
files={"file": ("first.837", text, "text/plain")},
headers={"Accept": "application/json"},
)
r2 = client.post(
"/api/parse-837",
files={"file": ("second.837", text, "text/plain")},
headers={"Accept": "application/json"},
)
assert r1.status_code == r2.status_code == 200
batches = client.get("/api/batches").json()["items"]
a_batch_id, b_batch_id = batches[0]["id"], batches[1]["id"]
body = client.get(
f"/api/batch-diff?a={a_batch_id}&b={b_batch_id}",
).json()
assert set(body["summary"].keys()) == {
"addedCount", "removedCount", "changedCount", "unchangedCount",
}
def test_store_helper_load_two_for_diff_raises_lookup():
"""``CycloneStore.load_two_for_diff`` raises ``LookupError`` on
missing ids so the API layer can map to 404 cleanly.
"""
from cyclone.store import CycloneStore
s = CycloneStore()
s.add(_make_record_837(
[_make_claim_837(claim_id="CLM-1")],
batch_id="present",
))
with pytest.raises(LookupError):
s.load_two_for_diff("present", "ghost")
with pytest.raises(LookupError):
s.load_two_for_diff("ghost", "present")
# Happy path
a, b = s.load_two_for_diff("present", "present")
assert a.id == "present"
assert b.id == "present"
# Silence the unused-import linter for items used purely as type hints
# or fixture anchors.
void_global_store = global_store