tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
Version-1.0.0
cyclone/backend/tests/test_api_gets.py
T
Nora f4bafc1c94 feat: History tab on Upload page with one-click Re-export ZIP 
Add a second tab to the Upload page that surfaces the persisted batch
archive and lets the user re-download any 837P batch as a ZIP without
re-parsing the original file.

- Backend: /api/batches now carries per-row claimIds (837P only).
  835 batches return an empty list, which the UI uses as the signal
  to hide the Re-export button on those rows. Avoids an extra
  round-trip to /api/batches/{id} per row.
- Frontend: BatchSummary.claimIds added to the list-endpoint type.
- Upload page: page body wrapped in Tabs.Root with a History trigger
  that mirrors ?tab= in the URL for deep-link round-trip. The
  History tab renders UploadHistory → HistoryTable → HistoryRow with
  a one-click Re-export ZIP button per 837P row. The button calls
  POST /api/batches/{id}/export-837 with the row's claim ids and
  downloads the ZIP via downloadBlob. Falls back to the in-memory
  parsedBatches store when the backend returns no rows so the tab
  stays useful in sample-data mode.
- Backend tests: claimIds present on 837P rows, empty on 835 rows.
- Frontend tests: 13 tests covering tab switching, URL deep-link,
  loading/error/empty states, the 837P-vs-835 button visibility
  split, the Re-export happy path, and the failure toast.
2026-06-24 13:57:12 -06:00

882 lines
32 KiB
Python
Raw Permalink Blame History

