litestar-testing

Litestar-specific testing patterns built on pytest + anyio. Covers:

• TestClient vs AsyncTestClient — constructor options, lifespan, and when to use each
• create_test_client and create_async_test_client — inline app factories for isolated handler/controller tests
• @pytest.mark.anyio setup and pytest-asyncio coexistence rules
• App + lifespan fixtures (on_startup / on_shutdown / lifespan)
• Mocking Guards and overriding DI dependencies (native Provide and Dishka Provider)
• Session testing with set_session_data and get_session_data
• WebSocket testing with websocket_connect and WebSocketTestSession
• Unit-testing Guards and dependencies with RequestFactory
• Live server subprocess testing with subprocess_async_client and subprocess_sync_client
• Integration with pytest-databases (see ../pytest-databases/SKILL.md) and polyfactory (see ../polyfactory/SKILL.md)
• Autowire discovery and cache isolation (see ../litestar-autowire/references/testing.md)

For JS-side testing (Vitest, Testing Library, Playwright), use the upstream Vitest docs and Litestar's own JS examples. Out of scope here.

Code Style Rules

• PEP 604 unions: T | None, never Optional[T]
• Test modules MAY use from __future__ import annotations — they are pure consumer code.
• Function-based tests (not class-based)
• One assertion concern per test
• Async Litestar tests use @pytest.mark.anyio by default; do not mix AnyIO and pytest-asyncio auto modes.
• Prefer AsyncTestClient for new code; TestClient only for legacy / sync-only flows.
• Always enter test clients as context managers (async with AsyncTestClient(...) / with TestClient(...)) — implicit lifespan startup without a context manager emits DeprecationWarning and is removed in Litestar 3.0.

Quick Reference

TestClient vs AsyncTestClient

Client	Base Class	Context Manager	Session Helpers	WebSocket Connect
AsyncTestClient	httpx.AsyncClient, BaseTestClient	async with AsyncTestClient(app=app) as client:	await client.set_session_data(...) / await client.get_session_data()	with await client.websocket_connect("/ws") as ws:
TestClient	httpx.Client, BaseTestClient	with TestClient(app=app) as client:	client.set_session_data(...) / client.get_session_data()	with client.websocket_connect("/ws") as ws:

Both clients accept the same constructor parameters:

from litestar.testing import AsyncTestClient, TestClient

AsyncTestClient(
    app=app,
    base_url="http://testserver.local",  # Must contain a dot ('.') or UserWarning is emitted
    raise_server_exceptions=True,  # False wraps unhandled server exceptions as HTTP 500 responses
    root_path="",
    backend="asyncio",  # "asyncio" | "trio"
    backend_options=None,
    session_config=None,  # Required for get_session_data() / set_session_data()
    timeout=None,
    cookies=None,
)

Both set follow_redirects=True and headers={"user-agent": "testclient"} by default. When testing template responses, TestClientTransport populates response.template and response.context on the returned httpx.Response.

from litestar.testing import AsyncTestClient


async def test_index(async_client: AsyncTestClient) -> None:
    resp = await async_client.get("/")
    assert resp.status_code == 200

from litestar.testing import TestClient


def test_index(client: TestClient) -> None:
    resp = client.get("/")
    assert resp.status_code == 200

create_test_client and create_async_test_client

For isolated route handler, controller, or router tests without bootstrapping the full application factory, use create_async_test_client (or create_test_client for sync tests). Both build a Litestar instance and return a configured test client.

• First positional argument route_handlers: a single handler, Controller subclass, Router, a sequence of them, or None.
• Keyword-only arguments forward Litestar(...) settings (dependencies, guards, middleware, plugins, stores, state, on_startup, on_shutdown, lifespan, exception_handlers, dto, return_dto, template_config, static_files_config, cors_config, csrf_config, compression_config, allowed_hosts, response_cache_config, logging_config, openapi_config, opt, parameters, path, security, tags, signature_namespace, signature_types, type_encoders, request_class, response_class, websocket_class, response_cookies, response_headers, before_request, after_request, after_response, before_send, after_exception, on_app_init, listeners, event_emitter_backend, cache_control, etag, include_in_schema, multipart_form_part_limit=1000, pdb_on_exception, experimental_features, debug=True) plus client parameters (backend="asyncio", backend_options=None, base_url="http://testserver.local", raise_server_exceptions=True, root_path="", session_config=None, timeout=None). Note: request_max_body_size, type_decoders, and debugger_module are only accepted on Litestar(...) directly — construct Litestar(...) and wrap it in AsyncTestClient(app=app) / TestClient(app=app) when testing those options.

