0
# Testing Utilities
1

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

4
## TestClient Class
5

6
```python { .api }
7
from starlette.testclient import TestClient
8
from starlette.types import ASGIApp
9
from httpx import Client
10
from typing import Any, Dict, Optional, Union, Mapping
11
import httpx
12

13
class TestClient(httpx.Client):
14
    """
15
    Test client for ASGI applications.
16
    
17
    Built on HTTPX for modern async HTTP testing with:
18
    - Automatic lifespan management
19
    - WebSocket testing support
20
    - File upload testing
21
    - Cookie and session management
22
    - Request/response inspection
23
    """
24
    
25
    def __init__(
26
        self,
27
        app: ASGIApp,
28
        base_url: str = "http://testserver",
29
        raise_server_exceptions: bool = True,
30
        root_path: str = "",
31
        backend: str = "asyncio",
32
        backend_options: Optional[Dict[str, Any]] = None,
33
        cookies: httpx.Cookies = None,
34
        headers: Mapping[str, str] = None,
35
        follow_redirects: bool = True,
36
        client: tuple[str, int] = ("testclient", 50000),
37
    ) -> None:
38
        """
39
        Initialize test client.
40
        
41
        Args:
42
            app: ASGI application to test
43
            base_url: Base URL for requests
44
            raise_server_exceptions: Raise server exceptions in tests
45
            root_path: ASGI root path
46
            backend: Async backend ("asyncio" or "trio")
47
            backend_options: Backend-specific options
48
            cookies: Default cookies for requests
49
            headers: Default headers for requests
50
            follow_redirects: Automatically follow redirects
51
            client: Client address tuple (host, port)
52
        """
53
    
54
    # HTTP Methods (inherited from httpx.Client)
55
    def get(
56
        self,
57
        url: str,
58
        *,
59
        params: Dict[str, Any] = None,
60
        headers: Mapping[str, str] = None,
61
        cookies: httpx.Cookies = None,
62
        auth: httpx.Auth = None,
63
        follow_redirects: bool = None,
64
        timeout: Union[float, httpx.Timeout] = None,
65
    ) -> httpx.Response:
66
        """Send GET request."""
67
    
68
    def post(
69
        self,
70
        url: str,
71
        *,
72
        content: Union[str, bytes] = None,
73
        data: Dict[str, Any] = None,
74
        files: Dict[str, Any] = None,
75
        json: Any = None,
76
        params: Dict[str, Any] = None,
77
        headers: Mapping[str, str] = None,
78
        cookies: httpx.Cookies = None,
79
        auth: httpx.Auth = None,
80
        follow_redirects: bool = None,
81
        timeout: Union[float, httpx.Timeout] = None,
82
    ) -> httpx.Response:
83
        """Send POST request."""
84
    
85
    def put(self, url: str, **kwargs) -> httpx.Response:
86
        """Send PUT request."""
87
    
88
    def patch(self, url: str, **kwargs) -> httpx.Response:
89
        """Send PATCH request."""
90
    
91
    def delete(self, url: str, **kwargs) -> httpx.Response:
92
        """Send DELETE request."""
93
    
94
    def head(self, url: str, **kwargs) -> httpx.Response:
95
        """Send HEAD request."""
96
    
97
    def options(self, url: str, **kwargs) -> httpx.Response:
98
        """Send OPTIONS request."""
99
    
100
    # WebSocket testing
101
    def websocket_connect(
102
        self,
103
        url: str,
104
        subprotocols: List[str] = None,
105
        **kwargs
106
    ) -> WebSocketTestSession:
107
        """
108
        Connect to WebSocket endpoint.
109
        
110
        Args:
111
            url: WebSocket URL
112
            subprotocols: List of subprotocols to negotiate
113
            **kwargs: Additional connection parameters
114
            
115
        Returns:
116
            WebSocketTestSession: Test session for WebSocket interaction
117
        """
118
    
119
    # Context manager support for lifespan events
120
    def __enter__(self) -> "TestClient":
121
        """Enter context manager and run lifespan startup."""
122
    
123
    def __exit__(self, *args) -> None:
124
        """Exit context manager and run lifespan shutdown."""
125
```
126

127
## WebSocket Testing
128

