Python unit tests

For HTTP or browser tests, see API tests and E2E tests.

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_prepare	session	override it to run migrations and seed data before the snapshot
pgmem_snapshot	session	a snapshot of the prepared template; override for custom snapshot options
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
pgmem_shared_fork	session	one prepared fork shared by @shadow_pg(fork=False) tests
pgmem_fixture_options	session	isolation, capacity, deadlines and optional read guard
pgmem_test_dsn	function	marker-selected fork, shared template or reset endpoint
pgmem_shared_dsn	session	writable stable endpoint for reset mode

See the lifecycle guide and runnable application examples.

Quickstart: prepare once, select isolation with a marker

pip install pgmem pytest "psycopg[binary]"

Save these two files and run pytest -q:

conftest.py

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

test_orders.py

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 runs once before the plugin takes a snapshot. Write cases use isolated forks by default. fork=False selects the shared prepared template; a write there affects other shared tests. The same db fixture works in both cases, and selection happens before it connects.

Keep a client and reset its database

If the application keeps a client for the session, build it against pgmem_shared_dsn and mark its serial tests with isolation="reset":

conftest.py

@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

The plugin resets the shared fork after each marked test, including test or fixture setup failures. The endpoint and sockets stay valid. Commit or roll back active transactions before teardown. This mode is for serial tests within one worker; pytest-xdist workers still have separate processes. Session fixtures cannot depend on the function-scoped pgmem_test_dsn.

Override session-scoped pgmem_fixture_options to configure isolation (fork, shared, reset), max_forks, fork_timeout (30 s), snapshot_timeout (30 s) and reset_timeout (5 s) and shared_read_only (false). The closest marker overrides the default; specify either fork=<bool> or isolation=<mode>. Explicit pgmem_dsn always acquires a fork. Advanced suites can still override pgmem_snapshot directly; that replaces the default preparation hook.

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.

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 define pgmem_prepare in conftest.py. @shadow_pg(prepare=True) sends ordinary SDK connections in this fixture to the template. When it returns, the plugin snapshots the prepared database; every test fork starts there.

conftest.py

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

Place @pytest.fixture outside @shadow_pg(prepare=True) as shown. The pgmem_server parameter gives the decorator its template target; the fixture body can ignore it. Commit or close every preparation connection before returning. To customize the snapshot limit or timeout, override pgmem_snapshot separately.

Route a test with @shadow_pg

After preparing the snapshot, mark a test to intercept new psycopg 3, psycopg2, and asyncpg connections. SQLAlchemy engines using these drivers follow the same route. The application can keep its usual connection arguments: pgmem replaces the destination while the test runs. The default creates a fresh fork and closes it after the test. Use fork=False to share one prepared fork across read-only tests in this pytest session.

psycopg 3

test_orders.py

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)  # connects to a fresh pgmem fork

psycopg2

test_orders.py

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)  # connects to a fresh pgmem fork
    finally:
        db.close()

asyncpg

test_orders.py

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

SQLAlchemy + psycopg

test_orders.py

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

SQLAlchemy + asyncpg

test_orders.py

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

Use @shadow_pg(fork=False) in any tab for a read-only test that shares a prepared fork. asyncpg.create_pool(...) works when the pool is created inside the test. SQLAlchemy’s default postgresql:// dialect uses psycopg2 and is also covered.

The marker leaves DATABASE_URL and libpq’s PG* connection variables unchanged. It redirects psycopg, psycopg2, and asyncpg connection calls to pgmem only while the test runs. An Engine created before the test can be reused: pooled connections from an earlier fork are invalidated on checkout. Connections already checked out, asyncpg pools created before the test, and connection functions copied with from psycopg import connect before the marker runs cannot be rerouted. For those patterns, pass pgmem_dsn explicitly. The shared fork does not enforce read-only access, so a write from a fork=False test affects later shared tests.

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 comes from memory: a quarter of the server process’s memory limit divided by a fork’s cost (shared_buffers plus about 32 MB), 28 on a 7 GB CI runner. 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, 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 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 a PostgreSQL cluster of its own, so a connection pool on one fork works as on a server. Commit or close connections to the template before snapshot(), or it fails with busy after 30 seconds.

Safe parallel setup for @shadow_pg

Run independent test cases in separate pytest-xdist processes, for example pytest -n 4 (install pytest-xdist first). Each worker builds its own template and forks; cases inside that worker execute sequentially. The driver monkeypatch is process-wide, so a plugin that runs test cases concurrently in threads within one worker is not safe. Wait for database work started by a test to finish before its marker is removed. Use fork=False only for read-only cases because they share one fork. Size the worker count and max_forks together for available memory.

Guard shared connections

Set shared_read_only before constructing clients. It guards marker-selected shared DSNs and class DSNs; isolated forks and pgmem_shared_dsn for reset mode remain writable. Use readonly_dsn(dsn) when building a guarded URL yourself. Raw fork URLs bypass this guard.

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

pgmem_prepare can run setup directly, as above, or return a callback accepting a template Server, as in the quickstart. Do not apply the same migrations in both forms.