Python ユニットテスト

API やブラウザからテストする場合は API テストと E2E テストを参照してください。

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

フィクスチャ	スコープ	中身
`pgmem_options`	session	`pgmem.start()` に渡すキーワード引数。`database` や `params` を変えるときに上書きする
`pgmem_process`	session	起動したバイナリ
`pgmem_server`	session	テンプレートサーバー
`pgmem_prepare`	session	スナップショット前にマイグレーションとシードを実行するときに上書きする
`pgmem_snapshot`	session	準備済みテンプレートのスナップショット。設定を変えるときに上書きする
`pgmem_fork`、`pgmem_dsn`	function	要求したテストごとのフォーク。teardown で閉じる
`pgmem_class_fork`、`pgmem_class_dsn`	class	テストクラスで共有するフォーク
`pgmem_shared_fork`	session	`@shadow_pg(fork=False)` のテストで共有する準備済みフォーク
`pgmem_fixture_options`	session	分離方式、上限、期限、読み取りガード
`pgmem_test_dsn`	function	マーカーで選ぶ fork、共有 template、reset 用の接続先
`pgmem_shared_dsn`	session	reset モード用の書き込み可能で固定した接続先

分離方式の選択とアプリ経由のサンプルも参照してください。

すぐに試す：一度準備してマーカーで分離を選ぶ

pip install pgmem pytest "psycopg[binary]"

次の 2 ファイルを保存し、pytest -q を実行します。

import psycopg
import pytest

@pytest.fixture(scope="session")
def pgmem_prepare():
    def prepare(server):
        with psycopg.connect(server.dsn) as conn:
            conn.execute("CREATE TABLE orders(id int)")
    return prepare

@pytest.fixture
def db(pgmem_test_dsn):
    with psycopg.connect(pgmem_test_dsn) as conn:
        yield conn

import pytest

def test_create(db):
    db.execute("INSERT INTO orders VALUES (1)")
    assert db.execute("SELECT count(*) FROM orders").fetchone()[0] == 1

@pytest.mark.pgmem(fork=False)
def test_read(db):
    assert db.execute("SELECT count(*) FROM orders").fetchone()[0] == 0

pgmem_prepare が一度走り、その後 plugin がスナップショットを作ります。書き込みテストは既定で独立したフォークを使います。fork=False は準備済みテンプレートを共有します。そこで書き込むと、ほかの共有テストにも影響します。どちらも同じ db fixture を使え、接続前に接続先が決まります。

クライアントを維持してデータベースを戻す

アプリがセッション中クライアントを保持する場合は、pgmem_shared_dsn で作り、直列実行するテストに isolation="reset" を指定します。

@pytest.fixture(scope="session")
def app_db(pgmem_shared_dsn):
    with psycopg.connect(pgmem_shared_dsn, autocommit=True) as conn:
        yield conn

@pytest.mark.pgmem(isolation="reset")
class TestOrders:
    def test_create(self, app_db):
        app_db.execute("INSERT INTO orders VALUES (1)")

    def test_clean(self, app_db):
        assert app_db.execute("SELECT count(*) FROM orders").fetchone()[0] == 0

plugin はマーカー付きテストの後に共有フォークをリセットします。テスト本体や fixture のセットアップが失敗した場合も実行し、接続先と socket は維持します。teardown 前に、開いたトランザクションを commit または rollback してください。このモードはワーカー内で直列実行するテスト用です。pytest-xdist のワーカー同士は別プロセスになります。session fixture から function スコープの pgmem_test_dsn には依存できません。

session スコープの pgmem_fixture_options を上書きすると、isolation（fork、shared、reset）、max_forks、fork_timeout（30 秒）、snapshot_timeout（30 秒）、reset_timeout（5 秒）、shared_read_only（False）を設定できます。最も近いマーカーが既定値より優先され、fork=<bool> と isolation=<mode> のどちらかを指定します。明示的な pgmem_dsn は常にフォークを取得します。高度な用途では pgmem_snapshot 自体も上書きできます。その場合、既定の準備 hook は置き換えられます。

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

テストごとに異なるスキーマが必要なら、新しいサーバーを使います。同じスキーマとシードデータを使うテストでは、セッションでスナップショットを 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_prepare を定義します。@shadow_pg(prepare=True) は、このフィクスチャ内の通常の SDK 接続をテンプレートに向けます。フィクスチャが戻るとプラグインがスナップショットを作り、すべてのテスト用フォークはそこから始まります。

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

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

@pytest.fixture(scope="session")
@shadow_pg(prepare=True)
def pgmem_prepare(pgmem_server):
    command.upgrade(Config("alembic.ini"), "head")
    load_seed()

上のように @pytest.fixture を @shadow_pg(prepare=True) の外側に置いてください。pgmem_server パラメーターはデコレーターへテンプレートを渡すためのもので、フィクスチャ本体では使わなくてもかまいません。戻る前に準備用の接続をコミットまたはクローズしてください。フォーク数の上限やタイムアウトを変える場合は、別途 pgmem_snapshot を上書きします。

