-
Notifications
You must be signed in to change notification settings - Fork 1
feat: add fastsqla-testing agent skill #32
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Closed
Closed
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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}] | ||
| ``` | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Seems very sqlite specific.