Testing

Testing a Litestar application is made simple by the testing utilities provided out of the box. Based on httpx, they come with a familiar interface and integrate seamlessly into synchronous or asynchronous tests.

Test Clients

Litestar provides 2 test clients:

• AsyncTestClient: An asynchronous test client to be used in asynchronous environments. It runs the application and client on an externally managed event loop. Ideal for testing asynchronous behaviour, or when dealing with asynchronous resources

• TestClient: A synchronous test client. It runs the application in a newly created event loop within a separate thread. Ideal when no async behaviour needs to be tested, and no external event loop is provided by the testing library

Let’s say we have a very simple app with a health check endpoint:

my_app/main.py

from litestar import Litestar, MediaType, get


@get(path="/health-check", media_type=MediaType.TEXT)
def health_check() -> str:
    return "healthy"


app = Litestar(route_handlers=[health_check])

We would then test it using the test client like so:

Sync

tests/test_health_check.py

from litestar.status_codes import HTTP_200_OK
from litestar.testing import TestClient

from my_app.main import app

app.debug = True


def test_health_check():
    with TestClient(app=app) as client:
        response = client.get("/health-check")
        assert response.status_code == HTTP_200_OK
        assert response.text == "healthy"

Async

tests/test_health_check.py

from litestar.status_codes import HTTP_200_OK
from litestar.testing import AsyncTestClient

from my_app.main import app

app.debug = True


async def test_health_check():
    async with AsyncTestClient(app=app) as client:
        response = await client.get("/health-check")
        assert response.status_code == HTTP_200_OK
        assert response.text == "healthy"

Since we would probably need to use the client in multiple places, it’s better to make it into a pytest fixture:

Sync

Python 3.8+

tests/conftest.py

from typing import TYPE_CHECKING, Iterator

import pytest

from litestar.testing import TestClient

from my_app.main import app

if TYPE_CHECKING:
    from litestar import Litestar

app.debug = True


@pytest.fixture(scope="function")
def test_client() -> Iterator[TestClient[Litestar]]:
    with TestClient(app=app) as client:
        yield client

Python 3.9+

tests/conftest.py

from typing import TYPE_CHECKING
from collections.abc import Iterator

import pytest

from litestar.testing import TestClient

from my_app.main import app

if TYPE_CHECKING:
    from litestar import Litestar

app.debug = True


@pytest.fixture(scope="function")
def test_client() -> Iterator[TestClient[Litestar]]:
    with TestClient(app=app) as client:
        yield client

Async

Python 3.8+

tests/conftest.py

from typing import TYPE_CHECKING, AsyncIterator

import pytest

from litestar.testing import AsyncTestClient

from my_app.main import app

if TYPE_CHECKING:
    from litestar import Litestar

app.debug = True


@pytest.fixture(scope="function")
async def test_client() -> AsyncIterator[AsyncTestClient[Litestar]]:
    async with AsyncTestClient(app=app) as client:
        yield client

Python 3.9+

tests/conftest.py

from typing import TYPE_CHECKING
from collections.abc import AsyncIterator

import pytest

from litestar.testing import AsyncTestClient

from my_app.main import app

if TYPE_CHECKING:
    from litestar import Litestar

app.debug = True


@pytest.fixture(scope="function")
async def test_client() -> AsyncIterator[AsyncTestClient[Litestar]]:
    async with AsyncTestClient(app=app) as client:
        yield client

We would then be able to rewrite our test like so:

Sync

tests/test_health_check.py

from collections.abc import Iterator

import pytest

from litestar import Litestar, MediaType, get
from litestar.status_codes import HTTP_200_OK
from litestar.testing import TestClient


@get(path="/health-check", media_type=MediaType.TEXT, sync_to_thread=False)
def health_check() -> str:
    return "healthy"


app = Litestar(route_handlers=[health_check], debug=True)


@pytest.fixture(scope="function")
def test_client() -> Iterator[TestClient[Litestar]]:
    with TestClient(app=app) as client:
        yield client


def test_health_check_with_fixture(test_client: TestClient[Litestar]) -> None:
    response = test_client.get("/health-check")
    assert response.status_code == HTTP_200_OK
    assert response.text == "healthy"

Async

tests/test_health_check.py

from collections.abc import AsyncIterator

import pytest

from litestar import Litestar, MediaType, get
from litestar.status_codes import HTTP_200_OK
from litestar.testing import AsyncTestClient


@get(path="/health-check", media_type=MediaType.TEXT, sync_to_thread=False)
def health_check() -> str:
    return "healthy"


app = Litestar(route_handlers=[health_check], debug=True)


@pytest.fixture(scope="function")
async def test_client() -> AsyncIterator[AsyncTestClient[Litestar]]:
    async with AsyncTestClient(app=app) as client:
        yield client


