Testing with pytest

Installing pgmem registers a pytest plugin. Its fixtures cover the prepared-database shapes; nothing needs to be added to conftest.py to get a server.

Fixture	Scope	What it is
pgmem_options	session	keyword arguments for pgmem.start(); override to set database or params
pgmem_process	session	the running binary
pgmem_server	session	the template server
pgmem_snapshot	session	a snapshot of the template; override it to run migrations first
pgmem_fork, pgmem_dsn	function	a fork per requesting test, closed at teardown
pgmem_class_fork, pgmem_class_dsn	class	one fork shared by a test class

Database lifecycle strategy

Use a fresh server when tests need different schemas. When tests share a schema and seed data, prepare one snapshot for the session. Read-only tests can share a class fork; request a per-test fork only in tests that write or need isolation.

A fresh server per test

When every test needs a different schema, start a process in a function-scoped fixture:

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

Each test pays a process start and its migrations.

Prepare once for the session

Override pgmem_options and pgmem_snapshot in conftest.py. Migrations and seed data run once per session; the snapshot is what every fork starts from.

conftest.py

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()

Share one fork across a read-only class

Tests that only read can share a fork per class.

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):
        ...

Caution

If one test in the class writes, the others see it, and the result depends on test order. Move writing tests to pgmem_dsn.

Fork only in tests that write

Request pgmem_dsn only in test functions that write or need isolation; pytest creates a fresh copy of the prepared database when that fixture is requested. Read-only tests in the same class can keep using pgmem_class_dsn and share their class fork. You do not need to make the per-test fork fixture automatic for the whole suite.

psycopg

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"

asyncpg

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()

SQLAlchemy

@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 the engine before the fork closes, so the pool does not hold connections to a server that is gone.

Several seed sets

A suite that needs two different prepared databases starts a second template server in its own session fixture:

@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

Parallel runs

With pytest-xdist, every worker is its own process and starts its own pgmem binary, so each worker prepares its own template and gets its own fork pool. Within a worker, max_forks on the session snapshot caps live forks; the default is the available CPU count. Forks hold separate data-directory copies and buffer caches, so raising both the xdist worker count and max_forks raises memory use.

The built-in fixture waits indefinitely when the pool is full. To set a limit and a deadline, override pgmem_snapshot and pgmem_fork in conftest.py:

conftest.py

@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 applies to this snapshot. fork(timeout=...) limits only the wait for a free slot and raises ProtocolError with code pool_timeout; omitting the timeout waits indefinitely. The timeout passed to snapshot() instead bounds its wait for open transactions to finish. See limits for the other timeout settings.

Note

A fork is one PostgreSQL session. A connection pool on one fork works, but its connections serialize at transaction boundaries. Commit or close connections to the template before snapshot(), or it fails with busy after 30 seconds.