Skip to content

Test Client

Esmerald comes with a test client for your application tests. It is not mandatory use it as every application and development team has its own way of testing it but just in case, it is provided.

Requirements

This section requires the esmerald testing suite to be installed. You can do it so by running:

$ pip install esmerald[test]

API Reference

Check the API Reference for the test client to understand more.

The test client

from esmerald.testclient import EsmeraldTestClient
from lilya.responses import HTMLResponse


async def app(scope, receive, send):
    assert scope["type"] == "http"
    response = HTMLResponse("<html><body>Hello, world!</body></html>")
    await response(scope, receive, send)


def test_application():
    client = EsmeraldTestClient(app)
    response = client.get("/")
    assert response.status_code == 200

The test client is very similar to its original as it extends it and adds extra unique and specifics for Esmerald and therefore the same examples and use cases will work.

You can use any of the httpx standard API like authentication, session cookies and file uploads.

from esmerald.testclient import EsmeraldTestClient
from lilya.responses import HTMLResponse


async def app(scope, receive, send):
    assert scope["type"] == "http"
    response = HTMLResponse("<html><body>Hello, world!</body></html>")
    await response(scope, receive, send)


client = EsmeraldTestClient(app)

# Set headers on the client for future requests
client.headers = {"Authorization": "..."}
response = client.get("/")

# Set headers for each request separately
response = client.get("/", headers={"Authorization": "..."})

And like Lilya, the same example to send files with EsmeraldTestClient.

from esmerald.testclient import EsmeraldTestClient
from lilya.responses import HTMLResponse


async def app(scope, receive, send):
    assert scope["type"] == "http"
    response = HTMLResponse("<html><body>Hello, world!</body></html>")
    await response(scope, receive, send)


client = EsmeraldTestClient(app)

# Send a single file
with open("example.txt", "rb") as f:
    response = client.post("/form", files={"file": f})


# Send multiple files
with open("example.txt", "rb") as f1:
    with open("example.png", "rb") as f2:
        files = {"file1": f1, "file2": ("filename", f2, "image/png")}
        response = client.post("/form", files=files)

httpx is a great library created by the same author of Django Rest Framework.

Info

By default the EsmeraldTestClient raise any exceptions that occur in the application. Occasionally you might want to test the content of 500 error responses, rather than allowing client to raise the server exception. In this case you should use client = EsmeraldTestClient(app, raise_server_exceptions=False).

Lifespan events

Note

Esmerald supports all the lifespan events available and therefore on_startup, on_shutdown and lifespan are also supported by EsmeraldTestClient but if you need to test these you will need to run EsmeraldTestClient as a context manager or otherwise the events will not be triggered when the EsmeraldTestClient is instantiated.

The framework also brings a ready to use functionality to be used as context manager for your tests.

Context manager create_client

This function is prepared to be used as a context manager for your tests and ready to use at any given time.

import pytest

from esmerald import Gateway, Include, Request, WebSocket, WebSocketGateway, get, websocket
from esmerald.enums import MediaType
from esmerald.permissions import AllowAny, DenyAll
from esmerald.responses import JSONResponse
from esmerald.testclient import create_client
from lilya.responses import Response


@get(path="/", permissions=[DenyAll])
async def deny_access(request: Request) -> JSONResponse:
    return JSONResponse("Hello, world")


@get(path="/", permissions=[AllowAny])
async def allow_access(request: Request) -> JSONResponse:
    return JSONResponse("Hello, world")


@get(path="/", media_type=MediaType.TEXT, status_code=200)
async def homepage(request: Request) -> Response:
    return Response("Hello, world")


@websocket(path="/")
async def websocket_endpoint(socket: WebSocket) -> None:
    await socket.accept()
    await socket.send_text("Hello, world!")
    await socket.close()


routes = [
    Gateway("/", handler=homepage, name="homepage"),
    Include(
        "/nested",
        routes=[
            Include(
                path="/test/",
                routes=[Gateway(path="/", handler=homepage, name="nested")],
            ),
            Include(
                path="/another",
                routes=[
                    Include(
                        path="/test",
                        routes=[Gateway(path="/{param}", handler=homepage, name="nested")],
                    )
                ],
            ),
        ],
    ),
    Include(
        "/static",
        app=Response("xxxxx", media_type=MediaType.PNG, status_code=200),
    ),
    WebSocketGateway("/ws", handler=websocket_endpoint, name="websocket_endpoint"),
    Gateway("/deny", handler=deny_access, name="deny_access"),
    Gateway("/allow", handler=allow_access, name="allow_access"),
]


@pytest.mark.filterwarnings(
    r"ignore"
    r":Trying to detect encoding from a tiny portion of \(5\) byte\(s\)\."
    r":UserWarning"
    r":charset_normalizer.api"
)
def test_router():
    with create_client(routes=routes) as client:
        response = client.get("/")
        assert response.status_code == 200
        assert response.text == "Hello, world"

        response = client.post("/")
        assert response.status_code == 405
        assert response.json()["detail"] == "Method POST not allowed."
        assert response.headers["content-type"] == MediaType.JSON

        response = client.get("/foo")
        assert response.status_code == 404
        assert response.json()["detail"] == "The resource cannot be found."

        response = client.get("/static/123")
        assert response.status_code == 200
        assert response.text == "xxxxx"

        response = client.get("/nested/test")
        assert response.status_code == 200
        assert response.text == "Hello, world"

        response = client.get("/nested/another/test/fluid")
        assert response.status_code == 200
        assert response.text == "Hello, world"

        with client.websocket_connect("/ws") as session:
            text = session.receive_text()
            assert text == "Hello, world!"

The tests work with both sync and async functions.

Info

The example above is used to also show the tests can be as complex as you desire and it will work with the context manager.