AI Tools for Automated Migration Testing 2026

# tests/test_migration_add_notification_enabled.py
import pytest
from alembic.config import Config
from alembic import command
from sqlalchemy import create_engine, text, inspect
from sqlalchemy.exc import IntegrityError

MIGRATION_ID = "abc123def456"  # ID of the migration under test
PREV_MIGRATION = "prev_migration_id"

@pytest.fixture(scope="module")
def db_engine():
    engine = create_engine("postgresql://test:test@localhost/test_migrations")
    yield engine
    engine.dispose()

@pytest.fixture
def migrated_db(db_engine):
    """Apply migration, yield engine, then rollback."""
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", str(db_engine.url))

    # Roll back to previous state
    command.downgrade(alembic_cfg, PREV_MIGRATION)

    # Seed test data before migration
    with db_engine.connect() as conn:
        conn.execute(text(
            "INSERT INTO users (email, created_at) VALUES "
            "('pre@example.com', NOW()), ('pre2@example.com', NOW())"
        ))
        conn.commit()

    # Apply the migration under test
    command.upgrade(alembic_cfg, MIGRATION_ID)

    yield db_engine

    # Cleanup
    command.downgrade(alembic_cfg, PREV_MIGRATION)

class TestUpgradeMigration:
    def test_column_exists_after_upgrade(self, migrated_db):
        inspector = inspect(migrated_db)
        columns = {c["name"]: c for c in inspector.get_columns("users")}
        assert "notification_enabled" in columns

    def test_column_is_boolean_not_nullable(self, migrated_db):
        inspector = inspect(migrated_db)
        col = next(c for c in inspector.get_columns("users")
                   if c["name"] == "notification_enabled")
        assert col["type"].__class__.__name__ == "BOOLEAN"
        assert col["nullable"] is False

    def test_existing_rows_have_default_true(self, migrated_db):
        with migrated_db.connect() as conn:
            result = conn.execute(text(
                "SELECT COUNT(*) FROM users WHERE notification_enabled IS NULL"
            ))
            null_count = result.scalar()
        assert null_count == 0

        with migrated_db.connect() as conn:
            result = conn.execute(text(
                "SELECT COUNT(*) FROM users WHERE notification_enabled = true"
            ))
            true_count = result.scalar()
        assert true_count >= 2  # our pre-seeded rows

    def test_new_row_inserts_without_column(self, migrated_db):
        with migrated_db.connect() as conn:
            conn.execute(text(
                "INSERT INTO users (email, created_at) VALUES ('new@example.com', NOW())"
            ))
            conn.commit()
            result = conn.execute(text(
                "SELECT notification_enabled FROM users WHERE email = 'new@example.com'"
            ))
            row = result.fetchone()
        assert row is not None

    def test_migration_uses_add_column_with_default_not_update(self, migrated_db):
        """Verify the migration doesn't do a separate UPDATE (which locks the table)."""
        migration_file = open("alembic/versions/abc123def456_add_notification.py").read()
        assert "op.execute" not in migration_file or "UPDATE" not in migration_file
        assert "server_default='true'" in migration_file


class TestDowngradeMigration:
    def test_column_removed_after_downgrade(self, db_engine):
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", str(db_engine.url))

        command.upgrade(alembic_cfg, MIGRATION_ID)
        command.downgrade(alembic_cfg, PREV_MIGRATION)

        inspector = inspect(db_engine)
        columns = {c["name"] for c in inspector.get_columns("users")}
        assert "notification_enabled" not in columns

    def test_existing_data_preserved_after_downgrade(self, db_engine):
        """Rows inserted before migration still exist after rollback."""
        with db_engine.connect() as conn:
            result = conn.execute(text(
                "SELECT COUNT(*) FROM users WHERE email = 'pre@example.com'"
            ))
        assert result.scalar() >= 1

# Test that existing 2M rows were backfilled correctly
# Test rollback removes column without data loss
# Test migration holds no exclusive lock (check for CONCURRENT index creation)
def test_migration():
    # Copilot will generate based on comments above
    ...

class MigrationStateMachine(TestCase):
 def test_pre_migration_state(self):
 # Verify schema before
 pass

 def test_post_migration_state(self):
 # Verify schema after
 pass

 def test_rollback_from_post_migration_state(self):
 # Verify rollback works
 pass

 def test_idempotent_migration(self):
 # Verify running migration twice is safe
 pass

def test_data_integrity_after_migration(migrated_db):
 """Verify data hasn't been corrupted by the migration."""
 # Compute checksum before
 before_query = text("""
 SELECT COUNT(*), SUM(id), MIN(created_at), MAX(created_at)
 FROM users
 """)

 # Verify counts match after migration
 with migrated_db.connect() as conn:
 result = conn.execute(before_query)
 count, id_sum, min_date, max_date = result.fetchone()

 assert count == 2 # Our seeded rows
 assert id_sum is not None