tessl/pypi-starlette

The little ASGI library that shines.

Overview

Eval results

Files

Testing Utilities

Name: tessl/pypi-starlette
Author: tessl

Starlette provides comprehensive testing utilities built on HTTPX, enabling easy testing of HTTP endpoints, WebSocket connections, middleware, authentication, and complete application workflows.

TestClient Class

from starlette.testclient import TestClient
from starlette.types import ASGIApp
from httpx import Client
from typing import Any, Dict, Optional, Union, Mapping
import httpx

class TestClient(httpx.Client):
    """
    Test client for ASGI applications.
    
    Built on HTTPX for modern async HTTP testing with:
    - Automatic lifespan management
    - WebSocket testing support
    - File upload testing
    - Cookie and session management
    - Request/response inspection
    """
    
    def __init__(
        self,
        app: ASGIApp,
        base_url: str = "http://testserver",
        raise_server_exceptions: bool = True,
        root_path: str = "",
        backend: str = "asyncio",
        backend_options: Optional[Dict[str, Any]] = None,
        cookies: httpx.Cookies = None,
        headers: Mapping[str, str] = None,
        follow_redirects: bool = True,
        client: tuple[str, int] = ("testclient", 50000),
    ) -> None:
        """
        Initialize test client.
        
        Args:
            app: ASGI application to test
            base_url: Base URL for requests
            raise_server_exceptions: Raise server exceptions in tests
            root_path: ASGI root path
            backend: Async backend ("asyncio" or "trio")
            backend_options: Backend-specific options
            cookies: Default cookies for requests
            headers: Default headers for requests
            follow_redirects: Automatically follow redirects
            client: Client address tuple (host, port)
        """
    
    # HTTP Methods (inherited from httpx.Client)
    def get(
        self,
        url: str,
        *,
        params: Dict[str, Any] = None,
        headers: Mapping[str, str] = None,
        cookies: httpx.Cookies = None,
        auth: httpx.Auth = None,
        follow_redirects: bool = None,
        timeout: Union[float, httpx.Timeout] = None,
    ) -> httpx.Response:
        """Send GET request."""
    
    def post(
        self,
        url: str,
        *,
        content: Union[str, bytes] = None,
        data: Dict[str, Any] = None,
        files: Dict[str, Any] = None,
        json: Any = None,
        params: Dict[str, Any] = None,
        headers: Mapping[str, str] = None,
        cookies: httpx.Cookies = None,
        auth: httpx.Auth = None,
        follow_redirects: bool = None,
        timeout: Union[float, httpx.Timeout] = None,
    ) -> httpx.Response:
        """Send POST request."""
    
    def put(self, url: str, **kwargs) -> httpx.Response:
        """Send PUT request."""
    
    def patch(self, url: str, **kwargs) -> httpx.Response:
        """Send PATCH request."""
    
    def delete(self, url: str, **kwargs) -> httpx.Response:
        """Send DELETE request."""
    
    def head(self, url: str, **kwargs) -> httpx.Response:
        """Send HEAD request."""
    
    def options(self, url: str, **kwargs) -> httpx.Response:
        """Send OPTIONS request."""
    
    # WebSocket testing
    def websocket_connect(
        self,
        url: str,
        subprotocols: List[str] = None,
        **kwargs
    ) -> WebSocketTestSession:
        """
        Connect to WebSocket endpoint.
        
        Args:
            url: WebSocket URL
            subprotocols: List of subprotocols to negotiate
            **kwargs: Additional connection parameters
            
        Returns:
            WebSocketTestSession: Test session for WebSocket interaction
        """
    
    # Context manager support for lifespan events
    def __enter__(self) -> "TestClient":
        """Enter context manager and run lifespan startup."""
    
    def __exit__(self, *args) -> None:
        """Exit context manager and run lifespan shutdown."""

WebSocket Testing

from starlette.testclient import WebSocketTestSession
from starlette.websockets import WebSocketDisconnect
from typing import Any, Dict

class WebSocketTestSession:
    """
    WebSocket test session for testing WebSocket endpoints.
    
    Provides methods for sending and receiving WebSocket messages
    in test scenarios with proper connection management.
    """
    
    def send(self, message: Dict[str, Any]) -> None:
        """Send raw WebSocket message."""
    
    def send_text(self, data: str) -> None:
        """Send text message."""
    
    def send_bytes(self, data: bytes) -> None:
        """Send binary message."""
    
    def send_json(self, data: Any, mode: str = "text") -> None:
        """Send JSON message."""
    
    def receive(self) -> Dict[str, Any]:
        """Receive raw WebSocket message."""
    
    def receive_text(self) -> str:
        """Receive text message."""
    
    def receive_bytes(self) -> bytes:
        """Receive binary message."""
    
    def receive_json(self, mode: str = "text") -> Any:
        """Receive JSON message."""
    
    def close(self, code: int = 1000, reason: str = None) -> None:
        """Close WebSocket connection."""

