feat(sp22): migration 0015 — drop inline UNIQUE on claims, add tests

Adds migration 0015_drop_claims_unique_constraint.sql that recreates the
claims table without the inline UNIQUE(batch_id, patient_control_number)
constraint, plus tests proving the migration runs cleanly and the
constraint is not re-introduced.
This commit is contained in:
Tyler
2026-06-23 16:11:42 -06:00
parent 0c81968d0c
commit 4fd55dc33e
3 changed files with 145 additions and 5 deletions
+5 -4
View File
@@ -51,20 +51,21 @@ def test_migration_0002_creates_acks_table():
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 14 after
user_version already at the latest version — currently 15 after
0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
providers/payers/clearhouse, SP10's 0008 payer_rejected,
SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
SP16's 0011 processed_inbound_files, SP17's 0012 db_backups,
SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id)."""
SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id,
SP22's 0015 drop_claims_unique_constraint)."""
with db.engine().begin() as c:
v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
assert v1 == 14
assert v1 == 15
# 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 == 14
assert v2 == 15
def test_add_ack_persists_row():
+66 -1
View File
@@ -112,4 +112,69 @@ def test_run_ignores_non_sql_files(
"SELECT name FROM sqlite_master WHERE type='table' AND name='should_not_exist'"
).all()
assert len(rows) == 0
assert _user_version(engine) == 1
assert _user_version(engine) == 1
def test_migration_latest_idempotent_on_fresh_db() -> None:
"""All migrations up to the current head run cleanly on a fresh DB,
and a second run is a no-op (no version bump). SP22 bumped the
expected head from 14 to 15 with the new UNIQUE-drop migration.
"""
engine = _fresh_engine(Path("/tmp/sp22_fresh_db_latest_test.db"))
db_migrate.run(engine)
v_after_first = _user_version(engine)
assert v_after_first == 15, f"expected head=15, got {v_after_first}"
db_migrate.run(engine)
assert _user_version(engine) == 15, "second run should not bump version"
def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""SP22: migration 0015 recreates the `claims` table without the inline
`UNIQUE(batch_id, patient_control_number)` constraint, so two claims in
one batch can share a patient_control_number (real 837P multi-claim
subscriber loops do this).
Discovery (2026-06-23): the inline UNIQUE does not exist in the current
production DB or in main's fresh-DB schema, so this migration is a
defensive no-op against the current state. The test still proves
migration correctness: (a) all migrations up to v15 run cleanly,
(b) two rows with the same (batch_id, patient_control_number) can
be inserted after the migration (proving no UNIQUE was re-introduced
by the table recreation).
"""
# Real migrations dir so the test exercises the actual 0015 file.
monkeypatch.setattr(
db_migrate,
"MIGRATIONS_DIR",
Path(__file__).parent.parent / "src" / "cyclone" / "migrations",
)
engine = _fresh_engine(tmp_path)
db_migrate.run(engine)
assert _user_version(engine) == 15, f"expected head=15, got {_user_version(engine)}"
# Two claims in one batch with the same patient_control_number
# must be insertable. If 0015's table recreation re-introduced a
# UNIQUE(batch_id, patient_control_number), this would raise
# IntegrityError. (The test also implicitly asserts the FK from
# claims to batches still works after the recreation.)
with engine.begin() as conn:
conn.exec_driver_sql(
"INSERT INTO batches (id) VALUES ('B1')"
)
conn.exec_driver_sql(
"INSERT INTO claims (id, batch_id, patient_control_number, charge_amount) "
"VALUES ('CLM-1', 'B1', 'SAME-PCN', 100)"
)
conn.exec_driver_sql(
"INSERT INTO claims (id, batch_id, patient_control_number, charge_amount) "
"VALUES ('CLM-2', 'B1', 'SAME-PCN', 200)"
)
rows = conn.exec_driver_sql(
"SELECT id, charge_amount FROM claims "
"WHERE patient_control_number='SAME-PCN' ORDER BY id"
).all()
assert [r[0] for r in rows] == ["CLM-1", "CLM-2"]
assert [float(r[1]) for r in rows] == [100.0, 200.0]