rename
This commit is contained in:
John Lancaster
@@ -0,0 +1,108 @@
|
||||
# AsyncIO Testing Patterns (Pytest, FastAPI, AnyIO)
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [pytest-asyncio configuration](https://pytest-asyncio.readthedocs.io/en/stable/reference/configuration.html)
|
||||
- [pytest-asyncio concepts](https://pytest-asyncio.readthedocs.io/en/stable/concepts.html)
|
||||
- [pytest-asyncio fixture loop scope how-to](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/change_fixture_loop.html)
|
||||
- [pytest-asyncio default fixture loop scope how-to](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/change_default_fixture_loop.html)
|
||||
- [AnyIO cancellation and cancel-scope safety](https://anyio.readthedocs.io/en/stable/cancellation.html)
|
||||
- [FastAPI async tests](https://fastapi.tiangolo.com/advanced/async-tests/)
|
||||
|
||||
## Agent Quick Path
|
||||
Use this reference when tests involve asynchronous fixtures, HTTP clients, task groups, or teardown failures.
|
||||
|
||||
1. Confirm async plugin mode in pytest config (`asyncio_mode`).
|
||||
2. Keep async fixture loop scope predictable, defaulting to `function` unless there is a measured need to broaden it.
|
||||
3. Prefer one async testing model per lane (pytest-asyncio or AnyIO-style markers), and keep it consistent.
|
||||
4. Keep async fixtures small and isolate stateful resources to the narrowest useful scope.
|
||||
5. If teardown errors mention cancel scopes or task groups, validate that setup and teardown run in the same task context.
|
||||
|
||||
## Baseline Configuration
|
||||
|
||||
Recommended defaults for most projects using `pytest-asyncio`:
|
||||
|
||||
```toml
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
asyncio_default_fixture_loop_scope = "function"
|
||||
```
|
||||
|
||||
Why:
|
||||
- [Strict mode](https://pytest-asyncio.readthedocs.io/en/stable/concepts.html#test-discovery-modes) is safer for multi-plugin environments, but [auto mode](https://pytest-asyncio.readthedocs.io/en/stable/concepts.html#test-discovery-modes) is often simpler when the suite is primarily asyncio-based.
|
||||
- [Function loop scope](https://pytest-asyncio.readthedocs.io/en/stable/reference/configuration.html#asyncio-default-fixture-loop-scope) minimizes cross-test coupling and avoids many lifecycle surprises.
|
||||
|
||||
If a fixture or test needs broader loop sharing, make it explicit instead of changing suite-wide defaults:
|
||||
|
||||
```python
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
|
||||
@pytest_asyncio.fixture(loop_scope="module")
|
||||
async def shared_resource():
|
||||
...
|
||||
|
||||
|
||||
@pytest.mark.asyncio(loop_scope="module")
|
||||
async def test_uses_shared_loop(shared_resource):
|
||||
...
|
||||
```
|
||||
|
||||
## FastAPI Endpoint Test Patterns
|
||||
|
||||
Use [FastAPI's async testing guidance](https://fastapi.tiangolo.com/advanced/async-tests/) as the default:
|
||||
|
||||
1. Use `httpx.AsyncClient` with `ASGITransport` for async endpoint tests.
|
||||
2. Mark async tests with one consistent marker style for the suite.
|
||||
3. If app lifespan hooks matter, add [LifespanManager](https://fastapi.tiangolo.com/advanced/async-tests/#httpx) support because `AsyncClient` alone does not trigger lifespan events.
|
||||
|
||||
Example:
|
||||
|
||||
```python
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_healthz(app):
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
||||
response = await client.get("/healthz")
|
||||
|
||||
assert response.status_code == 200
|
||||
```
|
||||
|
||||
## Fixture Design For Async Reliability
|
||||
|
||||
Apply these patterns first:
|
||||
|
||||
1. Keep async fixtures narrow (`function` scope by default).
|
||||
2. Keep one responsibility per fixture when possible.
|
||||
3. Prefer yield fixtures and pair each setup step with teardown in the same fixture.
|
||||
4. Avoid mixing many independent event-loop lifecycles in one fixture chain.
|
||||
|
||||
When using transports that manage internal task groups (for example, streaming clients), avoid patterns that risk splitting lifecycle across different task contexts.
|
||||
|
||||
## Troubleshooting Cancel-Scope Teardown Failures
|
||||
|
||||
When you see errors like `Attempted to exit cancel scope in a different task than it was entered in`, treat it as an async lifecycle-ownership issue first.
|
||||
|
||||
Checklist:
|
||||
|
||||
1. Verify fixture and test loop scopes are compatible and explicit.
|
||||
2. Confirm async resource setup and teardown are owned by the same fixture context.
|
||||
3. Reduce fixture scope (`module` or `session` -> `function`) to test for loop/task ownership drift.
|
||||
4. Ensure the suite uses one primary async plugin model for the failing lane.
|
||||
5. Re-run with focused selection and skip reasons to isolate first failing fixture:
|
||||
- `uv run --group test python -m pytest -m smoke tests/web -q -rs`
|
||||
|
||||
Relevant references:
|
||||
- [Avoiding cancel scope stack corruption](https://anyio.readthedocs.io/en/stable/cancellation.html#avoiding-cancel-scope-stack-corruption)
|
||||
- [pytest-asyncio configuration](https://pytest-asyncio.readthedocs.io/en/stable/reference/configuration.html)
|
||||
- [pytest fixture teardown behavior](https://docs.pytest.org/en/stable/how-to/fixtures.html#teardown-cleanup-aka-fixture-finalization)
|
||||
|
||||
## Commands Worth Remembering
|
||||
|
||||
- `uv run --group test python -m pytest --collect-only -q`
|
||||
- `uv run --group test python -m pytest -m smoke tests/web -q -rs`
|
||||
- `uv run --group test python -m pytest -m integration -q`
|
||||
- `uv run --group test python -m pytest -q`
|
||||
@@ -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,118 @@
|
||||
# Pytest Naming Conventions and Test Organization
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [Good integration practices](https://docs.pytest.org/en/stable/explanation/goodpractices.html)
|
||||
- [Changing standard (Python) test discovery](https://docs.pytest.org/en/stable/example/pythoncollection.html)
|
||||
- [How to use fixtures](https://docs.pytest.org/en/stable/how-to/fixtures.html)
|
||||
- [How to parametrize fixtures and test functions](https://docs.pytest.org/en/stable/how-to/parametrize.html)
|
||||
- [Marker examples](https://docs.pytest.org/en/stable/example/markers.html)
|
||||
|
||||
## Agent Quick Path
|
||||
Use this when creating or reorganizing test modules so naming and hierarchy stay predictable.
|
||||
|
||||
1. Mirror the product domain structure in `tests/` so ownership is obvious.
|
||||
2. Encode broad context in module and class names (`test_*.py`, `Test*`).
|
||||
3. Keep leaf test names short and behavior-focused (`test_*`).
|
||||
4. Use `class Test<Subject>:` only for grouping related scenarios.
|
||||
5. Place fixtures in the nearest `conftest.py` needed by scope.
|
||||
6. Separate expensive tests with markers first, directories second.
|
||||
|
||||
## Naming Conventions
|
||||
|
||||
### File and directory naming
|
||||
- Use lowercase snake_case for test file names: `test_user_service.py`.
|
||||
- Keep directories domain-oriented and stable over time: `tests/orders/`, `tests/billing/`.
|
||||
- Prefer descriptive test names over internal ticket numbers or implementation details.
|
||||
|
||||
### Test function naming
|
||||
- Prefer hierarchical naming: put broad context in folder/module/class, and keep the function name focused on the final assertion.
|
||||
- Keep pytest discovery prefixes intact:
|
||||
- modules start with `test_`
|
||||
- classes start with `Test`
|
||||
- functions start with `test_`
|
||||
- Start with user-visible behavior or contract, not private helper names.
|
||||
|
||||
Recommended pattern:
|
||||
- module: `test_<subject>.py`
|
||||
- class: `Test<Operation>` or `Test<Scenario>`
|
||||
- function: `test_<expected_outcome>`
|
||||
|
||||
Examples:
|
||||
- Flat (still valid): `test_create_order_rejects_invalid_currency`
|
||||
- Class-context: `TestOrder -> TestCreate -> test_rejects_invalid_currency`
|
||||
- Module-context: `test_order.py -> TestCreate -> test_rejects_invalid_currency`
|
||||
- Module + class context can similarly shorten:
|
||||
- `test_token.py -> TestRefresh -> test_rotates_session_id`
|
||||
- `test_user_list.py -> TestListUsers -> test_returns_empty_for_new_tenant`
|
||||
|
||||
### Test class naming
|
||||
- Use `class Test<SubjectOrScenario>:` for scenario grouping and context reduction.
|
||||
- Keep class names noun-focused (`TestOrderService`) rather than action-focused.
|
||||
- Avoid xUnit style setup inheritance when fixtures can express dependencies directly.
|
||||
|
||||
## Hierarchy and Organization Patterns
|
||||
|
||||
Two patterns work well; choose one and apply it consistently.
|
||||
|
||||
### Pattern A: Source-mirror hierarchy (default for product code ownership)
|
||||
|
||||
```text
|
||||
src/
|
||||
app/
|
||||
orders/service.py
|
||||
billing/invoice.py
|
||||
|
||||
tests/
|
||||
app/
|
||||
orders/test_service.py
|
||||
billing/test_invoice.py
|
||||
```
|
||||
|
||||
Use this when teams own modules by source path and want direct test-to-source mapping.
|
||||
|
||||
### Pattern B: Cost-lane hierarchy (default for CI policy clarity)
|
||||
|
||||
```text
|
||||
tests/
|
||||
unit/
|
||||
orders/test_service.py
|
||||
integration/
|
||||
api/test_orders.py
|
||||
persistence/test_order_repository.py
|
||||
smoke/
|
||||
test_health.py
|
||||
```
|
||||
|
||||
Use this when CI gating is based on cost lanes and marker filtering.
|
||||
|
||||
### Hybrid rule (recommended)
|
||||
- Keep a source-mirror tree for local ownership.
|
||||
- Add markers (`unit`, `integration`, `smoke`, `external`) for runtime policy.
|
||||
- Avoid duplicating both trees unless the repository already requires it.
|
||||
|
||||
## Fixture Placement Strategy
|
||||
- Put universal lightweight fixtures in `tests/conftest.py`.
|
||||
- Put domain fixtures in subtree `conftest.py` files close to where they are used.
|
||||
- Keep fixtures composable and explicit; avoid large fixture "god objects".
|
||||
- Use `yield` fixtures for teardown so cleanup is always paired with setup.
|
||||
|
||||
## Parametrize and ID Naming
|
||||
- Use `pytest.mark.parametrize` for behavior matrices instead of copy/paste tests.
|
||||
- Provide explicit `ids=` labels when case names are not obvious.
|
||||
- Keep IDs business-meaningful (`"expired-token"`, `"zero-balance"`) so failures are readable.
|
||||
|
||||
## Collection and Structure Checks
|
||||
Use these checks after introducing new test files or renaming modules:
|
||||
|
||||
- `uv run pytest --collect-only -q`
|
||||
- `uv run pytest -m unit -q`
|
||||
- `uv run pytest -m "not external" -q`
|
||||
|
||||
If collection surprises appear, verify file names, marker registration, and directory placement first.
|
||||
|
||||
## Common Anti-Patterns
|
||||
- Mixed naming styles (`testFoo.py`, `test_foo.py`, `foo_test.py`) in one repository.
|
||||
- Deep fixture chains that hide setup behavior.
|
||||
- Test names that encode implementation details instead of behavior.
|
||||
- Moving slow tests into `unit` directories without marker updates.
|
||||
- Sharing mutable module-level state across tests.
|
||||
@@ -0,0 +1,36 @@
|
||||
# Pytest Documentation Notes
|
||||
|
||||
!!! info "Primary sources"
|
||||
- [Good integration practices](https://docs.pytest.org/en/stable/explanation/goodpractices.html)
|
||||
- [Fixture how-to](https://docs.pytest.org/en/stable/how-to/fixtures.html)
|
||||
- [Marker examples](https://docs.pytest.org/en/stable/example/markers.html)
|
||||
- [Configuration reference](https://docs.pytest.org/en/stable/reference/customize.html)
|
||||
- [Flaky tests](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`
|
||||
- Naming conventions and test hierarchy: `naming-and-organization.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