129
```python { .api }
130
from starlette.testclient import WebSocketTestSession
131
from starlette.websockets import WebSocketDisconnect
132
from typing import Any, Dict
133

134
class WebSocketTestSession:
135
    """
136
    WebSocket test session for testing WebSocket endpoints.
137
    
138
    Provides methods for sending and receiving WebSocket messages
139
    in test scenarios with proper connection management.
140
    """
141
    
142
    def send(self, message: Dict[str, Any]) -> None:
143
        """Send raw WebSocket message."""
144
    
145
    def send_text(self, data: str) -> None:
146
        """Send text message."""
147
    
148
    def send_bytes(self, data: bytes) -> None:
149
        """Send binary message."""
150
    
151
    def send_json(self, data: Any, mode: str = "text") -> None:
152
        """Send JSON message."""
153
    
154
    def receive(self) -> Dict[str, Any]:
155
        """Receive raw WebSocket message."""
156
    
157
    def receive_text(self) -> str:
158
        """Receive text message."""
159
    
160
    def receive_bytes(self) -> bytes:
161
        """Receive binary message."""
162
    
163
    def receive_json(self, mode: str = "text") -> Any:
164
        """Receive JSON message."""
165
    
166
    def close(self, code: int = 1000, reason: str = None) -> None:
167
        """Close WebSocket connection."""
168

169
class WebSocketDenialResponse(httpx.Response, WebSocketDisconnect):
170
    """
171
    Exception raised when WebSocket connection is denied.
172
    
173
    Contains the HTTP response that was sent instead of
174
    accepting the WebSocket connection.
175
    """
176
    pass
177
```
178

179
## Basic Testing
180

181
### Simple HTTP Endpoint Testing
182

183
```python { .api }
184
from starlette.applications import Starlette
185
from starlette.routing import Route
186
from starlette.responses import JSONResponse
187
from starlette.testclient import TestClient
188

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

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

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

202
# Test functions
203
def test_homepage():
204
    client = TestClient(app)
205
    response = client.get("/")
206
    
207
    assert response.status_code == 200
208
    assert response.json() == {"message": "Hello World"}
209

210
def test_user_detail():
211
    client = TestClient(app)
212
    response = client.get("/users/123")
213
    
214
    assert response.status_code == 200
215
    assert response.json() == {"user_id": 123}
216

217
def test_not_found():
218
    client = TestClient(app)
219
    response = client.get("/nonexistent")
220
    
221
    assert response.status_code == 404
222
```
223

224
### Testing Different HTTP Methods
225

226
```python { .api }
227
async def api_endpoint(request):
228
    if request.method == "GET":
229
        return JSONResponse({"method": "GET"})
230
    elif request.method == "POST":
231
        data = await request.json()
232
        return JSONResponse({"method": "POST", "data": data})
233
    elif request.method == "PUT":
234
        data = await request.json()
235
        return JSONResponse({"method": "PUT", "updated": data})
236

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

241
def test_http_methods():
242
    client = TestClient(app)
243
    
244
    # Test GET
245
    response = client.get("/api")
246
    assert response.status_code == 200
247
    assert response.json()["method"] == "GET"
248
    
249
    # Test POST
250
    response = client.post("/api", json={"name": "test"})
251
    assert response.status_code == 200
252
    assert response.json()["method"] == "POST"
253
    assert response.json()["data"]["name"] == "test"
254
    
255
    # Test PUT
256
    response = client.put("/api", json={"id": 1, "name": "updated"})
257
    assert response.status_code == 200
258
    assert response.json()["method"] == "PUT"
259
```
260

261
### Testing Request Data
262

263
```python { .api }
264
async def form_endpoint(request):
265
    form = await request.form()
266
    return JSONResponse({
267
        "name": form.get("name"),
268
        "email": form.get("email"),
269
        "file_uploaded": bool(form.get("file"))
270
    })
271

272
def test_form_data():
273
    client = TestClient(app)
274
    
275
    # Test form data
276
    response = client.post("/form", data={
277
        "name": "John Doe",
278
        "email": "john@example.com"
279
    })
280
    
281
    assert response.status_code == 200
282
    assert response.json()["name"] == "John Doe"
283
    assert response.json()["email"] == "john@example.com"
284

285
def test_file_upload():
286
    client = TestClient(app)
287
    
288
    # Test file upload
289
    with open("test_file.txt", "w") as f:
290
        f.write("test content")
291
    
292
    with open("test_file.txt", "rb") as f:
293
        response = client.post("/form", 
294
            data={"name": "John"},
295
            files={"file": ("test.txt", f, "text/plain")}
296
        )
297
    
298
    assert response.status_code == 200
299
    assert response.json()["file_uploaded"] is True
300
```
301