class WebSocketDenialResponse(httpx.Response, WebSocketDisconnect):
    """
    Exception raised when WebSocket connection is denied.
    
    Contains the HTTP response that was sent instead of
    accepting the WebSocket connection.
    """
    pass

Basic Testing

Simple HTTP Endpoint Testing

from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import JSONResponse
from starlette.testclient import TestClient

# Application to test
async def homepage(request):
    return JSONResponse({"message": "Hello World"})

async def user_detail(request):
    user_id = request.path_params["user_id"]
    return JSONResponse({"user_id": int(user_id)})

app = Starlette(routes=[
    Route("/", homepage),
    Route("/users/{user_id:int}", user_detail),
])

# Test functions
def test_homepage():
    client = TestClient(app)
    response = client.get("/")
    
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_user_detail():
    client = TestClient(app)
    response = client.get("/users/123")
    
    assert response.status_code == 200
    assert response.json() == {"user_id": 123}

def test_not_found():
    client = TestClient(app)
    response = client.get("/nonexistent")
    
    assert response.status_code == 404

Testing Different HTTP Methods

async def api_endpoint(request):
    if request.method == "GET":
        return JSONResponse({"method": "GET"})
    elif request.method == "POST":
        data = await request.json()
        return JSONResponse({"method": "POST", "data": data})
    elif request.method == "PUT":
        data = await request.json()
        return JSONResponse({"method": "PUT", "updated": data})

app = Starlette(routes=[
    Route("/api", api_endpoint, methods=["GET", "POST", "PUT"]),
])

def test_http_methods():
    client = TestClient(app)
    
    # Test GET
    response = client.get("/api")
    assert response.status_code == 200
    assert response.json()["method"] == "GET"
    
    # Test POST
    response = client.post("/api", json={"name": "test"})
    assert response.status_code == 200
    assert response.json()["method"] == "POST"
    assert response.json()["data"]["name"] == "test"
    
    # Test PUT
    response = client.put("/api", json={"id": 1, "name": "updated"})
    assert response.status_code == 200
    assert response.json()["method"] == "PUT"

Testing Request Data

async def form_endpoint(request):
    form = await request.form()
    return JSONResponse({
        "name": form.get("name"),
        "email": form.get("email"),
        "file_uploaded": bool(form.get("file"))
    })

def test_form_data():
    client = TestClient(app)
    
    # Test form data
    response = client.post("/form", data={
        "name": "John Doe",
        "email": "john@example.com"
    })
    
    assert response.status_code == 200
    assert response.json()["name"] == "John Doe"
    assert response.json()["email"] == "john@example.com"

def test_file_upload():
    client = TestClient(app)
    
    # Test file upload
    with open("test_file.txt", "w") as f:
        f.write("test content")
    
    with open("test_file.txt", "rb") as f:
        response = client.post("/form", 
            data={"name": "John"},
            files={"file": ("test.txt", f, "text/plain")}
        )
    
    assert response.status_code == 200
    assert response.json()["file_uploaded"] is True

Advanced Testing

Testing with Lifespan Events

from contextlib import asynccontextmanager

# Application with lifespan
@asynccontextmanager
async def lifespan(app):
    # Startup
    app.state.database = {"connected": True}
    print("Database connected")
    
    yield
    
    # Shutdown  
    app.state.database = {"connected": False}
    print("Database disconnected")

app = Starlette(
    lifespan=lifespan,
    routes=routes
)

def test_with_lifespan():
    # Using context manager automatically handles lifespan
    with TestClient(app) as client:
        # Lifespan startup has run
        assert hasattr(client.app.state, "database")
        
        response = client.get("/")
        assert response.status_code == 200
        
    # Lifespan shutdown has run

def test_without_lifespan():
    # Create client without context manager
    client = TestClient(app)
    
    # Lifespan events don't run automatically
    response = client.get("/health")  # Simple endpoint
    assert response.status_code == 200
    
    # Manual cleanup
    client.close()

Testing Middleware

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware import Middleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        return response

app = Starlette(
    routes=routes,
    middleware=[
        Middleware(TimingMiddleware),
    ]
)

