Database DevOps for Test Automation: Flyway, Liquibase, and Schema Testing

-- V1__create_users_table.sql
CREATE TABLE users (
    id BIGSERIAL PRIMARY KEY,
    email VARCHAR(255) UNIQUE NOT NULL,
    username VARCHAR(100) UNIQUE NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_username ON users(username);

# tests/test_flyway_migrations.py
import pytest
import psycopg2
from flyway import Flyway

class FlywayMigrationTester:
    def __init__(self, db_config):
        self.db_config = db_config
        self.flyway = Flyway(
            url=f"jdbc:postgresql://{db_config['host']}:{db_config['port']}/{db_config['database']}",
            user=db_config['user'],
            password=db_config['password'],
            locations=["filesystem:migrations"]
        )

    def test_migration_execution(self):
        """Test migrations execute successfully"""
        # Clean database
        self.flyway.clean()

        # Run migrations
        result = self.flyway.migrate()

        assert result.success
        assert result.migrations_executed > 0
        assert result.target_schema_version is not None

    def test_migration_idempotency(self):
        """Test migrations are idempotent"""
        # Run migrations twice
        result1 = self.flyway.migrate()
        result2 = self.flyway.migrate()

        # Second run should not execute any migrations
        assert result2.migrations_executed == 0
        assert result1.target_schema_version == result2.target_schema_version

    def test_schema_validation(self):
        """Validate schema after migrations"""
        self.flyway.migrate()

        conn = psycopg2.connect(**self.db_config)
        cursor = conn.cursor()

        # Verify tables exist
        cursor.execute("""
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
        """)
        tables = [row[0] for row in cursor.fetchall()]
        assert 'users' in tables

        # Verify columns
        cursor.execute("""
            SELECT column_name, data_type, is_nullable
            FROM information_schema.columns
            WHERE table_name = 'users'
        """)
        columns = {row[0]: {'type': row[1], 'nullable': row[2]} for row in cursor.fetchall()}

        assert 'id' in columns
        assert columns['id']['type'] == 'bigint'
        assert columns['email']['nullable'] == 'NO'

        # Verify indexes
        cursor.execute("""
            SELECT indexname
            FROM pg_indexes
            WHERE tablename = 'users'
        """)
        indexes = [row[0] for row in cursor.fetchall()]
        assert 'idx_users_email' in indexes

        conn.close()

    def test_migration_rollback(self):
        """Test rollback to previous version"""
        # Migrate to latest
        self.flyway.migrate()

        # Record current version
        info = self.flyway.info()
        versions = [m.version for m in info.migrations if m.state == 'SUCCESS']

        if len(versions) > 1:
            target_version = versions[-2]

            # Rollback to previous version
            self.flyway.undo()

            # Verify rollback
            info_after = self.flyway.info()
            current_version = info_after.current_version
            assert current_version == target_version

# .gitlab-ci.yml
stages:
  - test
  - migrate

database_test:
  stage: test
  image: postgres:15
  services:
    - postgres:15
  variables:
    POSTGRES_DB: test_db
    POSTGRES_USER: test_user
    POSTGRES_PASSWORD: test_password
  before_script:
    - apt-get update && apt-get install -y python3-pip
    - pip3 install psycopg2-binary pytest flyway
  script:
    - pytest tests/test_flyway_migrations.py -v
  only:
    - merge_requests
    - main

migrate_staging:
  stage: migrate
  image: flyway/flyway:latest
  script:
    - flyway migrate
      -url=jdbc:postgresql://$DB_HOST:5432/$DB_NAME
      -user=$DB_USER
      -password=$DB_PASSWORD
      -locations=filesystem:migrations
      -validateMigrationNaming=true
      -outOfOrder=false
  environment:
    name: staging
  only:
    - main

migrate_production:
  stage: migrate
  image: flyway/flyway:latest
  script:
    - flyway info
      -url=jdbc:postgresql://$DB_HOST:5432/$DB_NAME
      -user=$DB_USER
      -password=$DB_PASSWORD
    - flyway migrate
      -url=jdbc:postgresql://$DB_HOST:5432/$DB_NAME
      -user=$DB_USER
      -password=$DB_PASSWORD
      -locations=filesystem:migrations
      -validateMigrationNaming=true
  environment:
    name: production
  when: manual
  only:
    - main

<!-- changesets/001-create-products-table.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<databaseChangeLog
    xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
        http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.0.xsd">

    <changeSet id="001" author="qa-team">
        <createTable tableName="products">
            <column name="id" type="BIGINT" autoIncrement="true">
                <constraints primaryKey="true" nullable="false"/>
            </column>
            <column name="name" type="VARCHAR(255)">
                <constraints nullable="false"/>
            </column>
            <column name="price" type="DECIMAL(10,2)">
                <constraints nullable="false"/>
            </column>
            <column name="stock" type="INT" defaultValue="0">
                <constraints nullable="false"/>
            </column>
            <column name="created_at" type="TIMESTAMP" defaultValueComputed="CURRENT_TIMESTAMP"/>
        </createTable>

        <createIndex indexName="idx_products_name" tableName="products">
            <column name="name"/>
        </createIndex>

        <rollback>
            <dropTable tableName="products"/>
        </rollback>
    </changeSet>

    <changeSet id="002" author="qa-team">
        <addColumn tableName="products">
            <column name="category" type="VARCHAR(100)"/>
        </addColumn>

        <rollback>
            <dropColumn tableName="products" columnName="category"/>
        </rollback>
    </changeSet>
</databaseChangeLog>

# tests/test_liquibase_migrations.py
import pytest
import subprocess
import psycopg2
from typing import Dict, List

class LiquibaseTester:
    def __init__(self, db_config: Dict, changelog_path: str):
        self.db_config = db_config
        self.changelog_path = changelog_path
        self.jdbc_url = f"jdbc:postgresql://{db_config['host']}:{db_config['port']}/{db_config['database']}"

    def run_liquibase_command(self, command: str) -> subprocess.CompletedProcess:
        """Execute Liquibase command"""
        cmd = [
            "liquibase",
            f"--url={self.jdbc_url}",
            f"--username={self.db_config['user']}",
            f"--password={self.db_config['password']}",
            f"--changeLogFile={self.changelog_path}",
            command
        ]
        return subprocess.run(cmd, capture_output=True, text=True)

    def test_changeset_validation(self):
        """Test changesets are valid"""
        result = self.run_liquibase_command("validate")
        assert result.returncode == 0, f"Validation failed: {result.stderr}"

    def test_update_execution(self):
        """Test update executes successfully"""
        # Clear database
        self.run_liquibase_command("dropAll")

        # Run update
        result = self.run_liquibase_command("update")
        assert result.returncode == 0, f"Update failed: {result.stderr}"

        # Verify changes were applied
        conn = psycopg2.connect(**self.db_config)
        cursor = conn.cursor()

        cursor.execute("""
            SELECT COUNT(*)
            FROM databasechangelog
        """)
        changeset_count = cursor.fetchone()[0]
        assert changeset_count > 0

        conn.close()

    def test_rollback_functionality(self):
        """Test rollback to previous tag"""
        # Apply all changes
        self.run_liquibase_command("update")

        # Tag current state
        self.run_liquibase_command("tag version-1.0")

        # Apply more changes (if any)
        self.run_liquibase_command("update")

        # Rollback to tag
        result = self.run_liquibase_command("rollback version-1.0")
        assert result.returncode == 0

        # Verify rollback
        conn = psycopg2.connect(**self.db_config)
        cursor = conn.cursor()

        cursor.execute("""
            SELECT tag
            FROM databasechangelog
            ORDER BY dateexecuted DESC
            LIMIT 1
        """)
        latest_tag = cursor.fetchone()
        assert latest_tag is not None

        conn.close()

    def test_rollback_sql_generation(self):
        """Test rollback SQL generation"""
        result = self.run_liquibase_command("rollbackSQL version-1.0")
        assert result.returncode == 0
        assert "DROP TABLE" in result.stdout or "ALTER TABLE" in result.stdout

    def test_diff_with_production(self):
        """Test schema diff between environments"""
        # This assumes you have a reference database
        cmd = [
            "liquibase",
            f"--url={self.jdbc_url}",
            f"--username={self.db_config['user']}",
            f"--password={self.db_config['password']}",
            "--referenceUrl=jdbc:postgresql://prod-db:5432/proddb",
            "--referenceUsername=prod_user",
            "--referencePassword=prod_pass",
            "diff"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)

        # Check for unexpected differences
        assert "Unexpected" not in result.stdout

# tests/test_schema_drift.py
import pytest
from sqlalchemy import create_engine, MetaData, inspect
from typing import Dict, Set

class SchemaDriftDetector:
    def __init__(self, database_url: str, expected_schema_path: str):
        self.engine = create_engine(database_url)
        self.expected_schema = self.load_expected_schema(expected_schema_path)

    def load_expected_schema(self, path: str) -> Dict:
        """Load expected schema definition"""
        import yaml
        with open(path, 'r') as f:
            return yaml.safe_load(f)

    def get_actual_schema(self) -> Dict:
        """Get actual database schema"""
        inspector = inspect(self.engine)
        schema = {
            'tables': {},
            'indexes': {},
            'foreign_keys': {}
        }

        for table_name in inspector.get_table_names():
            # Get columns
            columns = {}
            for column in inspector.get_columns(table_name):
                columns[column['name']] = {
                    'type': str(column['type']),
                    'nullable': column['nullable'],
                    'default': column.get('default')
                }
            schema['tables'][table_name] = columns

            # Get indexes
            indexes = []
            for index in inspector.get_indexes(table_name):
                indexes.append({
                    'name': index['name'],
                    'columns': index['column_names'],
                    'unique': index['unique']
                })
            schema['indexes'][table_name] = indexes

            # Get foreign keys
            foreign_keys = []
            for fk in inspector.get_foreign_keys(table_name):
                foreign_keys.append({
                    'name': fk['name'],
                    'constrained_columns': fk['constrained_columns'],
                    'referred_table': fk['referred_table'],
                    'referred_columns': fk['referred_columns']
                })
            schema['foreign_keys'][table_name] = foreign_keys

        return schema

    def test_no_schema_drift(self):
        """Verify actual schema matches expected schema"""
        actual = self.get_actual_schema()
        expected = self.expected_schema

        # Compare tables
        actual_tables = set(actual['tables'].keys())
        expected_tables = set(expected['tables'].keys())

        missing_tables = expected_tables - actual_tables
        extra_tables = actual_tables - expected_tables

        assert not missing_tables, f"Missing tables: {missing_tables}"
        assert not extra_tables, f"Unexpected tables: {extra_tables}"

        # Compare columns for each table
        for table_name in expected_tables:
            actual_columns = set(actual['tables'][table_name].keys())
            expected_columns = set(expected['tables'][table_name].keys())

            missing_cols = expected_columns - actual_columns
            extra_cols = actual_columns - expected_columns

            assert not missing_cols, f"Table {table_name} missing columns: {missing_cols}"
            assert not extra_cols, f"Table {table_name} has unexpected columns: {extra_cols}"

            # Compare column properties
            for col_name in expected_columns:
                actual_col = actual['tables'][table_name][col_name]
                expected_col = expected['tables'][table_name][col_name]

                assert actual_col['nullable'] == expected_col['nullable'], \
                    f"Column {table_name}.{col_name} nullable mismatch"

    def test_index_consistency(self):
        """Verify all expected indexes exist"""
        actual = self.get_actual_schema()
        expected = self.expected_schema

        for table_name, expected_indexes in expected['indexes'].items():
            actual_index_names = {idx['name'] for idx in actual['indexes'].get(table_name, [])}
            expected_index_names = {idx['name'] for idx in expected_indexes}

            missing_indexes = expected_index_names - actual_index_names
            assert not missing_indexes, f"Table {table_name} missing indexes: {missing_indexes}"

# tests/fixtures/test_data_seeder.py
import pytest
from faker import Faker
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

class TestDataSeeder:
    def __init__(self, database_url: str):
        self.engine = create_engine(database_url)
        self.Session = sessionmaker(bind=self.engine)
        self.faker = Faker()

    def seed_users(self, count: int = 100):
        """Seed test user data"""
        session = self.Session()

        for _ in range(count):
            user = {
                'email': self.faker.email(),
                'username': self.faker.user_name(),
                'first_name': self.faker.first_name(),
                'last_name': self.faker.last_name(),
                'created_at': self.faker.date_time_this_year()
            }
            session.execute(
                "INSERT INTO users (email, username, first_name, last_name, created_at) "
                "VALUES (:email, :username, :first_name, :last_name, :created_at)",
                user
            )

        session.commit()
        session.close()

    def seed_products(self, count: int = 50):
        """Seed test product data"""
        session = self.Session()

        categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports']

        for _ in range(count):
            product = {
                'name': self.faker.catch_phrase(),
                'price': round(self.faker.random.uniform(10, 1000), 2),
                'stock': self.faker.random_int(0, 500),
                'category': self.faker.random_element(categories),
                'description': self.faker.text(200)
            }
            session.execute(
                "INSERT INTO products (name, price, stock, category, description) "
                "VALUES (:name, :price, :stock, :category, :description)",
                product
            )

        session.commit()
        session.close()

    def clean_all_tables(self):
        """Clean all test data"""
        session = self.Session()

        # Disable foreign key checks
        session.execute("SET session_replication_role = 'replica';")

        # Get all tables
        result = session.execute("""
            SELECT tablename
            FROM pg_tables
            WHERE schemaname = 'public'
            AND tablename NOT IN ('databasechangelog', 'databasechangeloglock', 'flyway_schema_history')
        """)

        # Truncate all tables
        for row in result:
            table_name = row[0]
            session.execute(f"TRUNCATE TABLE {table_name} CASCADE")

        # Re-enable foreign key checks
        session.execute("SET session_replication_role = 'origin';")

        session.commit()
        session.close()

@pytest.fixture(scope="function")
def seeded_database(database_url):
    """Pytest fixture to seed and clean test database"""
    seeder = TestDataSeeder(database_url)

    # Seed data before test
    seeder.seed_users(100)
    seeder.seed_products(50)

    yield seeder

    # Clean data after test
    seeder.clean_all_tables()

# .github/workflows/database-ci.yml
name: Database CI/CD

on:
  pull_request:
    paths:
      - 'migrations/**'
      - 'changesets/**'
  push:
    branches:
      - main

jobs:
  test-migrations:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_PASSWORD: postgres
          POSTGRES_DB: test_db
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install pytest psycopg2-binary sqlalchemy pyyaml faker

      - name: Install Flyway
        run: |
          wget -qO- https://download.red-gate.com/maven/release/com/redgate/flyway/flyway-commandline/10.0.0/flyway-commandline-10.0.0-linux-x64.tar.gz | tar -xz
          sudo ln -s $(pwd)/flyway-10.0.0/flyway /usr/local/bin/flyway

      - name: Run Flyway tests
        env:
          DB_HOST: localhost
          DB_PORT: 5432
          DB_NAME: test_db
          DB_USER: postgres
          DB_PASSWORD: postgres
        run: |
          pytest tests/test_flyway_migrations.py -v

      - name: Run schema drift tests
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test_db
        run: |
          pytest tests/test_schema_drift.py -v

      - name: Generate migration report
        run: |
          flyway info \
            -url=jdbc:postgresql://localhost:5432/test_db \
            -user=postgres \
            -password=postgres \
            -locations=filesystem:migrations > migration-report.txt

      - name: Upload migration report
        uses: actions/upload-artifact@v3
        with:
          name: migration-report
          path: migration-report.txt

  deploy-staging:
    needs: test-migrations
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    environment: staging

    steps:
      - uses: actions/checkout@v3

      - name: Install Flyway
        run: |
          wget -qO- https://download.red-gate.com/maven/release/com/redgate/flyway/flyway-commandline/10.0.0/flyway-commandline-10.0.0-linux-x64.tar.gz | tar -xz
          sudo ln -s $(pwd)/flyway-10.0.0/flyway /usr/local/bin/flyway

      - name: Run migrations on staging
        env:
          DB_URL: ${{ secrets.STAGING_DB_URL }}
          DB_USER: ${{ secrets.STAGING_DB_USER }}
          DB_PASSWORD: ${{ secrets.STAGING_DB_PASSWORD }}
        run: |
          flyway migrate \
            -url=$DB_URL \
            -user=$DB_USER \
            -password=$DB_PASSWORD \
            -locations=filesystem:migrations