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:
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:
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"
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:
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
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
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
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:
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])
@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"
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])
@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"
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])
@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"
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])
@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 thecryptography
package. You can do so by installinglitestar
:
python3 -m pip install litestar[cryptography]
pipx install litestar[cryptography]
pdm add litestar[cryptography]
poetry add litestar[cryptography]
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"}
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"}
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"}
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"}
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"}
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.
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:
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:
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:
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:
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:
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:
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:
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:
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()