@pytest.mark.skip(reason="pytest-asyncio issue: https://github.com/pytest-dev/pytest-asyncio/issues/1191")
async def test_health_check_with_fixture(test_client: AsyncTestClient[Litestar]) -> None:
    response = await test_client.get("/health-check")
    assert response.status_code == HTTP_200_OK
    assert response.text == "healthy"

Deciding which test client to use

In most situations, it doesn’t make a functional difference, and just comes down to preference, as both clients offer the same API and capabilities. However, there are some situations where the way the clients run and interact with the application are important, specifically when testing in an asynchronous context.

A common issue when using anyio’s pytest plugin or pytest-asyncio to run asynchronous tests or fixtures, using the synchronous TestClient means that the application will run in a different event loop than the test or fixture. In practice, this can result in some difficult to debug and solve situations, especially when setting up async resources outside the application, for example when using the factory pattern.

The following example uses a shared instance of an httpx.AsyncClient. It uses the common factory function, which allows to customise the client for tests, for example to add authentication headers.

from collections.abc import AsyncIterable

import httpx
import pytest
import pytest_asyncio

from litestar import Litestar, get
from litestar.testing import TestClient


@get("/")
async def handler(http_client: httpx.AsyncClient) -> dict[str, int]:
    response = await http_client.get("https://example.org")
    return {"status": response.status_code}


def create_app(http_client: httpx.AsyncClient) -> Litestar:
    async def provide_http_client() -> httpx.AsyncClient:
        return http_client

    return Litestar([handler], dependencies={"http_client": provide_http_client})


@pytest_asyncio.fixture()
async def http_test_client() -> AsyncIterable[httpx.AsyncClient]:
    client = httpx.AsyncClient(headers={"Authorization": "something"})
    yield client
    await client.aclose()


@pytest.fixture()
def app(http_test_client: httpx.AsyncClient) -> Litestar:
    return create_app(http_client=http_test_client)


def test_handler(app: Litestar) -> None:
    with TestClient(app) as client:
        response = client.get("/")
        assert response.json() == {"status": 200}

Running this test will fail with a RuntimeError: Event loop is closed, when trying to close the AsyncClient instance. This is happening because:

• The http_test_client fixture sets up the client in event loop A

• The TestClient instance created within the test_handler test sets up event loop B and runs the application in it

• A call to http_client.get, the httpx.AsyncClient instance creates a new connection within loop B and attaches it to the client instance

• The TestClient instance closes event loop B

• The cleanup step of the http_test_client fixture calls httpx.AsyncClient.aclose() instance within loop A, which internally tries to close the connection made in the previous step. That connection however is still attached to loop B that was owned by the TestClient instance, and is now closed

This can easily fixed by switching the test from TestClient to AsyncTestClient:

from collections.abc import AsyncIterable

import httpx
import pytest
import pytest_asyncio

from litestar import Litestar, get
from litestar.testing import AsyncTestClient


@get("/")
async def handler(http_client: httpx.AsyncClient) -> dict[str, int]:
    response = await http_client.get("https://example.org")
    return {"status": response.status_code}


def create_app(http_client: httpx.AsyncClient) -> Litestar:
    async def provide_http_client() -> httpx.AsyncClient:
        return http_client

    return Litestar([handler], dependencies={"http_client": provide_http_client})


@pytest_asyncio.fixture()
async def http_test_client() -> AsyncIterable[httpx.AsyncClient]:
    client = httpx.AsyncClient(headers={"Authorization": "something"})
    yield client
    await client.aclose()


@pytest.fixture()
def app(http_test_client: httpx.AsyncClient) -> Litestar:
    return create_app(http_client=http_test_client)


async def test_handler(app: Litestar) -> None:
    async with AsyncTestClient(app) as client:
        response = await client.get("/")
        assert response.json() == {"status": 200}

Now the fixture, test and application code are all running within the same event loop, ensuring that all resources can be cleaned up properly without issues.

Showcasing the different running event loops when using TestClient

import asyncio

import pytest_asyncio

from litestar import Litestar, get
from litestar.testing import AsyncTestClient, TestClient


@get("/")
async def handler() -> dict[str, int]:
    return {"loop_id": id(asyncio.get_running_loop())}


@pytest_asyncio.fixture()
async def fixture_loop_id() -> int:
    return id(asyncio.get_running_loop())


def test_handler(fixture_loop_id: int) -> None:
    app = Litestar([handler])

    with TestClient(app) as client:
        response = client.get("/")
        assert response.json() == {"loop_id": fixture_loop_id}


async def test_handler_async(fixture_loop_id: int) -> None:
    app = Litestar([handler])

    async with AsyncTestClient(app) as client:
        response = await client.get("/")
        assert response.json() == {"loop_id": fixture_loop_id}