import pytest
from litestar import get
from litestar.testing import create_async_test_client


@get("/health")
async def health_check() -> dict[str, str]:
    return {"status": "ok"}


@pytest.mark.anyio
async def test_health_check() -> None:
    async with create_async_test_client(route_handlers=health_check) as client:
        resp = await client.get("/health")
        assert resp.status_code == 200
        assert resp.json() == {"status": "ok"}

anyio vs pytest-asyncio Setup

# conftest.py
import pytest


@pytest.fixture(scope="session")
def anyio_backend() -> str:
    return "asyncio"

# tests/test_x.py
import pytest


@pytest.mark.anyio
async def test_something() -> None: ...

Prefer @pytest.mark.anyio because Litestar's internal concurrency model is built on AnyIO. If a project already standardizes on pytest-asyncio, keep its mode explicit and do not enable both AnyIO and pytest-asyncio automatic modes or mix @pytest.mark.anyio and @pytest.mark.asyncio in the same test module.

App + Lifespan Fixture

# conftest.py
from collections.abc import AsyncGenerator
import pytest
from litestar import Litestar
from litestar.testing import AsyncTestClient

from app import create_app


@pytest.fixture
async def app() -> Litestar:
    return create_app()


@pytest.fixture
async def async_client(app: Litestar) -> AsyncGenerator[AsyncTestClient[Litestar], None]:
    async with AsyncTestClient(app=app) as client:
        yield client

async with AsyncTestClient(...) enters LifeSpanHandler, firing on_startup / on_shutdown hooks and plugin lifespan context managers (Vite, SAQ, SQLAlchemy, Channels, etc.). Without the context manager, implicit startup emits a DeprecationWarning and shutdown hooks never run.

Mocking Guards

Guards are functions of (connection: ASGIConnection, route_handler: BaseRouteHandler) -> None | Awaitable[None]. Test the real guard with fake identity or authorization providers, or unit-test the guard directly with RequestFactory. Build a fresh app with replacement providers; Litestar has no mutable app.dependency_overrides registry.

Gotcha (exclude_from_auth vs guards): Setting exclude_from_auth=True (or opt={"exclude_from_auth": True}) only bypasses AbstractAuthenticationMiddleware, leaving "user" and "auth" unset in connection.scope. Any guard attached at the Litestar, Router, or Controller layer still executes, and reading connection.user or connection.auth directly raises ImproperlyConfiguredException (500). App/Controller-wide guards must check route_handler.opt.get("exclude_from_auth") or connection.scope.get("user").

from collections.abc import AsyncGenerator
import pytest
from litestar.di import Provide
from litestar.testing import AsyncTestClient


@pytest.fixture
async def async_client() -> AsyncGenerator[AsyncTestClient, None]:
    fake_users_service = FakeUserService()

    async def provide_fake_users_service() -> UserService:
        return fake_users_service

    test_app = create_app(
        dependencies={
            "users_service": Provide(provide_fake_users_service),
        },
    )
    async with AsyncTestClient(app=test_app) as client:
        yield client

Overriding DI Dependencies (Match Your Stack)

Option A: Litestar Built-in DI (Provide + NamedDependency)

Pass replacement Provide entries to create_app(dependencies=...) or create_async_test_client(..., dependencies=...). Handler parameters resolved by name must use NamedDependency[T] (from litestar.di). For synchronous test providers, always specify sync_to_thread=False (or True) to avoid LitestarWarning.

from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock

import pytest
from litestar import post
from litestar.di import NamedDependency, Provide
from litestar.testing import AsyncTestClient, create_async_test_client


@post("/notify")
async def notify_handler(email_service: NamedDependency[AsyncMock]) -> dict[str, bool]:
    await email_service.send("user@example.com")
    return {"sent": True}


@pytest.fixture
async def client_with_email() -> AsyncGenerator[tuple[AsyncTestClient, AsyncMock], None]:
    fake_email = AsyncMock()
    async with create_async_test_client(
        route_handlers=[notify_handler],
        dependencies={
            "email_service": Provide(lambda: fake_email, sync_to_thread=False),
        },
    ) as client:
        yield client, fake_email

