Testing Python APIs with pytest: A Production Suite for Async Services
A commercial API earns money only while it keeps its promises, and the test suite is the machine that checks those promises before your customers do. This guide builds a complete pytest setup for an async Python API: event-loop configuration that does not fight you, fixtures for the app, the database and authenticated callers, transactional isolation so tests never leak state, contract tests that fail the build when your OpenAPI schema changes shape, and a CI job that finishes in under two minutes without flaking. Part of the Scaling and Operating Production Python APIs guide.
I am opinionated about two things here. First, test against a real Postgres, never SQLite — the dialect differences will hide bugs until a paying customer finds them. Second, isolate with a rolled-back transaction rather than truncating tables between tests, because it is roughly an order of magnitude faster and it scales to hundreds of tests without turning your pipeline into a coffee break.
Prerequisites
This guide assumes Python 3.11 or newer, a FastAPI application (0.115 or later) served by Uvicorn, and SQLAlchemy 2.0 with an async driver. If your data layer is not async yet, fix that first — the patterns below depend on it, and async database access with SQLAlchemy covers the engine and session setup this page builds on.
Install the test toolchain. Pin the majors; pytest-asyncio in particular has changed its configuration surface more than once, and an unpinned upgrade will break your suite on an unrelated Tuesday.
pip install \
"pytest>=8.3,<9" \
"pytest-asyncio>=0.24,<1.2" \
"pytest-xdist>=3.6" \
"httpx>=0.28" \
"asgi-lifespan>=2.1" \
"sqlalchemy[asyncio]>=2.0.36" \
"asyncpg>=0.30" \
"psycopg[binary]>=3.2" \
"alembic>=1.14" \
"jsonschema>=4.23" \
"respx>=0.22"
You also need a throwaway Postgres. Locally that is one command; in CI it is a service container. Use a separate database name from your development one so a stray --create-db run cannot destroy the data you were demoing yesterday.
docker run --rm -d -p 5433:5432 \
-e POSTGRES_PASSWORD=postgres \
-e POSTGRES_DB=api_test \
--name api-test-db postgres:16-alpine
Every knob in this guide comes from the environment. Nothing is hardcoded, which is what lets the same suite run on your laptop, in CI, and against an ephemeral preview database without editing a line of Python.
export TEST_DATABASE_URL="postgresql+asyncpg://postgres:postgres@localhost:5433/api_test"
export API_JWT_SECRET="local-dev-secret-not-for-production"
export API_JWT_ALGORITHM="HS256"
export APP_ENV="test"
export OPENAPI_SNAPSHOT_PATH="tests/contracts/openapi.json"
Step 1: Configure pytest-asyncio once, in pyproject.toml
Most async test pain traces back to a half-configured event loop. Set asyncio_mode = "auto" so every coroutine test runs without a decorator, and pin the default fixture loop scope to session so a session-scoped engine and a function-scoped test share one loop. Mismatched loop scopes are the single largest source of RuntimeError: ... attached to a different loop, and the fix is configuration, not code.
# pyproject.toml
[tool.pytest.ini_options]
minversion = "8.3"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
testpaths = ["tests"]
addopts = "-q --strict-markers --strict-config -p no:cacheprovider"
markers = [
"contract: validates responses against the committed OpenAPI schema",
"slow: excluded from the pre-commit run",
]
filterwarnings = ["error::DeprecationWarning"]
--strict-markers and filterwarnings = ["error::DeprecationWarning"] look pedantic until the day a library deprecation quietly changes serialization behaviour in a minor release. Turning deprecation warnings into failures means your suite tells you about the breakage during a routine dependency bump, not during an incident. Add a small meta-test that reads the config back with tomllib so nobody silently deletes the async mode while chasing an unrelated red build:
# tests/test_pytest_config.py
import tomllib
from pathlib import Path
def test_async_mode_is_auto() -> None:
config = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))
options = config["tool"]["pytest"]["ini_options"]
assert options["asyncio_mode"] == "auto"
assert options["asyncio_default_fixture_loop_scope"] == "session"
Step 2: Build the app fixture with a real lifespan
Your application almost certainly does work on startup — creating an HTTP client, warming a cache, opening a connection pool. If the tests skip that, they test a different program than the one you deploy. Drive the app through asgi-lifespan so startup and shutdown hooks actually run, and wrap it in an httpx AsyncClient over ASGITransport so requests never touch a real socket. In-process transport is what makes a 400-test suite finish in seconds; the detailed mechanics live in testing async FastAPI endpoints with httpx.
# tests/conftest.py
import os
from collections.abc import AsyncIterator
import pytest
from asgi_lifespan import LifespanManager
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from app.main import create_app
BASE_URL = os.getenv("TEST_BASE_URL", "http://testserver")
@pytest.fixture(scope="session")
async def app() -> AsyncIterator[FastAPI]:
os.environ.setdefault("APP_ENV", "test")
application = create_app()
async with LifespanManager(application, startup_timeout=30):
yield application
@pytest.fixture
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url=BASE_URL) as http:
yield http
Two details matter commercially. create_app() must be a factory, not a module-level singleton — a singleton bakes in whatever environment existed at import time and makes per-test configuration impossible. And startup_timeout should be generous but finite, so a hung startup fails in thirty seconds instead of burning a CI job's full six-hour allowance at your billing rate.
Step 3: Migrate a real Postgres, once per session
Run your actual Alembic migrations against the test database at session start. This costs a second or two and buys you a permanent regression test on the migration chain itself — the thing most likely to take production down during a deploy. Alembic's API is synchronous, so push it onto a worker thread rather than blocking the loop.
# tests/conftest.py (continued)
import asyncio
from alembic import command
from alembic.config import Config
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
TEST_DATABASE_URL = os.environ["TEST_DATABASE_URL"]
SYNC_URL = TEST_DATABASE_URL.replace("+asyncpg", "+psycopg")
def _upgrade_to_head() -> None:
config = Config(os.getenv("ALEMBIC_CONFIG", "alembic.ini"))
config.set_main_option("sqlalchemy.url", SYNC_URL)
command.upgrade(config, "head")
@pytest.fixture(scope="session")
async def async_engine() -> AsyncIterator[AsyncEngine]:
await asyncio.to_thread(_upgrade_to_head)
engine = create_async_engine(
TEST_DATABASE_URL,
pool_size=int(os.getenv("TEST_DB_POOL_SIZE", "5")),
max_overflow=0,
pool_pre_ping=True,
)
yield engine
await engine.dispose()
Set max_overflow=0 deliberately. In production an overflow pool hides leaks; in tests you want a leak to surface as an immediate timeout with a stack trace pointing at the offending fixture. If that error is already familiar, fixing connection pool exhaustion explains the production-side version of the same failure. The driver choice is not neutral either — asyncpg is materially faster under test parallelism, and asyncpg vs psycopg3 for FastAPI walks through the trade-off.
Step 4: Isolate every test inside a rolled-back transaction
Here is the technique that makes the whole suite fast and deterministic. Open one connection, begin a transaction on it, bind an AsyncSession to that connection with join_transaction_mode="create_savepoint", and roll the outer transaction back when the test ends. Application code can call session.commit() as often as it likes — each commit only releases a savepoint inside your outer transaction, so the data is visible to the test but never reaches the database permanently.
# tests/conftest.py (continued)
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
from app.db import get_session
@pytest.fixture
async def db_connection(async_engine: AsyncEngine) -> AsyncIterator[AsyncConnection]:
async with async_engine.connect() as connection:
transaction = await connection.begin()
yield connection
await transaction.rollback()
@pytest.fixture
async def db_session(db_connection: AsyncConnection) -> AsyncIterator[AsyncSession]:
session = AsyncSession(
bind=db_connection,
join_transaction_mode="create_savepoint",
expire_on_commit=False,
)
try:
yield session
finally:
await session.close()
@pytest.fixture(autouse=True)
def override_session(app: FastAPI, db_session: AsyncSession) -> AsyncIterator[None]:
async def _get_session() -> AsyncIterator[AsyncSession]:
yield db_session
app.dependency_overrides[get_session] = _get_session
yield
app.dependency_overrides.pop(get_session, None)
Making the override autouse is a judgement call I stand behind: it guarantees no test can accidentally hit the un-isolated session and leave rows behind. expire_on_commit=False keeps ORM objects usable after a commit, which is what you want when a test asserts on an object it just created through the API.
Step 5: Give tests an authenticated caller
Almost every endpoint worth billing for sits behind authentication, so make an authenticated client the default and an anonymous one the exception. Mint the token directly rather than driving the login endpoint on every test: signing a JWT costs microseconds, while a full login round-trip costs tens of milliseconds and couples unrelated tests to your auth flow. Keep exactly one test that exercises the real login path end to end, then mint for everything else. If you are still choosing a scheme, JWT vs API keys for Python APIs compares them on rotation cost and revocation.
# tests/conftest.py (continued)
from datetime import UTC, datetime, timedelta
import jwt # PyJWT
JWT_SECRET = os.environ["API_JWT_SECRET"]
JWT_ALGORITHM = os.getenv("API_JWT_ALGORITHM", "HS256")
def scopes_for(role: str) -> list[str]:
match role:
case "admin":
return ["orders:read", "orders:write", "billing:admin"]
case "customer":
return ["orders:read", "orders:write"]
case "readonly":
return ["orders:read"]
case _:
raise ValueError(f"unknown role: {role}")
def mint_token(subject: str, role: str, ttl_seconds: int = 300) -> str:
now = datetime.now(tz=UTC)
payload = {
"sub": subject,
"scopes": scopes_for(role),
"iat": now,
"exp": now + timedelta(seconds=ttl_seconds),
}
return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)
@pytest.fixture
def auth_client(client: AsyncClient) -> AsyncClient:
token = mint_token(os.getenv("TEST_USER_ID", "usr_test_001"), "customer")
client.headers["Authorization"] = f"Bearer {token}"
return client
@pytest.fixture
def admin_client(client: AsyncClient) -> AsyncClient:
token = mint_token(os.getenv("TEST_ADMIN_ID", "usr_admin_001"), "admin")
client.headers["Authorization"] = f"Bearer {token}"
return client
Resist the temptation to override the current_user dependency instead. Overriding skips signature verification, expiry checks and scope parsing — precisely the code that keeps one customer from reading another customer's data. The whole point of an auth test is to run that code. Reserve dependency overrides for the boundaries you genuinely cannot exercise in-process, and see handling API authentication in Python for the production side of the same middleware.
Step 6: Turn the OpenAPI schema into a contract test
Your published schema is a promise to paying integrators, and the cheapest way to keep that promise is to commit the generated schema and fail the build when it changes unexpectedly. Two tests do the job: one diffs the live schema against the committed snapshot, and one validates real response bodies against the schema components so drift between documentation and behaviour surfaces immediately. If your schema needs shaping before you snapshot it, documenting APIs with OpenAPI covers the generation side.
# tests/contracts/test_openapi_contract.py
import json
import os
from pathlib import Path
import pytest
from fastapi import FastAPI
from httpx import AsyncClient
from jsonschema import Draft202012Validator
SNAPSHOT = Path(os.getenv("OPENAPI_SNAPSHOT_PATH", "tests/contracts/openapi.json"))
def _canonical(schema: dict) -> str:
return json.dumps(schema, sort_keys=True, indent=2, ensure_ascii=False)
@pytest.mark.contract
def test_schema_matches_snapshot(app: FastAPI) -> None:
live = _canonical(app.openapi())
if os.getenv("UPDATE_OPENAPI_SNAPSHOT") == "1":
SNAPSHOT.parent.mkdir(parents=True, exist_ok=True)
SNAPSHOT.write_text(live + "\n", encoding="utf-8")
pytest.skip("snapshot refreshed")
committed = SNAPSHOT.read_text(encoding="utf-8").rstrip("\n")
assert live == committed, (
"OpenAPI schema changed. Review the diff, then re-run with "
"UPDATE_OPENAPI_SNAPSHOT=1 if the change is intentional."
)
@pytest.mark.contract
async def test_list_orders_matches_declared_schema(
app: FastAPI, auth_client: AsyncClient
) -> None:
document = app.openapi()
component = document["components"]["schemas"]["OrderPage"]
validator = Draft202012Validator({**component, "$defs": document["components"]["schemas"]})
response = await auth_client.get("/v1/orders", params={"limit": 10})
assert response.status_code == 200
validator.validate(response.json())
The UPDATE_OPENAPI_SNAPSHOT=1 escape hatch matters. Without it, contributors edit the snapshot by hand, get it subtly wrong, and stop trusting the test. With it, updating the contract becomes a deliberate one-command act that shows up as a reviewable diff in the pull request — which is exactly where a breaking change should be caught. Pair this with a rule about which diffs are acceptable: adding an optional field is additive and safe, while removing a field, narrowing a type or tightening a required list breaks customers. That policy is the subject of versioning and evolving public APIs, and deprecating an API endpoint without breaking customers covers the retirement path.
Step 7: Run it in CI without flakes
Parallelise with pytest-xdist, give each worker its own database, and stop there. Most flakiness in an API suite comes from three places: shared mutable state between workers, real network calls, and wall-clock time. Kill all three at once. Give every xdist worker a distinct database name derived from the PYTEST_XDIST_WORKER environment variable, block outbound HTTP so a forgotten live call fails loudly instead of intermittently, and freeze anything that depends on the clock.
# tests/conftest.py (continued)
import socket
def worker_database_url() -> str:
base = os.environ["TEST_DATABASE_URL"]
worker = os.getenv("PYTEST_XDIST_WORKER", "gw0")
return f"{base}_{worker}" if worker != "master" else base
@pytest.fixture(scope="session", autouse=True)
def block_outbound_network() -> None:
"""Fail fast if a test tries to open a real socket."""
if os.getenv("ALLOW_TEST_NETWORK") == "1":
return
real_connect = socket.socket.connect
allowed = {"127.0.0.1", "::1", "localhost"}
def guarded(self, address, *args, **kwargs):
host = address[0] if isinstance(address, tuple) else str(address)
if host not in allowed:
raise RuntimeError(f"blocked outbound connection to {host}")
return real_connect(self, address, *args, **kwargs)
socket.socket.connect = guarded
That socket guard turns "the payment provider was slow today" into a deterministic failure with a stack trace. Replace those calls with recorded doubles — mocking external APIs with respx covers the httpx-native way to do it, and for billing specifically, testing Stripe integrations with test clocks shows how to simulate a renewal cycle without waiting a month.
# .github/workflows/test.yml
name: test
on: [push, pull_request]
jobs:
pytest:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16-alpine
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: api_test
ports: ["5432:5432"]
options: >-
--health-cmd pg_isready --health-interval 5s
--health-timeout 5s --health-retries 10
env:
TEST_DATABASE_URL: postgresql+asyncpg://postgres:postgres@localhost:5432/api_test
API_JWT_SECRET: ${{ secrets.TEST_JWT_SECRET }}
APP_ENV: test
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
- run: pip install -r requirements-dev.txt
- run: pytest -n 4 --dist loadscope --maxfail=1
Use --dist loadscope rather than the default. It keeps all tests in a module on the same worker, so module-scoped fixtures build once instead of once per worker, and it removes an entire family of ordering surprises.
Configuration reference
Every variable below has a safe default for local work and a specific recommendation for CI. Treat the CI column as the contract for your pipeline definition.
| Variable | Default | Recommended in CI |
|---|---|---|
TEST_DATABASE_URL | none, required | service container URL |
APP_ENV | dev | test |
API_JWT_SECRET | none, required | repository secret |
API_JWT_ALGORITHM | HS256 | HS256 or RS256 |
TEST_DB_POOL_SIZE | 5 | 5 per worker |
OPENAPI_SNAPSHOT_PATH | tests/contracts/openapi.json | same, committed |
UPDATE_OPENAPI_SNAPSHOT | unset | never set |
ALLOW_TEST_NETWORK | unset | never set |
PYTEST_ADDOPTS | unset | -n 4 --dist loadscope |
Two of these deserve emphasis. UPDATE_OPENAPI_SNAPSHOT must never be set in CI, because a pipeline that regenerates its own contract cannot detect a breaking change. ALLOW_TEST_NETWORK exists only for a nightly smoke job that deliberately hits a provider sandbox; if you find it set on the main test job, someone was silencing a failure rather than fixing it.
Gotchas and failure modes
Symptom: RuntimeError: Task got Future attached to a different loop. You created an async resource in a session-scoped fixture while pytest-asyncio ran fixtures on a function-scoped loop. Fix it by setting asyncio_default_fixture_loop_scope = "session" in pyproject.toml, and make sure the engine, the connection and the client all live in the same or a narrower scope than the loop. Never call asyncio.get_event_loop() yourself inside a fixture.
Symptom: tests pass alone, fail in a full run, and leave rows behind. Your application code opened its own session instead of the injected one — usually a background task, a Celery producer, or a module-level async_sessionmaker. The savepoint trick only isolates work that flows through the overridden dependency. Audit for direct engine use, and route everything through get_session. Making the override autouse catches most of these on the first run.
Symptom: everything is green locally, and a customer reports a 500 in production. You tested against SQLite. JSONB, ON CONFLICT DO UPDATE, array columns, case-sensitive collation and transactional DDL all behave differently, and none of those differences show up until real traffic hits. Run the same Postgres major version you deploy; a service container costs about fifteen seconds of job time and removes an entire category of production incident.
Symptom: startup-only resources are None inside tests. You built the client with a bare ASGITransport and never triggered lifespan events, so the HTTP pool, cache client or metrics registry your handlers expect was never created. Wrap the app in LifespanManager as shown in step 2. The same class of bug bites people who mount routers directly instead of going through their app factory.
Symptom: an intermittent failure every few dozen runs, always in a date or ordering assertion. Either a test asserts on datetime.now() crossing a boundary, or it depends on the insertion order of rows with equal timestamps. Freeze the clock at a fixed instant for time-sensitive tests, and always add an explicit ORDER BY with a tiebreaker column to any query whose order you assert on. A flake that costs one re-run a day costs a small team several hours a month in lost focus.
Verification
Prove the isolation actually works before you trust it. Write a pair of tests that would fail if state leaked, and run them in both orders:
# tests/test_isolation.py
from httpx import AsyncClient
from sqlalchemy import func, select
from app.models import Order
async def test_creates_one_order(auth_client: AsyncClient, db_session) -> None:
response = await auth_client.post("/v1/orders", json={"sku": "PRO-1", "qty": 2})
assert response.status_code == 201
total = await db_session.scalar(select(func.count()).select_from(Order))
assert total == 1
async def test_database_starts_empty(db_session) -> None:
total = await db_session.scalar(select(func.count()).select_from(Order))
assert total == 0
Run them shuffled and repeatedly. If test_database_starts_empty ever fails, isolation is broken somewhere and every other result in the suite is suspect:
pytest tests/test_isolation.py -p no:randomly -v
pytest tests/test_isolation.py::test_database_starts_empty tests/test_isolation.py::test_creates_one_order -v
psql "$TEST_DATABASE_URL_PSQL" -c "select count(*) from orders;" # expect 0
Then verify the contract gate does its job by deliberately breaking it. Delete a response field from a Pydantic model and run pytest -m contract. You should see the snapshot assertion fail with a diff naming the removed property. If it passes, your snapshot is stale or the marker is not selecting the test — either way, the gate was providing false confidence, which is worse than no gate.
Cost and performance at scale
A well-built API suite is cheap enough that the argument against running it never comes up. On a standard two-core GitHub-hosted runner at roughly $0.008 per minute, a 412-test suite of the shape described here takes about 372 seconds sequentially and about 112 seconds at -n 4 — call it two and a half cents per push, or under fifteen dollars a month for a team pushing thirty times a day. Against that, a single breaking schema change shipped to a paying integrator costs a support thread, an emergency deploy, and sometimes a refund. The economics are not close.
The scaling curve matters when you pick a worker count. Going from one worker to four cuts the run by 70%, but four to eight buys only another twelve seconds because the single Postgres container becomes the constraint — you are paying for runner cores that sit idle waiting on fsync. Match workers to available vCPUs, not to ambition. If you genuinely need more parallelism, the next lever is a larger runner with a tmpfs-backed database volume, which typically halves database time again for a few dollars a month. Watch the trend rather than the absolute number: a suite that creeps from ninety seconds to four minutes over a quarter is telling you that fixtures are doing too much setup per test, and the usual culprit is a fixture that should be session-scoped but is not. The same discipline that keeps your cost per API request predictable applies to your pipeline: measure it, attribute it, and keep the feedback loop under two minutes so shipping stays fast.
FAQ
How much does a test suite like this cost to run each month? On hosted CI at roughly $0.008 per runner-minute, a two-minute parallel run costs about 1.6 cents. Thirty pushes a day lands near fifteen dollars a month, and a self-hosted runner on a small VPS drops that to the price of the box. Compare that with a single breaking change reaching integrators and the suite pays for itself the first time it catches one.
Is contract testing worth it before I have paying customers? Yes, and it gets cheaper the earlier you start. Committing the schema snapshot on day one costs a few minutes; retrofitting it after twenty integrators depend on undocumented field names means you first have to discover what you accidentally promised. The snapshot also becomes your changelog source when you start publishing versioned releases.
Will testing against real Postgres slow my pipeline enough to matter? A service container adds roughly twelve to twenty seconds of startup and a few milliseconds per transactional test. That is inside the noise of a two-minute build. SQLite is faster but tests a different database than the one holding your customers' money, which is a trade that only looks good until the first dialect bug.
How do I test key rotation and expiry without waiting for tokens to age? Mint tokens with an explicit short TTL in the fixture and assert on the 401, rather than sleeping. For rotation, sign one token with the outgoing secret and one with the incoming secret, then assert both verify during the overlap window and only the new one verifies afterwards. That is the same overlap window you use in production.
What is the migration risk of adopting transactional isolation on an existing suite? Low, and it surfaces real bugs. The tests that break are the ones already depending on leaked state or on code paths that bypass the injected session — both of which are defects. Convert one test module at a time, keep the old truncation fixture available behind a marker during the transition, and delete it once the suite is green.
Related
Same track:
- Testing Async FastAPI Endpoints with httpx — the request-level mechanics behind the
clientfixture. - Mocking External APIs with respx — replace the outbound calls your network guard now blocks.
- Async Database Access with SQLAlchemy — the engine and session setup these fixtures wrap.
- Versioning and Evolving Public APIs — what to do when the contract test correctly fails.
Other tracks:
- Documenting APIs with OpenAPI — generating the schema your snapshot test guards.
- Zero-Downtime Deploys for Python APIs — what a green suite should trigger next.