Testing websockets

Litestar’s test client enhances the httpx client to support websockets. To test a websocket endpoint, you can use the websocket_connect method on the test client. The method returns a websocket connection object that you can use to send and receive messages, see an example below for json:

For more information, see also the WebSocket class in the API documentation and the websocket documentation.

Sync

from typing import Any

from litestar import WebSocket, websocket
from litestar.testing import create_test_client


def test_websocket() -> None:
    @websocket(path="/ws")
    async def websocket_handler(socket: WebSocket[Any, Any, Any]) -> None:
        await socket.accept()
        recv = await socket.receive_json()
        await socket.send_json({"message": recv})
        await socket.close()

    with create_test_client(route_handlers=[websocket_handler]) as client, client.websocket_connect("/ws") as ws:
        ws.send_json({"hello": "world"})
        data = ws.receive_json()
        assert data == {"message": {"hello": "world"}}

Async

from typing import Any

from litestar import WebSocket, websocket
from litestar.testing import create_async_test_client


async def test_websocket() -> None:
    @websocket(path="/ws")
    async def websocket_handler(socket: WebSocket[Any, Any, Any]) -> None:
        await socket.accept()
        recv = await socket.receive_json()
        await socket.send_json({"message": recv})
        await socket.close()

    async with (
        create_async_test_client(route_handlers=[websocket_handler]) as client,
        await client.websocket_connect("/ws") as ws,
    ):
        await ws.send_json({"hello": "world"})
        data = await ws.receive_json()
        assert data == {"message": {"hello": "world"}}

Using sessions

If you are using session middleware for session persistence across requests, then you might want to inject or inspect session data outside a request. For this, TestClient provides two methods:

• set_session_data

• get_session_data

Sync

Setting session data

from typing import Any

from litestar import Litestar, Request, get
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.testing import TestClient

session_config = ServerSideSessionConfig()


@get(path="/test", sync_to_thread=False)
def get_session_data(request: Request) -> dict[str, Any]:
    return request.session


app = Litestar(route_handlers=[get_session_data], middleware=[session_config.middleware], debug=True)


def test_get_session_data() -> None:
    with TestClient(app=app, session_config=session_config) as client:
        client.set_session_data({"foo": "bar"})
        assert client.get("/test").json() == {"foo": "bar"}

Getting session data

from litestar import Litestar, Request, post
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.testing import TestClient

session_config = ServerSideSessionConfig()


@post(path="/test", sync_to_thread=False)
def set_session_data(request: Request) -> None:
    request.session["foo"] = "bar"


app = Litestar(route_handlers=[set_session_data], middleware=[session_config.middleware], debug=True)

with TestClient(app=app, session_config=session_config) as client:
    client.post("/test").json()
    assert client.get_session_data() == {"foo": "bar"}

Async

Setting session data

from typing import Any

from litestar import Litestar, Request, get
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.testing import AsyncTestClient

session_config = ServerSideSessionConfig()


@get(path="/test", sync_to_thread=False)
def get_session_data(request: Request) -> dict[str, Any]:
    return request.session


app = Litestar(route_handlers=[get_session_data], middleware=[session_config.middleware], debug=True)


async def test_get_session_data() -> None:
    async with AsyncTestClient(app=app, session_config=session_config) as client:
        await client.set_session_data({"foo": "bar"})
        res = await client.get("/test")
        assert res.json() == {"foo": "bar"}

Getting session data

from litestar import Litestar, Request, post
from litestar.middleware.session.server_side import ServerSideSessionConfig
from litestar.testing import AsyncTestClient

session_config = ServerSideSessionConfig()


@post(path="/test", sync_to_thread=False)
def set_session_data(request: Request) -> None:
    request.session["foo"] = "bar"


app = Litestar(route_handlers=[set_session_data], middleware=[session_config.middleware], debug=True)


async def test_set_session_data() -> None:
    async with AsyncTestClient(app=app, session_config=session_config) as client:
        await client.post("/test")
        assert await client.get_session_data() == {"foo": "bar"}

Running async functions on TestClient

When using the synchronous TestClient, it runs the application in a separate thread, which provides the event loop. For this, it makes use of anyio.BlockingPortal.

TestClient makes this portal public, so it can be used to run arbitrary asynchronous code in the same event loop as the application:

Using a blocking portal

from concurrent.futures import Future, wait

import anyio

from litestar.testing import create_test_client