302
## Advanced Testing
303

304
### Testing with Lifespan Events
305

306
```python { .api }
307
from contextlib import asynccontextmanager
308

309
# Application with lifespan
310
@asynccontextmanager
311
async def lifespan(app):
312
    # Startup
313
    app.state.database = {"connected": True}
314
    print("Database connected")
315
    
316
    yield
317
    
318
    # Shutdown  
319
    app.state.database = {"connected": False}
320
    print("Database disconnected")
321

322
app = Starlette(
323
    lifespan=lifespan,
324
    routes=routes
325
)
326

327
def test_with_lifespan():
328
    # Using context manager automatically handles lifespan
329
    with TestClient(app) as client:
330
        # Lifespan startup has run
331
        assert hasattr(client.app.state, "database")
332
        
333
        response = client.get("/")
334
        assert response.status_code == 200
335
        
336
    # Lifespan shutdown has run
337

338
def test_without_lifespan():
339
    # Create client without context manager
340
    client = TestClient(app)
341
    
342
    # Lifespan events don't run automatically
343
    response = client.get("/health")  # Simple endpoint
344
    assert response.status_code == 200
345
    
346
    # Manual cleanup
347
    client.close()
348
```
349

350
### Testing Middleware
351

352
```python { .api }
353
from starlette.middleware.base import BaseHTTPMiddleware
354
from starlette.middleware import Middleware
355
import time
356

357
class TimingMiddleware(BaseHTTPMiddleware):
358
    async def dispatch(self, request, call_next):
359
        start_time = time.time()
360
        response = await call_next(request)
361
        process_time = time.time() - start_time
362
        response.headers["X-Process-Time"] = str(process_time)
363
        return response
364

365
app = Starlette(
366
    routes=routes,
367
    middleware=[
368
        Middleware(TimingMiddleware),
369
    ]
370
)
371

372
def test_middleware():
373
    client = TestClient(app)
374
    response = client.get("/")
375
    
376
    # Check middleware added header
377
    assert "X-Process-Time" in response.headers
378
    
379
    # Verify timing is reasonable
380
    process_time = float(response.headers["X-Process-Time"])
381
    assert process_time > 0
382
    assert process_time < 1.0  # Should be under 1 second
383
```
384

385
### Testing Authentication
386

387
```python { .api }
388
from starlette.middleware.authentication import AuthenticationMiddleware
389
from starlette.authentication import AuthenticationBackend, AuthCredentials, SimpleUser
390

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

399
app = Starlette(
400
    routes=[
401
        Route("/public", lambda r: JSONResponse({"public": True})),
402
        Route("/protected", protected_endpoint),
403
    ],
404
    middleware=[
405
        Middleware(AuthenticationMiddleware, backend=TestAuthBackend()),
406
    ]
407
)
408

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

413
def test_public_endpoint():
414
    client = TestClient(app)
415
    response = client.get("/public")
416
    assert response.status_code == 200
417

418
def test_protected_without_auth():
419
    client = TestClient(app)
420
    response = client.get("/protected")
421
    assert response.status_code == 401
422

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

431
def test_invalid_token():
432
    client = TestClient(app)
433
    headers = {"Authorization": "Bearer invalid-token"}
434
    response = client.get("/protected", headers=headers)
435
    
436
    assert response.status_code == 401
437
```
438

439
## WebSocket Testing
440

441
### Basic WebSocket Testing
442

