0
# Testing
1

2
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.
3

4
## Capabilities
5

6
### Test Clients
7

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

10
```python { .api }
11
class TestClient:
12
    def __init__(
13
        self,
14
        app: Litestar,
15
        *,
16
        base_url: str = "http://testserver",
17
        backend: Literal["asyncio", "trio"] = "asyncio",
18
        backend_options: dict[str, Any] | None = None,
19
        cookies: httpx.Cookies | dict[str, str] | None = None,
20
        headers: dict[str, str] | None = None,
21
        follow_redirects: bool = False,
22
        **kwargs: Any,
23
    ):
24
        """
25
        Synchronous test client for Litestar applications.
26

27
        Parameters:
28
        - app: Litestar application instance
29
        - base_url: Base URL for requests
30
        - backend: Async backend to use
31
        - backend_options: Backend-specific options
32
        - cookies: Default cookies for requests
33
        - headers: Default headers for requests
34
        - follow_redirects: Whether to follow redirects automatically
35
        - **kwargs: Additional httpx.Client options
36
        """
37

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

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

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

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

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

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

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

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

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

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

70
class AsyncTestClient:
71
    def __init__(
72
        self,
73
        app: Litestar,
74
        *,
75
        base_url: str = "http://testserver",
76
        backend: Literal["asyncio", "trio"] = "asyncio",
77
        backend_options: dict[str, Any] | None = None,
78
        cookies: httpx.Cookies | dict[str, str] | None = None,
79
        headers: dict[str, str] | None = None,
80
        follow_redirects: bool = False,
81
        **kwargs: Any,
82
    ):
83
        """
84
        Asynchronous test client for Litestar applications.
85

86
        Parameters: Same as TestClient
87
        """
88

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

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

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

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

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

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

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

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

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

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

122
### WebSocket Test Session
123

124
Test utilities for WebSocket connections.
125

126
```python { .api }
127
class WebSocketTestSession:
128
    def __init__(self, client: TestClient | AsyncTestClient):
129
        """
130
        WebSocket test session.
131

132
        Parameters:
133
        - client: Test client instance
134
        """
135

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

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

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

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

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

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

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

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

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

164
### Test Client Factory Functions
165

166
Convenience functions for creating test clients.
167

168
```python { .api }
169
def create_test_client(
170
    route_handlers: ControllerRouterHandler | Sequence[ControllerRouterHandler],
171
    *,
172
    backend: Literal["asyncio", "trio"] = "asyncio",
173
    **litestar_kwargs: Any,
174
) -> TestClient:
175
    """
176
    Create a synchronous test client with route handlers.
177

178
    Parameters:
179
    - route_handlers: Route handlers to test
180
    - backend: Async backend to use
181
    - **litestar_kwargs: Additional Litestar constructor arguments
182

183
    Returns:
184
    Configured TestClient instance
185
    """
186

187
async def create_async_test_client(
188
    route_handlers: ControllerRouterHandler | Sequence[ControllerRouterHandler],
189
    *,
190
    backend: Literal["asyncio", "trio"] = "asyncio",
191
    **litestar_kwargs: Any,
192
) -> AsyncTestClient:
193
    """
194
    Create an asynchronous test client with route handlers.
195

196
    Parameters:
197
    - route_handlers: Route handlers to test
198
    - backend: Async backend to use  
199
    - **litestar_kwargs: Additional Litestar constructor arguments
200

201
    Returns:
202
    Configured AsyncTestClient instance
203
    """
204
```
205

206
### Request Factory
207

208
Factory for creating test request objects.
209

210
```python { .api }
211
class RequestFactory:
212
    def __init__(self, app: Litestar | None = None):
213
        """
214
        Request factory for creating test requests.
215

216
        Parameters:
217
        - app: Litestar application instance
218
        """
219

220
    def get(
221
        self,
222
        path: str = "/",
223
        *,
224
        headers: dict[str, str] | None = None,
225
        query_params: dict[str, str] | None = None,
226
        **kwargs: Any,
227
    ) -> Request:
228
        """Create a GET request."""
229

230
    def post(
231
        self,
232
        path: str = "/",
233
        *,
234
        headers: dict[str, str] | None = None,
235
        json: Any = None,
236
        data: dict[str, Any] | None = None,
237
        **kwargs: Any,
238
    ) -> Request:
239
        """Create a POST request."""
240

241
    def put(
242
        self,
243
        path: str = "/",
244
        **kwargs: Any,
245
    ) -> Request:
246
        """Create a PUT request."""
247

248
    def patch(
249
        self,
250
        path: str = "/",
251
        **kwargs: Any,
252
    ) -> Request:
253
        """Create a PATCH request."""
254

255
    def delete(
256
        self,
257
        path: str = "/",
258
        **kwargs: Any,
259
    ) -> Request:
260
        """Create a DELETE request."""
261

262
    def websocket(
263
        self,
264
        path: str = "/",
265
        *,
266
        headers: dict[str, str] | None = None,
267
        query_params: dict[str, str] | None = None,
268
        **kwargs: Any,
269
    ) -> WebSocket:
270
        """Create a WebSocket connection."""
271
```
272