def test_with_portal() -> None:
    """This example shows how to manage asynchronous tasks using a portal.

    The test function itself is not async. Asynchronous functions are executed and awaited using the portal.
    """

    async def get_float(value: float) -> float:
        await anyio.sleep(value)
        return value

    with create_test_client(route_handlers=[]) as test_client:
        # start a background task with the portal
        future: Future[float] = test_client.blocking_portal.start_task_soon(get_float, 0.25)
        # do other work
        assert test_client.blocking_portal.call(get_float, 0.1) == 0.1
        # wait for the background task to complete
        wait([future])
        assert future.done()
        assert future.result() == 0.25

Creating a test app

Litestar also offers a helper function called create_test_client which first creates an instance of Litestar and then a test client using it. There are multiple use cases for this helper - when you need to check generic logic that is decoupled from a specific Litestar app, or when you want to test endpoints in isolation.

my_app/tests/test_health_check.py

from litestar.status_codes import HTTP_200_OK
from litestar.testing import create_test_client

from my_app.main import health_check

def test_health_check():
    with create_test_client([health_check]) as client:
        response = client.get("/health-check")
        assert response.status_code == HTTP_200_OK
        assert response.text == "healthy"

Running a live server

The test clients make use of HTTPX’s ability to directly call into an ASGI app, without having to run an actual server. In most cases this is sufficient but there are some exceptions where this won’t work, due to the limitations of the emulated client-server communication.

For example, when using server-sent events with an infinite generator, it will lock up the test client, since HTTPX tries to consume the full response before returning a request.

Litestar offers two helper functions, litestar.testing.subprocess_sync_client() and litestar.testing.subprocess_async_client() that will launch a Litestar instance with in a subprocess and set up an httpx client for running tests. You can either load your actual app file or create subsets from it as you would with the regular test client setup:

"""
Assemble components into an app that shall be tested
"""

from collections.abc import AsyncGenerator

from litestar import Litestar, get
from litestar.response import ServerSentEvent
from litestar.types import SSEData


async def generator(topic: str) -> AsyncGenerator[SSEData, None]:
    count = 0
    while count < 2:
        yield topic
        count += 1


@get("/notify/{topic:str}")
async def get_notified(topic: str) -> ServerSentEvent:
    return ServerSentEvent(generator(topic), event_type="Notifier")


app = Litestar(route_handlers=[get_notified])

"""
Test the app running in a subprocess
"""

import asyncio
import pathlib
import sys
from collections.abc import AsyncIterator

import httpx
import httpx_sse
import pytest

from litestar.testing import subprocess_async_client

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())


ROOT = pathlib.Path(__file__).parent


@pytest.fixture(name="async_client")
async def fx_async_client() -> AsyncIterator[httpx.AsyncClient]:
    async with subprocess_async_client(workdir=ROOT, app="subprocess_sse_app:app") as client:
        yield client


async def test_subprocess_async_client(async_client: httpx.AsyncClient) -> None:
    """Demonstrates functionality of the async client with an infinite SSE source that cannot be tested with the
    regular async test client.
    """
    topic = "demo"

    async with httpx_sse.aconnect_sse(async_client, "GET", f"/notify/{topic}") as event_source:
        async for event in event_source.aiter_sse():
            assert event.data == topic
            break

RequestFactory

Another helper is the RequestFactory class, which creates instances of litestar.connection.request.Request. The use case for this helper is when you need to test logic that expects to receive a request object.

For example, lets say we wanted to unit test a guard function in isolation, to which end we’ll reuse the examples from the route guards documentation:

my_app/guards.py

from litestar import Request
from litestar.exceptions import NotAuthorizedException
from litestar.handlers.base import BaseRouteHandler


def secret_token_guard(request: Request, route_handler: BaseRouteHandler) -> None:
    if (
        route_handler.opt.get("secret")
        and not request.headers.get("Secret-Header", "") == route_handler.opt["secret"]
    ):
        raise NotAuthorizedException()

We already have our route handler in place:

my_app/secret.py

from os import environ

from litestar import get

from my_app.guards import secret_token_guard


@get(path="/secret", guards=[secret_token_guard], opt={"secret": environ.get("SECRET")})
def secret_endpoint() -> None: ...

We could thus test the guard function like so:

tests/guards/test_secret_token_guard.py

import pytest

from litestar.exceptions import NotAuthorizedException
from litestar.testing import RequestFactory

from my_app.guards import secret_token_guard
from my_app.secret import secret_endpoint

request = RequestFactory().get("/")


def test_secret_token_guard_failure_scenario():
    copied_endpoint_handler = secret_endpoint.copy()
    copied_endpoint_handler.opt["secret"] = None
    with pytest.raises(NotAuthorizedException):
        secret_token_guard(request=request, route_handler=copied_endpoint_handler)


def test_secret_token_guard_success_scenario():
    copied_endpoint_handler = secret_endpoint.copy()
    copied_endpoint_handler.opt["secret"] = "super-secret"
    secret_token_guard(request=request, route_handler=copied_endpoint_handler)