Option B: Dishka DI (Provider + make_async_container)

When the project uses Dishka, subclass Provider in tests and pass the test provider after LitestarProvider() and production providers in make_async_container(LitestarProvider(), AppProvider(), TestProvider()) — later providers override earlier ones for matching return types. Route handlers using FromDishka[T] must be decorated with @inject (or @inject_websocket for WebSocket handlers, or registered on a DishkaRouter) before calling setup_dishka(container=container, app=app).

from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock

import pytest
from dishka import Provider, Scope, make_async_container, provide
from dishka.integrations.litestar import (
    FromDishka,
    LitestarProvider,
    inject,
    setup_dishka,
)
from litestar import Litestar, post
from litestar.testing import AsyncTestClient


class FakeEmailProvider(Provider):
    scope = Scope.REQUEST

    def __init__(self, mock_email: AsyncMock) -> None:
        super().__init__()
        self._mock_email = mock_email

    @provide
    def email_service(self) -> EmailService:
        return self._mock_email


@post("/notify")
@inject
async def notify_dishka(email_service: FromDishka[EmailService]) -> dict[str, bool]:
    await email_service.send("user@example.com")
    return {"sent": True}


@pytest.fixture
async def dishka_client() -> AsyncGenerator[tuple[AsyncTestClient[Litestar], AsyncMock], None]:
    mock_email = AsyncMock(spec=EmailService)
    container = make_async_container(
        LitestarProvider(),
        AppProvider(),
        FakeEmailProvider(mock_email),
    )
    app = Litestar(route_handlers=[notify_dishka])
    setup_dishka(container=container, app=app)
    async with AsyncTestClient(app=app) as client:
        yield client, mock_email
    await container.close()

• DishkaRouter controller monkey-patching isolation in tests: DishkaRouter.register() (used directly or via AutowirePlugin(AutowireConfig(..., integrations=["dishka"]))) mutates ControllerSubclass.get_route_handlers in-place on the controller class (ControllerSubclass.__dict__["get_route_handlers"] = _inject_route_handlers(...)). If an earlier test in the pytest session registers a Controller subclass on a DishkaRouter (e.g., via create_app()), any subsequent isolated unit test mounting that same Controller class on a plain Litestar or create_async_test_client without setup_dishka(...) fails at request time with AttributeError: 'State' object has no attribute 'dishka_container' (or KeyError: 'dishka_container'). Isolate by snapshotting get_route_handlers with monkeypatch.setattr(MyController, "get_route_handlers", MyController.get_route_handlers) before DishkaRouter registration, or wire a test Dishka container with setup_dishka(container=container, app=app).
• Runtime type hints with from __future__ import annotations: @inject, @inject_websocket, and DishkaRouter.register call typing.get_type_hints(func) at decoration/registration time. Keep FromDishka[...] and injected service types at module top-level (func.__globals__), never inside if TYPE_CHECKING: or local test-function scopes when from __future__ import annotations is active.

Autowire Cache Isolation (clear_autowire_cache())

When testing applications that use litestar-autowire (AutowirePlugin / AutowireConfig), discovery caches imported domain modules and optional-module misses in process-local caches. Use an autouse fixture that calls clear_autowire_cache() before and after each test (and evicts temporary module names from sys.modules when generating packages under tmp_path):

from collections.abc import Iterator
import pytest
from litestar_autowire import clear_autowire_cache


@pytest.fixture(autouse=True)
def isolate_autowire_cache() -> Iterator[None]:
    clear_autowire_cache()
    yield
    clear_autowire_cache()

Session Testing (set_session_data and get_session_data)

To seed or inspect session data in tests, pass the same session_config (ServerSideSessionConfig from litestar.middleware.session.server_side or CookieBackendConfig from litestar.middleware.session.client_side) to both the app's middleware=[session_config.middleware] and the test client's session_config=session_config. Omitting session_config on the client raises ImproperlyConfiguredException.

• On AsyncTestClient: await client.set_session_data(data) and await client.get_session_data() are async.
• On TestClient: client.set_session_data(data) and client.get_session_data() are sync.

import pytest
from litestar import Request, get, post
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.testing import create_async_test_client

session_config = ServerSideSessionConfig()


@get("/me")
async def read_session(request: Request) -> dict[str, str]:
    return dict(request.session)