`@shadow_pg` でテストの接続先を切り替える

スナップショットを準備した後、テストにマーカーを付けると、新しく作る psycopg 3、psycopg2、asyncpg の接続先を pgmem に切り替えます。これらのドライバを使う SQLAlchemy の Engine も対象です。アプリケーション側の接続引数は普段のままで使えます。既定ではテストごとに新しいフォークを作り、終了後に閉じます。fork=False は同じ pytest セッションの読み取り専用テストで、準備済みフォークを 1 つ共有します。

import psycopg
from pgmem import shadow_pg

@shadow_pg
def test_create_order():
    with psycopg.connect("postgresql://app@db.example/app") as db:
        create_order(db)  # 接続先はテスト専用の pgmem フォーク

import psycopg2
from pgmem import shadow_pg

@shadow_pg
def test_create_order():
    db = psycopg2.connect("postgresql://app@db.example/app")
    try:
        with db:
            create_order(db)  # 接続先はテスト専用の pgmem フォーク
    finally:
        db.close()

import asyncpg
import pytest
from pgmem import shadow_pg

@shadow_pg
@pytest.mark.asyncio
async def test_create_order():
    db = await asyncpg.connect("postgresql://app@db.example/app")
    try:
        await create_order(db)
    finally:
        await db.close()

from sqlalchemy import create_engine
from pgmem import shadow_pg

@shadow_pg
def test_create_order():
    engine = create_engine("postgresql+psycopg://app@db.example/app")
    try:
        with engine.begin() as db:
            create_order(db)
    finally:
        engine.dispose()

import pytest
from sqlalchemy.ext.asyncio import create_async_engine
from pgmem import shadow_pg

@shadow_pg
@pytest.mark.asyncio
async def test_create_order():
    engine = create_async_engine("postgresql+asyncpg://app@db.example/app")
    try:
        async with engine.begin() as db:
            await create_order(db)
    finally:
        await engine.dispose()

どのタブでも、読み取り専用テストに @shadow_pg(fork=False) を使えば準備済みフォークを共有できます。asyncpg.create_pool(...) はテスト内でプールを作る場合に使えます。SQLAlchemy の既定の postgresql:// URL は psycopg2 を使い、こちらも対象です。

マーカーは既存の DATABASE_URL と libpq の PG* 接続変数を変更しません。psycopg、psycopg2、asyncpg の接続呼び出しをテスト中だけ pgmem へ向けます。テスト前に作った Engine は再利用でき、前のフォークのプール接続は取り出すときに無効化します。ただし、すでに取り出した接続、テスト前に作った asyncpg プール、マーカーが動く前に from psycopg import connect でコピーした関数の接続先は切り替えられません。その場合は pgmem_dsn を明示的に渡してください。共有フォークはデータベース側で書き込みを禁止していないため、fork=False のテストが書き込むと後続の共有テストに影響します。

読み取り専用のクラスで 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 がフォーク数を制限し、既定値はメモリから決まり、サーバープロセスのメモリ上限の 4 分の 1 をフォーク 1 個のコスト（shared_buffers + 約 32MB）で割った値で、7GB の CI ランナーなら 28 です。フォークはそれぞれデータディレクトリとバッファキャッシュを複製するため、xdist のワーカー数と max_forks の両方を増やすとメモリ使用量も増えます。

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

@pytest.fixture(scope="session")
def pgmem_snapshot(pgmem_server, pgmem_prepare):
    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 は、開いているトランザクションが終わるのを待つ時間です。その他のタイムアウトは制限事項を参照してください。

`@shadow_pg` を安全に並列実行する

独立したテストケースは pytest-xdist の別プロセスで実行します。pytest-xdist をインストールした上で、例えば pytest -n 4 を使います。各ワーカーは独自のテンプレートとフォークを持ち、ワーカー内のケースは逐次実行されます。ドライバのモンキーパッチはプロセス全体に効くため、1 ワーカー内でケースをスレッド並列に実行するプラグインとは併用しません。テストが開始した DB 作業はマーカーの解除前に完了させてください。fork=False はフォークを共有する読み取り専用ケースだけに使います。ワーカー数と max_forks はメモリ容量に合わせて調整します。

共有接続の書き込みを防ぐ

クライアントを作る前に shared_read_only を設定します。マーカーで shared を選んだ DSN とクラス用 DSN を保護します。独立したフォークと reset 用の pgmem_shared_dsn は書き込み可能です。自分で URL を作る場合は readonly_dsn(dsn) を使えます。生のフォーク URL にはガードを付けません。

@pytest.fixture(scope="session")
def pgmem_fixture_options():
    return {"isolation": "shared", "shared_read_only": True}

pgmem_prepare は上のように fixture 内で準備を実行するか、quickstart のようにテンプレートの Server を受け取る callback を返せます。同じ migration を両方で実行しないでください。