Python API tests

For tests that send requests to an application, keep one pgmem fork per pytest worker. @shadow_pg(live=True) restores its data before each test and routes new connections to that fork. Create the HTTP client for each test. Tests in one worker must run sequentially. Separate pytest-xdist workers have separate pgmem processes and application clients.

Prepare schema and seed data once with pgmem_prepare, as in unit tests. The live fixtures are:

Fixture	Scope	Purpose
pgmem_live_fork	session	stable server and DSN for the application
pgmem_live_snapshots	session	mapping of dataset names to snapshots; override to add datasets
pgmem_live_baseline	session	the default dataset; capture() adds what the application wrote while starting
pgmem_live_db	function	reset the live server to the selected snapshot before the test

The live fork occupies one slot in pgmem_snapshot.max_forks for the whole session. Allow another slot if this worker also uses a per-test fork or @shadow_pg(fork=False).

Connect an HTTP test client

@shadow_pg(live=True) restores the stable pgmem endpoint before each test and routes new driver connections to it. Leave the application’s datasource configuration unchanged. Keep the client fixture function-scoped so app lifespan and pool creation happen while the marker is active. The normal application configuration must still use a PostgreSQL driver.

FastAPI

conftest.py

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

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

Using TestClient as a context manager runs the app’s lifespan for each test.

Starlette

conftest.py

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

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

Flask

conftest.py

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’s test client calls the WSGI app directly; the database still uses the same stable pgmem endpoint.

Django

conftest.py

import os
import pytest

@pytest.fixture
def client():
    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myapp.settings")
    import django
    django.setup()  # use the application's normal PostgreSQL settings
    from django.test import Client
    return Client()

Run plain pytest functions with Django’s request Client. Do not use Django TestCase, pytest-django’s db or transactional_db fixtures, or its live_server with this lifecycle: they create, roll back, or flush a test database themselves. Application code may still use transaction.atomic() and savepoints. Keep this setup in a test environment where Django has not already initialized its database connections. For a separately launched Django server, see E2E tests.

Import shadow_pg from pgmem in the test module. The same test shape works for each client:

test_orders.py

@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

Adapt request bodies to the application’s API: FastAPI and Starlette often accept json=, while Flask and Django clients also support form data. The client starts while the marker is active and closes after the test.

Select a dataset

Override pgmem_live_snapshots to add snapshots from the same pgmem process, database, and user. Commit or close seed-loading connections before taking a snapshot.

conftest.py

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

test_orders.py

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

Reset happens before each test, including after a failed test. It waits for open transactions and raises ProtocolError(code="busy") after the timeout. Finish background work and close or commit transactions before the next test. The endpoint and connections survive; after restore, connection sessions are fresh, so session settings and temporary tables do not persist.

Note

This fixture controls the database contents between sequential tests. If two tests send requests concurrently to the same application process, their data cannot be isolated by this reset; use separate workers and application processes.

Capture application startup state

An application that creates its schema while it starts writes into the live fork after pgmem_snapshot was taken, so the first reset would undo it. Request pgmem_live_baseline where the application is started and call capture() once it is up; later resets restore that state.

conftest.py

@pytest.fixture(scope="session")
def app_url(pgmem_live_fork, pgmem_live_baseline):
    process = start_app(dsn=pgmem_live_fork.dsn)  # Alembic, create_all or migrate runs here
    wait_until_ready(process)
    pgmem_live_baseline.capture()  # the application's own schema becomes the default dataset
    try:
        yield "http://127.0.0.1:8000"
    finally:
        process.terminate()
        process.wait()

capture() keeps the first snapshot it takes; recapture() replaces it. Named datasets from pgmem_live_snapshots are unaffected. The captured snapshot holds one more copy of the data directory for the session, and like every snapshot it waits for open transactions and raises ProtocolError(code="busy") after the timeout.

Safe parallel execution

Use pytest -n 4 with pytest-xdist to run API test cases in separate worker processes. Each worker owns its pgmem process and stable live fork; keep the HTTP client function-scoped as above. Within one worker, @shadow_pg(live=True) resets that fork before a case, so do not overlap two cases against its app or start a thread-based parallel test runner. Finish background jobs and close transactions before the next reset. A single external app server shared by all workers would defeat this isolation: each worker needs its own app instance and database target.