@post("/login")
async def write_session(request: Request, data: dict[str, str]) -> None:
    request.session["user_id"] = data["user_id"]


@pytest.mark.anyio
async def test_session_roundtrip() -> None:
    async with create_async_test_client(
        route_handlers=[read_session, write_session],
        middleware=[session_config.middleware],
        session_config=session_config,
    ) as client:
        await client.set_session_data({"user_id": "u-123"})
        resp = await client.get("/me")
        assert resp.json() == {"user_id": "u-123"}

        await client.post("/login", json={"user_id": "u-456"})
        assert await client.get_session_data() == {"user_id": "u-456"}

WebSocket Testing (websocket_connect and WebSocketTestSession)

Both TestClient and AsyncTestClient expose websocket_connect(url, subprotocols=None, params=None, headers=None, cookies=None, auth=..., follow_redirects=..., timeout=..., extensions=None) -> WebSocketTestSession.

• Critical distinction: On AsyncTestClient, websocket_connect is async def (must be awaited), while the returned WebSocketTestSession is always a synchronous context manager (with await client.websocket_connect(...) as ws:) with synchronous send/receive methods:
  • Send: ws.send(data, mode="text", encoding="utf-8"), ws.send_text(data, encoding="utf-8"), ws.send_bytes(data, encoding="utf-8"), ws.send_json(data, mode="text"), ws.send_msgpack(data)
  • Receive: ws.receive(block=True, timeout=None), ws.receive_text(block=True, timeout=None), ws.receive_bytes(block=True, timeout=None), ws.receive_json(mode="text", block=True, timeout=None), ws.receive_msgpack(block=True, timeout=None)
  • Close / Disconnect: ws.close(code=WS_1000_NORMAL_CLOSURE); server-initiated close causes ws.receive*() to raise litestar.exceptions.WebSocketDisconnect (inspect .code and .detail).
  • Metadata attributes: ws.accepted_subprotocol, ws.extra_headers, ws.scope.

import pytest
from litestar import websocket_listener
from litestar.testing import create_async_test_client


@websocket_listener("/ws")
async def echo_ws(data: dict[str, str]) -> dict[str, str]:
    return {"echo": data["msg"]}


@pytest.mark.anyio
async def test_websocket_echo() -> None:
    async with create_async_test_client(route_handlers=[echo_ws]) as client:
        with await client.websocket_connect("/ws") as ws:
            ws.send_json({"msg": "hello"})
            assert ws.receive_json() == {"echo": "hello"}

SSE (ServerSentEvent) and ChannelsPlugin Testing

• SSE (ServerSentEvent): Stream events with async with client.stream("GET", "/sse") as resp: and iterate resp.aiter_lines() (or resp.iter_lines() on TestClient).
• ChannelsPlugin: Configure ChannelsPlugin(backend=MemoryChannelsBackend(), channels=[...], create_ws_route_handlers=True) and enter async with AsyncTestClient(app=app) as client: so the plugin lifespan starts the backend. Publish via channels_plugin.publish(data, channels=[...]) (or await channels_plugin.wait_published(data, channels=[...])) and consume via websocket_connect or async with channels_plugin.start_subscription([...]) as subscriber:.

from collections.abc import AsyncGenerator
import pytest
from litestar import get
from litestar.channels import ChannelsPlugin
from litestar.channels.backends.memory import MemoryChannelsBackend
from litestar.response.sse import ServerSentEvent, ServerSentEventMessage
from litestar.testing import create_async_test_client


@get("/events")
async def stream_events() -> ServerSentEvent:
    async def generator() -> AsyncGenerator[ServerSentEventMessage, None]:
        yield ServerSentEventMessage(data="ready", event="status")

    return ServerSentEvent(generator())


@pytest.mark.anyio
async def test_sse_stream() -> None:
    async with create_async_test_client(route_handlers=[stream_events]) as client:
        async with client.stream("GET", "/events") as resp:
            assert resp.status_code == 200
            lines = [line async for line in resp.aiter_lines() if line]
            assert lines == ["event: status", "data: ready"]


@pytest.mark.anyio
async def test_channels_websocket() -> None:
    channels_plugin = ChannelsPlugin(
        backend=MemoryChannelsBackend(),
        channels=["alerts"],
        create_ws_route_handlers=True,
    )
    async with create_async_test_client(plugins=[channels_plugin]) as client:
        with await client.websocket_connect("/alerts") as ws:
            channels_plugin.publish({"msg": "fired"}, channels=["alerts"])
            assert ws.receive_json() == {"msg": "fired"}

