CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-litestar

Litestar is a powerful, flexible yet opinionated ASGI web framework specifically focused on building high-performance APIs.

—

Pending
Overview
Eval results
Files

testing.mddocs/

Testing

Test client implementations for testing Litestar applications. Litestar provides comprehensive testing utilities including synchronous and asynchronous test clients with support for HTTP requests, WebSocket connections, and integration testing.

Capabilities

Test Clients

Test clients provide a convenient way to test Litestar applications without running a full server.

class TestClient:
    def __init__(
        self,
        app: Litestar,
        *,
        base_url: str = "http://testserver",
        backend: Literal["asyncio", "trio"] = "asyncio",
        backend_options: dict[str, Any] | None = None,
        cookies: httpx.Cookies | dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
        follow_redirects: bool = False,
        **kwargs: Any,
    ):
        """
        Synchronous test client for Litestar applications.

        Parameters:
        - app: Litestar application instance
        - base_url: Base URL for requests
        - backend: Async backend to use
        - backend_options: Backend-specific options
        - cookies: Default cookies for requests
        - headers: Default headers for requests
        - follow_redirects: Whether to follow redirects automatically
        - **kwargs: Additional httpx.Client options
        """

    # HTTP methods
    def get(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send GET request."""

    def post(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send POST request."""

    def put(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send PUT request."""

    def patch(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send PATCH request."""

    def delete(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send DELETE request."""

    def head(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send HEAD request."""

    def options(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send OPTIONS request."""

    def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
        """Send request with specified method."""

    # Context manager support
    def __enter__(self) -> TestClient:
        """Enter test client context."""

    def __exit__(self, *args: Any) -> None:
        """Exit test client context."""

class AsyncTestClient:
    def __init__(
        self,
        app: Litestar,
        *,
        base_url: str = "http://testserver",
        backend: Literal["asyncio", "trio"] = "asyncio",
        backend_options: dict[str, Any] | None = None,
        cookies: httpx.Cookies | dict[str, str] | None = None,
        headers: dict[str, str] | None = None,
        follow_redirects: bool = False,
        **kwargs: Any,
    ):
        """
        Asynchronous test client for Litestar applications.

        Parameters: Same as TestClient
        """

    # Async HTTP methods
    async def get(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send GET request asynchronously."""

    async def post(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send POST request asynchronously."""

    async def put(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send PUT request asynchronously."""

    async def patch(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send PATCH request asynchronously."""

    async def delete(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send DELETE request asynchronously."""

    async def head(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send HEAD request asynchronously."""

    async def options(self, url: str, **kwargs: Any) -> httpx.Response:
        """Send OPTIONS request asynchronously."""

    async def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
        """Send request with specified method asynchronously."""

    # Async context manager support
    async def __aenter__(self) -> AsyncTestClient:
        """Enter async test client context."""

    async def __aexit__(self, *args: Any) -> None:
        """Exit async test client context."""

WebSocket Test Session

Test utilities for WebSocket connections.

class WebSocketTestSession:
    def __init__(self, client: TestClient | AsyncTestClient):
        """
        WebSocket test session.

        Parameters:
        - client: Test client instance
        """

    async def __aenter__(self) -> WebSocketTestSession:
        """Enter WebSocket test session."""

    async def __aexit__(self, *args: Any) -> None:
        """Exit WebSocket test session."""

    async def send_text(self, data: str) -> None:
        """Send text data over WebSocket."""

    async def send_bytes(self, data: bytes) -> None:
        """Send binary data over WebSocket."""

    async def send_json(self, data: Any) -> None:
        """Send JSON data over WebSocket."""

    async def receive_text(self) -> str:
        """Receive text data from WebSocket."""

    async def receive_bytes(self) -> bytes:
        """Receive binary data from WebSocket."""

    async def receive_json(self) -> Any:
        """Receive JSON data from WebSocket."""

    async def close(self, code: int = 1000) -> None:
        """Close the WebSocket connection."""

Test Client Factory Functions

Convenience functions for creating test clients.

def create_test_client(
    route_handlers: ControllerRouterHandler | Sequence[ControllerRouterHandler],
    *,
    backend: Literal["asyncio", "trio"] = "asyncio",
    **litestar_kwargs: Any,
) -> TestClient:
    """
    Create a synchronous test client with route handlers.

    Parameters:
    - route_handlers: Route handlers to test
    - backend: Async backend to use
    - **litestar_kwargs: Additional Litestar constructor arguments

    Returns:
    Configured TestClient instance
    """

async def create_async_test_client(
    route_handlers: ControllerRouterHandler | Sequence[ControllerRouterHandler],
    *,
    backend: Literal["asyncio", "trio"] = "asyncio",
    **litestar_kwargs: Any,
) -> AsyncTestClient:
    """
    Create an asynchronous test client with route handlers.

    Parameters:
    - route_handlers: Route handlers to test
    - backend: Async backend to use  
    - **litestar_kwargs: Additional Litestar constructor arguments

    Returns:
    Configured AsyncTestClient instance
    """

Request Factory

Factory for creating test request objects.

class RequestFactory:
    def __init__(self, app: Litestar | None = None):
        """
        Request factory for creating test requests.

        Parameters:
        - app: Litestar application instance
        """

    def get(
        self,
        path: str = "/",
        *,
        headers: dict[str, str] | None = None,
        query_params: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> Request:
        """Create a GET request."""

    def post(
        self,
        path: str = "/",
        *,
        headers: dict[str, str] | None = None,
        json: Any = None,
        data: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> Request:
        """Create a POST request."""

    def put(
        self,
        path: str = "/",
        **kwargs: Any,
    ) -> Request:
        """Create a PUT request."""

    def patch(
        self,
        path: str = "/",
        **kwargs: Any,
    ) -> Request:
        """Create a PATCH request."""

    def delete(
        self,
        path: str = "/",
        **kwargs: Any,
    ) -> Request:
        """Create a DELETE request."""

    def websocket(
        self,
        path: str = "/",
        *,
        headers: dict[str, str] | None = None,
        query_params: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> WebSocket:
        """Create a WebSocket connection."""

Subprocess Test Clients

Test clients that run the application in a separate process for true integration testing.

def subprocess_sync_client(
    app: Litestar,
    *,
    port: int = 0,
    host: str = "127.0.0.1",
    **kwargs: Any,
) -> TestClient:
    """
    Create a test client that runs the app in a subprocess.

    Parameters:
    - app: Litestar application
    - port: Port to bind to (0 for random port)
    - host: Host to bind to
    - **kwargs: Additional test client options

    Returns:
    TestClient connected to subprocess server
    """

def subprocess_async_client(
    app: Litestar,
    *,
    port: int = 0,
    host: str = "127.0.0.1",
    **kwargs: Any,
) -> AsyncTestClient:
    """
    Create an async test client that runs the app in a subprocess.

    Parameters:
    - app: Litestar application
    - port: Port to bind to (0 for random port)
    - host: Host to bind to
    - **kwargs: Additional test client options

    Returns:
    AsyncTestClient connected to subprocess server
    """

Usage Examples

Basic HTTP Testing

import pytest
from litestar import Litestar, get, post
from litestar.testing import TestClient
from litestar.status_codes import HTTP_200_OK, HTTP_201_CREATED

@get("/health")
def health_check() -> dict[str, str]:
    return {"status": "healthy"}

@post("/users")
def create_user(data: dict) -> dict:
    return {"id": 123, "name": data["name"]}

app = Litestar(route_handlers=[health_check, create_user])

def test_health_check():
    with TestClient(app=app) as client:
        response = client.get("/health")
        assert response.status_code == HTTP_200_OK
        assert response.json() == {"status": "healthy"}

def test_create_user():
    with TestClient(app=app) as client:
        response = client.post("/users", json={"name": "Alice"})
        assert response.status_code == HTTP_201_CREATED
        assert response.json() == {"id": 123, "name": "Alice"}

Async Testing

import pytest
from litestar.testing import AsyncTestClient

@pytest.mark.asyncio
async def test_async_health_check():
    async with AsyncTestClient(app=app) as client:
        response = await client.get("/health")
        assert response.status_code == HTTP_200_OK
        assert response.json() == {"status": "healthy"}

@pytest.mark.asyncio
async def test_async_create_user():
    async with AsyncTestClient(app=app) as client:
        response = await client.post("/users", json={"name": "Bob"})
        assert response.status_code == HTTP_201_CREATED
        assert response.json() == {"id": 123, "name": "Bob"}

WebSocket Testing

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

@websocket("/ws")
async def websocket_handler(websocket: WebSocket) -> None:
    await websocket.accept()
    message = await websocket.receive_text()
    await websocket.send_text(f"Echo: {message}")
    await websocket.close()

def test_websocket():
    with create_test_client(route_handlers=[websocket_handler]) as client:
        with client.websocket_connect("/ws") as websocket:
            websocket.send_text("Hello WebSocket")
            data = websocket.receive_text()
            assert data == "Echo: Hello WebSocket"

Authentication Testing

from litestar.security.jwt import JWTAuth
from litestar.testing import create_test_client

# JWT auth setup (simplified)
jwt_auth = JWTAuth(
    token_secret="test-secret",
    retrieve_user_handler=lambda token, connection: {"id": 1, "name": "Test User"},
)

@get("/profile")
def get_profile(request: Request) -> dict:
    return {"user": request.user}

def test_authenticated_endpoint():
    app = Litestar(
        route_handlers=[get_profile],
        on_app_init=[jwt_auth.on_app_init],
    )
    
    with TestClient(app=app) as client:
        # Get token
        token = jwt_auth.create_token(identifier="1")
        
        # Make authenticated request
        response = client.get(
            "/profile",
            headers={"Authorization": f"Bearer {token}"}
        )
        
        assert response.status_code == HTTP_200_OK
        assert response.json() == {"user": {"id": 1, "name": "Test User"}}

def test_unauthenticated_request():
    with TestClient(app=app) as client:
        response = client.get("/profile")
        assert response.status_code == 401

Database Testing with Fixtures

import pytest
from litestar import Litestar, get, post, Dependency
from litestar.testing import create_async_test_client
from unittest.mock import AsyncMock

# Mock database
class MockDatabase:
    def __init__(self):
        self.users = {}
        self.next_id = 1
    
    async def create_user(self, name: str) -> dict:
        user = {"id": self.next_id, "name": name}
        self.users[self.next_id] = user
        self.next_id += 1
        return user
    
    async def get_user(self, user_id: int) -> dict | None:
        return self.users.get(user_id)

@pytest.fixture
def mock_db():
    return MockDatabase()

def create_app(db: MockDatabase) -> Litestar:
    @post("/users")
    async def create_user(data: dict, db: MockDatabase = Dependency(lambda: db)) -> dict:
        return await db.create_user(data["name"])

    @get("/users/{user_id:int}")
    async def get_user(user_id: int, db: MockDatabase = Dependency(lambda: db)) -> dict:
        user = await db.get_user(user_id)
        if not user:
            raise NotFoundException("User not found")
        return user

    return Litestar(route_handlers=[create_user, get_user])

@pytest.mark.asyncio
async def test_user_crud(mock_db):
    app = create_app(mock_db)
    
    async with create_async_test_client(app) as client:
        # Create user
        response = await client.post("/users", json={"name": "Alice"})
        assert response.status_code == HTTP_201_CREATED
        user = response.json()
        assert user["name"] == "Alice"
        user_id = user["id"]
        
        # Get user
        response = await client.get(f"/users/{user_id}")
        assert response.status_code == HTTP_200_OK
        assert response.json() == user
        
        # Get non-existent user
        response = await client.get("/users/999")
        assert response.status_code == 404

File Upload Testing

from litestar import post, Request
from litestar.testing import TestClient
import io

@post("/upload")
async def upload_file(request: Request) -> dict:
    form_data = await request.form()
    uploaded_file = form_data["file"]
    return {
        "filename": uploaded_file.filename,
        "size": len(await uploaded_file.read()),
    }

def test_file_upload():
    with TestClient(app=Litestar(route_handlers=[upload_file])) as client:
        file_content = b"test file content"
        files = {"file": ("test.txt", io.BytesIO(file_content), "text/plain")}
        
        response = client.post("/upload", files=files)
        assert response.status_code == HTTP_200_OK
        assert response.json() == {
            "filename": "test.txt",
            "size": len(file_content)
        }

Custom Headers and Cookies Testing

def test_custom_headers():
    @get("/headers")
    def get_headers(request: Request) -> dict:
        return {"custom_header": request.headers.get("X-Custom-Header")}
    
    with TestClient(app=Litestar(route_handlers=[get_headers])) as client:
        response = client.get(
            "/headers",
            headers={"X-Custom-Header": "test-value"}
        )
        assert response.json() == {"custom_header": "test-value"}

def test_cookies():
    @get("/cookies")
    def get_cookies(request: Request) -> dict:
        return {"session_id": request.cookies.get("session_id")}
    
    with TestClient(app=Litestar(route_handlers=[get_cookies])) as client:
        response = client.get(
            "/cookies",
            cookies={"session_id": "abc123"}
        )
        assert response.json() == {"session_id": "abc123"}

Integration Testing with Real Server

from litestar.testing import subprocess_async_client
import pytest

@pytest.mark.asyncio
async def test_integration():
    """Test with real server running in subprocess."""
    app = Litestar(route_handlers=[health_check])
    
    async with subprocess_async_client(app) as client:
        response = await client.get("/health")
        assert response.status_code == HTTP_200_OK
        assert response.json() == {"status": "healthy"}

Types

# Test client types
TestClientBackend = Literal["asyncio", "trio"]

# HTTP response type from httpx
HTTPXResponse = httpx.Response

# Route handler types for testing
ControllerRouterHandler = Controller | Router | BaseRouteHandler

# Test context managers
TestClientContext = TestClient
AsyncTestClientContext = AsyncTestClient

Install with Tessl CLI

npx tessl i tessl/pypi-litestar

docs

application-routing.md

configuration.md

dto.md

exceptions.md

http-handlers.md

index.md

middleware.md

openapi.md

plugins.md

request-response.md

security.md

testing.md

websocket.md

tile.json