Tyler
218 lines
9.8 KiB
Python
218 lines
9.8 KiB
Python
"""Tests for the PRAGMA user_version migration runner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
import sqlalchemy.exc
|
|
|
|
from cyclone import db_migrate
|
|
|
|
|
|
def _fresh_engine(tmp_path: Path) -> sa.Engine:
|
|
return sa.create_engine(f"sqlite:///{tmp_path}/m.db", future=True)
|
|
|
|
|
|
def _user_version(engine: sa.Engine) -> int:
|
|
with engine.connect() as c:
|
|
return c.exec_driver_sql("PRAGMA user_version").scalar() or 0
|
|
|
|
|
|
def test_run_on_empty_db_bumps_to_latest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine)
|
|
|
|
assert _user_version(engine) == 1
|
|
with engine.connect() as c:
|
|
rows = c.exec_driver_sql("SELECT name FROM sqlite_master WHERE type='table' AND name='t'").all()
|
|
assert len(rows) == 1
|
|
|
|
|
|
def test_run_is_idempotent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t (id INTEGER);\n")
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine)
|
|
db_migrate.run(engine) # second call should be a no-op
|
|
|
|
assert _user_version(engine) == 1
|
|
|
|
|
|
def test_run_skips_already_applied_migrations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_initial.sql").write_text("-- version: 1\nCREATE TABLE t1 (id INTEGER);\n")
|
|
(tmp_path / "0002_add_col.sql").write_text("-- version: 2\nALTER TABLE t1 ADD COLUMN name TEXT;\n")
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine)
|
|
assert _user_version(engine) == 2
|
|
|
|
# Modify 0002 (a no-op for SQLite ALTER, but we want to verify the
|
|
# runner doesn't re-run already-applied migrations).
|
|
db_migrate.run(engine)
|
|
assert _user_version(engine) == 2
|
|
|
|
|
|
def test_run_raises_on_missing_version_header(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_bad.sql").write_text("CREATE TABLE t (id INTEGER);\n") # no header
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
with pytest.raises(RuntimeError, match="missing '-- version"):
|
|
db_migrate.run(engine)
|
|
|
|
|
|
def test_run_rolls_back_version_on_failed_statement(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""If a DDL statement fails, the version bump must also roll back."""
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_initial.sql").write_text(
|
|
"-- version: 1\nCREATE TABLE t (id INTEGER);\n"
|
|
)
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine) # applies 0001, user_version=1
|
|
assert _user_version(engine) == 1
|
|
|
|
# Now drop in a second migration whose DDL conflicts with the first.
|
|
# The runner should apply 0001 and 0002 in one call; 0002 fails, the
|
|
# transaction rolls back, and user_version must stay at 1 (not bump to 2).
|
|
(tmp_path / "0002_bad.sql").write_text(
|
|
"-- version: 2\nCREATE TABLE t (id INTEGER);\n" # table already exists!
|
|
)
|
|
|
|
with pytest.raises(sqlalchemy.exc.SQLAlchemyError):
|
|
db_migrate.run(engine)
|
|
|
|
assert _user_version(engine) == 1 # bumped atomically, rolled back on failure
|
|
|
|
|
|
def test_run_ignores_non_sql_files(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""README.md and similar files in MIGRATIONS_DIR must not be executed."""
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
(tmp_path / "0001_initial.sql").write_text(
|
|
"-- version: 1\nCREATE TABLE t (id INTEGER);\n"
|
|
)
|
|
(tmp_path / "README.md").write_text("Don't run me!\nCREATE TABLE should_not_exist (x INT);\n")
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine)
|
|
|
|
with engine.connect() as c:
|
|
rows = c.exec_driver_sql(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='should_not_exist'"
|
|
).all()
|
|
assert len(rows) == 0
|
|
assert _user_version(engine) == 1
|
|
|
|
|
|
def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Migration 0013 must drop the inline UNIQUE(batch_id, patient_control_number)
|
|
on claims by recreating the table. Idempotent and preserves data.
|
|
|
|
We copy the real 0013 file (so this test exercises the real migration, not
|
|
a synthetic one) and stub 0001-0012 with a synthetic 0001 that has the
|
|
historical inline UNIQUE. If 0013 doesn't exist as a real file, the runner
|
|
will leave user_version at 1 and the test will fail.
|
|
"""
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
|
|
# Synthetic 0001: claims with INLINE UNIQUE(batch_id, patient_control_number).
|
|
# The schema mirrors the real 0001 column shape so 0013's
|
|
# ``INSERT INTO claims_new SELECT * FROM claims`` works against this fixture.
|
|
(tmp_path / "0001_initial.sql").write_text(
|
|
"-- version: 1\n"
|
|
"CREATE TABLE batches (id TEXT PRIMARY KEY, kind TEXT NOT NULL, "
|
|
"input_filename TEXT NOT NULL, parsed_at DATETIME NOT NULL);\n"
|
|
"CREATE TABLE claims ("
|
|
"id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
|
|
"patient_control_number TEXT NOT NULL, service_date_from DATE, service_date_to DATE,"
|
|
"charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0, provider_npi TEXT, payer_id TEXT,"
|
|
"state TEXT NOT NULL DEFAULT 'submitted', state_before_reversal TEXT,"
|
|
"matched_remittance_id TEXT REFERENCES remittances(id), raw_json TEXT,"
|
|
"rejection_reason TEXT, rejected_at TIMESTAMP, resubmit_count INTEGER NOT NULL DEFAULT 0,"
|
|
"state_changed_at TIMESTAMP, payer_rejected_at TEXT, payer_rejected_reason TEXT,"
|
|
"payer_rejected_status_code TEXT, payer_rejected_by_277ca_id TEXT,"
|
|
"payer_rejected_acknowledged_at TEXT, payer_rejected_acknowledged_actor TEXT,"
|
|
"UNIQUE (batch_id, patient_control_number));\n"
|
|
)
|
|
|
|
# Copy the REAL 0013 from the source tree. If it's missing, the test fails
|
|
# because user_version will stay at 1 and we'll never get to assert it == 13.
|
|
real_migrations = Path(db_migrate.__file__).parent / "migrations"
|
|
real_0013 = real_migrations / "0013_drop_claims_unique_constraint.sql"
|
|
assert real_0013.exists(), (
|
|
f"Real migration 0013 missing at {real_0013} — this test is meant to "
|
|
f"exercise the real migration, not a synthetic one."
|
|
)
|
|
(tmp_path / "0013_drop_claims_unique_constraint.sql").write_text(real_0013.read_text())
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine)
|
|
assert _user_version(engine) == 13, (
|
|
"0013 did not apply; user_version did not reach 13. Either the migration "
|
|
"file is missing, or it has a syntax error."
|
|
)
|
|
|
|
# Before-0013 assertion: with the inline UNIQUE, two claims sharing
|
|
# (batch_id, patient_control_number) on the same batch must fail. The
|
|
# recreation must remove that constraint, so this insert must now succeed.
|
|
with engine.begin() as c:
|
|
c.exec_driver_sql(
|
|
"INSERT INTO batches(id, kind, input_filename, parsed_at) "
|
|
"VALUES ('b1', '837p', 'x.txt', '2026-01-01')"
|
|
)
|
|
c.exec_driver_sql(
|
|
"INSERT INTO claims(id, batch_id, patient_control_number) VALUES ('c1', 'b1', 'M')"
|
|
)
|
|
c.exec_driver_sql(
|
|
"INSERT INTO claims(id, batch_id, patient_control_number) VALUES ('c2', 'b1', 'M')"
|
|
)
|
|
ids = [r[0] for r in c.exec_driver_sql("SELECT id FROM claims ORDER BY id").all()]
|
|
assert ids == ["c1", "c2"], (
|
|
"Both claims with same (batch_id, patient_control_number) must persist "
|
|
"after 0013 — the UNIQUE(batch_id, patient_control_number) is gone."
|
|
)
|
|
|
|
|
|
def test_migration_0013_is_idempotent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Re-running db_migrate.run on a v13 DB is a no-op."""
|
|
monkeypatch.setattr(db_migrate, "MIGRATIONS_DIR", tmp_path)
|
|
|
|
# Use synthetic 0001 + copy the real 0013 so we exercise the real file.
|
|
(tmp_path / "0001_initial.sql").write_text(
|
|
"-- version: 1\n"
|
|
"CREATE TABLE batches (id TEXT PRIMARY KEY, kind TEXT NOT NULL, "
|
|
"input_filename TEXT NOT NULL, parsed_at DATETIME NOT NULL);\n"
|
|
"CREATE TABLE claims ("
|
|
"id TEXT PRIMARY KEY, batch_id TEXT NOT NULL REFERENCES batches(id) ON DELETE CASCADE,"
|
|
"patient_control_number TEXT NOT NULL, service_date_from DATE, service_date_to DATE,"
|
|
"charge_amount NUMERIC(12, 2) NOT NULL DEFAULT 0, provider_npi TEXT, payer_id TEXT,"
|
|
"state TEXT NOT NULL DEFAULT 'submitted', state_before_reversal TEXT,"
|
|
"matched_remittance_id TEXT REFERENCES remittances(id), raw_json TEXT,"
|
|
"rejection_reason TEXT, rejected_at TIMESTAMP, resubmit_count INTEGER NOT NULL DEFAULT 0,"
|
|
"state_changed_at TIMESTAMP, payer_rejected_at TEXT, payer_rejected_reason TEXT,"
|
|
"payer_rejected_status_code TEXT, payer_rejected_by_277ca_id TEXT,"
|
|
"payer_rejected_acknowledged_at TEXT, payer_rejected_acknowledged_actor TEXT,"
|
|
"UNIQUE (batch_id, patient_control_number));\n"
|
|
)
|
|
real_migrations = Path(db_migrate.__file__).parent / "migrations"
|
|
real_0013 = real_migrations / "0013_drop_claims_unique_constraint.sql"
|
|
assert real_0013.exists(), "Real 0013 missing — test cannot exercise it."
|
|
(tmp_path / "0013_drop_claims_unique_constraint.sql").write_text(real_0013.read_text())
|
|
|
|
engine = _fresh_engine(tmp_path)
|
|
db_migrate.run(engine) # applies both
|
|
version_after_first = _user_version(engine)
|
|
assert version_after_first == 13
|
|
db_migrate.run(engine) # second call: no-op
|
|
assert _user_version(engine) == version_after_first |