Unit Testing with RequestFactory

Use RequestFactory to construct real litestar.connection.Request instances for unit-testing Guards, dependencies, or request helpers without spinning up an ASGI transport.

• Constructor: RequestFactory(app=None, server="test.org", port=3000, root_path="", scheme="http", handler_kwargs=None)
• Methods: .get(), .post(), .put(), .patch(), .delete() accepting path="/", headers=None, cookies=None, session=None, user=None, auth=None, query_params=None, state=None, path_params=None, http_version="1.1", route_handler=None, plus data=None and request_media_type=RequestEncodingType.JSON on .post() / .put() / .patch().
• Synthetic Request scope rule: In Litestar, connection.app / request.app reads request.scope["litestar_app"] (not request.scope["app"]). RequestFactory(app=app) automatically sets scope["litestar_app"] = app (or a default Litestar instance when app=None). If constructing a synthetic Request(scope={...}) or ASGIConnection(scope={...}) directly without RequestFactory, always include "litestar_app": app in the scope dict or accessing .app raises KeyError: 'litestar_app'.

import pytest
from litestar.connection import ASGIConnection
from litestar.exceptions import PermissionDeniedException
from litestar.handlers import BaseRouteHandler
from litestar.testing import RequestFactory


def requires_admin(connection: ASGIConnection, _: BaseRouteHandler) -> None:
    if not connection.user or connection.user.get("role") != "admin":
        raise PermissionDeniedException("Admin required")


def test_requires_admin_guard() -> None:
    factory = RequestFactory()
    denied_req = factory.get("/admin", user={"role": "viewer"})
    with pytest.raises(PermissionDeniedException):
        requires_admin(denied_req, denied_req.route_handler)

    allowed_req = factory.get("/admin", user={"role": "admin"})
    requires_admin(allowed_req, allowed_req.route_handler)

Live Subprocess Clients (subprocess_async_client / subprocess_sync_client)

For end-to-end smoke tests that verify litestar --app <app> run CLI startup and real socket binding, use subprocess_async_client(workdir, app, capture_output=True) (yields httpx.AsyncClient) or subprocess_sync_client(workdir, app, capture_output=True) (yields httpx.Client).

from pathlib import Path
import pytest
from litestar.testing import subprocess_async_client


@pytest.mark.anyio
async def test_live_subprocess_server(tmp_path: Path) -> None:
    app_file = tmp_path / "main.py"
    app_file.write_text(
        "from litestar import Litestar, get\n"
        "@get('/ping')\n"
        "def ping() -> dict[str, str]: return {'pong': 'ok'}\n"
        "app = Litestar([ping])\n",
        encoding="utf-8",
    )
    async with subprocess_async_client(workdir=tmp_path, app="main:app") as client:
        resp = await client.get("/ping")
        assert resp.status_code == 200

Integration with pytest-databases

Combine pytest-databases fixtures with the app fixture. See ../pytest-databases/SKILL.md.

# conftest.py
pytest_plugins = ["pytest_databases.docker.postgres"]


@pytest.fixture
async def app(postgres_service) -> Litestar:
    from app import create_app
    from app.config import Settings

    settings = Settings(
        database_url=f"postgresql+asyncpg://{postgres_service.user}:{postgres_service.password}@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
    )
    return create_app(settings=settings)

The postgres_service fixture starts a Postgres container. Inject its connection details into the app config.

Request Bodies

Body Type	Pass via
JSON	client.post("/", json={...})
Form	client.post("/", data={...})
Multipart (file upload)	client.post("/", files={"file": ("name.txt", b"content", "text/plain")})
Raw bytes	client.post("/", content=b"...")
Custom content-type	client.post("/", content=b"...", headers={"Content-Type": "..."})

async def test_create_user(async_client: AsyncTestClient) -> None:
    resp = await async_client.post(
        "/api/users",
        json={"name": "Alice", "email": "alice@example.com"},
    )
    assert resp.status_code == 201
    body = resp.json()
    assert body["name"] == "Alice"

Headers, Cookies, Auth

# Header
resp = await async_client.get("/", headers={"Authorization": "Bearer token"})

