Python API テスト

アプリへ HTTP リクエストを送るテストでは、pytest ワーカーごとに pgmem の fork を 1 つ維持します。@shadow_pg(live=True) が各テスト前にデータを復元し、新規接続をこの fork へ向けます。HTTP クライアントはテストごとに作ります。同じワーカー内のテストは逐次実行してください。pytest-xdist の各ワーカーは別の pgmem プロセスとアプリクライアントを持ちます。

スキーマとシードデータはユニットテストと同様に pgmem_prepare で一度準備します。HTTP テストで使う fixture は次のとおりです。

fixture	スコープ	用途
`pgmem_live_fork`	session	アプリ用の固定サーバーと DSN
`pgmem_live_snapshots`	session	データセット名から snapshot への対応。追加するときに上書きする
`pgmem_live_baseline`	session	既定データセット。`capture()` でアプリ起動時の書き込みを取り込む
`pgmem_live_db`	function	テスト前に選択した snapshot へ復元する

常駐する fork は pgmem_snapshot.max_forks の枠をセッション中 1 つ使います。同じワーカーでテスト単位の fork や @shadow_pg(fork=False) も使う場合は、その分の枠を確保してください。

HTTP テストクライアントを接続する

@shadow_pg(live=True) は各テストの前に同じ pgmem エンドポイントを復元し、ドライバの新規接続をそこへ向けます。アプリのデータソース設定は変更しません。クライアント fixture を関数スコープにし、アプリの lifespan と接続プールの作成がマーカーの有効期間内に入るようにします。アプリの通常設定は PostgreSQL ドライバを使う必要があります。

import pytest
from fastapi.testclient import TestClient
from myapp import create_app

@pytest.fixture
def client():
    with TestClient(create_app()) as client:
        yield client

TestClient をコンテキストマネージャーとして使うと、各テストでアプリの lifespan が実行されます。

import pytest
from starlette.testclient import TestClient
from myapp import create_app

@pytest.fixture
def client():
    with TestClient(create_app()) as client:
        yield client

import pytest
from myapp import create_app

@pytest.fixture
def client():
    app = create_app()
    app.config.update(TESTING=True)
    with app.test_client() as client:
        yield client

Flask のテストクライアントは WSGI アプリを直接呼びます。DB の接続先には同じ固定 pgmem サーバーを使います。

import os
import pytest

@pytest.fixture
def client():
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myapp.settings")
    import django
    django.setup()  # アプリの通常の PostgreSQL 設定を使う
    from django.test import Client
    return Client()

Django のリクエスト Client と通常の pytest 関数を使います。この方式では Django の TestCase、pytest-django の db・transactional_db fixture、live_server を併用しません。これらはテスト DB の作成、ロールバック、テーブルの消去を独自に管理します。アプリコード内の transaction.atomic() やセーブポイントは使えます。Django がまだ DB 接続を初期化していないテスト環境で設定してください。別プロセスの Django サーバーについては E2E テストを参照してください。

テストモジュールでは from pgmem import shadow_pg を追加します。どのクライアントもテストの形は同じです。

@shadow_pg(live=True)
def test_create_order(client):
    response = client.post("/orders", data={"sku": "A"})
    assert response.status_code == 201

@shadow_pg(live=True)
def test_list_starts_from_seed(client):
    assert client.get("/orders").status_code == 200

リクエストの形式はアプリの API に合わせてください。FastAPI と Starlette では json=、Flask と Django ではフォームデータも使えます。クライアントはマーカーの有効期間内で起動し、テスト終了時に閉じます。

データセットを選ぶ

pgmem_live_snapshots を上書きし、同じ pgmem プロセス、データベース名、ユーザーの snapshot を追加します。シード投入用の接続は snapshot 作成前にコミットするか閉じてください。

@pytest.fixture(scope="session")
def pgmem_live_snapshots(pgmem_server, pgmem_snapshot):
    load_paid_orders(pgmem_server.dsn)
    return {"default": pgmem_snapshot, "paid": pgmem_server.snapshot()}

@pytest.mark.pgmem_dataset("paid")
@shadow_pg(live=True)
def test_paid_orders(client):
    assert client.get("/orders/paid").status_code == 200

復元は失敗したテストの後も含め、各テストの前に行います。開いたトランザクションが残っていると待機し、期限に達すると ProtocolError(code="busy") で失敗します。次のテストまでにバックグラウンド処理を完了し、トランザクションをコミットまたは終了してください。接続先と接続自体は維持されますが、復元後の DB セッションは新しくなるため、セッション設定と一時テーブルは引き継がれません。

アプリ起動時の状態を取り込む

アプリが起動時にスキーマを作る場合、その書き込みは pgmem_snapshot を取った後に行われるため、最初の復元で消えます。lifespan で動く Alembic、Django の migrate、SQLAlchemy の create_all などが該当します。アプリを起動する fixture で pgmem_live_baseline を要求し、起動を待ってから capture() を呼ぶと、以降の復元はその状態に戻ります。

@pytest.fixture(scope="session")
def app_url(pgmem_live_fork, pgmem_live_baseline):
    process = start_app(dsn=pgmem_live_fork.dsn)  # ここで Alembic や create_all、migrate が走る
    wait_until_ready(process)
    pgmem_live_baseline.capture()  # アプリ自身が作ったスキーマが既定データセットになる
    try:
        yield "http://127.0.0.1:8000"
    finally:
        process.terminate()
        process.wait()

capture() は最初に取った snapshot を使い続けます。取り直すときは recapture() を呼びます。pgmem_live_snapshots の名前付きデータセットには影響しません。取り込んだ snapshot はセッション中、データディレクトリのコピーを 1 つ増やします。ほかの snapshot と同じく開いたトランザクションを待ち、期限に達すると ProtocolError(code="busy") で失敗します。

安全な並列実行

API テストのケースは pytest-xdist の別ワーカープロセスに分けます。例えば pytest -n 4 です。各ワーカーが独自の pgmem プロセスと固定 live fork を持ち、上の HTTP クライアントは関数スコープのままにします。1 ワーカー内では @shadow_pg(live=True) がケース前に同じフォークを復元するため、2 ケースのリクエストを重ねたり、スレッド並列のテストランナーを使ったりしないでください。次の復元前にバックグラウンド処理とトランザクションを終わらせます。全ワーカーが 1 つの外部アプリサーバーを共有すると分離できないため、アプリと DB の接続先もワーカーごとに用意します。