Tyler
296 lines
9.8 KiB
Python
296 lines
9.8 KiB
Python
"""Tests for ORM model definitions and CRUD."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import enum
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from cyclone import db
|
|
from cyclone.db import Base, Batch, CasAdjustment, Claim, ClaimState, Remittance
|
|
|
|
|
|
@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()
|
|
|
|
|
|
def test_claim_state_enum_values():
|
|
assert ClaimState.SUBMITTED.value == "submitted"
|
|
assert ClaimState.RECEIVED.value == "received"
|
|
assert ClaimState.PAID.value == "paid"
|
|
assert ClaimState.PARTIAL.value == "partial"
|
|
assert ClaimState.DENIED.value == "denied"
|
|
assert ClaimState.RECONCILED.value == "reconciled"
|
|
assert ClaimState.REVERSED.value == "reversed"
|
|
# 7 values total.
|
|
assert len(list(ClaimState)) == 7
|
|
|
|
|
|
def test_create_and_query_batch():
|
|
with db.SessionLocal()() as s:
|
|
b = Batch(
|
|
id="b1",
|
|
kind="837p",
|
|
input_filename="test.txt",
|
|
parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
|
|
totals_json={"total_claims": 1},
|
|
validation_json={"passed": True, "warnings": [], "errors": []},
|
|
raw_result_json={"_": "stub"},
|
|
)
|
|
s.add(b)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Batch, "b1")
|
|
assert loaded is not None
|
|
assert loaded.kind == "837p"
|
|
assert loaded.totals_json == {"total_claims": 1}
|
|
|
|
|
|
def test_create_and_query_claim_with_default_state():
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
c = Claim(
|
|
id="CLM-1",
|
|
batch_id="b1",
|
|
patient_control_number="CLM-1",
|
|
service_date_from=date(2026, 6, 1),
|
|
service_date_to=date(2026, 6, 1),
|
|
charge_amount=Decimal("124.00"),
|
|
provider_npi="1234567890",
|
|
payer_id="SKCO0",
|
|
state=ClaimState.SUBMITTED,
|
|
)
|
|
s.add(c)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Claim, "CLM-1")
|
|
assert loaded is not None
|
|
assert loaded.state == ClaimState.SUBMITTED
|
|
assert loaded.charge_amount == Decimal("124.00")
|
|
assert loaded.service_date_from == date(2026, 6, 1)
|
|
|
|
|
|
def test_claim_uses_db_default_when_state_omitted():
|
|
"""The `state` column has a DB-side DEFAULT of 'submitted' (per migration
|
|
and ORM `default=`). When the caller omits `state=`, the loaded value
|
|
must be ClaimState.SUBMITTED, not None."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
c = Claim(
|
|
id="CLM-1",
|
|
batch_id="b1",
|
|
patient_control_number="CLM-1",
|
|
charge_amount=Decimal("0"),
|
|
)
|
|
s.add(c)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Claim, "CLM-1")
|
|
assert loaded is not None
|
|
assert loaded.state == ClaimState.SUBMITTED
|
|
|
|
|
|
def test_state_before_reversal_round_trips():
|
|
"""`state_before_reversal` is nullable but stores a ClaimState enum.
|
|
Setting it to PAID must round-trip as ClaimState.PAID after commit+reload."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
c = Claim(
|
|
id="CLM-1",
|
|
batch_id="b1",
|
|
patient_control_number="CLM-1",
|
|
charge_amount=Decimal("50.00"),
|
|
state_before_reversal=ClaimState.PAID,
|
|
)
|
|
s.add(c)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Claim, "CLM-1")
|
|
assert loaded is not None
|
|
assert loaded.state_before_reversal == ClaimState.PAID
|
|
|
|
|
|
def test_batch_cascade_delete_drops_claims():
|
|
"""Deleting a Batch must cascade-delete its claims (FK is ON DELETE CASCADE
|
|
in migration; ORM relationship uses cascade='all, delete-orphan' which makes
|
|
SQLAlchemy emit the child DELETE at flush time even without
|
|
PRAGMA foreign_keys=ON)."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
s.add(Claim(
|
|
id="CLM-1",
|
|
batch_id="b1",
|
|
patient_control_number="CLM-1",
|
|
charge_amount=Decimal("0"),
|
|
))
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
batch = s.get(Batch, "b1")
|
|
assert batch is not None
|
|
s.delete(batch)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
assert s.get(Batch, "b1") is None
|
|
assert s.get(Claim, "CLM-1") is None
|
|
|
|
|
|
def test_create_and_query_remittance():
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="835", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
r = Remittance(
|
|
id="CLP-1",
|
|
batch_id="b1",
|
|
payer_claim_control_number="CLP-1",
|
|
status_code="1",
|
|
total_charge=Decimal("124.00"),
|
|
total_paid=Decimal("124.00"),
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
service_date=date(2026, 6, 1),
|
|
is_reversal=False,
|
|
)
|
|
s.add(r)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Remittance, "CLP-1")
|
|
assert loaded is not None
|
|
assert loaded.status_code == "1"
|
|
assert loaded.total_paid == Decimal("124.00")
|
|
assert loaded.is_reversal is False
|
|
assert loaded.adjustment_amount == Decimal("0") # default
|
|
|
|
|
|
def test_cas_adjustment_aggregation():
|
|
"""Sum of CAS rows for a Remittance equals expected total."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="835", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
r = Remittance(
|
|
id="CLP-1", batch_id="b1",
|
|
payer_claim_control_number="CLP-1", status_code="1",
|
|
total_charge=Decimal("124.00"), total_paid=Decimal("62.00"),
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
)
|
|
s.add(r)
|
|
s.flush()
|
|
s.add_all([
|
|
CasAdjustment(remittance_id="CLP-1", group_code="CO", reason_code="45",
|
|
amount=Decimal("30.00")),
|
|
CasAdjustment(remittance_id="CLP-1", group_code="PR", reason_code="1",
|
|
amount=Decimal("32.00")),
|
|
])
|
|
s.commit()
|
|
|
|
from sqlalchemy import select, func
|
|
with db.SessionLocal()() as s:
|
|
total = s.execute(
|
|
select(func.sum(CasAdjustment.amount))
|
|
.where(CasAdjustment.remittance_id == "CLP-1")
|
|
).scalar_one()
|
|
assert total == Decimal("62.00")
|
|
|
|
|
|
def test_remittance_raw_json_round_trips_dict():
|
|
"""JSONText on Remittance.raw_json round-trips a dict, not a string."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="835", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
r = Remittance(
|
|
id="CLP-1", batch_id="b1", payer_claim_control_number="CLP-1",
|
|
status_code="1",
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
raw_json={"x12_segments": ["CLP", "NM1", "SVC"], "version": "005010X221A1"},
|
|
)
|
|
s.add(r)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Remittance, "CLP-1")
|
|
assert loaded.raw_json == {"x12_segments": ["CLP", "NM1", "SVC"], "version": "005010X221A1"}
|
|
|
|
|
|
def test_remittance_unique_constraint_rejects_duplicate_pcn():
|
|
"""UNIQUE (batch_id, payer_claim_control_number) blocks duplicates."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="835", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
s.add(Remittance(
|
|
id="CLP-1", batch_id="b1", payer_claim_control_number="SAME",
|
|
status_code="1",
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
s.add(Remittance(
|
|
id="CLP-2", batch_id="b1", payer_claim_control_number="SAME",
|
|
status_code="2",
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
with pytest.raises(IntegrityError):
|
|
s.commit()
|
|
|
|
|
|
def test_remittance_cascade_delete_drops_cas_adjustments():
|
|
"""Deleting a Remittance drops its CAS rows via ORM cascade."""
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id="b1", kind="835", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
))
|
|
r = Remittance(
|
|
id="CLP-1", batch_id="b1", payer_claim_control_number="CLP-1",
|
|
status_code="1", total_paid=Decimal("62.00"),
|
|
received_at=datetime(2026, 6, 19, tzinfo=timezone.utc),
|
|
)
|
|
s.add(r)
|
|
s.flush()
|
|
s.add(CasAdjustment(remittance_id="CLP-1", group_code="CO",
|
|
reason_code="45", amount=Decimal("30.00")))
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
r = s.get(Remittance, "CLP-1")
|
|
s.delete(r)
|
|
s.commit()
|
|
|
|
with db.SessionLocal()() as s:
|
|
from sqlalchemy import select
|
|
cas = s.execute(
|
|
select(CasAdjustment).where(CasAdjustment.remittance_id == "CLP-1")
|
|
).all()
|
|
assert len(cas) == 0
|