def test_middleware():
    client = TestClient(app)
    response = client.get("/")
    
    # Check middleware added header
    assert "X-Process-Time" in response.headers
    
    # Verify timing is reasonable
    process_time = float(response.headers["X-Process-Time"])
    assert process_time > 0
    assert process_time < 1.0  # Should be under 1 second

Testing Authentication

from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.authentication import AuthenticationBackend, AuthCredentials, SimpleUser

class TestAuthBackend(AuthenticationBackend):
    async def authenticate(self, conn):
        # Simple test authentication
        auth_header = conn.headers.get("Authorization")
        if auth_header == "Bearer valid-token":
            return AuthCredentials(["authenticated"]), SimpleUser("testuser")
        return None

app = Starlette(
    routes=[
        Route("/public", lambda r: JSONResponse({"public": True})),
        Route("/protected", protected_endpoint),
    ],
    middleware=[
        Middleware(AuthenticationMiddleware, backend=TestAuthBackend()),
    ]
)

@requires("authenticated")
async def protected_endpoint(request):
    return JSONResponse({"user": request.user.display_name})

def test_public_endpoint():
    client = TestClient(app)
    response = client.get("/public")
    assert response.status_code == 200

def test_protected_without_auth():
    client = TestClient(app)
    response = client.get("/protected")
    assert response.status_code == 401

def test_protected_with_auth():
    client = TestClient(app)
    headers = {"Authorization": "Bearer valid-token"}
    response = client.get("/protected", headers=headers)
    
    assert response.status_code == 200
    assert response.json()["user"] == "testuser"

def test_invalid_token():
    client = TestClient(app)
    headers = {"Authorization": "Bearer invalid-token"}
    response = client.get("/protected", headers=headers)
    
    assert response.status_code == 401

WebSocket Testing

Basic WebSocket Testing

from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket

async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    # Echo messages
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"Echo: {message}")
    except WebSocketDisconnect:
        pass

app = Starlette(routes=[
    WebSocketRoute("/ws", websocket_endpoint),
])

def test_websocket():
    client = TestClient(app)
    
    with client.websocket_connect("/ws") as websocket:
        # Send message
        websocket.send_text("Hello")
        
        # Receive echo
        data = websocket.receive_text()
        assert data == "Echo: Hello"
        
        # Send another message
        websocket.send_text("World")
        data = websocket.receive_text()
        assert data == "Echo: World"

JSON WebSocket Testing

