diff --git a/docs/module/pytest.md b/docs/module/pytest.md index c52746f..80d7f6a 100644 --- a/docs/module/pytest.md +++ b/docs/module/pytest.md @@ -1,6 +1,6 @@ # Pytest -Testing helpers for FastAPI applications with async client, database sessions, and parallel worker support. +Testing helpers for FastAPI applications: async HTTP client, database sessions, and parallel worker support. ## Installation @@ -14,13 +14,9 @@ Testing helpers for FastAPI applications with async client, database sessions, a pip install "fastapi-toolsets[pytest]" ``` -## Overview +## Async client -The `pytest` module provides utilities for setting up async test clients, managing test database sessions, and supporting parallel test execution with `pytest-xdist`. - -## Creating an async client - -Use [`create_async_client`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_async_client) to get an `httpx.AsyncClient` configured for your FastAPI app: +Use [`create_async_client`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_async_client) to get an `httpx.AsyncClient` bound to your FastAPI app: ```python from fastapi_toolsets.pytest import create_async_client @@ -38,9 +34,20 @@ async def http_client(db_session): yield c ``` -## Database sessions in tests +Any extra keyword arguments are forwarded to `httpx.AsyncClient`, so you can set default headers, authentication, timeouts, and more: -Use [`create_db_session`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_db_session) to create an isolated `AsyncSession` for a test, combined with [`create_worker_database`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_worker_database) to set up a per-worker database: +```python +async with create_async_client( + app=app, + headers={"X-Api-Key": "secret"}, + timeout=10, +) as c: + ... +``` + +## Database sessions + +Use [`create_worker_database`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_worker_database) + [`create_db_session`](../reference/pytest.md#fastapi_toolsets.pytest.utils.create_db_session) to get a fully isolated `AsyncSession` for each test: ```python from fastapi_toolsets.pytest import create_worker_database, create_db_session @@ -61,25 +68,42 @@ async def db_session(worker_db_url): yield session ``` +`create_worker_database` connects without specifying a database (asyncpg falls back to the username), so the target test database does not need to exist beforehand. + !!! info - In this example, the database is reset between each test using the argument `cleanup=True`. + `cleanup=True` truncates all tables between tests via `TRUNCATE … RESTART IDENTITY CASCADE`, which is faster than dropping and recreating tables. + +### Engine and session options + +Pass `engine_kwargs` or `session_kwargs` to forward options to the underlying SQLAlchemy primitives: + +```python +async with create_db_session( + database_url=worker_db_url, + base=Base, + engine_kwargs={"pool_size": 5, "connect_args": {"timeout": 10}}, + session_kwargs={"autoflush": False}, +) as session: + ... +``` + +## Parallel testing with pytest-xdist + +The fixtures above work with `pytest-xdist` out of the box. Each worker gets its own database suffixed with the worker name (e.g. `myapp_gw0`, `myapp_gw1`). Use [`worker_database_url`](../reference/pytest.md#fastapi_toolsets.pytest.utils.worker_database_url) to derive the per-worker URL manually if needed: ```python from fastapi_toolsets.pytest import worker_database_url -url = worker_database_url("postgresql+asyncpg://user:pass@localhost/test_db", default_test_db="test") -# e.g. "postgresql+asyncpg://user:pass@localhost/test_db_gw0" under xdist +url = worker_database_url("postgresql+asyncpg://user:pass@localhost/myapp", default_test_db="test") +# → "postgresql+asyncpg://user:pass@localhost/myapp_gw0" under xdist +# → "postgresql+asyncpg://user:pass@localhost/myapp_test" otherwise ``` -## Parallel testing with pytest-xdist +## Manual table cleanup -The examples above are already compatible with parallel test execution with `pytest-xdist`. - -## Cleaning up tables - -If you want to manually clean up a database you can use [`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables), this will truncate all tables between tests for fast isolation: +[`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables) truncates all tables in a single statement and can be called directly when you need more control: ```python from fastapi_toolsets.db import cleanup_tables diff --git a/src/fastapi_toolsets/pytest/utils.py b/src/fastapi_toolsets/pytest/utils.py index 5242063..4d153d4 100644 --- a/src/fastapi_toolsets/pytest/utils.py +++ b/src/fastapi_toolsets/pytest/utils.py @@ -7,7 +7,7 @@ from typing import Any from httpx import ASGITransport, AsyncClient from sqlalchemy import text -from sqlalchemy.engine import make_url +from sqlalchemy.engine import URL, make_url from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, @@ -63,6 +63,8 @@ def worker_database_url(database_url: str, default_test_db: str) -> str: async def create_worker_database( database_url: str, default_test_db: str = "test_db", + *, + server_url: str | None = None, ) -> AsyncGenerator[str, None]: """Create and drop a per-worker database for pytest-xdist isolation. @@ -74,10 +76,13 @@ async def create_worker_database( name (e.g. ``_gw0``). Otherwise it is suffixed with *default_test_db*. Args: - database_url: Original database connection URL (used as the server - connection and as the base for the worker database name). + database_url: Original database connection URL (used as the base for + the worker database name). default_test_db: Suffix appended to the database name when ``PYTEST_XDIST_WORKER`` is not set. Defaults to ``"test_db"``. + server_url: URL used for server-level DDL (must point to an existing + database on the same server). Defaults to *database_url* with the + database omitted, letting asyncpg fall back to the username. Yields: The worker-specific database URL. @@ -86,7 +91,7 @@ async def create_worker_database( ```python from fastapi_toolsets.pytest import create_worker_database, create_db_session - DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/test_db" + DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/myapp" @pytest.fixture(scope="session") async def worker_db_url(): @@ -107,11 +112,21 @@ async def create_worker_database( worker_db_name = make_url(worker_url).database assert worker_db_name is not None - engine = create_async_engine(database_url, isolation_level="AUTOCOMMIT") + _parsed = make_url(database_url) + _server_url = server_url or URL.create( + drivername=_parsed.drivername, + username=_parsed.username, + password=_parsed.password, + host=_parsed.host, + port=_parsed.port, + query=_parsed.query, + ).render_as_string(hide_password=False) + + engine = create_async_engine(_server_url, isolation_level="AUTOCOMMIT") try: async with engine.connect() as conn: await conn.execute(text(f"DROP DATABASE IF EXISTS {worker_db_name}")) - await create_database(db_name=worker_db_name, server_url=database_url) + await create_database(db_name=worker_db_name, server_url=_server_url) yield worker_url @@ -126,6 +141,7 @@ async def create_async_client( app: Any, base_url: str = "http://test", dependency_overrides: dict[Callable[..., Any], Callable[..., Any]] | None = None, + **kwargs: Any, ) -> AsyncGenerator[AsyncClient, None]: """Create an async httpx client for testing FastAPI applications. @@ -135,6 +151,9 @@ async def create_async_client( dependency_overrides: Optional mapping of original dependencies to their test replacements. Applied via ``app.dependency_overrides`` before yielding and cleaned up after. + **kwargs: Additional keyword arguments forwarded to + :class:`httpx.AsyncClient` (e.g. ``headers``, ``cookies``, + ``auth``, ``timeout``). Yields: An AsyncClient configured for the app. @@ -182,7 +201,9 @@ async def create_async_client( transport = ASGITransport(app=app) try: - async with AsyncClient(transport=transport, base_url=base_url) as client: + async with AsyncClient( + transport=transport, base_url=base_url, **kwargs + ) as client: yield client finally: if dependency_overrides: @@ -199,6 +220,8 @@ async def create_db_session( expire_on_commit: bool = False, drop_tables: bool = True, cleanup: bool = False, + engine_kwargs: dict[str, Any] | None = None, + session_kwargs: dict[str, Any] | None = None, ) -> AsyncGenerator[AsyncSession, None]: """Create a database session for testing. @@ -213,6 +236,12 @@ async def create_db_session( drop_tables: Drop tables after test. Defaults to True. cleanup: Truncate all tables after test using :func:`cleanup_tables`. Defaults to False. + engine_kwargs: Additional keyword arguments forwarded to + :func:`sqlalchemy.ext.asyncio.create_async_engine` + (e.g. ``pool_size``, ``connect_args``). + session_kwargs: Additional keyword arguments forwarded to + :class:`sqlalchemy.ext.asyncio.async_sessionmaker` + (e.g. ``autoflush``, ``class_``). Yields: An AsyncSession ready for database operations. @@ -237,15 +266,17 @@ async def create_db_session( await db_session.commit() ``` """ - engine = create_async_engine(database_url, echo=echo) + engine = create_async_engine(database_url, echo=echo, **(engine_kwargs or {})) try: - # Create tables async with engine.begin() as conn: await conn.run_sync(base.metadata.create_all) session_maker = async_sessionmaker( - engine, expire_on_commit=expire_on_commit, class_=EventSession + engine, + expire_on_commit=expire_on_commit, + class_=EventSession, + **(session_kwargs or {}), ) async with session_maker() as session: yield session diff --git a/tests/test_pytest.py b/tests/test_pytest.py index f89910b..47e4fbb 100644 --- a/tests/test_pytest.py +++ b/tests/test_pytest.py @@ -278,6 +278,21 @@ class TestCreateAsyncClient: # Overrides should be cleaned up assert original_dep not in app.dependency_overrides + @pytest.mark.anyio + async def test_kwargs_forwarded_to_async_client(self): + """Extra kwargs are forwarded to AsyncClient (e.g. default headers).""" + from fastapi import Request + + app = FastAPI() + + @app.get("/headers") + async def headers_endpoint(request: Request): + return {"x-custom": request.headers.get("x-custom")} + + async with create_async_client(app, headers={"X-Custom": "sentinel"}) as client: + response = await client.get("/headers") + assert response.json() == {"x-custom": "sentinel"} + class TestCreateDbSession: """Tests for create_db_session helper.""" @@ -355,6 +370,22 @@ class TestCreateDbSession: result = await session.execute(select(Role)) assert result.all() == [] + @pytest.mark.anyio + async def test_engine_kwargs_forwarded(self): + """engine_kwargs are forwarded to create_async_engine.""" + async with create_db_session( + DATABASE_URL, Base, engine_kwargs={"pool_pre_ping": True} + ) as session: + assert isinstance(session, AsyncSession) + + @pytest.mark.anyio + async def test_session_kwargs_forwarded(self): + """session_kwargs are forwarded to async_sessionmaker.""" + async with create_db_session( + DATABASE_URL, Base, session_kwargs={"autoflush": False} + ) as session: + assert session.autoflush is False + @pytest.mark.anyio async def test_get_transaction_commits_visible_to_separate_session(self): """Data written via get_transaction() is committed and visible to other sessions.""" @@ -535,6 +566,66 @@ class TestCreateWorkerDatabase: assert result.scalar() is None await engine.dispose() + @pytest.mark.anyio + async def test_works_when_database_url_db_does_not_exist( + self, monkeypatch: pytest.MonkeyPatch + ): + """Succeeds even when the database named in database_url does not exist. + + Regression test: the old code connected the DDL engine to database_url + itself, which failed when that database had not been created yet. + """ + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw_noexist") + nonexistent_url = ( + make_url(DATABASE_URL) + .set(database="no_such_db") + .render_as_string(hide_password=False) + ) + expected_db = make_url( + worker_database_url(nonexistent_url, default_test_db="unused") + ).database + + async with create_worker_database(nonexistent_url) as url: + assert make_url(url).database == expected_db + + engine = create_async_engine(DATABASE_URL, isolation_level="AUTOCOMMIT") + async with engine.connect() as conn: + result = await conn.execute( + text("SELECT 1 FROM pg_database WHERE datname = :name"), + {"name": expected_db}, + ) + assert result.scalar() == 1 + await engine.dispose() + + engine = create_async_engine(DATABASE_URL, isolation_level="AUTOCOMMIT") + async with engine.connect() as conn: + result = await conn.execute( + text("SELECT 1 FROM pg_database WHERE datname = :name"), + {"name": expected_db}, + ) + assert result.scalar() is None + await engine.dispose() + + @pytest.mark.anyio + async def test_explicit_server_url(self, monkeypatch: pytest.MonkeyPatch): + """Explicit server_url is used instead of the auto-derived one.""" + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw_explicit_srv") + expected_db = make_url( + worker_database_url(DATABASE_URL, default_test_db="unused") + ).database + + async with create_worker_database(DATABASE_URL, server_url=DATABASE_URL) as url: + assert make_url(url).database == expected_db + + engine = create_async_engine(DATABASE_URL, isolation_level="AUTOCOMMIT") + async with engine.connect() as conn: + result = await conn.execute( + text("SELECT 1 FROM pg_database WHERE datname = :name"), + {"name": expected_db}, + ) + assert result.scalar() == 1 + await engine.dispose() + class _LocalBase(DeclarativeBase): pass