# Cookie
async_client.cookies.set("session", "abc123")
resp = await async_client.get("/")

# Per-request cookies
resp = await async_client.get("/", cookies={"session": "abc123"})

HTMX Requests

async def test_htmx_partial(async_client: AsyncTestClient) -> None:
    resp = await async_client.get(
        "/items/list",
        headers={"HX-Request": "true", "HX-Target": "#item-list"},
    )
    assert resp.status_code == 200
    assert "<ul" in resp.text

Response Assertions

# Status
assert resp.status_code == 200

# Body
assert resp.json() == {"id": 1, "name": "Alice"}

# Headers
assert resp.headers["content-type"].startswith("application/json")
assert "HX-Trigger" in resp.headers

# Cookies (set by server)
assert "session" in resp.cookies

Parametrize

import pytest


@pytest.mark.parametrize(
    "payload, expected_status",
    [
        ({"name": "valid", "email": "a@b.co"}, 201),
        ({"name": "", "email": "a@b.co"}, 400),
        ({"name": "valid", "email": "not-email"}, 400),
    ],
)
@pytest.mark.anyio
async def test_create_user_validation(async_client, payload, expected_status):
    resp = await async_client.post("/api/users", json=payload)
    assert resp.status_code == expected_status

Coverage

pytest --cov=src --cov-report=html
pytest --cov=src --cov-fail-under=90

<workflow>

Workflow

Step 1: Set Up anyio Backend

Add anyio_backend fixture to conftest.py returning "asyncio". Mark async tests with @pytest.mark.anyio.

Step 2: App + Client Fixtures

Build an app fixture that returns a fresh Litestar instance per test (or per session if no shared state). Build an async_client fixture that wraps the app in AsyncTestClient via async with (or use create_async_test_client for isolated handler/controller tests).

Step 3: Add Database Fixtures

If the app talks to a DB, layer in pytest-databases (postgres_service, mysql_service, etc.) and pass connection details into the app config. See ../pytest-databases/SKILL.md.

Step 4: Override DI for Externals

Mock EmailService, HTTP clients, and other side-effect-laden dependencies by constructing a fresh app or test client with replacement Provide instances (or a Dishka override Provider passed to make_async_container). Avoid real network calls in tests.

Step 5: Test Guards and Sessions

Build a fresh app with fake identity providers, use RequestFactory to unit-test Guards directly, and pass session_config to the test client when seeding or asserting session state via set_session_data / get_session_data.

Step 6: Write Tests

• One assertion concern per test.
• Use @pytest.mark.parametrize for input variations.
• Use AsyncTestClient for new code (with await client.websocket_connect(...) as ws: for WebSockets).
• Include HTMX / Inertia headers when testing those paths.

Step 7: Verify Coverage

pytest --cov=src --cov-fail-under=90. Cover handlers, services, Guards, and at least one happy-path + one error-path per route.

</workflow> <guardrails>

Guardrails

• Use @pytest.mark.anyio for new Litestar async tests — keep pytest-asyncio only when a project already uses it explicitly, and never mix auto modes.
• Always enter AsyncTestClient / TestClient as a context manager — without async with / with, plugin lifespans (Vite, SAQ, SQLAlchemy) do not run cleanly and Litestar emits a DeprecationWarning for implicit startup (removed in 3.0).
• Prefer AsyncTestClient over TestClient for new tests — the async client matches Litestar's runtime model.
• Pass session_config to the client when using get_session_data / set_session_data — omitting session_config raises ImproperlyConfiguredException.
• Await websocket_connect on AsyncTestClient, then enter WebSocketTestSession with sync with — with await async_client.websocket_connect("/ws") as ws:; WebSocketTestSession methods (send_json, receive_json, etc.) are synchronous.
• Mock side effects via DI override, not patching — keeps tests isolated from import order and global state.
• Build a fresh app for dependency replacements — Litestar has no mutable dependency-override registry, and shared app mutation races under parallel tests.
• Use pytest-databases for real DB testing — never mock SQLAlchemy / sqlspec internals; assertions on mocked queries don't catch real bugs.
• Function-based tests — no class-based test containers unless absolutely needed for shared setup.
• One assertion concern per test — failures should pinpoint a single behavior.
• Don't share state between tests — fresh app + fresh DB per test (or per module with explicit cleanup).
• Test the HTMX path with HX-Request: true — handlers that branch on request.htmx need both branches covered.
• Mock email via backend="memory" / InMemoryBackend — see ../litestar-email/SKILL.md.

