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

290 lines
8.8 KiB
Python
Raw Blame History

"""Tests for the in-memory batch store."""
from __future__ import annotations
from decimal import Decimal
from cyclone.parsers.models import (
BatchSummary,
BillingProvider,
ClaimHeader,
ClaimOutput,
Envelope,
Payer,
ParseResult,
Subscriber,
ValidationIssue,
ValidationReport,
)
from cyclone.parsers.models_835 import ClaimPayment
from cyclone.store import BatchRecord, InMemoryStore, store
# ---------------------------------------------------------------------------
# Stubs (match real model field names)
# ---------------------------------------------------------------------------
def _make_claim(claim_id: str = "CLM-1") -> ClaimOutput:
"""Minimal ClaimOutput stub. Real parser not needed here."""
return ClaimOutput(
claim_id=claim_id,
control_number="0001",
transaction_date="2026-06-19",
billing_provider=BillingProvider(
name="Test",
npi="1234567890",
tax_id=None,
address=None,
),
subscriber=Subscriber(
first_name="Jane",
last_name="Doe",
member_id="M1",
dob=None,
gender=None,
address=None,
),
payer=Payer(name="Test Payer", id="P1"),
claim=ClaimHeader(
claim_id=claim_id,
total_charge=Decimal("0.00"),
place_of_service="11",
facility_code_qualifier=None,
frequency_code="1",
provider_signature=None,
assignment=None,
release_of_info=None,
prior_auth=None,
),
diagnoses=[],
service_lines=[],
validation=ValidationReport(passed=True, errors=[], warnings=[]),
raw_segments=[],
)
def _make_result() -> ParseResult:
return ParseResult(
envelope=Envelope(
sender_id="SENDER",
receiver_id="RECEIVER",
control_number="0001",
transaction_date="2026-06-19",
transaction_time=None,
implementation_guide=None,
),
claims=[_make_claim("CLM-1")],
summary=BatchSummary(
input_file="test.txt",
control_number="0001",
transaction_date="2026-06-19",
total_claims=1,
passed=1,
failed=0,
failed_claim_ids=[],
issues_by_rule={},
output_dir=None,
),
)
# ---------------------------------------------------------------------------
# Task 1: BatchRecord + InMemoryStore
# ---------------------------------------------------------------------------
def test_module_singleton_store_exists():
"""The module-level `store` should be importable and be an InMemoryStore."""
assert isinstance(store, InMemoryStore)
def test_in_memory_store_starts_empty():
s = InMemoryStore()
assert s.list() == []
def test_add_and_get_round_trip():
s = InMemoryStore()
result = _make_result()
rec = BatchRecord(
id="abc123",
kind="837p",
input_filename="test.txt",
parsed_at="2026-06-19T12:00:00Z",
result=result,
)
s.add(rec)
assert s.get("abc123") is rec
assert s.list() == [rec]
def test_get_missing_returns_none():
s = InMemoryStore()
assert s.get("does-not-exist") is None
# ---------------------------------------------------------------------------
# Task 2: Mappers
# ---------------------------------------------------------------------------
def test_to_ui_claim_maps_status_submitted():
from cyclone.store import to_ui_claim
claim = _make_claim("CLM-1")
out = to_ui_claim(claim, batch_id="b1", parsed_at="2026-06-19T12:00:00Z")
assert out["id"] == "CLM-1"
assert out["status"] == "submitted" # passed=True, frequency_code="1"
assert out["providerNpi"] == "1234567890"
assert out["batchId"] == "b1"
def test_to_ui_claim_maps_status_pending_when_warnings():
from cyclone.store import to_ui_claim
claim = _make_claim("CLM-2")
claim.validation = ValidationReport(
passed=True, errors=[],
warnings=[ValidationIssue(rule="W001", severity="warning", message="soft")],
)
out = to_ui_claim(claim, batch_id="b1", parsed_at="2026-06-19T12:00:00Z")
assert out["status"] == "pending"
def test_to_ui_claim_maps_status_denied_when_failed():
from cyclone.store import to_ui_claim
claim = _make_claim("CLM-3")
claim.validation = ValidationReport(
passed=False,
errors=[ValidationIssue(rule="E001", severity="error", message="hard")],
warnings=[],
)
out = to_ui_claim(claim, batch_id="b1", parsed_at="2026-06-19T12:00:00Z")
assert out["status"] == "denied"
def test_to_ui_remittance_maps_status_received_for_paid():
from cyclone.store import to_ui_remittance
cp = ClaimPayment(
payer_claim_control_number="PCN-1",
status_code="1",
status_label="Processed as Primary",
total_charge=Decimal("100.00"),
total_paid=Decimal("80.00"),
patient_responsibility=Decimal("0.00"),
claim_filing_indicator="MC",
original_claim_id=None,
frequency_code=None,
facility_type=None,
service_payments=[],
)
out = to_ui_remittance(cp, batch_id="b1", parsed_at="2026-06-19T12:00:00Z")
assert out["status"] == "received"
def test_to_ui_provider_aggregates_claims():
from cyclone.store import to_ui_provider
out = to_ui_provider(
npi="1234567890", name="Acme Clinic", tax_id="99-9999999",
address="1 Main", city="Denver", state="CO", zip="80202", phone="303-555-0100",
claim_count=5, outstanding_ar=250.0,
)
assert out["npi"] == "1234567890"
assert out["claimCount"] == 5
def test_to_activity_event_uses_iso_timestamp():
from cyclone.store import to_activity_event
out = to_activity_event(
id="a1", kind="claim_submitted",
message="Claim CLM-1 submitted",
timestamp="2026-06-19T12:00:00Z",
npi="1234567890", amount=100.0,
)
assert out["id"] == "a1"
assert out["kind"] == "claim_submitted"
assert out["amount"] == 100.0
# ---------------------------------------------------------------------------
# Task 3: Iterators
# ---------------------------------------------------------------------------
def test_iter_claims_filters_by_status():
s = InMemoryStore()
result = _make_result()
a = _make_claim("A")
b = _make_claim("B")
b.validation = ValidationReport(
passed=True, errors=[],
warnings=[ValidationIssue(rule="W1", severity="warning", message="x")],
)
result.claims = [a, b]
s.add(BatchRecord(
id="1", kind="837p", input_filename="a.txt",
parsed_at="2026-06-19T12:00:00Z", result=result,
))
out = list(s.iter_claims(status="submitted"))
assert all(c["status"] == "submitted" for c in out)
assert any(c["id"] == "A" for c in out)
out_pending = list(s.iter_claims(status="pending"))
assert all(c["status"] == "pending" for c in out_pending)
assert any(c["id"] == "B" for c in out_pending)
def test_iter_claims_filters_by_batch_id():
s = InMemoryStore()
s.add(BatchRecord(
id="b1", kind="837p", input_filename="a.txt",
parsed_at="2026-06-19T12:00:00Z", result=_make_result(),
))
out = list(s.iter_claims(batch_id="b1"))
assert all(c["batchId"] == "b1" for c in out)
def test_iter_claims_sorts_by_billed_amount_desc():
s = InMemoryStore()
result = _make_result()
result.claims = [
_make_claim("cheap"),
_make_claim("expensive"),
]
result.claims[0].claim.total_charge = Decimal("50.00")
result.claims[1].claim.total_charge = Decimal("500.00")
s.add(BatchRecord(
id="b1", kind="837p", input_filename="a.txt",
parsed_at="2026-06-19T12:00:00Z", result=result,
))
out = list(s.iter_claims(sort="billedAmount", order="desc"))
assert out[0]["id"] == "expensive"
assert out[1]["id"] == "cheap"
def test_distinct_providers_aggregates_claims():
s = InMemoryStore()
result = _make_result()
result.claims = [_make_claim("A"), _make_claim("B")]
s.add(BatchRecord(
id="b1", kind="837p", input_filename="a.txt",
parsed_at="2026-06-19T12:00:00Z", result=result,
))
out = s.distinct_providers()
assert len(out) == 1
assert out[0]["npi"] == "1234567890"
assert out[0]["claimCount"] == 2
def test_recent_activity_returns_newest_first():
s = InMemoryStore()
s.add(BatchRecord(
id="b1", kind="837p", input_filename="a.txt",
parsed_at="2026-06-19T11:00:00Z", result=_make_result(),
))
s.add(BatchRecord(
id="b2", kind="837p", input_filename="b.txt",
parsed_at="2026-06-19T12:00:00Z", result=_make_result(),
))
out = s.recent_activity(limit=10)
assert out[0]["timestamp"] == "2026-06-19T12:00:00Z"
assert out[1]["timestamp"] == "2026-06-19T11:00:00Z"
Reference in New Issue View Git Blame Copy Permalink