litestar-org/litestar-skills/litestar-testing

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

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(route_handlers=..., **litestar_and_client_kwargs) (or create_test_client for sync tests). Both build a Litestar instance and return a configured test client.

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"

Prefer @pytest.mark.anyio because Litestar's internal concurrency model is built on AnyIO. Do not 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 & Overriding DI Dependencies (Match Your Stack)

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.

• 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]. For synchronous test providers, always specify sync_to_thread=False (or True) to avoid LitestarWarning.
• Option B — Dishka DI (`Provider` + `make_async_container`): Subclass Provider in tests and pass the test provider after LitestarProvider() and production providers in make_async_container(LitestarProvider(), AppProvider(), FakeEmailProvider(mock_email)) — later providers override earlier ones for matching return types. Route handlers using FromDishka[T] must be decorated with @inject (or @inject_websocket, or registered on a DishkaRouter) before calling setup_dishka(container=container, app=app). See Async Testing for full FakeEmailProvider and dishka_client fixture patterns.

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

Session Testing (set_session_data and get_session_data)

Pass the same session_config (ServerSideSessionConfig or CookieBackendConfig) to both middleware=[session_config.middleware] and session_config=session_config on the test client (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, they 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"})
        assert (await client.get("/me")).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, SSE (ServerSentEvent), and ChannelsPlugin Testing

• WebSockets (`websocket_connect`): 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 methods (ws.send_json(...), ws.receive_json(...), ws.close(...)). Server-initiated close causes ws.receive*() to raise litestar.exceptions.WebSocketDisconnect.
• SSE (`ServerSentEvent`): Stream events with async with client.stream("GET", "/sse") as resp: and iterate resp.aiter_lines().
• `ChannelsPlugin`: Configure ChannelsPlugin(backend=MemoryChannelsBackend(), channels=["alerts"], create_ws_route_handlers=True) and enter async with create_async_test_client(plugins=[channels_plugin]) as client:. Publish via channels_plugin.publish(data, channels=["alerts"]) and consume via websocket_connect.

from collections.abc import AsyncGenerator
import pytest
from litestar import get, websocket_listener
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
@websocket_listener("/ws")
async def echo_ws(data: dict[str, str]) -> dict[str, str]:
    return {"echo": data["msg"]}
@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_ws_sse_and_channels() -> None:
    channels = ChannelsPlugin(backend=MemoryChannelsBackend(), channels=["alerts"], create_ws_route_handlers=True)
    async with create_async_test_client(route_handlers=[echo_ws, stream_events], plugins=[channels]) as client:
        with await client.websocket_connect("/ws") as ws:
            ws.send_json({"msg": "hello"})
            assert ws.receive_json() == {"echo": "hello"}
        async with client.stream("GET", "/events") as resp:
            lines = [line async for line in resp.aiter_lines() if line]
            assert lines == ["event: status", "data: ready"]
        with await client.websocket_connect("/alerts") as ws:
            channels.publish({"msg": "fired"}, channels=["alerts"])
            assert ws.receive_json() == {"msg": "fired"}

Unit Testing with RequestFactory & Live Subprocess Clients