</guardrails> <validation>

Validation Checkpoint

Before delivering Litestar tests, verify:

• anyio_backend fixture returns "asyncio"
• Async tests use @pytest.mark.anyio (no mixed AnyIO / pytest-asyncio auto modes)
• AsyncTestClient / create_async_test_client is wrapped in async with (lifespan fires)
• session_config is passed to the test client whenever get_session_data / set_session_data is called
• WebSocket tests on AsyncTestClient use with await client.websocket_connect(...) as ws:
• DI dependencies (email, HTTP clients) are overridden via Provide or Dishka Provider, not patched
• DB-dependent tests use pytest-databases fixtures
• Guards either pass real auth, use a fresh app with fake identity providers, or are unit-tested with RequestFactory
• One assertion concern per test; parametrize for input variations
• HTMX-targeted handlers have tests with HX-Request: true
• Coverage gate (--cov-fail-under) is set in CI

</validation> <example>

Example

Task: Test an account creation endpoint that hits Postgres, sends a welcome email via SAQ, and is guarded by an auth check.

# conftest.py
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, Mock

import pytest
from litestar import Litestar
from litestar.di import Provide
from litestar.testing import AsyncTestClient

pytest_plugins = ["pytest_databases.docker.postgres"]


@pytest.fixture
def anyio_backend() -> str:
    return "asyncio"


@pytest.fixture
async def app(postgres_service) -> tuple[Litestar, AsyncMock]:
    from app import create_app
    from app.config import Settings

    fake_queue = AsyncMock()
    fake_task_queues = Mock()
    fake_task_queues.get.return_value = fake_queue

    async def provide_fake_task_queues() -> Mock:
        return fake_task_queues

    settings = Settings(
        database_url=(
            f"postgresql+asyncpg://{postgres_service.user}:{postgres_service.password}"
            f"@{postgres_service.host}:{postgres_service.port}/{postgres_service.database}"
        ),
    )
    return (
        create_app(
            settings=settings,
            dependencies={
                "task_queues": Provide(provide_fake_task_queues),
            },
        ),
        fake_queue,
    )


@pytest.fixture
async def async_client(
    app: tuple[Litestar, AsyncMock],
) -> AsyncGenerator[tuple[AsyncTestClient, AsyncMock], None]:
    test_app, fake_queue = app
    async with AsyncTestClient(app=test_app) as client:
        yield client, fake_queue

# tests/test_accounts.py
import pytest


@pytest.mark.anyio
async def test_create_account_persists_and_queues_email(async_client):
    client, fake_queue = async_client

    resp = await client.post(
        "/api/accounts",
        json={"email": "alice@example.com", "name": "Alice"},
    )

    assert resp.status_code == 201
    body = resp.json()
    assert body["email"] == "alice@example.com"
    fake_queue.enqueue.assert_awaited_once()
    args, kwargs = fake_queue.enqueue.await_args
    assert args[0] == "send_welcome_email"
    assert kwargs["email"] == "alice@example.com"


@pytest.mark.anyio
@pytest.mark.parametrize(
    "payload, expected_status",
    [
        ({"email": "valid@example.com", "name": "Valid"}, 201),
        ({"email": "", "name": "Valid"}, 400),
        ({"email": "valid@example.com", "name": ""}, 400),
    ],
)
async def test_create_account_validation(async_client, payload, expected_status):
    client, _ = async_client
    resp = await client.post("/api/accounts", json=payload)
    assert resp.status_code == expected_status

</example>

---

References Index

• Async Testing — anyio setup, async fixtures, create_async_test_client, session/WebSocket/RequestFactory/subprocess helpers, and common pitfalls.
• litestar — Litestar fundamentals.
• pytest-databases — Container-based DB fixtures.
• polyfactory — Model, dataclass, and msgspec test data factories.
• litestar-email — backend="memory" and InMemoryBackend for email tests.
• litestar-saq — Mocking task queues.

Official References

• https://docs.litestar.dev/2/usage/testing.html
• https://github.com/litestar-org/litestar/tree/v2.25.0
• https://docs.pytest.org/en/stable/
• https://anyio.readthedocs.io/en/stable/testing.html

Shared Styleguide Baseline

• General Principles
• Testing
• Python
• Litestar