From ed6068f398cb3bbc129edb2ce29da2d4b5de06d7 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Fri, 19 Jun 2026 17:22:34 -0500 Subject: [PATCH] pytest improvements --- docs/skills/pytest-scaffolding/SKILL.md | 197 +++++++------- .../references/fastapi-testing.md | 236 +++++++++++++++++ .../references/pytest-docs copy.md | 35 +++ .../references/sqlalchemy-testing.md | 246 ++++++++++++++++++ 4 files changed, 606 insertions(+), 108 deletions(-) create mode 100644 docs/skills/pytest-scaffolding/references/fastapi-testing.md create mode 100644 docs/skills/pytest-scaffolding/references/pytest-docs copy.md create mode 100644 docs/skills/pytest-scaffolding/references/sqlalchemy-testing.md diff --git a/docs/skills/pytest-scaffolding/SKILL.md b/docs/skills/pytest-scaffolding/SKILL.md index f9fce7f..90b36ea 100644 --- a/docs/skills/pytest-scaffolding/SKILL.md +++ b/docs/skills/pytest-scaffolding/SKILL.md @@ -1,136 +1,117 @@ --- name: pytest-scaffolding -description: "Scaffold a maintainable, hierarchical pytest suite for core functionality first, then extend safely. Use when setting up tests, organizing fixtures by dependency, mirroring src structure in tests, or enforcing fast-by-default test runs." -argument-hint: "Target scope (for example: app/services/job, app/ai, or full repo)" +description: "Scaffold a maintainable, hierarchical pytest suite with fast defaults and clear escalation paths for FastAPI and SQLAlchemy tests. Use when creating or reorganizing tests, defining fixture/marker boundaries, or making test strategy progressively discoverable." +argument-hint: "Target scope plus stack details (pure Python, FastAPI, SQLAlchemy sync, SQLAlchemy async, or mixed)" --- # Pytest Scaffolding -Create test scaffolding that is: -- Hierarchical: test layout roughly mirrors source layout. -- Fast by default: most tests run in under a second total for core units. -- Dependency-aware: slow/external dependencies are isolated behind markers and fixture scope. -- Extensible: minimal initial skeleton supports adding detailed tests later without refactors. +Create test scaffolding that stays fast for daily work and scales safely as dependencies increase. -This repository currently uses: -- `uv run pytest` as the canonical test invocation. -- `pyproject.toml` pytest config under `[tool.pytest.ini_options]`. -- strict marker checking (`--strict-markers`). +This skill is optimized for progressive discoverability: +1. Start with the shortest path in this file. +2. Load exactly one deeper reference only when a decision requires it. +3. Continue only as far as needed for the current task. -Load [pytest references](./references/pytest-docs.md) when you need detailed rules. +Repository defaults: +- `uv run pytest` is the canonical invocation. +- pytest settings live in `pyproject.toml` under `[tool.pytest.ini_options]`. +- strict marker checking is expected (`--strict-markers`). -## When To Use -- Bootstrapping tests for a new or existing Python repo. -- Reorganizing tests that have become flat, slow, or difficult to extend. -- Defining fixture boundaries before writing many assertions. -- Creating only the first-layer scaffold for core behavior (not exhaustive coverage yet). +## Discovery Ladder -## Inputs To Collect -1. Target test scope: full repo, package, or module. -2. Dependency profile: pure Python, DB, network/API, filesystem, UI/browser. -3. Runtime expectation: what must be instant vs allowed to be slower. -4. CI policy: which marker groups must block merges. +### Level 0: Scope And Stack Triage (always) +Collect: +1. Target scope (repo, package, module). +2. Stack shape (pure Python, FastAPI, SQLAlchemy sync, SQLAlchemy async, or mixed). +3. Speed target (what must stay instant). +4. CI gate policy (which marker groups block merge). -If these are missing, ask concise clarifying questions before editing. +If any are missing, ask concise clarifying questions before scaffolding. -## Workflow -1. Map source tree to test tree. -2. Classify tests by dependency cost. -3. Create minimal directories and placeholder test modules. -4. Create fixture layers (`tests/conftest.py` plus local `conftest.py` in subtrees only when needed). -5. Register markers and default selection behavior. -6. Run collection and fast path tests. -7. Report gaps and next extension points. +### Level 1: Core pytest scaffold (default) +Use this for all stacks first: +1. Mirror `src/` into `tests/` with one starter file per core module. +2. Classify test intent by cost: + - `unit`: no DB/network/filesystem side effects. + - `integration`: framework, DB, or multi-layer contracts. + - `smoke`: thin critical-path checks. +3. Scaffold each new module with: + - one happy-path test, + - one failure/edge test, + - TODO anchors for deeper assertions. +4. Keep fixtures layered: + - global lightweight fixtures in `tests/conftest.py`, + - domain fixtures in subtree `conftest.py` only when needed. +5. Register markers early: `unit`, `integration`, `smoke`, `slow`, `external`. +6. Validate in order: + - `uv run pytest --collect-only -q` + - `uv run pytest -m unit -q` + - `uv run pytest -q` when dependencies are available. -## Step 1: Map Source To Tests -Create a mirrored structure rooted at `tests/` that follows major source concepts. +Load next reference only if needed: +- Baseline details and rationale: [pytest-docs.md](./references/pytest-docs.md) +- Condensed quick path variant: [pytest-docs copy.md](./references/pytest-docs copy.md) -Example mapping pattern: -- `src/app/services/job.py` -> `tests/app/services/test_job.py` -- `src/app/ai/graphs/transcription.py` -> `tests/app/ai/graphs/test_transcription.py` -- `src/app/api/routes.py` -> `tests/app/api/test_routes.py` +### Level 2: FastAPI branch (only for HTTP/dependency/lifespan concerns) +Escalate here when testing API routes, dependency injection boundaries, or app lifespan behavior. -Rules: -- One initial test module per core source module. -- Prefer `test_.py` naming. -- Keep directory mirrors shallow first; add deeper modules only where behavior is complex. +Apply these defaults: +1. Prefer `TestClient` with sync `def` tests for route behavior. +2. Use `AsyncClient` + `@pytest.mark.anyio` only when test logic must await other async work. +3. Prefer `app.dependency_overrides` over patching internals. +4. Reset dependency overrides in teardown after every test/fixture. +5. For startup/shutdown semantics: + - use `TestClient` as context manager, or + - use `LifespanManager` with async client. -## Step 2: Classify By Dependency Cost -Assign each test module to one initial class: -- `unit`: no DB/network/filesystem side effects; instant execution. -- `integration`: touches DB, HTTP stack, workflow runtime, or external services. -- `smoke`: thin end-to-end confidence checks. +Marker intent in FastAPI-heavy suites: +- `unit`: service logic without HTTP/DB. +- `integration`: route + DI + DB contract checks. +- `smoke`: one request per critical user path. -Decision logic: -- If logic can run with fakes/stubs, make it `unit`. -- If contract with framework/DB is essential, make it `integration`. -- If validating a user-critical path across layers, make it `smoke`. +Reference: [fastapi-testing.md](./references/fastapi-testing.md) -## Step 3: Scaffold Minimal Test Modules -For each target module, scaffold: -- import section -- one happy-path test function -- one error/edge test function -- TODO comments indicating detail expansion points +### Level 3: SQLAlchemy branch (only for DB transaction/session design) +Escalate here when session lifecycle, transaction isolation, or async ORM behavior matters. -Keep assertions minimal but behavior-focused. Avoid large fixtures in module files. +Apply these defaults: +1. Create engine once per test session. +2. Open connection + outer transaction per test. +3. Bind session with `join_transaction_mode="create_savepoint"`. +4. Allow code under test to call `commit()` safely; rollback outer transaction at test end. +5. Keep unit tests DB-free; DB tests belong under `integration`. -## Step 4: Fixture Layering Strategy -Use fixture scopes based on cost: -- `function` scope by default. -- broader scopes (`module`/`session`) only for expensive setup with clear teardown. +Async additions: +- use async fixtures and `@pytest.mark.anyio`. +- set `expire_on_commit=False` for `AsyncSession`. +- avoid implicit lazy IO; use eager loading (`selectinload`) or explicit refresh. -Layer fixtures by directory: -- `tests/conftest.py`: global, lightweight fixtures only (factories, deterministic defaults). -- subtree `conftest.py`: domain-specific fixtures (API client, DB session, AI runtime stubs). +SQLite in-memory with threaded test clients: +- use `StaticPool` when required by thread/connection sharing. -Guidelines: -- Prefer yield fixtures for setup/teardown. -- Keep fixtures atomic (one state-changing responsibility per fixture). -- Avoid autouse except for truly universal behavior. +Reference: [sqlalchemy-testing.md](./references/sqlalchemy-testing.md) -## Step 5: Marker Taxonomy And Config -Ensure marker names are explicit and registered in `pyproject.toml` because strict markers are enabled. +## Branching Logic Summary +- If pure logic can be faked cleanly, keep in `unit`. +- If framework/DB contract is the behavior under test, use `integration`. +- If external service credentials/network is required, gate behind `external`. +- If suite slows down, split by marker before broadening fixture scope. +- If async relationship access raises `MissingGreenlet`, switch to eager loading strategy. -Recommended baseline markers: -- `unit` -- `integration` -- `smoke` -- `slow` -- `external` (requires network/service credentials) - -Default run strategy: -- Fast local path: run only `unit` by default in day-to-day iteration. -- Full validation path: run all markers in CI or pre-release checks. - -## Step 6: Execution And Verification -Run commands in this order: -1. `uv run pytest --collect-only -q` -2. `uv run pytest -m unit -q` -3. `uv run pytest -q` (if dependencies are available) - -Optional targeted runs: -- by node id for one test -- by `-k` expression for focused iteration - -## Step 7: Completion Checks +## Completion Checks A scaffold pass is complete when all are true: -1. Every core source area has at least one corresponding test module. -2. Unit tests run quickly and deterministically. -3. Integration/external tests are isolated by marker and fixture boundaries. -4. No unregistered marker warnings/errors. -5. `tests/` structure is understandable without extra documentation. -6. A clear TODO path exists for deepening assertions later. +1. Core source areas map to clear test modules. +2. Fast path (`-m unit`) is deterministic and quick. +3. Integration and external paths are isolated by fixtures and markers. +4. No unregistered-marker failures occur. +5. Structure is understandable without extra oral context. +6. Clear TODO extension points exist for deeper assertions. -## Branching Scenarios -- If external APIs are required: provide stubs/mocks for unit tests; guard real calls behind `external` marker. -- If DB is required: build a dedicated integration fixture layer and keep unit tests DB-free. -- If tests become slow: split slow tests via marker and widen fixture scope only where safe. -- If naming conflicts appear: keep unique test module names or package test directories explicitly. - -## Output Format -When applying this skill, provide: +## Output Contract +When this skill is applied, return: 1. Proposed test tree diff. 2. Marker and fixture plan. -3. Exact commands for fast path and full path. -4. Risks/open questions before writing detailed assertions. +3. Exact fast-path and full-path commands. +4. Which reference level was loaded and why. +5. Risks or open questions before expanding assertions. diff --git a/docs/skills/pytest-scaffolding/references/fastapi-testing.md b/docs/skills/pytest-scaffolding/references/fastapi-testing.md new file mode 100644 index 0000000..e0f73fc --- /dev/null +++ b/docs/skills/pytest-scaffolding/references/fastapi-testing.md @@ -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/) diff --git a/docs/skills/pytest-scaffolding/references/pytest-docs copy.md b/docs/skills/pytest-scaffolding/references/pytest-docs copy.md new file mode 100644 index 0000000..735e951 --- /dev/null +++ b/docs/skills/pytest-scaffolding/references/pytest-docs copy.md @@ -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` diff --git a/docs/skills/pytest-scaffolding/references/sqlalchemy-testing.md b/docs/skills/pytest-scaffolding/references/sqlalchemy-testing.md new file mode 100644 index 0000000..69014cf --- /dev/null +++ b/docs/skills/pytest-scaffolding/references/sqlalchemy-testing.md @@ -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/)