"""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_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> 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. SP27 Task 11 bumped it to 16 with the claims.matched_remittance_id index (drift-check perf). SP27 Task 17 bumped it to 17 with the patient_control_number backfill UPDATE (the migration runner now applies DML too). SP28 bumped it to 18 with the claim_acks join table. SP32 bumped it to 19 with rendering_provider_npi + service_provider_npi on claims and remittances. SP37 bumped it to 20 with batch transaction_set_control_number. SP39 bumped it to 21 with the resubmissions audit table. SP41 bumped it to 22 with submission_dedup. """ engine = _fresh_engine(tmp_path) db_migrate.run(engine) v_after_first = _user_version(engine) assert v_after_first == 23, f"expected head=23, got {v_after_first}" db_migrate.run(engine) assert _user_version(engine) == 23, "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) == 23, f"expected head=23, 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, kind, input_filename, parsed_at) " "VALUES ('B1', '837p', 'test.txt', '2026-01-01 00:00:00')" ) 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] # --------------------------------------------------------------------------- # SP39: migration 0021 creates the resubmissions table # --------------------------------------------------------------------------- def test_migration_0021_creates_resubmissions_table(tmp_path: Path) -> None: """SP39: 0021_resubmissions.sql adds the resubmissions audit table with the documented columns + unique constraint on (claim_id, interchange_control_number).""" # Point the runner at the REAL migrations dir so we exercise 0021. real_dir = Path(db_migrate.__file__).parent / "migrations" import cyclone.db_migrate as real_migrate_mod real_dir_str = str(real_dir) engine = _fresh_engine(tmp_path) # monkeypatch the module-level MIGRATIONS_DIR import importlib monkey_save = real_migrate_mod.MIGRATIONS_DIR real_migrate_mod.MIGRATIONS_DIR = Path(real_dir_str) try: db_migrate.run(engine) finally: real_migrate_mod.MIGRATIONS_DIR = monkey_save # Verify the table exists with the expected columns + unique index. with engine.connect() as conn: version = conn.exec_driver_sql("PRAGMA user_version").scalar() assert version >= 21 cols = conn.exec_driver_sql( "PRAGMA table_info(resubmissions)" ).all() col_names = {row[1] for row in cols} assert col_names == { "id", "claim_id", "batch_id", "resubmitted_at", "source_corrected_path", "interchange_control_number", "group_control_number", } idx_rows = conn.exec_driver_sql( "SELECT name, sql FROM sqlite_master " "WHERE type='index' AND tbl_name='resubmissions'" ).all() idx_names = {row[0] for row in idx_rows} assert "ix_resubmissions_claim_id" in idx_names assert "ix_resubmissions_batch_id" in idx_names assert "ux_resubmissions_claim_icn" in idx_names