443
```python { .api }
444
from starlette.routing import WebSocketRoute
445
from starlette.websockets import WebSocket
446

447
async def websocket_endpoint(websocket: WebSocket):
448
    await websocket.accept()
449
    
450
    # Echo messages
451
    try:
452
        while True:
453
            message = await websocket.receive_text()
454
            await websocket.send_text(f"Echo: {message}")
455
    except WebSocketDisconnect:
456
        pass
457

458
app = Starlette(routes=[
459
    WebSocketRoute("/ws", websocket_endpoint),
460
])
461

462
def test_websocket():
463
    client = TestClient(app)
464
    
465
    with client.websocket_connect("/ws") as websocket:
466
        # Send message
467
        websocket.send_text("Hello")
468
        
469
        # Receive echo
470
        data = websocket.receive_text()
471
        assert data == "Echo: Hello"
472
        
473
        # Send another message
474
        websocket.send_text("World")
475
        data = websocket.receive_text()
476
        assert data == "Echo: World"
477
```
478

479
### JSON WebSocket Testing
480

481
```python { .api }
482
async def json_websocket(websocket: WebSocket):
483
    await websocket.accept()
484
    
485
    try:
486
        while True:
487
            data = await websocket.receive_json()
488
            
489
            # Process different message types
490
            if data["type"] == "ping":
491
                await websocket.send_json({"type": "pong"})
492
            elif data["type"] == "echo":
493
                await websocket.send_json({
494
                    "type": "echo_response",
495
                    "original": data["message"]
496
                })
497
    except WebSocketDisconnect:
498
        pass
499

500
def test_json_websocket():
501
    client = TestClient(app)
502
    
503
    with client.websocket_connect("/ws/json") as websocket:
504
        # Test ping/pong
505
        websocket.send_json({"type": "ping"})
506
        response = websocket.receive_json()
507
        assert response["type"] == "pong"
508
        
509
        # Test echo
510
        websocket.send_json({
511
            "type": "echo", 
512
            "message": "Hello JSON"
513
        })
514
        response = websocket.receive_json()
515
        assert response["type"] == "echo_response"
516
        assert response["original"] == "Hello JSON"
517
```
518

519
### WebSocket Authentication Testing
520

521
```python { .api }
522
async def auth_websocket(websocket: WebSocket):
523
    # Check authentication
524
    token = websocket.query_params.get("token")
525
    if token != "valid-token":
526
        await websocket.close(code=1008, reason="Unauthorized")
527
        return
528
    
529
    await websocket.accept()
530
    await websocket.send_text("Authenticated successfully")
531

532
def test_websocket_auth():
533
    client = TestClient(app)
534
    
535
    # Test without token
536
    with pytest.raises(WebSocketDenialResponse):
537
        with client.websocket_connect("/ws/auth"):
538
            pass
539
    
540
    # Test with invalid token
541
    with pytest.raises(WebSocketDenialResponse):
542
        with client.websocket_connect("/ws/auth?token=invalid"):
543
            pass
544
    
545
    # Test with valid token
546
    with client.websocket_connect("/ws/auth?token=valid-token") as websocket:
547
        message = websocket.receive_text()
548
        assert message == "Authenticated successfully"
549
```
550

551
## Testing Utilities and Helpers
552

553
### Custom Test Client
554

555
```python { .api }
556
class CustomTestClient(TestClient):
557
    """Extended test client with helper methods."""
558
    
559
    def __init__(self, app, **kwargs):
560
        super().__init__(app, **kwargs)
561
        self.auth_token = None
562
    
563
    def authenticate(self, token: str):
564
        """Set authentication token for subsequent requests."""
565
        self.auth_token = token
566
        self.headers["Authorization"] = f"Bearer {token}"
567
    
568
    def get_json(self, url: str, **kwargs) -> dict:
569
        """GET request expecting JSON response."""
570
        response = self.get(url, **kwargs)
571
        assert response.status_code == 200
572
        return response.json()
573
    
574
    def post_json(self, url: str, data: dict, **kwargs) -> dict:
575
        """POST JSON data and expect JSON response."""
576
        response = self.post(url, json=data, **kwargs)
577
        assert response.status_code in (200, 201)
578
        return response.json()
579
    
580
    def assert_status(self, url: str, expected_status: int, method: str = "GET"):
581
        """Assert endpoint returns expected status code."""
582
        response = getattr(self, method.lower())(url)
583
        assert response.status_code == expected_status
584

585
# Usage
586
def test_with_custom_client():
587
    client = CustomTestClient(app)
588
    
589
    # Test authentication
590
    client.authenticate("valid-token")
591
    user_data = client.get_json("/user/profile")
592
    assert "username" in user_data
593
    
594
    # Test status codes
595
    client.assert_status("/", 200)
596
    client.assert_status("/nonexistent", 404)
597
```
598

