Testing¶

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

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"

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)

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

Testing¶

Test Client¶

Testing websockets¶

Using sessions¶

Using a blocking portal¶

Creating a test app¶

Running a live server¶

RequestFactory¶

Using polyfactory¶