async def json_websocket(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            data = await websocket.receive_json()
            
            # Process different message types
            if data["type"] == "ping":
                await websocket.send_json({"type": "pong"})
            elif data["type"] == "echo":
                await websocket.send_json({
                    "type": "echo_response",
                    "original": data["message"]
                })
    except WebSocketDisconnect:
        pass

def test_json_websocket():
    client = TestClient(app)
    
    with client.websocket_connect("/ws/json") as websocket:
        # Test ping/pong
        websocket.send_json({"type": "ping"})
        response = websocket.receive_json()
        assert response["type"] == "pong"
        
        # Test echo
        websocket.send_json({
            "type": "echo", 
            "message": "Hello JSON"
        })
        response = websocket.receive_json()
        assert response["type"] == "echo_response"
        assert response["original"] == "Hello JSON"

WebSocket Authentication Testing

async def auth_websocket(websocket: WebSocket):
    # Check authentication
    token = websocket.query_params.get("token")
    if token != "valid-token":
        await websocket.close(code=1008, reason="Unauthorized")
        return
    
    await websocket.accept()
    await websocket.send_text("Authenticated successfully")

def test_websocket_auth():
    client = TestClient(app)
    
    # Test without token
    with pytest.raises(WebSocketDenialResponse):
        with client.websocket_connect("/ws/auth"):
            pass
    
    # Test with invalid token
    with pytest.raises(WebSocketDenialResponse):
        with client.websocket_connect("/ws/auth?token=invalid"):
            pass
    
    # Test with valid token
    with client.websocket_connect("/ws/auth?token=valid-token") as websocket:
        message = websocket.receive_text()
        assert message == "Authenticated successfully"

Testing Utilities and Helpers

Custom Test Client

class CustomTestClient(TestClient):
    """Extended test client with helper methods."""
    
    def __init__(self, app, **kwargs):
        super().__init__(app, **kwargs)
        self.auth_token = None
    
    def authenticate(self, token: str):
        """Set authentication token for subsequent requests."""
        self.auth_token = token
        self.headers["Authorization"] = f"Bearer {token}"
    
    def get_json(self, url: str, **kwargs) -> dict:
        """GET request expecting JSON response."""
        response = self.get(url, **kwargs)
        assert response.status_code == 200
        return response.json()
    
    def post_json(self, url: str, data: dict, **kwargs) -> dict:
        """POST JSON data and expect JSON response."""
        response = self.post(url, json=data, **kwargs)
        assert response.status_code in (200, 201)
        return response.json()
    
    def assert_status(self, url: str, expected_status: int, method: str = "GET"):
        """Assert endpoint returns expected status code."""
        response = getattr(self, method.lower())(url)
        assert response.status_code == expected_status

# Usage
def test_with_custom_client():
    client = CustomTestClient(app)
    
    # Test authentication
    client.authenticate("valid-token")
    user_data = client.get_json("/user/profile")
    assert "username" in user_data
    
    # Test status codes
    client.assert_status("/", 200)
    client.assert_status("/nonexistent", 404)

Test Fixtures

import pytest
import tempfile
import os

@pytest.fixture
def client():
    """Test client fixture."""
    return TestClient(app)

@pytest.fixture
def authenticated_client():
    """Authenticated test client fixture."""
    client = TestClient(app)
    client.headers["Authorization"] = "Bearer valid-token"
    return client

@pytest.fixture
def temp_upload_dir():
    """Temporary directory for file uploads."""
    with tempfile.TemporaryDirectory() as temp_dir:
        yield temp_dir

@pytest.fixture
def sample_file():
    """Sample file for upload testing."""
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
        f.write("Sample file content")
        temp_path = f.name
    
    yield temp_path
    
    # Cleanup
    os.unlink(temp_path)

# Usage
def test_with_fixtures(client, authenticated_client, sample_file):
    # Test public endpoint
    response = client.get("/public")
    assert response.status_code == 200
    
    # Test authenticated endpoint
    response = authenticated_client.get("/protected")
    assert response.status_code == 200
    
    # Test file upload
    with open(sample_file, 'rb') as f:
        response = authenticated_client.post(
            "/upload",
            files={"file": f}
        )
    assert response.status_code == 200

Mocking External Dependencies

import pytest
from unittest.mock import AsyncMock, patch

# Application with external dependency
async def external_api_endpoint(request):
    data = await external_api_call()
    return JSONResponse(data)

async def external_api_call():
    # This would normally make HTTP request to external service
    pass

def test_with_mocked_dependency():
    with patch('myapp.external_api_call') as mock_api:
        mock_api.return_value = {"mocked": True}
        
        client = TestClient(app)
        response = client.get("/external")
        
        assert response.status_code == 200
        assert response.json()["mocked"] is True
        mock_api.assert_called_once()

@pytest.fixture
def mock_database():
    """Mock database fixture."""
    with patch('myapp.database') as mock_db:
        mock_db.fetch_user.return_value = {
            "id": 1,
            "username": "testuser"
        }
        yield mock_db

def test_with_mocked_database(mock_database):
    client = TestClient(app)
    response = client.get("/users/1")
    
    assert response.status_code == 200
    mock_database.fetch_user.assert_called_with(1)

Performance Testing

Response Time Testing

import time
import statistics

def test_response_times():
    client = TestClient(app)
    
    times = []
    for _ in range(50):
        start = time.time()
        response = client.get("/")
        end = time.time()
        
        assert response.status_code == 200
        times.append(end - start)
    
    # Performance assertions
    avg_time = statistics.mean(times)
    max_time = max(times)
    p95_time = statistics.quantiles(times, n=20)[18]  # 95th percentile
    
    assert avg_time < 0.1, f"Average response time too high: {avg_time}"
    assert max_time < 0.5, f"Max response time too high: {max_time}"
    assert p95_time < 0.2, f"95th percentile too high: {p95_time}"

Concurrent Request Testing

import asyncio
import httpx

async def test_concurrent_requests():
    """Test application under concurrent load."""
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        # Make 100 concurrent requests
        tasks = []
        for i in range(100):
            task = client.get(f"/api/data?id={i}")
            tasks.append(task)
        
        # Wait for all requests to complete
        responses = await asyncio.gather(*tasks)
        
        # Check all responses succeeded
        for response in responses:
            assert response.status_code == 200
        
        print(f"Completed {len(responses)} concurrent requests")

Starlette's testing utilities provide comprehensive tools for testing all aspects of web applications, from simple endpoints to complex WebSocket interactions, with proper mocking, fixtures, and performance validation capabilities.

Install with Tessl CLI