Tyler
c8bc2c73e6
Replace the in-memory InMemoryStore with a SQLAlchemy-backed CycloneStore that persists Batch, Claim, Remittance, Match, and ActivityEvent rows to a configurable DB engine (sqlite by default, overridable via CYCLONE_DB_URL). Public API preserved: add / get_batch / iter_claims / iter_remittances / distinct_providers / recent_activity. New T10/T12 stubs: list_unmatched, manual_match, manual_unmatch. Backward-compat shims for tests that called ._batches.clear() or acquired ._lock as a context manager. Idempotency: add() does a per-row session.get(Claim, c.id) (and s.get(Remittance, ...)) check before each insert; duplicates are skipped with a warning. This makes re-uploading the same fixture idempotent instead of raising IntegrityError on the PK. Auto-init DB fixture (backend/tests/conftest.py) sets CYCLONE_DB_URL to a per-test sqlite file and calls db.init_db() once per test, replacing the old module-scoped in-memory fixture.
123 lines
3.6 KiB
Python
123 lines
3.6 KiB
Python
"""Tests for the SQLAlchemy-backed CycloneStore.
|
|
|
|
Public API is preserved from the in-memory version: add / get_batch /
|
|
iter_claims / iter_remittances / distinct_providers / recent_activity.
|
|
New methods: list_unmatched / manual_match / manual_unmatch.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from cyclone import db
|
|
from cyclone.store import BatchRecord, CycloneStore, store
|
|
from cyclone.parsers.models import (
|
|
Address,
|
|
BatchSummary,
|
|
BillingProvider,
|
|
ClaimHeader,
|
|
ClaimOutput,
|
|
Envelope,
|
|
Payer,
|
|
ParseResult,
|
|
Subscriber,
|
|
ValidationReport,
|
|
)
|
|
|
|
|
|
@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 _make_claim_837(claim_id: str = "CLM-1", charge: str = "124.00") -> ClaimOutput:
|
|
return ClaimOutput(
|
|
claim_id=claim_id,
|
|
control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
billing_provider=BillingProvider(
|
|
name="Test", npi="1234567890",
|
|
),
|
|
# Member_id doubles as patient_control_number in the DB; the
|
|
# ``uq_claims_batch_pcn`` unique constraint requires it to be
|
|
# distinct within a batch, so derive it from the claim id.
|
|
subscriber=Subscriber(
|
|
first_name="Jane", last_name="Doe", 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="1", place_of_service="11",
|
|
),
|
|
diagnoses=[],
|
|
service_lines=[],
|
|
validation=ValidationReport(passed=True, errors=[], warnings=[]),
|
|
raw_segments=[],
|
|
)
|
|
|
|
|
|
def _make_result_837() -> ParseResult:
|
|
return ParseResult(
|
|
envelope=Envelope(
|
|
sender_id="S", receiver_id="R", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
),
|
|
claims=[_make_claim_837("CLM-1"), _make_claim_837("CLM-2", "88.50")],
|
|
summary=BatchSummary(
|
|
input_file="test.txt", control_number="0001",
|
|
transaction_date=date(2026, 6, 19),
|
|
total_claims=2, passed=2, failed=0,
|
|
),
|
|
)
|
|
|
|
|
|
def _make_batch_record(kind: str = "837p", result=None, batch_id: str = "b-uuid-1"):
|
|
return BatchRecord(
|
|
id=batch_id, kind=kind, input_filename="test.txt",
|
|
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
|
|
result=result or _make_result_837(),
|
|
)
|
|
|
|
|
|
def test_module_singleton():
|
|
assert isinstance(store, CycloneStore)
|
|
|
|
|
|
def test_add_837_persists_batch_and_claims():
|
|
s = CycloneStore()
|
|
rec = _make_batch_record()
|
|
s.add(rec)
|
|
# New session confirms persistence.
|
|
with db.SessionLocal()() as session:
|
|
from cyclone.db import Batch, Claim, ClaimState
|
|
b = session.get(Batch, "b-uuid-1")
|
|
assert b is not None
|
|
assert b.kind == "837p"
|
|
claims = session.query(Claim).all()
|
|
assert len(claims) == 2
|
|
assert all(c.state == ClaimState.SUBMITTED for c in claims)
|
|
|
|
|
|
def test_iter_claims_returns_dicts():
|
|
s = CycloneStore()
|
|
s.add(_make_batch_record())
|
|
rows = s.iter_claims()
|
|
assert len(rows) == 2
|
|
assert all("id" in r and "state" in r for r in rows)
|
|
|
|
|
|
def test_persistence_across_session():
|
|
s1 = CycloneStore()
|
|
s1.add(_make_batch_record(batch_id="b-x"))
|
|
|
|
s2 = CycloneStore() # fresh instance, same engine
|
|
loaded = s2.get_batch("b-x")
|
|
assert loaded is not None
|
|
assert loaded["kind"] == "837p" |