273
### Subprocess Test Clients
274

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

277
```python { .api }
278
def subprocess_sync_client(
279
    app: Litestar,
280
    *,
281
    port: int = 0,
282
    host: str = "127.0.0.1",
283
    **kwargs: Any,
284
) -> TestClient:
285
    """
286
    Create a test client that runs the app in a subprocess.
287

288
    Parameters:
289
    - app: Litestar application
290
    - port: Port to bind to (0 for random port)
291
    - host: Host to bind to
292
    - **kwargs: Additional test client options
293

294
    Returns:
295
    TestClient connected to subprocess server
296
    """
297

298
def subprocess_async_client(
299
    app: Litestar,
300
    *,
301
    port: int = 0,
302
    host: str = "127.0.0.1",
303
    **kwargs: Any,
304
) -> AsyncTestClient:
305
    """
306
    Create an async test client that runs the app in a subprocess.
307

308
    Parameters:
309
    - app: Litestar application
310
    - port: Port to bind to (0 for random port)
311
    - host: Host to bind to
312
    - **kwargs: Additional test client options
313

314
    Returns:
315
    AsyncTestClient connected to subprocess server
316
    """
317
```
318

319
## Usage Examples
320

321
### Basic HTTP Testing
322

323
```python
324
import pytest
325
from litestar import Litestar, get, post
326
from litestar.testing import TestClient
327
from litestar.status_codes import HTTP_200_OK, HTTP_201_CREATED
328

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

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

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

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

345
def test_create_user():
346
    with TestClient(app=app) as client:
347
        response = client.post("/users", json={"name": "Alice"})
348
        assert response.status_code == HTTP_201_CREATED
349
        assert response.json() == {"id": 123, "name": "Alice"}
350
```
351

352
### Async Testing
353

354
```python
355
import pytest
356
from litestar.testing import AsyncTestClient
357

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

365
@pytest.mark.asyncio
366
async def test_async_create_user():
367
    async with AsyncTestClient(app=app) as client:
368
        response = await client.post("/users", json={"name": "Bob"})
369
        assert response.status_code == HTTP_201_CREATED
370
        assert response.json() == {"id": 123, "name": "Bob"}
371
```
372

373
### WebSocket Testing
374

375
```python
376
from litestar import websocket, WebSocket
377
from litestar.testing import create_test_client
378

379
@websocket("/ws")
380
async def websocket_handler(websocket: WebSocket) -> None:
381
    await websocket.accept()
382
    message = await websocket.receive_text()
383
    await websocket.send_text(f"Echo: {message}")
384
    await websocket.close()
385

386
def test_websocket():
387
    with create_test_client(route_handlers=[websocket_handler]) as client:
388
        with client.websocket_connect("/ws") as websocket:
389
            websocket.send_text("Hello WebSocket")
390
            data = websocket.receive_text()
391
            assert data == "Echo: Hello WebSocket"
392
```
393

394
### Authentication Testing
395

396
```python
397
from litestar.security.jwt import JWTAuth
398
from litestar.testing import create_test_client
399

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

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

410
def test_authenticated_endpoint():
411
    app = Litestar(
412
        route_handlers=[get_profile],
413
        on_app_init=[jwt_auth.on_app_init],
414
    )
415
    
416
    with TestClient(app=app) as client:
417
        # Get token
418
        token = jwt_auth.create_token(identifier="1")
419
        
420
        # Make authenticated request
421
        response = client.get(
422
            "/profile",
423
            headers={"Authorization": f"Bearer {token}"}
424
        )
425
        
426
        assert response.status_code == HTTP_200_OK
427
        assert response.json() == {"user": {"id": 1, "name": "Test User"}}
428

429
def test_unauthenticated_request():
430
    with TestClient(app=app) as client:
431
        response = client.get("/profile")
432
        assert response.status_code == 401
433
```
434

435
### Database Testing with Fixtures
436

437
```python
438
import pytest
439
from litestar import Litestar, get, post, Dependency
440
from litestar.testing import create_async_test_client
441
from unittest.mock import AsyncMock
442

443
# Mock database
444
class MockDatabase:
445
    def __init__(self):
446
        self.users = {}
447
        self.next_id = 1
448
    
449
    async def create_user(self, name: str) -> dict:
450
        user = {"id": self.next_id, "name": name}
451
        self.users[self.next_id] = user
452
        self.next_id += 1
453
        return user
454
    
455
    async def get_user(self, user_id: int) -> dict | None:
456
        return self.users.get(user_id)
457

458
@pytest.fixture
459
def mock_db():
460
    return MockDatabase()
461

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

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

474
    return Litestar(route_handlers=[create_user, get_user])
475

476
@pytest.mark.asyncio
477
async def test_user_crud(mock_db):
478
    app = create_app(mock_db)
479
    
480
    async with create_async_test_client(app) as client:
481
        # Create user
482
        response = await client.post("/users", json={"name": "Alice"})
483
        assert response.status_code == HTTP_201_CREATED
484
        user = response.json()
485
        assert user["name"] == "Alice"
486
        user_id = user["id"]
487
        
488
        # Get user
489
        response = await client.get(f"/users/{user_id}")
490
        assert response.status_code == HTTP_200_OK
491
        assert response.json() == user
492
        
493
        # Get non-existent user
494
        response = await client.get("/users/999")
495
        assert response.status_code == 404
496
```
497

498
### File Upload Testing
499

500
```python
501
from litestar import post, Request
502
from litestar.testing import TestClient
503
import io
504

505
@post("/upload")
506
async def upload_file(request: Request) -> dict:
507
    form_data = await request.form()
508
    uploaded_file = form_data["file"]
509
    return {
510
        "filename": uploaded_file.filename,
511
        "size": len(await uploaded_file.read()),
512
    }
513

514
def test_file_upload():
515
    with TestClient(app=Litestar(route_handlers=[upload_file])) as client:
516
        file_content = b"test file content"
517
        files = {"file": ("test.txt", io.BytesIO(file_content), "text/plain")}
518
        
519
        response = client.post("/upload", files=files)
520
        assert response.status_code == HTTP_200_OK
521
        assert response.json() == {
522
            "filename": "test.txt",
523
            "size": len(file_content)
524
        }
525
```
526

527
### Custom Headers and Cookies Testing
528

529
```python
530
def test_custom_headers():
531
    @get("/headers")
532
    def get_headers(request: Request) -> dict:
533
        return {"custom_header": request.headers.get("X-Custom-Header")}
534
    
535
    with TestClient(app=Litestar(route_handlers=[get_headers])) as client:
536
        response = client.get(
537
            "/headers",
538
            headers={"X-Custom-Header": "test-value"}
539
        )
540
        assert response.json() == {"custom_header": "test-value"}
541

542
def test_cookies():
543
    @get("/cookies")
544
    def get_cookies(request: Request) -> dict:
545
        return {"session_id": request.cookies.get("session_id")}
546
    
547
    with TestClient(app=Litestar(route_handlers=[get_cookies])) as client:
548
        response = client.get(
549
            "/cookies",
550
            cookies={"session_id": "abc123"}
551
        )
552
        assert response.json() == {"session_id": "abc123"}
553
```
554

555
### Integration Testing with Real Server
556

557
```python
558
from litestar.testing import subprocess_async_client
559
import pytest
560

561
@pytest.mark.asyncio
562
async def test_integration():
563
    """Test with real server running in subprocess."""
564
    app = Litestar(route_handlers=[health_check])
565
    
566
    async with subprocess_async_client(app) as client:
567
        response = await client.get("/health")
568
        assert response.status_code == HTTP_200_OK
569
        assert response.json() == {"status": "healthy"}
570
```
571

572
## Types
573

574
```python { .api }
575
# Test client types
576
TestClientBackend = Literal["asyncio", "trio"]
577

578
# HTTP response type from httpx
579
HTTPXResponse = httpx.Response
580

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

584
# Test context managers
585
TestClientContext = TestClient
586
AsyncTestClientContext = AsyncTestClient
587
```