Testing#

Testing is a first class citizen in Litestar, which offers several powerful testing utilities out of the box.

Test Client#

Litestar’s test client is built on top of the httpx library. To use the test client you should pass to it an instance of Litestar as the app kwarg.

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:

tests/test_health_check.py#
from litestar.status_codes import HTTP_200_OK
from litestar.testing import TestClient

from my_app.main import app


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"
tests/test_health_check.py#
from litestar.status_codes import HTTP_200_OK
from litestar.testing import AsyncTestClient

from my_app.main import app


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:

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


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


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


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


@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:

tests/test_health_check.py#
from typing 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])


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


@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"
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])


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


@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"
tests/test_health_check.py#
from typing 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])


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


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


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"
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])


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


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


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"

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:

Attention

  • The Session Middleware must be enabled in Litestar app provided to the TestClient to use sessions.

  • If you are using the ClientSideSessionBackend you need to install the cryptography package. You can do so by installing litestar:

Using pip#
python3 -m pip install litestar[cryptography]
Using pipx#
pipx install litestar[cryptography]
Using PDM#
pdm add litestar[cryptography]
Using Poetry#
poetry add litestar[cryptography]
Setting session data#
from typing import Any, Dict

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])


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"}
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])


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])

with TestClient(app=app, session_config=session_config) as client:
    client.post("/test").json()
    assert client.get_session_data() == {"foo": "bar"}
Setting session data#
from typing import Any, Dict

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])


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"}
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])


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])


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"}

Using a blocking portal#

The TestClient uses a feature of anyio called a Blocking Portal.

The anyio.abc.BlockingPortal allows TestClient to execute asynchronous functions using a synchronous call. TestClient creates a blocking portal to manage Litestar’s async logic, and it allows TestClient’s API to remain fully synchronous.

Any tests that are using an instance of TestClient can also make use of the blocking portal to execute asynchronous functions without the test itself being asynchronous.

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, test_client.portal() as portal:
        # start a background task with the portal
        future: Future[float] = portal.start_task_soon(get_float, 0.25)
        # do other work
        assert 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.

You can pass to this helper all the kwargs accepted by the litestar constructor, with the route_handlers kwarg being required. Yet unlike the Litestar app, which expects route_handlers to be a list, here you can also pass individual values.

For example, you can do this:

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(route_handlers=[health_check]) as client:
        response = client.get("/health-check")
        assert response.status_code == HTTP_200_OK
        assert response.text == "healthy"

But also this:

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(route_handlers=health_check) as client:
        response = client.get("/health-check")
        assert response.status_code == HTTP_200_OK
        assert response.text == "healthy"

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)

Using polyfactory#

Polyfactory offers an easy and powerful way to generate mock data from pydantic models and dataclasses.

Let’s say we have an API that talks to an external service and retrieves some data:

main.py#
from typing import Protocol, runtime_checkable

from polyfactory.factories.pydantic import BaseModel
from litestar import get


class Item(BaseModel):
    name: str


@runtime_checkable
class Service(Protocol):
    def get(self) -> Item: ...


@get(path="/item")
def get_item(service: Service) -> Item:
    return service.get()

We could test the /item route like so:

tests/conftest.py#
import pytest

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

from my_app.main import Service, Item, get_item


@pytest.fixture()
def item():
    return Item(name="Chair")


def test_get_item(item: Item):
    class MyService(Service):
        def get_one(self) -> Item:
            return item

    with create_test_client(
        route_handlers=get_item, dependencies={"service": Provide(lambda: MyService())}
    ) as client:
        response = client.get("/item")
        assert response.status_code == HTTP_200_OK
        assert response.json() == item.dict()

While we can define the test data manually, as is done in the above, this can be quite cumbersome. That’s where polyfactory library comes in. It generates mock data for pydantic models and dataclasses based on type annotations. With it, we could rewrite the above example like so:

main.py#
from typing import Protocol, runtime_checkable

import pytest
from pydantic import BaseModel
from polyfactory.factories.pydantic_factory import ModelFactory
from litestar.status_codes import HTTP_200_OK
from litestar import get
from litestar.di import Provide
from litestar.testing import create_test_client


class Item(BaseModel):
    name: str


@runtime_checkable
class Service(Protocol):
    def get_one(self) -> Item: ...


@get(path="/item")
def get_item(service: Service) -> Item:
    return service.get_one()


class ItemFactory(ModelFactory[Item]):
    model = Item


@pytest.fixture()
def item():
    return ItemFactory.build()


def test_get_item(item: Item):
    class MyService(Service):
        def get_one(self) -> Item:
            return item

    with create_test_client(
        route_handlers=get_item, dependencies={"service": Provide(lambda: MyService())}
    ) as client:
        response = client.get("/item")
        assert response.status_code == HTTP_200_OK
        assert response.json() == item.dict()