pytest improvements
This commit is contained in:
John Lancaster
@@ -0,0 +1,236 @@
|
||||
# [FastAPI Testing](https://fastapi.tiangolo.com/tutorial/testing/)
|
||||
|
||||
Best practices for testing FastAPI applications with pytest.
|
||||
|
||||
## Agent Quick Path
|
||||
Use this sequence before reading the full reference:
|
||||
|
||||
1. If test is pure route behavior, use `TestClient` and plain `def` tests.
|
||||
2. If test must `await` other async work, use `AsyncClient` + `@pytest.mark.anyio`.
|
||||
3. Prefer `app.dependency_overrides` over `mock.patch`.
|
||||
4. Reset overrides after each test/fixture teardown.
|
||||
5. For startup/shutdown logic, use `TestClient` as context manager or `LifespanManager` with async client.
|
||||
|
||||
Decision rules:
|
||||
- Need DB contract verification: choose integration tests and override `get_db`/`get_session`.
|
||||
- Need pure business logic checks: keep tests HTTP-free (`unit`).
|
||||
- Need one critical path sanity check: one endpoint per `smoke` test.
|
||||
|
||||
---
|
||||
|
||||
## Tools
|
||||
|
||||
| Tool | Purpose |
|
||||
|------|---------|
|
||||
| `fastapi.testclient.TestClient` | Synchronous HTTP test client (wraps HTTPX, built on Starlette) |
|
||||
| `httpx.AsyncClient` + `ASGITransport` | Async client for tests that `await` other async code |
|
||||
| `app.dependency_overrides` | Replace any `Depends()` dependency for the duration of a test |
|
||||
| `anyio` / `pytest-anyio` | Run async test functions with `@pytest.mark.anyio` |
|
||||
|
||||
Install deps: `httpx`, `anyio` (or `pytest-anyio`).
|
||||
|
||||
---
|
||||
|
||||
## Synchronous Tests (Preferred Default)
|
||||
|
||||
Use `TestClient` for route tests that don't need to `await` anything else.
|
||||
Test functions are plain `def` — no `async def`, no `await`.
|
||||
|
||||
```python
|
||||
from fastapi.testclient import TestClient
|
||||
from myapp.main import app
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
def test_read_item_returns_200():
|
||||
response = client.get("/items/foo", headers={"X-Token": "secret"})
|
||||
assert response.status_code == 200
|
||||
assert response.json()["id"] == "foo"
|
||||
|
||||
def test_read_item_bad_token_returns_400():
|
||||
response = client.get("/items/foo", headers={"X-Token": "wrong"})
|
||||
assert response.status_code == 400
|
||||
```
|
||||
|
||||
Rules:
|
||||
- One `TestClient` per test module is fine (stateless between calls).
|
||||
- Pass headers, query params, JSON body, or form data the same way as HTTPX/requests.
|
||||
- Do not pass Pydantic models directly; use `.model_dump()` or `jsonable_encoder`.
|
||||
|
||||
---
|
||||
|
||||
## Async Tests
|
||||
|
||||
Use `AsyncClient` only when the test itself needs to `await` other coroutines
|
||||
(e.g. querying a real async DB after an API call to verify side effects).
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from myapp.main import app
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_root_async():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
||||
response = await ac.get("/")
|
||||
assert response.status_code == 200
|
||||
```
|
||||
|
||||
Rules:
|
||||
- Mark with `@pytest.mark.anyio`; register the `anyio` marker in `pyproject.toml`.
|
||||
- `AsyncClient` does **not** trigger lifespan events by default; use `asgi-lifespan`'s
|
||||
`LifespanManager` when startup/shutdown matters.
|
||||
- Instantiate objects that require an event loop (e.g. async DB clients) inside
|
||||
async functions, not at module level.
|
||||
|
||||
---
|
||||
|
||||
## Dependency Overrides (Preferred Over Mocking)
|
||||
|
||||
`app.dependency_overrides` is the idiomatic FastAPI seam — use it instead of
|
||||
patching internals with `unittest.mock`.
|
||||
|
||||
```python
|
||||
from fastapi.testclient import TestClient
|
||||
from myapp.main import app
|
||||
from myapp.deps import get_current_user
|
||||
|
||||
client = TestClient(app)
|
||||
|
||||
def fake_user():
|
||||
return {"id": 1, "name": "Test User"}
|
||||
|
||||
def test_protected_route_with_fake_user():
|
||||
app.dependency_overrides[get_current_user] = fake_user
|
||||
response = client.get("/me")
|
||||
app.dependency_overrides = {} # always reset after the test
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == "Test User"
|
||||
```
|
||||
|
||||
Or reset cleanly with `autouse=False` fixture teardown:
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from myapp.main import app
|
||||
|
||||
@pytest.fixture()
|
||||
def override_user():
|
||||
app.dependency_overrides[get_current_user] = lambda: {"id": 1, "name": "Test User"}
|
||||
yield
|
||||
app.dependency_overrides = {}
|
||||
```
|
||||
|
||||
Rules:
|
||||
- Override at the lowest-level dependency that owns the external boundary
|
||||
(e.g. `get_db`, `get_current_user`, `get_settings`).
|
||||
- Always reset `app.dependency_overrides` after each test or fixture teardown.
|
||||
- Prefer a real in-process fake (e.g. in-memory SQLite session) over a mock object.
|
||||
|
||||
---
|
||||
|
||||
## Database Testing
|
||||
|
||||
The preferred pattern is a real SQLite (or test Postgres) session injected via
|
||||
dependency override, not a mock.
|
||||
|
||||
```python
|
||||
# tests/conftest.py
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from fastapi.testclient import TestClient
|
||||
from myapp.main import app
|
||||
from myapp.db.session import get_db
|
||||
from myapp.db.models import Base
|
||||
|
||||
TEST_DB_URL = "sqlite:///./test.db"
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def engine():
|
||||
e = create_engine(TEST_DB_URL, connect_args={"check_same_thread": False})
|
||||
Base.metadata.create_all(bind=e)
|
||||
yield e
|
||||
Base.metadata.drop_all(bind=e)
|
||||
|
||||
@pytest.fixture()
|
||||
def db_session(engine):
|
||||
Session = sessionmaker(bind=engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.rollback()
|
||||
session.close()
|
||||
|
||||
@pytest.fixture()
|
||||
def client(db_session):
|
||||
app.dependency_overrides[get_db] = lambda: db_session
|
||||
yield TestClient(app)
|
||||
app.dependency_overrides = {}
|
||||
```
|
||||
|
||||
Rules:
|
||||
- Prefer `session`-scoped engine creation; `function`-scoped session with rollback per test.
|
||||
- Keep unit tests DB-free; use this pattern only in `integration`-marked tests.
|
||||
- For async SQLAlchemy, mirror the same pattern using `AsyncEngine` / `AsyncSession`.
|
||||
|
||||
---
|
||||
|
||||
## Lifespan and Startup Events
|
||||
|
||||
`TestClient` triggers lifespan events (startup/shutdown) when used as a context manager:
|
||||
|
||||
```python
|
||||
def test_with_lifespan():
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
```
|
||||
|
||||
For `AsyncClient`, use `asgi-lifespan`:
|
||||
|
||||
```python
|
||||
from asgi_lifespan import LifespanManager
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_with_async_lifespan():
|
||||
async with LifespanManager(app):
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
||||
response = await ac.get("/health")
|
||||
assert response.status_code == 200
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Marker Strategy for FastAPI Tests
|
||||
|
||||
| Marker | When to use |
|
||||
|--------|------------|
|
||||
| `unit` | Pure service/utility logic with no HTTP or DB calls |
|
||||
| `integration` | `TestClient` + real DB session via dependency override |
|
||||
| `smoke` | One `TestClient` call per critical user path, no DB reset |
|
||||
| `external` | Tests that call real third-party APIs (skip in CI by default) |
|
||||
|
||||
---
|
||||
|
||||
## Quick Reference: Sending Data
|
||||
|
||||
| What to send | Parameter |
|
||||
|---|---|
|
||||
| Path / query param | Part of the URL string |
|
||||
| JSON body | `json={"key": "value"}` |
|
||||
| Form data | `data={"field": "value"}` |
|
||||
| Headers | `headers={"X-Token": "..."}` |
|
||||
| Cookies | `cookies={"session": "..."}` |
|
||||
| File upload | `files={"file": ("name.txt", b"content", "text/plain")}` |
|
||||
|
||||
---
|
||||
|
||||
## Official Docs
|
||||
|
||||
- [Testing tutorial](https://fastapi.tiangolo.com/tutorial/testing/)
|
||||
- [Async tests](https://fastapi.tiangolo.com/advanced/async-tests/)
|
||||
- [Testing dependencies with overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
|
||||
- [Testing a database (SQLModel)](https://fastapi.tiangolo.com/how-to/testing-database/)
|
||||
- [Testing lifespan events](https://fastapi.tiangolo.com/advanced/testing-events/)
|
||||
- [Testing WebSockets](https://fastapi.tiangolo.com/advanced/testing-websockets/)
|
||||
- [HTTPX docs](https://www.python-httpx.org/)
|
||||
@@ -0,0 +1,35 @@
|
||||
# Pytest Documentation Notes
|
||||
|
||||
Primary references used:
|
||||
- https://docs.pytest.org/en/stable/explanation/goodpractices.html
|
||||
- https://docs.pytest.org/en/stable/how-to/fixtures.html
|
||||
- https://docs.pytest.org/en/stable/example/markers.html
|
||||
- https://docs.pytest.org/en/stable/reference/customize.html
|
||||
- https://docs.pytest.org/en/stable/explanation/flaky.html
|
||||
|
||||
## Agent Quick Path
|
||||
Use this file when you need fast pytest scaffolding defaults without framework-specific details.
|
||||
|
||||
1. Mirror source layout under `tests/`.
|
||||
2. Keep fixtures small and explicit; default to `function` scope.
|
||||
3. Register markers up front in `pyproject.toml`.
|
||||
4. Validate structure first with `uv run pytest --collect-only -q`.
|
||||
5. Run fast lane with `uv run pytest -m unit -q`.
|
||||
|
||||
Load other references only when needed:
|
||||
- FastAPI routes/dependency injection/lifespan: `fastapi-testing.md`
|
||||
- SQLAlchemy sessions/transactions/DB fixtures: `sqlalchemy-testing.md`
|
||||
|
||||
## Practical Guidance For This Skill
|
||||
- Use src-aligned test layout and keep test discovery conventional.
|
||||
- Keep fixtures small, composable, and explicit; use `yield` for teardown.
|
||||
- Register custom markers and keep strict marker validation on.
|
||||
- Separate quick unit runs from slower integration/external runs.
|
||||
- Minimize flakiness by controlling shared state and avoiding hidden dependencies.
|
||||
- Use `--collect-only` and marker-filtered runs to validate scaffold quality early.
|
||||
|
||||
## Commands Worth Remembering
|
||||
- `uv run pytest --collect-only -q`
|
||||
- `uv run pytest -m unit -q`
|
||||
- `uv run pytest -m "not external" -q`
|
||||
- `uv run pytest -q`
|
||||
@@ -0,0 +1,246 @@
|
||||
# [SQLAlchemy 2.x Testing](https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
|
||||
|
||||
Best practices for testing SQLAlchemy ORM code (sync and async) with pytest.
|
||||
|
||||
## Agent Quick Path
|
||||
Use this path first; read deeper sections only when needed.
|
||||
|
||||
1. Create engine once per test session.
|
||||
2. Open connection + outer transaction per test function.
|
||||
3. Bind `Session`/`AsyncSession` to that connection with `join_transaction_mode="create_savepoint"`.
|
||||
4. Let code under test call `commit()` safely; rollback outer transaction after test.
|
||||
5. Inject session into FastAPI via dependency override and always clear overrides.
|
||||
|
||||
Branching logic:
|
||||
- Sync stack: use `create_engine` + `Session` fixtures.
|
||||
- Async stack: use `create_async_engine` + `AsyncSession` fixtures + `pytest.mark.anyio`.
|
||||
- SQLite in-memory with threaded client: use `StaticPool` when required by framework threading behavior.
|
||||
- Async relationship access fails (`MissingGreenlet`): eager load (`selectinload`) or explicit refresh.
|
||||
|
||||
---
|
||||
|
||||
## Core Concepts
|
||||
|
||||
| Concept | Preferred approach |
|
||||
|---|---|
|
||||
| Isolate tests from production DB | Use an in-memory SQLite engine per test |
|
||||
| Prevent data leaking between tests | Roll back at the connection level after each test |
|
||||
| Inject test DB into FastAPI | Override the `get_db` / `get_session` dependency |
|
||||
| Create schema for tests | Call `Base.metadata.create_all(engine)` once per session |
|
||||
| Avoid lazy-load errors in async | Use `expire_on_commit=False`; use `selectinload()` for relationships |
|
||||
|
||||
---
|
||||
|
||||
## Sync SQLAlchemy + FastAPI (Recommended Pattern)
|
||||
|
||||
The canonical 2.0 pattern joins a test `Session` into an external transaction on a shared `Connection`, then rolls back after each test. This means `session.commit()` calls within the code under test are "committed" to a savepoint, not the real transaction, and are fully undone after the test.
|
||||
|
||||
```python
|
||||
# tests/conftest.py
|
||||
import pytest
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from myapp.main import app
|
||||
from myapp.db.session import get_db
|
||||
from myapp.db.models import Base
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def engine():
|
||||
e = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
||||
Base.metadata.create_all(bind=e)
|
||||
yield e
|
||||
Base.metadata.drop_all(bind=e)
|
||||
|
||||
@pytest.fixture()
|
||||
def db_session(engine):
|
||||
connection = engine.connect()
|
||||
transaction = connection.begin()
|
||||
session = Session(bind=connection, join_transaction_mode="create_savepoint")
|
||||
yield session
|
||||
session.close()
|
||||
transaction.rollback()
|
||||
connection.close()
|
||||
|
||||
@pytest.fixture()
|
||||
def client(db_session):
|
||||
app.dependency_overrides[get_db] = lambda: db_session
|
||||
yield TestClient(app)
|
||||
app.dependency_overrides = {}
|
||||
```
|
||||
|
||||
Key points:
|
||||
- `join_transaction_mode="create_savepoint"` means each `session.commit()` inside code under test issues a SAVEPOINT release, not a real COMMIT — everything is rolled back when the test ends.
|
||||
- `scope="session"` engine + `scope="function"` session/connection gives fast table creation with full per-test isolation.
|
||||
- SQLite in-memory (`sqlite://`) is preferred: no files, no cleanup, fast.
|
||||
|
||||
---
|
||||
|
||||
## Async SQLAlchemy + FastAPI
|
||||
|
||||
For `AsyncSession` / `AsyncEngine`, the setup mirrors the sync version but uses async fixtures and `pytest-anyio`.
|
||||
|
||||
```python
|
||||
# tests/conftest.py
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
||||
from fastapi.testclient import TestClient # sync client still works
|
||||
from httpx import AsyncClient, ASGITransport
|
||||
|
||||
from myapp.main import app
|
||||
from myapp.db.session import get_db
|
||||
from myapp.db.models import Base
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def async_engine():
|
||||
e = create_async_engine("sqlite+aiosqlite://", connect_args={"check_same_thread": False})
|
||||
async with e.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield e
|
||||
async with e.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
await e.dispose()
|
||||
|
||||
@pytest.fixture()
|
||||
async def async_session(async_engine):
|
||||
async with async_engine.connect() as conn:
|
||||
await conn.begin()
|
||||
session = AsyncSession(bind=conn, join_transaction_mode="create_savepoint",
|
||||
expire_on_commit=False)
|
||||
yield session
|
||||
await session.close()
|
||||
await conn.rollback()
|
||||
|
||||
@pytest.fixture()
|
||||
async def async_client(async_session):
|
||||
async def override_get_db():
|
||||
yield async_session
|
||||
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
|
||||
yield ac
|
||||
app.dependency_overrides = {}
|
||||
```
|
||||
|
||||
Notes:
|
||||
- Async fixtures need `@pytest.mark.anyio` on the test or `anyio_backend` session fixture.
|
||||
- `expire_on_commit=False` prevents expired-attribute access on objects after `await session.commit()`.
|
||||
- Install: `aiosqlite`, `anyio[asyncio]`, `httpx`.
|
||||
|
||||
---
|
||||
|
||||
## SQLModel Pattern (FastAPI + SQLModel)
|
||||
|
||||
SQLModel's official test pattern uses `StaticPool` + in-memory SQLite, with separate `session` and `client` pytest fixtures.
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
from sqlmodel.pool import StaticPool
|
||||
|
||||
from myapp.main import app, get_session # get_session is the SQLModel dependency
|
||||
|
||||
@pytest.fixture(name="session")
|
||||
def session_fixture():
|
||||
engine = create_engine(
|
||||
"sqlite://",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
SQLModel.metadata.create_all(engine)
|
||||
with Session(engine) as session:
|
||||
yield session
|
||||
|
||||
@pytest.fixture(name="client")
|
||||
def client_fixture(session: Session):
|
||||
app.dependency_overrides[get_session] = lambda: session
|
||||
yield TestClient(app)
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
# --- Tests ---
|
||||
|
||||
def test_create_hero(client: TestClient):
|
||||
response = client.post("/heroes/", json={"name": "Deadpond", "secret_name": "Dive Wilson"})
|
||||
assert response.status_code == 200
|
||||
|
||||
def test_read_heroes(session: Session, client: TestClient):
|
||||
# Directly insert test data via the session — no HTTP call needed for setup
|
||||
from myapp.models import Hero
|
||||
session.add(Hero(name="Test Hero", secret_name="Hidden"))
|
||||
session.commit()
|
||||
response = client.get("/heroes/")
|
||||
assert response.status_code == 200
|
||||
assert len(response.json()) == 1
|
||||
```
|
||||
|
||||
Rules:
|
||||
- Use `StaticPool` so a single in-memory SQLite connection is shared across threads (required by `TestClient`'s threading model).
|
||||
- Both the `client` fixture and test functions can receive the same `session` — insert data directly for controlled setup rather than via the API.
|
||||
- Always call `app.dependency_overrides.clear()` in fixture teardown (after `yield`).
|
||||
|
||||
---
|
||||
|
||||
## Async Session: Avoiding Implicit I/O
|
||||
|
||||
SQLAlchemy async has strict rules about lazy loading — attributes that would trigger IO on access will raise an error.
|
||||
|
||||
```python
|
||||
# WRONG — will raise MissingGreenlet / lazy load error
|
||||
result = await session.execute(select(User))
|
||||
user = result.scalars().one()
|
||||
print(user.posts) # lazy load, fails in async context
|
||||
|
||||
# RIGHT — use selectinload() to load relationships eagerly
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
result = await session.execute(
|
||||
select(User).options(selectinload(User.posts))
|
||||
)
|
||||
user = result.scalars().one()
|
||||
print(user.posts) # already loaded, no IO needed
|
||||
```
|
||||
|
||||
Other strategies:
|
||||
- `AsyncAttrs` mixin: access any attribute as an awaitable via `await obj.awaitable_attrs.relationship_name`.
|
||||
- `write_only` relationships: never loaded implicitly; queried explicitly.
|
||||
- `await session.refresh(obj, ["attribute_name"])`: force-load a specific attribute after the fact.
|
||||
|
||||
---
|
||||
|
||||
## Fixture Scope Decision Table
|
||||
|
||||
| What to scope | `scope` | Reason |
|
||||
|---|---|---|
|
||||
| Engine + DDL (`create_all`) | `session` | Expensive; shared across all tests |
|
||||
| Connection + Transaction | `function` | Rolled back per test for isolation |
|
||||
| Session | `function` | One transaction per test |
|
||||
| TestClient / AsyncClient | `function` | Depends on session; recreated per test |
|
||||
|
||||
---
|
||||
|
||||
## Common Pitfalls
|
||||
|
||||
| Problem | Cause | Fix |
|
||||
|---|---|---|
|
||||
| `MissingGreenlet` in async | Lazy-loaded relationship accessed outside awaitable context | Use `selectinload()` or `AsyncAttrs.awaitable_attrs` |
|
||||
| `RuntimeError: Event loop is closed` | `AsyncEngine` not disposed | Call `await engine.dispose()` in fixture teardown |
|
||||
| Tests share state / data bleeds | Session not rolled back | Use `join_transaction_mode="create_savepoint"` + rollback pattern |
|
||||
| `StaticPool` not used with SQLite in-memory | TestClient spawns threads that get separate in-memory DBs | Always add `poolclass=StaticPool` for in-memory SQLite |
|
||||
| `expire_on_commit=True` (default) breaks async | Accessing attributes after commit triggers lazy IO | Set `expire_on_commit=False` on AsyncSession |
|
||||
| Not resetting `dependency_overrides` | Override persists into next test | Always clear in fixture teardown, after `yield` |
|
||||
|
||||
---
|
||||
|
||||
## Official Docs
|
||||
|
||||
- [Joining a Session into an External Transaction (test suites)](https://docs.sqlalchemy.org/en/20/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites)
|
||||
- [Asynchronous I/O (asyncio)](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html)
|
||||
- [Preventing Implicit IO when Using AsyncSession](https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html#preventing-implicit-io-when-using-asyncsession)
|
||||
- [SQLModel: Test Applications with FastAPI](https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/)
|
||||
- [FastAPI: Testing Dependencies with Overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
|
||||
Reference in New Issue
Block a user