pytest への組み込み

pgmem をインストールすると pytest プラグインが登録されます。準備済みデータベースを使う形はそのフィクスチャでまかなえるので、サーバーを用意するだけなら conftest.py に何も書く必要はありません。

フィクスチャ	スコープ	中身
`pgmem_options`	session	`pgmem.start()` に渡すキーワード引数。`database` や `params` を変えるときに上書きする
`pgmem_process`	session	起動したバイナリ
`pgmem_server`	session	テンプレートサーバー
`pgmem_snapshot`	session	テンプレートのスナップショット。マイグレーションを先に流すときに上書きする
`pgmem_fork`、`pgmem_dsn`	function	要求したテストごとのフォーク。teardown で閉じる
`pgmem_class_fork`、`pgmem_class_dsn`	class	テストクラスで共有するフォーク

データベースのライフサイクル

テストごとに異なるスキーマが必要なら、新しいサーバーを使います。同じスキーマとシードデータを使うテストでは、セッションでスナップショットを 1 つ準備します。読み取り専用テストはクラス用フォークを共有し、テストごとのフォークは書き込みや分離が必要なテストだけで要求します。

テストごとに新しいサーバー

テストごとにスキーマが違うなら、function スコープのフィクスチャでプロセスを起動します。

import pgmem
import pytest

@pytest.fixture
def fresh_dsn():
    with pgmem.start(database="app") as pg:
        apply_schema(pg.template.dsn)
        yield pg.template.dsn

テストごとにプロセスの起動とマイグレーションのコストがかかります。

セッションで一度だけ準備する

conftest.py で pgmem_options と pgmem_snapshot を上書きします。マイグレーションとシードはセッションで一度だけ走り、そのスナップショットがすべてのフォークの出発点になります。

import pytest
from alembic import command
from alembic.config import Config

@pytest.fixture(scope="session")
def pgmem_options():
    return {"database": "app"}

@pytest.fixture(scope="session")
def pgmem_snapshot(pgmem_server):
    url = pgmem_server.dsn.replace("postgres://", "postgresql+psycopg://", 1)
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", url)
    command.upgrade(cfg, "head")
    load_seed(pgmem_server.dsn)
    return pgmem_server.snapshot()

読み取り専用のクラスで 1 つのフォークを共有する

読むだけのテストは、クラス単位でフォークを共有できます。

class TestReports:
    def test_total(self, pgmem_class_dsn):
        with psycopg.connect(pgmem_class_dsn) as conn:
            assert conn.execute("SELECT count(*) FROM orders").fetchone()[0] == 10_000

    def test_top_customer(self, pgmem_class_dsn):
        ...

書き込むテストだけでフォークする

書き込みや分離が必要なテスト関数だけで pgmem_dsn を要求します。pytest はこのフィクスチャが使われたときに、準備済みデータベースの複製を作ります。同じクラスの読み取り専用テストは pgmem_class_dsn を使い、クラス用のフォークを共有できます。スイート全体にテストごとのフォークを自動適用する必要はありません。

def test_cancel_order(pgmem_dsn):
    with psycopg.connect(pgmem_dsn) as conn:
        cancel_order(conn, order_id=1)
        status = conn.execute("SELECT status FROM orders WHERE id = 1").fetchone()[0]
    assert status == "cancelled"

import pytest

@pytest.mark.asyncio
async def test_cancel_order(pgmem_dsn):
    conn = await asyncpg.connect(pgmem_dsn)
    try:
        await cancel_order(conn, 1)
    finally:
        await conn.close()

@pytest.fixture
def session(pgmem_dsn):
    engine = create_engine(pgmem_dsn.replace("postgres://", "postgresql+psycopg://", 1))
    with Session(engine) as s:
        yield s
    engine.dispose()

def test_cancel_order(session):
    cancel_order(session, 1)

フォークが閉じる前にエンジンを dispose し、なくなったサーバーへの接続をプールに残さないようにします。

シードの組を複数使う

準備済みデータベースが 2 種類必要なスイートでは、専用の session フィクスチャで 2 つ目のテンプレートサーバーを起動します。

@pytest.fixture(scope="session")
def audit_snapshot(pgmem_process):
    server = pgmem_process.start_server("audit")
    apply_audit_schema(server.dsn)
    return server.snapshot()

@pytest.fixture
def audit_dsn(audit_snapshot):
    with audit_snapshot.fork() as fork:
        yield fork.dsn

並列実行

pytest-xdist では各ワーカーが別プロセスで、それぞれ pgmem バイナリを起動し、テンプレートとフォークプールを持ちます。ワーカー内ではセッション用スナップショットの max_forks がフォーク数を制限し、既定値は利用可能な CPU 数です。フォークはそれぞれデータディレクトリとバッファキャッシュを複製するため、xdist のワーカー数と max_forks の両方を増やすとメモリ使用量も増えます。

プールが満杯のとき、組み込みフィクスチャは空き枠を無期限に待ちます。上限と期限を設定するには、conftest.py で pgmem_snapshot と pgmem_fork を上書きします。

@pytest.fixture(scope="session")
def pgmem_snapshot(pgmem_server):
    return pgmem_server.snapshot(max_forks=4, timeout=30.0)

@pytest.fixture
def pgmem_fork(pgmem_snapshot):
    with pgmem_snapshot.fork(timeout=30.0) as fork:
        yield fork

max_forks はこのスナップショットに適用されます。fork(timeout=...) は空き枠を待つ時間だけを制限し、期限を超えると ProtocolError の code pool_timeout で失敗します。timeout を省略すれば無期限に待ちます。snapshot() の timeout は、開いているトランザクションが終わるのを待つ時間です。その他のタイムアウトは制限事項を参照してください。