Posted inPython

Pythonのインテグレーションテスト：Testcontainersで「モックは成功、本番は失敗」の不安を解消する

Python tutorial - IT technology blog
Python tutorial - IT technology blog

なぜモックではなく「本物」を使うべきなのか？

ローカルでのテストは100%パスしたのに、ステージング環境にデプロイした途端、SQLエラーが多発したという経験はありませんか？私自身、モックを多用しすぎて苦い経験をしたことがあります。テストを高速化するためにDatabaseやRedisをモック化することがよくありますが、モックは往々にして「お行儀が良すぎ」て、実際の環境の厳しさを反映してくれません。

例えば、PostgresのJSONB機能や複雑なWindow Functionsを使用したクエリを考えてみてください。モックは構文の間違いを指摘してはくれません. 本物のPostgresと対峙して初めて、エラーが露呈します。私が以前参加したプロジェクトでは、モックから実環境でのテストに切り替えたことで、CI段階でデータベース関連のバグを40%削減することができました。

Testcontainersはこの問題を解決するために登場しました。わずか数行のPythonコードで、Postgres、Redis、Kafkaなどのコンテナを丸ごとテスト環境に持ち込むことができます。

クイックスタート：5分で本物のPostgresを使ってテストを実行する

まず、マシンにDockerがインストールされている必要があります。手動でdocker pullを行う必要はありません。Testcontainersがすべて処理してくれます。以下のライブラリをインストールしましょう。

pip install pytest testcontainers sqlalchemy psycopg2-binary

テストケース内で「本物」のデータベースを初期化する方法は以下の通りです：

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

def test_postgres_connection():
    # Postgres 15のコンテナを初期化
    with PostgresContainer("postgres:15") as postgres:
        # 動的な接続文字列を取得（衝突を避けるためポートはランダムにマッピングされる）
        engine = create_engine(postgres.get_connection_url())
        
        with engine.connect() as conn:
            result = conn.execute("SELECT 1").fetchone()
            assert result[0] == 1
            print("\nPostgresへの接続に成功しました！")

pytestコマンドを実行すると、Testcontainersは自動的にDocker Hubからイメージをプルします。コンテナを起動してテストを実行し、終了後は自動的にクリーンアップしてくれます。初回はイメージのダウンロードに30秒ほどかかるかもしれませんが、2回目以降は非常に高速です。

Pytest Fixturesでレベルアップする

上記のようなwithブロックは小さなスクリプトには便利です。しかし、50個のテストファイルがあり、それぞれで新しいコンテナを起動していては非常に時間がかかります。最適化のために、scope="session"を指定したPytest Fixturesを使用しましょう。

conftest.pyの構成

プロジェクト全体で共有できるように、インフラの設定をconftest.pyにまとめます：

import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres_db():
    # コンテナはテスト実行プロセス全体で1回だけ初期化される
    postgres = PostgresContainer("postgres:15")
    postgres.start()
    
    yield postgres
    
    postgres.stop() # すべてのテストが終了した後にクリーンアップ

これで、テストの記述が非常に楽になります。関数の引数にpostgres_dbを渡すだけです：

def test_create_user(postgres_db):
    db_url = postgres_db.get_connection_url()
    # ここで本物のDBにユーザーを保存するロジックを実行する
    assert "postgres" in db_url

DatabaseとRedisの組み合わせ（フルスタックテスト）

モダンなFastAPIアプリでは、保存用のPostgresとキャッシュ用のRedisを組み合わせるのが一般的です。Testcontainersは、複数の種類のコンテナを同時に組み合わせることもスムーズにサポートしています。

from testcontainers.redis import RedisContainer
import redis

@pytest.fixture(scope="session")
def redis_client():
    with RedisContainer("redis:7") as container:
        host = container.get_container_host_ip()
        port = container.get_exposed_port(6379)
        client = redis.Redis(host=host, port=port)
        yield client

def test_cache_logic(redis_client):
    redis_client.set("session_key", "active")
    assert redis_client.get("session_key") == b"active"

GitHub Actionsで実行する場合、RunnerがDockerをサポートしていることを確認するだけです。テスト環境が本番環境と同一になり、サーバー上だけで発生する謎のエラーを完全に排除できます。

実戦経験：テストを亀のように遅くさせないために

テストケースが10件から1000件に増えると、速度が死活問題になります。多くのプロジェクトを経て得たヒントをいくつか紹介します：

  • Session Scopeを優先する： テスト全体でコンテナを共有しましょう。テスト関数ごとにPostgresを再起動すると、実行時間が10倍に増える可能性があります。
  • クリーンアップ戦略： DBを共有するため、前のテストケースのデータが次のテストケースに悪影響を与えることがあります。コンテナを削除するのではなく、各テスト関数が終わるたびにテーブルをTRUNCATEするフィクスチャを使用しましょう。
  • Wait Strategyの設定： コンテナがrunning状態でも、内部のDBが初期化中であることがあります。wait_forを使用して、データベースが実際に接続を受け入れられる状態になるまで待ちましょう。
  • ログによるデバッグ： テストが不可解な失敗をした場合は、container.get_logs()を使用してください。モックでは決して分からない、DBのメモリ制限やエンコーディング設定のミスが原因であることがあります。

Testcontainersに移行すると、テストの起動に数分余計にかかるかもしれません。しかし、自分のコードが実際のデータベースで動作したという安心感は計り知れません。本番環境でバグが出たときに、「あれ、自分のマシンでは動いたんだけどな？」と困惑しながら説明する必要はもうなくなります。

Share:
Tags: