Testcontainersとpytestによるデータベーステストパイプライン:スキーマとマイグレーションのバグを本番環境に届く前に検出する

Database tutorial - IT technology blog
Database tutorial - IT technology blog

クイックスタート:5分で動かす

6ヶ月前、失敗したマイグレーションが本番データベースを45分間オフラインにした。ユニットテストはすべてパスしていた——モックに対して実行されていたからだ。本物のPostgreSQLは、制約の順序について別の考えを持っていた。その事故がきっかけで、Testcontainersを使った適切な統合テストパイプラインを構築することを決意した。

まず依存関係をインストールする:

pip install testcontainers pytest sqlalchemy psycopg2-binary alembic

次に、実際のPostgreSQLコンテナを起動する最小限のテスト:

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def postgres_container():
    with PostgresContainer("postgres:16") as pg:
        yield pg

def test_basic_connection(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    with engine.connect() as conn:
        result = conn.execute(text("SELECT version()"))
        version = result.scalar()
    assert "PostgreSQL 16" in version

実行する:

pytest test_db.py -v

初回実行時はDockerイメージをダウンロードする——約170MBだ。それ以降は、コンテナが2〜3秒で起動する。これはモックでもPostgresコスプレのSQLiteでもなく、本物のPostgreSQLだ。モックでパスしたマイグレーションが本番データベースをダウンさせるまで、その違いは軽視されがちだ。

詳細解説:実際にテストできること

マイグレーションのエンドツーエンドテスト

本当にクリーンなデータベースに対してAlembicマイグレーションを実行する——それこそがこのセットアップの価値が発揮される場面だ。ユニットテストでは見逃すことが多い:存在しないカラムを参照するマイグレーション、SQLiteが暗黙的に受け入れるPostgreSQL固有の構文、本物のエンジンでのみ問題になる制約の順序など。

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text
from alembic.config import Config
from alembic import command

@pytest.fixture(scope="session")
def migrated_db():
    with PostgresContainer("postgres:16") as pg:
        engine = create_engine(pg.get_connection_url())

        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
        command.upgrade(alembic_cfg, "head")

        yield engine

def test_users_table_schema(migrated_db):
    with migrated_db.connect() as conn:
        result = conn.execute(text(
            "SELECT column_name, data_type "
            "FROM information_schema.columns "
            "WHERE table_name = 'users' ORDER BY ordinal_position"
        ))
        columns = {row[0]: row[1] for row in result}

    assert "id" in columns
    assert "email" in columns
    assert columns["created_at"] == "timestamp with time zone"

この方法で3つのマイグレーションバグを検出した。本番では手動対応が必要になっていたはずのものだ:間違った方向を向いた外部キー制約、高トラフィックな検索カラムのインデックス欠如、そしてTEXTが必要なところに指定されたVARCHAR(255)。これらはいずれもモックテストでは表面化しなかっただろう。

本物のデータに対するクエリロジックのテスト

モックデータベースは「クエリが呼ばれた」ことをアサートできる。本物のデータベースは「クエリが正しいデータを返す」ことをアサートできる。大きな違いがある。

from sqlalchemy.orm import Session
from your_app.models import User, Order

@pytest.fixture
def db_session(migrated_db):
    with Session(migrated_db) as session:
        yield session
        session.rollback()  # 各テスト後にクリーンアップ

def test_user_total_completed_orders(db_session):
    # Arrange: 実際のテストデータを挿入
    user = User(email="[email protected]", name="テストユーザー")
    db_session.add(user)

    for i in range(3):
        order = Order(user=user, amount=10.00 * (i + 1), status="completed")
        db_session.add(order)

    db_session.add(Order(user=user, amount=99.00, status="pending"))
    db_session.flush()

    # Act: アプリケーション層から実際のクエリを実行
    result = db_session.execute(text(
        "SELECT SUM(amount) FROM orders "
        "WHERE user_id = :uid AND status = 'completed'"
    ), {"uid": user.id}).scalar()

    # Assert: pendingステータスの注文は含まれてはいけない
    assert result == 60.00

フィクスチャのteardownで呼ばれるsession.rollback()が、スキーマを削除・再作成することなく各テストを独立した状態に保つ。高速で予測可能だ。

高度な使い方:CI/CDパイプラインの完全構築

セッションスコープのコンテナとテストごとの分離

50個のデータベーステストを逐次実行すると、CIが完全に止まってしまう。セッションごとに1つのコンテナを使い——マイグレーションを1回だけ実行し——データ分離のためにテストごとにトランザクションロールバックを行う。以下がそのパターンだ:

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from alembic.config import Config
from alembic import command

@pytest.fixture(scope="session")
def db_engine():
    with PostgresContainer("postgres:16") as pg:
        engine = create_engine(
            pg.get_connection_url(),
            pool_size=10,
            max_overflow=20
        )
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
        command.upgrade(alembic_cfg, "head")
        yield engine

@pytest.fixture
def db_session(db_engine):
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

並列実行のためにpytest-xdistを追加する:

pip install pytest-xdist
pytest -n 4 tests/integration/

このパターンにより、テストスイートの実行時間が8分から90秒に短縮された。

GitHub Actionsとの統合

TestcontainersにはDockerが必要だ。GitHub ActionsにはDockerがすでに搭載されているため、CI設定は最小限で済む:

# .github/workflows/db-tests.yml
name: データベース統合テスト

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Pythonのセットアップ
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: 依存関係のインストール
        run: pip install -r requirements.txt

      - name: データベース統合テストの実行
        run: pytest tests/integration/ -v --tb=short
        env:
          TESTCONTAINERS_RYUK_DISABLED: "false"

マイグレーションロールバックのテスト

緊急ロールバックはただでさえストレスが多い。障害対応の最中にダウングレードスクリプトがエラーを出すと、さらに悪化する。すべてのマイグレーションを両方向でテストしよう:

def test_migration_upgrade_and_downgrade(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", postgres_container.get_connection_url())

    # 最新バージョンにアップグレード
    command.upgrade(alembic_cfg, "head")

    # 1ステップダウングレード — エラーは発生しないはず
    command.downgrade(alembic_cfg, "-1")

    # 冪等性を確認するため再度アップグレード
    command.upgrade(alembic_cfg, "head")

    with engine.connect() as conn:
        result = conn.execute(text("SELECT COUNT(*) FROM alembic_version"))
        assert result.scalar() == 1

6ヶ月間の本番運用から得た実践的なヒント

フィクスチャのスコープを意図的に分ける。コンテナとマイグレーションにはscope="session"を使い(繰り返すとコストが高い)、データ分離にはトランザクションロールバックと組み合わせてscope="function"を使う(コストが低い)。この組み合わせにより、テストスイートが8分から90秒に短縮された。

テストではデータベースのバージョンを固定する。postgres:latestではなくpostgres:16を使おう。新しいメジャーバージョンがリリースされてもテストがランダムに壊れなくなる。本番環境で動いているバージョンと完全に一致させること。

マイグレーションごとにスキーマアサーションテストを書く。mainブランチにマージされるすべてのマイグレーションには、期待するスキーマ状態を検証するテストを対応づけよう。データベースがどのように進化してきたかの生きた記録となり、CIを離れる前にリグレッションを検出できる。

# migration 005_add_user_preferences.py と一緒に追加されたテスト
def test_migration_005_schema(migrated_db):
    with migrated_db.connect() as conn:
        result = conn.execute(text(
            "SELECT data_type FROM information_schema.columns "
            "WHERE table_name = 'user_preferences' AND column_name = 'settings'"
        ))
        data_type = result.scalar()

    # jsonではなくjsonbを明示的にアサート — GINインデックスサポートが必要
    assert data_type == "jsonb"

シードデータスクリプトをテストする。アプリケーションに参照データやデフォルト設定が含まれている場合、それらのスクリプトを実行してレコード数を検証するフィクスチャを用意しよう。この方法で、ステージング環境に到達する前に壊れたシードスクリプトを検出できた。

テストフィクスチャを慎重に準備する。統合テスト用にエッジケースのデータセットが必要になることがある——CSVエクスポートや匿名化された本番スナップショットの参照データなどだ。データインポート用にCSVをJSONに素早く変換する必要があるときは、toolcraft.app/ja/tools/data/csv-to-jsonを使っている——ブラウザ上で完全に動作するため、データが外部に出ない。CSVに匿名化されたテストレコードが含まれていてどこにもアップロードできない場合、これは重要な点だ。

CI上でのコンテナ起動にリトライを追加する。GitHub Actionsでは、Dockerの準備に数秒余計にかかることがある。新しいランナーで断続的な起動失敗が見られる場合は、コンテナフィクスチャをシンプルなリトライでラップしよう。

このパイプラインが6ヶ月間、すべてのプルリクエストで動き続けてきた。スキーマのバグ、制約の方向の誤り、インデックスの欠如、そしてローカルのSQLiteではパスしたもののステージング環境の本物のPostgreSQLで爆発した2件のマイグレーションを検出できた。パイプラインはCIに約90秒を追加する。少なくとも4件の本番インシデントから救ってくれた。十分な価値がある。

Share: