From a6ae5aa9662a43f0ae0edc2f58efd0fc1cf988f3 Mon Sep 17 00:00:00 2001 From: Hadrien David Date: Thu, 26 Feb 2026 12:55:00 -0500 Subject: [PATCH] feat: add fastsqla-testing agent skill --- skills/fastsqla-testing/SKILL.md | 380 +++++++++++++++++++++++++++++++ 1 file changed, 380 insertions(+) create mode 100644 skills/fastsqla-testing/SKILL.md diff --git a/skills/fastsqla-testing/SKILL.md b/skills/fastsqla-testing/SKILL.md new file mode 100644 index 0000000..82dc936 --- /dev/null +++ b/skills/fastsqla-testing/SKILL.md @@ -0,0 +1,380 @@ +--- +name: fastsqla-testing +description: > + Patterns for testing FastSQLA applications with pytest. Covers the async fixture + chain (tmp SQLite → patched env → engine → table setup → FastAPI app → ASGI client), + critical teardown to prevent mapper leaks between tests, direct DB verification with + a separate session, and SQLModel integration test marks. +--- + +# Testing FastSQLA Applications + +FastSQLA tests run as async integration tests against a per-test temporary SQLite database. The fixture chain is **order-sensitive** — each fixture depends on the previous one. + +## Test Dependencies + +Install the testing dependencies: + +``` +pip install pytest pytest-asyncio httpx asgi-lifespan aiosqlite faker +``` + +- **pytest-asyncio**: async test and fixture support +- **httpx** + **asgi-lifespan**: ASGI client without a running server +- **aiosqlite**: async SQLite driver for isolated per-test databases +- **faker**: generates realistic test data (provides a `faker` pytest fixture) + +## pytest Configuration + +```toml +[tool.pytest.ini_options] +asyncio_mode = "auto" +asyncio_default_fixture_loop_scope = "function" +``` + +- `asyncio_mode = "auto"` — all `async def` tests and fixtures run without `@pytest.mark.asyncio` +- `asyncio_default_fixture_loop_scope = "function"` — each test gets its own event loop + +## Fixture Chain + +The fixtures form a strict dependency chain. **Ordering matters** — creating tables must happen before `Base.prepare()` runs during the app lifespan, and mapper cleanup must run after every test. + +``` +sqlalchemy_url → environ → engine → setup_tear_down → app → client + ↘ session (direct DB verification) +``` + +### 1. `sqlalchemy_url` — Per-test temporary database + +Each test gets its own SQLite file in pytest's `tmp_path`: + +```python +from pytest import fixture + + +@fixture +def sqlalchemy_url(tmp_path): + return f"sqlite+aiosqlite:///{tmp_path}/test.db" +``` + +### 2. `environ` — Patched environment with `clear=True` + +Patches `os.environ` so FastSQLA's lifespan reads the test database URL. **`clear=True` is critical** — it prevents any stray `SQLALCHEMY_*` variables from the host environment from interfering: + +```python +from unittest.mock import patch +from pytest import fixture + + +@fixture +def environ(sqlalchemy_url): + values = {"PYTHONASYNCIODEBUG": "1", "SQLALCHEMY_URL": sqlalchemy_url} + with patch.dict("os.environ", values=values, clear=True): + yield values +``` + +### 3. `engine` — Async engine with teardown + +Creates a standalone engine for direct DB operations (table setup, data seeding, verification). Disposed after the test: + +```python +from pytest import fixture +from sqlalchemy.ext.asyncio import create_async_engine + + +@fixture +async def engine(environ): + engine = create_async_engine(environ["SQLALCHEMY_URL"]) + yield engine + await engine.dispose() +``` + +### 4. `setup_tear_down` — Create tables via raw SQL + +Tables **must** be created with raw SQL before `Base.prepare()` runs (which happens during the ASGI lifespan). This is because `Base` inherits from `DeferredReflection` — it reflects existing tables rather than creating them: + +```python +from pytest import fixture +from sqlalchemy import text + + +@fixture(autouse=True) +async def setup_tear_down(engine): + async with engine.connect() as conn: + await conn.execute( + text(""" + CREATE TABLE hero ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT UNIQUE NOT NULL, + secret_identity TEXT NOT NULL, + age INTEGER NOT NULL + ) + """) + ) +``` + +For tests that need seed data (e.g., pagination), insert rows in the same fixture using Core operations: + +```python +from sqlalchemy import MetaData, Table + + +@fixture(autouse=True) +async def setup_tear_down(engine, faker): + async with engine.connect() as conn: + await conn.execute(text("CREATE TABLE user (...)")) + metadata = MetaData() + user = await conn.run_sync( + lambda sync_conn: Table("user", metadata, autoload_with=sync_conn) + ) + await conn.execute( + user.insert(), + [{"email": faker.email(), "name": faker.name()} for _ in range(42)], + ) + await conn.commit() +``` + +### 5. `app` — FastAPI application with models and routes + +The base `app` fixture creates a FastAPI app with the FastSQLA lifespan. Each test module **overrides** this fixture to register its own ORM models and routes: + +**Base fixture** (in `tests/integration/conftest.py`): + +```python +from pytest import fixture +from fastapi import FastAPI + + +@fixture +def app(environ): + from fastsqla import lifespan + app = FastAPI(lifespan=lifespan) + return app +``` + +**Per-module override** (depends on `setup_tear_down` to ensure tables exist first): + +```python +from pytest import fixture +from sqlalchemy.orm import Mapped, mapped_column + + +@fixture +def app(setup_tear_down, app): + from fastsqla import Base, Item, Session + + class User(Base): + __tablename__ = "user" + id: Mapped[int] = mapped_column(primary_key=True) + email: Mapped[str] = mapped_column(unique=True) + name: Mapped[str] + + @app.post("/users", response_model=Item[UserModel]) + async def create_user(user_in: UserIn, session: Session): + user = User(**user_in.model_dump()) + session.add(user) + await session.flush() + return {"data": user} + + return app +``` + +**Key detail**: The overriding `app` fixture takes `setup_tear_down` as a parameter to enforce that tables are created before `Base.prepare()` runs. + +### 6. `client` — ASGI test client + +Uses `asgi-lifespan` to trigger the app's startup/shutdown events (which runs `Base.prepare()`), then wraps it with httpx for HTTP requests: + +```python +from asgi_lifespan import LifespanManager +from httpx import AsyncClient, ASGITransport +from pytest import fixture + + +@fixture +async def client(app): + async with LifespanManager(app) as manager: + transport = ASGITransport(app=manager.app) + async with AsyncClient(transport=transport, base_url="http://app") as client: + yield client +``` + +## Critical Teardown — Preventing Mapper Leaks + +**This is the most common testing pitfall.** Without this autouse fixture, ORM model definitions from one test leak into subsequent tests, causing `ArgumentError: Class already has a primary mapper defined` or `InvalidRequestError`: + +```python +from pytest import fixture + + +@fixture(autouse=True) +def tear_down(): + from sqlalchemy.orm import clear_mappers + from fastsqla import Base + + yield + + Base.metadata.clear() + clear_mappers() +``` + +**Why both calls?** +- `Base.metadata.clear()` removes all table definitions from the shared `MetaData` +- `clear_mappers()` removes all class-to-table mapper configurations + +This fixture is **synchronous** and **autouse** — it runs automatically after every test without being explicitly requested. + +## Direct Data Verification + +Use a separate `session` fixture bound to the same engine to query the database directly, bypassing the application layer. This verifies that the app actually persisted data: + +```python +from pytest import fixture +from sqlalchemy.ext.asyncio import AsyncSession + + +@fixture +async def session(engine): + async with engine.connect() as conn: + yield AsyncSession(bind=conn) +``` + +Usage in a test: + +```python +from sqlalchemy import text + + +async def test_user_is_persisted(client, session): + payload = {"email": "bob@bob.com", "name": "Bobby"} + res = await client.post("/users", json=payload) + assert res.status_code == 201 + + rows = (await session.execute(text("SELECT * FROM user"))).mappings().all() + assert rows == [{"id": 1, **payload}] +``` + +## SQLModel Integration Tests + +Tests requiring SQLModel use a custom marker and an autouse fixture that skips them when SQLModel is not installed: + +### Register the marker + +```python +def pytest_configure(config): + config.addinivalue_line( + "markers", "require_sqlmodel: skip test when sqlmodel is not installed." + ) +``` + +### Auto-skip fixture + +```python +from pytest import fixture, skip + +try: + import sqlmodel +except ImportError: + is_sqlmodel_installed = False +else: + is_sqlmodel_installed = True + + +@fixture(autouse=True) +def check_sqlmodel(request): + marker = request.node.get_closest_marker("require_sqlmodel") + if marker and not is_sqlmodel_installed: + skip(f"{request.node.nodeid} requires sqlmodel which is not installed.") +``` + +### Mark an entire module + +```python +from pytest import mark + +pytestmark = mark.require_sqlmodel +``` + +### SQLModel model definition + +SQLModel models need `__table_args__ = {"extend_existing": True}` since the table was already created via raw SQL and reflected by `Base.prepare()`: + +```python +from sqlmodel import Field, SQLModel + + +class Hero(SQLModel, table=True): + __table_args__ = {"extend_existing": True} + id: int | None = Field(default=None, primary_key=True) + name: str + secret_identity: str + age: int +``` + +## Integration Test Example + +A complete test that creates a resource via POST and verifies it was persisted: + +```python +from http import HTTPStatus +from pydantic import BaseModel, ConfigDict +from pytest import fixture +from sqlalchemy import select, text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Mapped, mapped_column +from fastapi import HTTPException + + +@fixture(autouse=True) +async def setup_tear_down(engine): + async with engine.connect() as conn: + await conn.execute( + text(""" + CREATE TABLE user ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + email TEXT UNIQUE NOT NULL, + name TEXT NOT NULL + ) + """) + ) + + +@fixture +def app(setup_tear_down, app): + from fastsqla import Base, Item, Session + + class User(Base): + __tablename__ = "user" + id: Mapped[int] = mapped_column(primary_key=True) + email: Mapped[str] = mapped_column(unique=True) + name: Mapped[str] + + class UserIn(BaseModel): + email: str + name: str + + class UserModel(UserIn): + model_config = ConfigDict(from_attributes=True) + id: int + + @app.post("/users", response_model=Item[UserModel], status_code=HTTPStatus.CREATED) + async def create_user(user_in: UserIn, session: Session): + user = User(**user_in.model_dump()) + session.add(user) + try: + await session.flush() + except IntegrityError: + raise HTTPException(status_code=400) + return {"data": user} + + return app + + +async def test_create_and_verify(client, session): + payload = {"email": "bob@bob.com", "name": "Bobby"} + res = await client.post("/users", json=payload) + assert res.status_code == HTTPStatus.CREATED + + all_users = (await session.execute(text("SELECT * FROM user"))).mappings().all() + assert all_users == [{"id": 1, **payload}] +```