599
### Test Fixtures
600

601
```python { .api }
602
import pytest
603
import tempfile
604
import os
605

606
@pytest.fixture
607
def client():
608
    """Test client fixture."""
609
    return TestClient(app)
610

611
@pytest.fixture
612
def authenticated_client():
613
    """Authenticated test client fixture."""
614
    client = TestClient(app)
615
    client.headers["Authorization"] = "Bearer valid-token"
616
    return client
617

618
@pytest.fixture
619
def temp_upload_dir():
620
    """Temporary directory for file uploads."""
621
    with tempfile.TemporaryDirectory() as temp_dir:
622
        yield temp_dir
623

624
@pytest.fixture
625
def sample_file():
626
    """Sample file for upload testing."""
627
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt') as f:
628
        f.write("Sample file content")
629
        temp_path = f.name
630
    
631
    yield temp_path
632
    
633
    # Cleanup
634
    os.unlink(temp_path)
635

636
# Usage
637
def test_with_fixtures(client, authenticated_client, sample_file):
638
    # Test public endpoint
639
    response = client.get("/public")
640
    assert response.status_code == 200
641
    
642
    # Test authenticated endpoint
643
    response = authenticated_client.get("/protected")
644
    assert response.status_code == 200
645
    
646
    # Test file upload
647
    with open(sample_file, 'rb') as f:
648
        response = authenticated_client.post(
649
            "/upload",
650
            files={"file": f}
651
        )
652
    assert response.status_code == 200
653
```
654

655
### Mocking External Dependencies
656

657
```python { .api }
658
import pytest
659
from unittest.mock import AsyncMock, patch
660

661
# Application with external dependency
662
async def external_api_endpoint(request):
663
    data = await external_api_call()
664
    return JSONResponse(data)
665

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

670
def test_with_mocked_dependency():
671
    with patch('myapp.external_api_call') as mock_api:
672
        mock_api.return_value = {"mocked": True}
673
        
674
        client = TestClient(app)
675
        response = client.get("/external")
676
        
677
        assert response.status_code == 200
678
        assert response.json()["mocked"] is True
679
        mock_api.assert_called_once()
680

681
@pytest.fixture
682
def mock_database():
683
    """Mock database fixture."""
684
    with patch('myapp.database') as mock_db:
685
        mock_db.fetch_user.return_value = {
686
            "id": 1,
687
            "username": "testuser"
688
        }
689
        yield mock_db
690

691
def test_with_mocked_database(mock_database):
692
    client = TestClient(app)
693
    response = client.get("/users/1")
694
    
695
    assert response.status_code == 200
696
    mock_database.fetch_user.assert_called_with(1)
697
```
698

699
## Performance Testing
700

701
### Response Time Testing
702

703
```python { .api }
704
import time
705
import statistics
706

707
def test_response_times():
708
    client = TestClient(app)
709
    
710
    times = []
711
    for _ in range(50):
712
        start = time.time()
713
        response = client.get("/")
714
        end = time.time()
715
        
716
        assert response.status_code == 200
717
        times.append(end - start)
718
    
719
    # Performance assertions
720
    avg_time = statistics.mean(times)
721
    max_time = max(times)
722
    p95_time = statistics.quantiles(times, n=20)[18]  # 95th percentile
723
    
724
    assert avg_time < 0.1, f"Average response time too high: {avg_time}"
725
    assert max_time < 0.5, f"Max response time too high: {max_time}"
726
    assert p95_time < 0.2, f"95th percentile too high: {p95_time}"
727
```
728

729
### Concurrent Request Testing
730

731
```python { .api }
732
import asyncio
733
import httpx
734

735
async def test_concurrent_requests():
736
    """Test application under concurrent load."""
737
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
738
        # Make 100 concurrent requests
739
        tasks = []
740
        for i in range(100):
741
            task = client.get(f"/api/data?id={i}")
742
            tasks.append(task)
743
        
744
        # Wait for all requests to complete
745
        responses = await asyncio.gather(*tasks)
746
        
747
        # Check all responses succeeded
748
        for response in responses:
749
            assert response.status_code == 200
750
        
751
        print(f"Completed {len(responses)} concurrent requests")
752
```
753

754
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.