"""Tests for the 6 new GET endpoints."""
from __future__ import annotations
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from cyclone.api import app
from cyclone.store import store as global_store
FIXTURE_837 = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt"
@pytest.fixture(autouse=True)
def clear_store():
with global_store._lock:
global_store._batches.clear()
yield
with global_store._lock:
global_store._batches.clear()
@pytest.fixture
def client() -> TestClient:
return TestClient(app)
@pytest.fixture
def seeded_store():
"""One parsed 837P fixture batch."""
text = FIXTURE_837.read_text()
client = TestClient(app)
client.post(
"/api/parse-837",
files={"file": ("x.txt", text, "text/plain")},
headers={"Accept": "application/json"},
)
return client
JSON = {"Accept": "application/json"}
# --------------------------------------------------------------------------- #
# /api/batches
# --------------------------------------------------------------------------- #
def test_batches_returns_empty_list_when_no_parses(client: TestClient):
resp = client.get("/api/batches", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body == {"items": [], "total": 0, "returned": 0, "has_more": False}
def test_batches_returns_summary_after_parse(seeded_store):
resp = seeded_store.get("/api/batches", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body["total"] == 1
assert body["returned"] == 1
assert body["items"][0]["kind"] == "837p"
assert body["items"][0]["inputFilename"] == "x.txt"
def test_batches_includes_claim_ids_for_837p(seeded_store):
"""The Upload page's History tab renders a Re-export ZIP button per
row that fires ``POST /api/batches/{id}/export-837`` with the row's
claim ids. Carry those ids in the list response so the UI doesn't
need an extra round-trip per row to fetch them."""
batches = seeded_store.get("/api/batches", headers=JSON).json()["items"]
assert len(batches) == 1
item = batches[0]
assert item["kind"] == "837p"
# claimIds is a non-empty list of the same claim ids that live
# inside the parsed result — see `seeded_store` in conftest.py.
assert isinstance(item["claimIds"], list)
assert len(item["claimIds"]) == 2
full = seeded_store.get(f"/api/batches/{item['id']}").json()
assert sorted(item["claimIds"]) == sorted(c["claim_id"] for c in full["claims"])
def test_batches_claim_ids_empty_for_835(client: TestClient, tmp_path):
"""835 has no re-export endpoint — the field is always ``[]`` so the
History tab can use it as the signal to hide the Re-export button."""
src = Path(__file__).parent / "fixtures" / "minimal_835.txt"
files = {"file": ("minimal_835.txt", src.read_bytes(), "text/plain")}
r = client.post(
"/api/parse-835",
params={"payer": "co_medicaid_835"},
files=files,
headers=JSON,
)
assert r.status_code == 200, r.text
body = client.get("/api/batches", headers=JSON).json()
assert body["total"] == 1
item = body["items"][0]
assert item["kind"] == "835"
assert item["claimIds"] == []
# --------------------------------------------------------------------------- #
# /api/batches/{id}
# --------------------------------------------------------------------------- #
def test_get_batch_by_id_returns_full_result(seeded_store):
batches = seeded_store.get("/api/batches", headers=JSON).json()["items"]
bid = batches[0]["id"]
resp = seeded_store.get(f"/api/batches/{bid}")
assert resp.status_code == 200
body = resp.json()
assert "envelope" in body
assert "claims" in body
assert len(body["claims"]) == 2
def test_get_batch_404_when_missing(client: TestClient):
resp = client.get("/api/batches/does-not-exist")
assert resp.status_code == 404
# --------------------------------------------------------------------------- #
# /api/claims
# --------------------------------------------------------------------------- #
def test_claims_returns_empty_list_when_no_parses(client: TestClient):
resp = client.get("/api/claims", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
assert body["total"] == 0
def test_claims_filters_by_status_submitted(seeded_store):
resp = seeded_store.get("/api/claims?status=submitted", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert all(c["status"] == "submitted" for c in body["items"])
assert body["total"] >= 1
def test_claims_filters_by_batch_id(seeded_store):
bid = seeded_store.get("/api/batches", headers=JSON).json()["items"][0]["id"]
resp = seeded_store.get(f"/api/claims?batch_id={bid}", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert all(c["batchId"] == bid for c in body["items"])
def test_claims_sorts_by_billed_amount_desc(seeded_store):
resp = seeded_store.get("/api/claims?sort=billedAmount&order=desc", headers=JSON)
assert resp.status_code == 200
body = resp.json()
amounts = [c["billedAmount"] for c in body["items"]]
assert amounts == sorted(amounts, reverse=True)
def test_claims_pagination_limit_offset(seeded_store):
resp1 = seeded_store.get("/api/claims?limit=1&offset=0", headers=JSON)
body1 = resp1.json()
assert body1["returned"] == 1
assert body1["has_more"] is True
# --------------------------------------------------------------------------- #
# /api/remittances
# --------------------------------------------------------------------------- #
def test_remittances_returns_empty_list_when_no_parses(client: TestClient):
resp = client.get("/api/remittances", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
def test_remittances_filters_by_batch_id(seeded_store):
"""No 835 fixture parsed yet — but the endpoint should return 200 with empty."""
resp = seeded_store.get("/api/remittances?batch_id=nope", headers=JSON)
assert resp.status_code == 200
assert resp.json()["items"] == []
# --------------------------------------------------------------------------- #
# /api/remittances list payload includes adjustments array (SP3 P2 follow-up)
# --------------------------------------------------------------------------- #
def test_list_remittances_includes_adjustments_array(client: TestClient) -> None:
"""SP3 P2 follow-up: the list endpoint must include `adjustments` so the
UI's expandable row has data when a remittance is opened from the list
(not just from the detail endpoint)."""
_seed_835_remittance(
pcns=["PCN-LIST-1"],
cas_rows={"PCN-LIST-1": [("CO", "45", "50.00", ""), ("PR", "1", "10.00", "")]},
)
resp = client.get("/api/remittances", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert len(body["items"]) == 1
item = body["items"][0]
assert "adjustments" in item, item
assert len(item["adjustments"]) == 2
# Verify labels come through (resolved by cas_codes.reason_label).
labels = {a["reason"]: a["label"] for a in item["adjustments"]}
assert "45" in labels and "1" in labels
assert labels["45"] # non-empty
assert labels["1"]
def test_list_remittances_empty_adjustments_is_empty_array(client: TestClient) -> None:
"""A remittance with no CAS rows must still expose `adjustments: []`."""
_seed_835_remittance(pcns=["PCN-LIST-EMPTY"], cas_rows=None)
resp = client.get("/api/remittances", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert len(body["items"]) == 1
assert body["items"][0]["adjustments"] == []
# --------------------------------------------------------------------------- #
# /api/providers
# --------------------------------------------------------------------------- #
def test_providers_empty_when_no_parses(client: TestClient):
resp = client.get("/api/providers", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
def test_providers_returns_one_per_distinct_npi(seeded_store):
resp = seeded_store.get("/api/providers", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert len(body["items"]) == 1
assert body["items"][0]["npi"] == "1881068062"
# --------------------------------------------------------------------------- #
# /api/activity
# --------------------------------------------------------------------------- #
def test_activity_empty_when_no_parses(client: TestClient):
resp = client.get("/api/activity", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert body["items"] == []
def test_activity_returns_one_event_per_claim(seeded_store):
resp = seeded_store.get("/api/activity", headers=JSON)
assert resp.status_code == 200
body = resp.json()
assert len(body["items"]) == 2 # co_medicaid_837p.txt has 2 claims
assert all(e["kind"] == "claim_submitted" for e in body["items"])
# --------------------------------------------------------------------------- #
# NDJSON streaming on the list endpoints
# --------------------------------------------------------------------------- #
def _assert_ndjson_envelope(resp, expected_total: int) -> list[dict]:
"""Common helper: parse NDJSON body, return the item payloads."""
import json as _json
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/x-ndjson"), resp.headers
lines = [l for l in resp.text.split("\n") if l.strip()]
events = [_json.loads(l) for l in lines]
items = [e["data"] for e in events if e["type"] == "item"]
summaries = [e for e in events if e["type"] == "summary"]
assert len(summaries) == 1, events
s = summaries[0]["data"]
assert s["total"] == expected_total
assert s["returned"] == len(items)
return items
def test_batches_ndjson_returns_parseable_lines(seeded_store):
resp = seeded_store.get(
"/api/batches",
headers={"Accept": "application/x-ndjson"},
)
items = _assert_ndjson_envelope(resp, expected_total=1)
assert len(items) == 1
assert items[0]["kind"] == "837p"
def test_claims_ndjson_returns_parseable_lines(seeded_store):
resp = seeded_store.get(
"/api/claims",
headers={"Accept": "application/x-ndjson"},
)
items = _assert_ndjson_envelope(resp, expected_total=2)
assert len(items) == 2
def test_remittances_ndjson_returns_parseable_lines(seeded_store):
resp = seeded_store.get(
"/api/remittances",
headers={"Accept": "application/x-ndjson"},
)
# 0 batches parsed, 0 remits
_assert_ndjson_envelope(resp, expected_total=0)
def test_providers_ndjson_returns_parseable_lines(seeded_store):
resp = seeded_store.get(
"/api/providers",
headers={"Accept": "application/x-ndjson"},
)
items = _assert_ndjson_envelope(resp, expected_total=1)
assert items[0]["npi"] == "1881068062"
def test_activity_ndjson_returns_parseable_lines(seeded_store):
resp = seeded_store.get(
"/api/activity",
headers={"Accept": "application/x-ndjson"},
)
items = _assert_ndjson_envelope(resp, expected_total=2)
assert all(it["kind"] == "claim_submitted" for it in items)
def test_claims_json_default_wraps_in_envelope(seeded_store):
"""Accept: application/json → wrapped JSON response (items/total/returned/has_more)."""
resp = seeded_store.get(
"/api/claims",
headers={"Accept": "application/json"},
)
assert resp.status_code == 200
body = resp.json()
assert set(body.keys()) == {"items", "total", "returned", "has_more"}
assert body["total"] == 2
assert body["returned"] == 2
def test_claims_ndjson_wins_over_json_when_both_accepted(seeded_store):
"""When both NDJSON and JSON are in Accept, the explicit NDJSON hint wins."""
resp = seeded_store.get(
"/api/claims",
headers={"Accept": "application/x-ndjson, application/json"},
)
assert resp.headers["content-type"].startswith("application/x-ndjson")
def test_claims_default_accept_returns_json(seeded_store):
"""No Accept header → JSON envelope (per spec 6.2: JSON is the default for
list endpoints; NDJSON is the opt-in)."""
resp = seeded_store.get("/api/claims")
body = resp.json()
assert "items" in body
assert "total" in body
assert "returned" in body
assert "has_more" in body
# --------------------------------------------------------------------------- #
# /api/reconciliation/unmatched (T14)
# --------------------------------------------------------------------------- #
def test_get_reconciliation_unmatched_empty(client: TestClient):
"""With no parsed batches, both sides of the unmatched list are empty."""
resp = client.get("/api/reconciliation/unmatched")
assert resp.status_code == 200
assert resp.json() == {"claims": [], "remittances": []}
def test_get_reconciliation_unmatched_returns_orphans(client: TestClient):
"""An unmatched Claim (left) and an unmatched Remittance (right) both surface.
The 837 subscriber's ``member_id="M1"`` intentionally differs from the
835 remit's ``payer_claim_control_number="PCN-OTHER"`` so the auto-matcher
in ``reconcile.run`` does not pair them on ingest — we want them to land
in the unmatched list so the endpoint has something to return.
"""
from datetime import date, datetime, timezone
from decimal import Decimal
from cyclone.parsers.models import (
BatchSummary, BillingProvider, ClaimHeader, ClaimOutput,
Envelope, Payer, ParseResult, Subscriber, ValidationReport,
)
from cyclone.parsers.models_835 import (
BatchSummary as BatchSummary835,
ClaimPayment, Envelope as Envelope835,
FinancialInfo, Payer835, ParseResult835, Payee835,
ReassociationTrace,
)
from cyclone.store import BatchRecord837, BatchRecord835
# Add a single 837 Claim. member_id="M1" so PCN-based auto-match won't pair it.
co = ClaimOutput(
claim_id="CLM-1",
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="M1",
),
payer=Payer(name="Test Payer", id="P1"),
claim=ClaimHeader(
claim_id="CLM-1", total_charge=Decimal("100"),
frequency_code="1", place_of_service="11",
),
diagnoses=[],
service_lines=[],
validation=ValidationReport(passed=True, errors=[], warnings=[]),
raw_segments=[],
)
global_store.add(BatchRecord837(
id="b-837", kind="837p", input_filename="c.txt",
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
result=ParseResult(
claims=[co],
summary=BatchSummary(
input_file="c.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=1, passed=1, failed=0,
),
),
))
# Add a single 835 Remittance with PCN="PCN-OTHER" (won't auto-match "M1").
cp = ClaimPayment(
payer_claim_control_number="PCN-OTHER",
status_code="1",
total_charge=Decimal("50"), total_paid=Decimal("50"),
service_payments=[],
)
global_store.add(BatchRecord835(
id="b-835", kind="835", input_filename="e.txt",
parsed_at=datetime(2026, 6, 19, 12, 5, tzinfo=timezone.utc),
result=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="e.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=1, passed=1, failed=0,
),
),
))
resp = client.get("/api/reconciliation/unmatched")
assert resp.status_code == 200
body = resp.json()
assert len(body["claims"]) == 1
assert body["claims"][0]["id"] == "CLM-1"
assert len(body["remittances"]) == 1
assert body["remittances"][0]["payerClaimControlNumber"] == "PCN-OTHER"
# --------------------------------------------------------------------------- #
# /api/reconciliation/match + /unmatch (T15)
# --------------------------------------------------------------------------- #
def test_post_match_happy_path(client: TestClient):
"""Match an unmatched claim + remit; both removed from unmatched bucket.
Setup is fully self-contained: we add a Claim (837 ingest) AND a
Remittance (835 ingest) with mismatched member_id/PCN so the auto-
matcher in ``reconcile.run`` does not pair them on ingest, then
POST to /api/reconciliation/match. Asserts the 200 response shape:
``body["claim"]["state"] == "paid"`` (charge == paid triggers
PAID in ``reconcile.apply_payment``) and the match row records
``strategy="manual"``.
The plan's snippet used stale Pydantic v1 dict fields for
``ClaimOutput`` / ``ParseResult``; the current models are typed
BaseModel instances, so we mirror the pattern from
``test_get_reconciliation_unmatched_returns_orphans`` and
``tests/test_store_reconcile.py::test_manual_match_pairs_claim_with_orphan_remit``.
"""
from datetime import date, datetime, timezone
from decimal import Decimal
from cyclone.parsers.models import (
BatchSummary, BillingProvider, ClaimHeader, ClaimOutput,
Envelope, Payer, ParseResult, Subscriber, ValidationReport,
)
from cyclone.parsers.models_835 import (
BatchSummary as BatchSummary835,
ClaimPayment, Envelope as Envelope835,
FinancialInfo, Payer835, ParseResult835, Payee835,
ReassociationTrace,
)
from cyclone.store import BatchRecord837, BatchRecord835
# 837 Claim. member_id="M1" so the 835 PCN-based auto-match can't pair them.
co = ClaimOutput(
claim_id="CLM-1",
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="M1",
),
payer=Payer(name="Test Payer", id="P1"),
claim=ClaimHeader(
claim_id="CLM-1", total_charge=Decimal("100"),
frequency_code="1", place_of_service="11",
),
diagnoses=[],
service_lines=[],
validation=ValidationReport(passed=True, errors=[], warnings=[]),
raw_segments=[],
)
global_store.add(BatchRecord837(
id="b-837", kind="837p", input_filename="c.txt",
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
result=ParseResult(
claims=[co],
summary=BatchSummary(
input_file="c.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=1, passed=1, failed=0,
),
),
))
# 835 Remittance. PCN="CLM-1" but member_id on the claim side is "M1",
# so the auto-matcher in reconcile.run does NOT pair them; we want
# an orphan so manual_match has work to do. charge == paid == 100
# so apply_payment picks ClaimState.PAID.
cp = ClaimPayment(
payer_claim_control_number="CLM-1",
status_code="1",
total_charge=Decimal("100"),
total_paid=Decimal("100"),
service_payments=[],
)
global_store.add(BatchRecord835(
id="b-835", kind="835", input_filename="era.txt",
parsed_at=datetime(2026, 6, 19, 12, 5, tzinfo=timezone.utc),
result=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,
),
),
))
out = client.get("/api/reconciliation/unmatched").json()
assert len(out["claims"]) == 1
assert len(out["remittances"]) == 1
claim_id = out["claims"][0]["id"]
remit_id = out["remittances"][0]["id"]
r = client.post(
"/api/reconciliation/match",
json={"claim_id": claim_id, "remit_id": remit_id},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["claim"]["state"] == "paid"
assert body["match"]["strategy"] == "manual"
assert body["claim"]["matchedRemittanceId"] == remit_id
assert body["match"]["claimId"] == claim_id
assert body["match"]["remittanceId"] == remit_id
# Both sides now filtered out of the unmatched list.
out2 = client.get("/api/reconciliation/unmatched").json()
assert out2["claims"] == []
assert out2["remittances"] == []
def test_post_match_conflict_returns_409(client: TestClient):
"""Matching an already-matched claim returns 409 with ``already_matched``.
Self-contained: builds its own Claim + Remittance via direct ORM
inserts (skips the 837/835 ingest path so the test doesn't depend
on the previous test's data — pytest's default file ordering runs
``test_post_match_happy_path`` first and that test drains the
unmatched bucket, so reading from
``/api/reconciliation/unmatched`` would yield nothing here and
the plan's ``pytest.skip(...)`` would silently no-op the test).
"""
from datetime import datetime, timezone
from decimal import Decimal
from cyclone import db
from cyclone.db import Batch, Claim, Remittance
with db.SessionLocal()() as s:
s.add(Batch(
id="b-conflict", kind="837p", input_filename="x",
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
))
s.add(Claim(
id="CLM-CONFLICT", batch_id="b-conflict",
patient_control_number="CLM-CONFLICT",
charge_amount=Decimal("100"),
state="submitted",
matched_remittance_id=None,
))
s.add(Remittance(
id="CLP-CONFLICT", batch_id="b-conflict",
payer_claim_control_number="PCN-CONFLICT",
status_code="1", total_charge=Decimal("100"),
total_paid=Decimal("100"),
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
))
s.commit()
# First match succeeds.
r1 = client.post(
"/api/reconciliation/match",
json={"claim_id": "CLM-CONFLICT", "remit_id": "CLP-CONFLICT"},
)
assert r1.status_code == 200, r1.text
# Second match for the same claim → 409.
r2 = client.post(
"/api/reconciliation/match",
json={"claim_id": "CLM-CONFLICT", "remit_id": "CLP-CONFLICT"},
)
assert r2.status_code == 409, r2.text
detail = r2.json()["detail"]
assert detail["error"] == "already_matched"
def test_post_unmatch_removes_match(client: TestClient):
"""Unmatch reverses a manual pair and returns the claim to ``submitted``.
The plan snippet creates a Claim with ``state="paid"`` and a
``matched_remittance_id`` set, but **no** corresponding ``Match``
row. ``manual_unmatch`` is built to handle this defensive case —
``matched_remittance_id IS NOT NULL`` lets it through the first
guard, then with no ``Match`` rows it falls back to restoring
``ClaimState.SUBMITTED`` (the default prior state).
We also need an 835 batch row for the remittance's ``batch_id`` FK;
the remittance here is detached from auto-match (no Match row) so
we just need the batch row to exist.
"""
from datetime import datetime, timezone
from decimal import Decimal
from cyclone import db
from cyclone.db import Batch, Claim, Remittance
with db.SessionLocal()() as s:
s.add(Batch(
id="b-unmatch", kind="837p", input_filename="x",
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
))
s.add(Claim(
id="CLM-X", batch_id="b-unmatch",
patient_control_number="CLM-X",
charge_amount=Decimal("100"),
state="paid",
matched_remittance_id="CLP-Y:b-unmatch",
))
s.add(Remittance(
id="CLP-Y:b-unmatch", batch_id="b-unmatch",
payer_claim_control_number="CLM-X",
status_code="1",
total_charge=Decimal("100"),
total_paid=Decimal("100"),
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
))
s.commit()
r = client.post(
"/api/reconciliation/unmatch",
json={"claim_id": "CLM-X"},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["claim"]["state"] == "submitted"
assert body["claim"]["matchedRemittanceId"] is None
# No Match rows existed before the call → 0 deleted.
assert body["deletedMatches"] == 0
# --------------------------------------------------------------------------- #
# /api/remittances/{remittance_id} (T7 / SP3 P2)
# --------------------------------------------------------------------------- #
def test_get_remittance_404_when_missing(client: TestClient) -> None:
"""A missing remittance returns 404 (not 500)."""
resp = client.get("/api/remittances/does-not-exist")
assert resp.status_code == 404
def _seed_835_remittance(
*,
pcns: list[str],
cas_rows: dict[str, list[tuple[str, str, str, str]]] | None = None,
status_code: str = "1",
total_charge: str = "100",
total_paid: str = "100",
) -> None:
"""Helper: insert one 835 batch with one ClaimPayment per PCN.
``cas_rows`` maps ``pcn -> [(group, reason, amount, qty_or_None), ...]``
so a single CAS row maps cleanly to ``CasAdjustment``.
"""
from datetime import date, datetime, timezone
from decimal import Decimal
from cyclone.parsers.models_835 import (
BatchSummary as BatchSummary835,
ClaimAdjustment as ClaimAdjustment835,
ClaimPayment,
Envelope as Envelope835,
FinancialInfo,
Payer835,
ParseResult835,
Payee835,
ReassociationTrace,
ServicePayment,
)
from cyclone.store import BatchRecord835
claims = []
for pcn in pcns:
adjustments: list[ClaimAdjustment835] = []
rows = (cas_rows or {}).get(pcn, [])
for group, reason, amount, qty in rows:
adjustments.append(ClaimAdjustment835(
group_code=group, reason_code=reason,
amount=Decimal(amount),
quantity=Decimal(qty) if qty else None,
))
sp = ServicePayment(
line_number=1,
procedure_qualifier="HC",
procedure_code="T1019",
modifiers=[],
charge=Decimal(total_charge),
payment=Decimal(total_paid),
units=Decimal("1"),
unit_type="UN",
adjustments=adjustments,
)
claims.append(ClaimPayment(
payer_claim_control_number=pcn,
status_code=status_code,
total_charge=Decimal(total_charge),
total_paid=Decimal(total_paid),
service_payments=[sp] if rows else [],
))
global_store.add(BatchRecord835(
id=f"b-835-{pcns[0]}", kind="835", input_filename="era.txt",
parsed_at=datetime(2026, 6, 19, 12, 5, tzinfo=timezone.utc),
result=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(total_paid),
credit_debit_flag="C", payment_method=None,
),
trace=ReassociationTrace(
trace_type_code="1", trace_number="0001",
originating_company_id="S",
),
payer=Payer835(name="Colorado Medicaid", id="SKCO0"),
payee=Payee835(name="TOC, Inc.", npi="1881068062"),
claims=claims,
summary=BatchSummary835(
input_file="era.txt", control_number="0001",
transaction_date=date(2026, 6, 19),
total_claims=len(claims), passed=len(claims), failed=0,
),
),
))
def test_get_remittance_with_empty_adjustments(client: TestClient) -> None:
"""A remittance with no CAS rows returns ``adjustments == []``."""
_seed_835_remittance(pcns=["PCN-EMPTY"])
resp = client.get("/api/remittances/PCN-EMPTY")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["id"] == "PCN-EMPTY"
assert body["adjustments"] == []
def test_get_remittance_with_adjustments_returns_labeled(client: TestClient) -> None:
"""A remittance with two CAS rows surfaces both with the correct labels.
Seeds CO-45 (the fee-schedule code) and PR-1 (deductible) so we
exercise both a CO group code and a PR group code, plus both a
quantified and non-quantified row.
"""
_seed_835_remittance(
pcns=["PCN-LABELED"],
cas_rows={
"PCN-LABELED": [
("CO", "45", "50.00", ""), # no quantity
("PR", "1", "10.00", "1"), # quantified
],
},
)
resp = client.get("/api/remittances/PCN-LABELED")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["id"] == "PCN-LABELED"
adjs = body["adjustments"]
assert len(adjs) == 2
by_key = {(a["group"], a["reason"]): a for a in adjs}
assert ("CO", "45") in by_key
assert ("PR", "1") in by_key
co45 = by_key[("CO", "45")]
assert co45["amount"] == 50.00
assert co45["quantity"] is None
# Label is non-empty and isn't the "Unknown" fallback (both are in the dict).
assert co45["label"] and "Unknown" not in co45["label"]
pr1 = by_key[("PR", "1")]
assert pr1["amount"] == 10.00
assert pr1["quantity"] == 1
assert pr1["label"] and "Unknown" not in pr1["label"]
# The CO and PR rows must have distinct labels (different codes → different meanings).
assert co45["label"] != pr1["label"]
def test_get_remittance_basic_fields_preserved(client: TestClient) -> None:
"""claimId, payerName, paidAmount, status all surface unchanged."""
_seed_835_remittance(
pcns=["PCN-FIELDS"],
cas_rows={"PCN-FIELDS": [("CO", "45", "12.34", "")]},
total_paid="75.50",
)
resp = client.get("/api/remittances/PCN-FIELDS")
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["id"] == "PCN-FIELDS"
assert body["payerName"] == "Colorado Medicaid"
assert body["paidAmount"] == 75.50
assert body["status"] == "received"
assert body["batchId"] # populated; not pinned to a specific value
assert body["receivedDate"] # populated
assert body["parsedAt"] # populated
Reference in New Issue View Git Blame Copy Permalink