Tyler
b5e27927e0
Three related changes for real CO Medicaid data: 1. Drop UNIQUE(batch_id, patient_control_number) on claims and UNIQUE(batch_id, payer_claim_control_number) on remittances. The X12 spec allows multiple CLM segments per 2000B subscriber loop and 835 ERAs can repeat a payer_claim_control_number for reversals. Claim/ remittance identity is provided by the primary key (claims.id = CLM01, remittances.id = CLP01). 2. Add validator rule R835_MULTI_BPR warning for files with multiple BPR segments (CO Medicaid split-payment pattern). The parser already sums BPR02 paid_amounts; this surfaces the non-standard data to operators. 3. Skip R835_BAL_BPR_vs_CLP04 when BPR01='I' (Information Only 835). In that mode BPR02 is informational and the per-claim CLP04 totals are authoritative — a diff is expected, not an error. Migration 0003 handles the drop with IF EXISTS so fresh DBs skip cleanly. Updates affected tests to reflect new schema (no UNIQUE constraint on batch_id + patient_control_number / payer_claim_control_number). Fixes test_api_835::test_prodfile_round_trip_persists_separately which was failing on real production data.
126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""Tests for the 999 ACK migration + ORM + store helpers (SP3 P3 T13)."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
|
|
from cyclone import db, db_migrate
|
|
from cyclone.db import Ack
|
|
from cyclone.store import store
|
|
|
|
|
|
@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_batch(batch_id: str = "b-1") -> None:
|
|
"""Insert a stub batches row so the acks.source_batch_id FK resolves."""
|
|
from cyclone.db import Batch
|
|
with db.SessionLocal()() as s:
|
|
s.add(Batch(
|
|
id=batch_id, kind="837p", input_filename="x",
|
|
parsed_at=datetime(2026, 6, 20, 12, 0, tzinfo=timezone.utc),
|
|
))
|
|
s.commit()
|
|
|
|
|
|
def test_migration_0002_creates_acks_table():
|
|
"""On a fresh DB the migration runner must create the `acks` table."""
|
|
inspector = sa.inspect(db.engine())
|
|
assert "acks" in inspector.get_table_names()
|
|
# The index declared in 0002_acks.sql must also exist.
|
|
indexes = {ix["name"] for ix in inspector.get_indexes("acks")}
|
|
assert "ix_acks_source_batch_id" in indexes
|
|
# And the column list matches the spec.
|
|
cols = {c["name"] for c in inspector.get_columns("acks")}
|
|
expected = {
|
|
"id", "source_batch_id", "accepted_count", "rejected_count",
|
|
"received_count", "ack_code", "parsed_at", "raw_json",
|
|
}
|
|
assert expected <= cols # spec columns are all present
|
|
|
|
|
|
def test_migration_latest_idempotent_on_fresh_db():
|
|
"""Re-running the migration on the same DB must be a no-op (PRAGMA
|
|
user_version already at the latest version — currently 3 after the
|
|
0003 constraint drop was added)."""
|
|
with db.engine().begin() as c:
|
|
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
assert v1 == 3
|
|
# A second run should not raise and should not bump the version.
|
|
db_migrate.run(db.engine())
|
|
with db.engine().begin() as c:
|
|
v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
assert v2 == 3
|
|
|
|
|
|
def test_add_ack_persists_row():
|
|
_make_batch("b-1")
|
|
row = store.add_ack(
|
|
source_batch_id="b-1",
|
|
accepted_count=3,
|
|
rejected_count=1,
|
|
received_count=4,
|
|
ack_code="P",
|
|
raw_json={"envelope": {"control_number": "0001"}, "set_count": 1},
|
|
)
|
|
assert row.id is not None
|
|
assert row.source_batch_id == "b-1"
|
|
assert row.accepted_count == 3
|
|
assert row.rejected_count == 1
|
|
assert row.received_count == 4
|
|
assert row.ack_code == "P"
|
|
assert row.raw_json == {"envelope": {"control_number": "0001"}, "set_count": 1}
|
|
# Round-trip: fetch from the same session.
|
|
with db.SessionLocal()() as s:
|
|
loaded = s.get(Ack, row.id)
|
|
assert loaded is not None
|
|
assert loaded.ack_code == "P"
|
|
assert loaded.raw_json == {"envelope": {"control_number": "0001"}, "set_count": 1}
|
|
|
|
|
|
def test_list_acks_newest_first():
|
|
_make_batch("b-1")
|
|
# Insert two rows in a known order; the second must come first.
|
|
a1 = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=1, rejected_count=0,
|
|
received_count=1, ack_code="A", raw_json={"order": 1},
|
|
)
|
|
a2 = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=0, rejected_count=1,
|
|
received_count=1, ack_code="R", raw_json={"order": 2},
|
|
)
|
|
rows = store.list_acks()
|
|
assert len(rows) == 2
|
|
# Newest first means a2 should be at index 0 (auto-increment id desc).
|
|
assert rows[0].id == a2.id
|
|
assert rows[1].id == a1.id
|
|
assert rows[0].ack_code == "R"
|
|
assert rows[1].ack_code == "A"
|
|
|
|
|
|
def test_get_ack_returns_none_for_missing():
|
|
assert store.get_ack(99999) is None
|
|
|
|
|
|
def test_get_ack_returns_row_when_present():
|
|
_make_batch("b-1")
|
|
row = store.add_ack(
|
|
source_batch_id="b-1", accepted_count=1, rejected_count=0,
|
|
received_count=1, ack_code="A", raw_json={"hello": "world"},
|
|
)
|
|
fetched = store.get_ack(row.id)
|
|
assert fetched is not None
|
|
assert fetched.id == row.id
|
|
assert fetched.source_batch_id == "b-1"
|
|
assert fetched.raw_json == {"hello": "world"}
|