0
# Testing Utilities
1

2
BlackSheep provides comprehensive testing utilities for unit testing, integration testing, and end-to-end testing of web applications. The testing framework supports async testing, request simulation, and complete application testing without network overhead.
3

4
## TestClient
5

6
The `TestClient` class is the primary testing utility that allows you to make HTTP requests to your application without starting a server.
7

8
### Basic Testing Setup
9

10
```python { .api }
11
import pytest
12
import asyncio
13
from blacksheep.testing import TestClient
14
from blacksheep import Application, json, Request, FromJSON
15

16
# Sample application for testing
17
app = Application()
18

19
@app.get("/")
20
async def home():
21
    return json({"message": "Hello, World!"})
22

23
@app.get("/users/{user_id:int}")
24
async def get_user(user_id: int):
25
    return json({"id": user_id, "name": f"User {user_id}"})
26

27
@app.post("/users")
28
async def create_user(data: FromJSON[dict]):
29
    return json({"created": True, "data": data.value})
30

31
# Basic test example
32
async def test_basic_endpoints():
33
    client = TestClient(app)
34
    
35
    # Test GET endpoint
36
    response = await client.get("/")
37
    assert response.status == 200
38
    data = await response.json()
39
    assert data["message"] == "Hello, World!"
40
    
41
    # Test GET with parameters
42
    response = await client.get("/users/123")
43
    assert response.status == 200
44
    data = await response.json()
45
    assert data["id"] == 123
46
    
47
    # Test POST with JSON
48
    response = await client.post("/users", json={"name": "Alice"})
49
    assert response.status == 200
50
    data = await response.json()
51
    assert data["created"] is True
52

53
# Run test
54
asyncio.run(test_basic_endpoints())
55
```
56

57
### PyTest Integration
58

59
```python { .api }
60
import pytest
61
from blacksheep.testing import TestClient
62

63
@pytest.fixture
64
def client():
65
    """Create test client fixture"""
66
    return TestClient(app)
67

68
@pytest.mark.asyncio
69
async def test_home_endpoint(client):
70
    response = await client.get("/")
71
    assert response.status == 200
72
    
73
    data = await response.json()
74
    assert "message" in data
75

76
@pytest.mark.asyncio  
77
async def test_user_creation(client):
78
    user_data = {"name": "Alice", "email": "alice@example.com"}
79
    response = await client.post("/users", json=user_data)
80
    
81
    assert response.status == 200
82
    result = await response.json()
83
    assert result["created"] is True
84
    assert result["data"]["name"] == "Alice"
85

86
@pytest.mark.asyncio
87
async def test_not_found(client):
88
    response = await client.get("/nonexistent")
89
    assert response.status == 404
90
```
91

92
### Request Methods
93

94
```python { .api }
95
from blacksheep.testing import TestClient
96
from blacksheep import JSONContent, TextContent, FormContent
97

98
async def test_http_methods():
99
    client = TestClient(app)
100
    
101
    # GET request
102
    response = await client.get("/users")
103
    assert response.status == 200
104
    
105
    # GET with query parameters
106
    response = await client.get("/users", params={"page": 1, "limit": 10})
107
    # Equivalent to: /users?page=1&limit=10
108
    
109
    # POST request with JSON
110
    response = await client.post("/users", json={"name": "Bob"})
111
    
112
    # POST with custom content
113
    json_content = JSONContent({"user": {"name": "Charlie"}})
114
    response = await client.post("/users", content=json_content)
115
    
116
    # POST with form data
117
    response = await client.post("/login", data={"username": "user", "password": "pass"})
118
    
119
    # PUT request
120
    response = await client.put("/users/123", json={"name": "Updated Name"})
121
    
122
    # DELETE request
123
    response = await client.delete("/users/123")
124
    
125
    # PATCH request
126
    response = await client.patch("/users/123", json={"email": "new@example.com"})
127
    
128
    # HEAD request
129
    response = await client.head("/users/123")
130
    assert response.status == 200
131
    # No response body for HEAD requests
132
    
133
    # OPTIONS request
134
    response = await client.options("/users")
135
```
136

137
### Headers and Cookies
138

139
```python { .api }
140
async def test_headers_and_cookies():
141
    client = TestClient(app)
142
    
143
    # Custom headers
144
    headers = {"Authorization": "Bearer token123", "X-API-Version": "1.0"}
145
    response = await client.get("/protected", headers=headers)
146
    
147
    # Cookies
148
    cookies = {"session_id": "abc123", "theme": "dark"}
149
    response = await client.get("/dashboard", cookies=cookies)
150
    
151
    # Both headers and cookies
152
    response = await client.get(
153
        "/api/data",
154
        headers={"Authorization": "Bearer token"},
155
        cookies={"session": "value"}
156
    )
157
    
158
    # Check response headers
159
    assert response.headers.get_first(b"Content-Type") == b"application/json"
160
    
161
    # Check response cookies
162
    set_cookie_header = response.headers.get_first(b"Set-Cookie")
163
    if set_cookie_header:
164
        print(f"Server set cookie: {set_cookie_header.decode()}")
165
```
166

167
## File Upload Testing
168

169
Test file upload endpoints with multipart form data.
170

171
### File Upload Tests
172

173
```python { .api }
174
from blacksheep import MultiPartFormData, FormPart
175
from io import BytesIO
176

177
# Application with file upload
178
@app.post("/upload")
179
async def upload_file(files: FromFiles):
180
    uploaded_files = files.value
181
    return json({
182
        "files_received": len(uploaded_files),
183
        "filenames": [f.file_name.decode() if f.file_name else "unknown" 
184
                     for f in uploaded_files]
185
    })
186

187
async def test_file_upload():
188
    client = TestClient(app)
189
    
190
    # Create test file content
191
    file_content = b"Test file content"
192
    
193
    # Create form parts
194
    text_part = FormPart(b"title", b"My Document")
195
    file_part = FormPart(
196
        name=b"document",
197
        data=file_content,
198
        content_type=b"text/plain",
199
        file_name=b"test.txt"
200
    )
201
    
202
    # Create multipart form data
203
    multipart = MultiPartFormData([text_part, file_part])
204
    
205
    # Upload file
206
    response = await client.post("/upload", content=multipart)
207
    assert response.status == 200
208
    
209
    result = await response.json()
210
    assert result["files_received"] == 1
211
    assert "test.txt" in result["filenames"]
212

213
# Alternative: Upload with files parameter
214
async def test_file_upload_simple():
215
    client = TestClient(app)
216
    
217
    # Simple file upload
218
    file_data = BytesIO(b"Simple file content")
219
    response = await client.post(
220
        "/upload",
221
        files={"document": ("simple.txt", file_data, "text/plain")}
222
    )
223
    
224
    assert response.status == 200
225
```
226

227
## Authentication Testing
228

229
Test authentication and authorization features.
230

231
### Authentication Tests
232

233
```python { .api }
234
from blacksheep import auth, allow_anonymous
235
from blacksheep.server.authentication.jwt import JWTBearerAuthentication
236
from guardpost import Identity
237

238
# Application with auth
239
auth_app = Application()
240
auth_strategy = auth_app.use_authentication()
241

242
@auth_app.get("/public")
243
@allow_anonymous
244
async def public_endpoint():
245
    return json({"public": True})
246

247
@auth_app.get("/protected")
248
@auth()
249
async def protected_endpoint(request: Request):
250
    identity = request.identity
251
    return json({"user_id": identity.id, "authenticated": True})
252

253
async def test_authentication():
254
    client = TestClient(auth_app)
255
    
256
    # Test public endpoint
257
    response = await client.get("/public")
258
    assert response.status == 200
259
    
260
    # Test protected endpoint without auth (should fail)
261
    response = await client.get("/protected")
262
    assert response.status == 401
263
    
264
    # Test protected endpoint with auth
265
    headers = {"Authorization": "Bearer valid-jwt-token"}
266
    response = await client.get("/protected", headers=headers)
267
    # Note: This will depend on your JWT validation logic
268
```
269

270
### Mock Authentication
271

272
```python { .api }
273
from unittest.mock import AsyncMock, patch
274

275
async def test_mock_authentication():
276
    client = TestClient(auth_app)
277
    
278
    # Mock the authentication handler
279
    mock_identity = Identity({"sub": "user123", "name": "Test User"})
280
    
281
    with patch.object(auth_strategy, 'authenticate', return_value=mock_identity):
282
        response = await client.get("/protected")
283
        assert response.status == 200
284
        
285
        data = await response.json()
286
        assert data["user_id"] == "user123"
287
        assert data["authenticated"] is True
288
```
289

290
## Database Testing
291

292
Test applications that use databases with proper setup and teardown.
293

294
### Database Test Setup
295

296
```python { .api }
297
import pytest
298
from blacksheep import Application, FromServices
299
from rodi import Container
300

301
# Sample service and repository
302
class UserRepository:
303
    def __init__(self):
304
        self.users = {}
305
        self.next_id = 1
306
    
307
    async def create_user(self, user_data: dict) -> dict:
308
        user = {"id": self.next_id, **user_data}
309
        self.users[self.next_id] = user
310
        self.next_id += 1
311
        return user
312
    
313
    async def get_user(self, user_id: int) -> dict:
314
        return self.users.get(user_id)
315

316
# Application with database
317
db_app = Application()
318

319
# Configure DI
320
container = Container()
321
container.add_singleton(UserRepository)
322
db_app.services = container
323

324
@db_app.post("/users")
325
async def create_user_endpoint(
326
    data: FromJSON[dict],
327
    repo: FromServices[UserRepository]
328
):
329
    user = await repo.create_user(data.value)
330
    return json(user)
331

332
@db_app.get("/users/{user_id:int}")
333
async def get_user_endpoint(
334
    user_id: int,
335
    repo: FromServices[UserRepository]
336
):
337
    user = await repo.get_user(user_id)
338
    if not user:
339
        return Response(404)
340
    return json(user)
341

342
@pytest.fixture
343
async def db_client():
344
    """Create test client with fresh database"""
345
    # Create fresh application instance for each test
346
    test_app = Application()
347
    test_container = Container()
348
    test_container.add_singleton(UserRepository)
349
    test_app.services = test_container
350
    
351
    # Re-register routes (in real app, you'd organize this better)
352
    test_app.post("/users")(create_user_endpoint)
353
    test_app.get("/users/{user_id:int}")(get_user_endpoint)
354
    
355
    return TestClient(test_app)
356

357
@pytest.mark.asyncio
358
async def test_user_crud(db_client):
359
    # Create user
360
    user_data = {"name": "Alice", "email": "alice@example.com"}
361
    response = await db_client.post("/users", json=user_data)
362
    assert response.status == 200
363
    
364
    created_user = await response.json()
365
    user_id = created_user["id"]
366
    
367
    # Get user
368
    response = await db_client.get(f"/users/{user_id}")
369
    assert response.status == 200
370
    
371
    retrieved_user = await response.json()
372
    assert retrieved_user["name"] == "Alice"
373
    assert retrieved_user["email"] == "alice@example.com"
374
    
375
    # Get non-existent user
376
    response = await db_client.get("/users/999")
377
    assert response.status == 404
378
```
379

380
## WebSocket Testing
381

382
Test WebSocket endpoints and real-time functionality.
383

384
### WebSocket Test Setup
385

386
```python { .api }
387
from blacksheep.testing import TestClient
388
from blacksheep import WebSocket, WebSocketState
389

390
# WebSocket application
391
ws_app = Application()
392

393
@ws_app.ws("/echo")
394
async def echo_handler(websocket: WebSocket):
395
    await websocket.accept()
396
    
397
    try:
398
        while True:
399
            message = await websocket.receive_text()
400
            await websocket.send_text(f"Echo: {message}")
401
    except WebSocketDisconnectError:
402
        pass
403

404
@ws_app.ws("/json-echo")
405
async def json_echo_handler(websocket: WebSocket):
406
    await websocket.accept()
407
    
408
    try:
409
        while True:
410
            data = await websocket.receive_json()
411
            await websocket.send_json({"echo": data})
412
    except WebSocketDisconnectError:
413
        pass
414

415
async def test_websocket_echo():
416
    client = TestClient(ws_app)
417
    
418
    # Test WebSocket connection
419
    async with client.websocket("/echo") as websocket:
420
        # Send message
421
        await websocket.send_text("Hello WebSocket!")
422
        
423
        # Receive echo
424
        response = await websocket.receive_text()
425
        assert response == "Echo: Hello WebSocket!"
426
        
427
        # Send another message
428
        await websocket.send_text("Second message")
429
        response = await websocket.receive_text()
430
        assert response == "Echo: Second message"
431

432
async def test_websocket_json():
433
    client = TestClient(ws_app)
434
    
435
    async with client.websocket("/json-echo") as websocket:
436
        # Send JSON data
437
        test_data = {"message": "Hello", "count": 42}
438
        await websocket.send_json(test_data)
439
        
440
        # Receive JSON response
441
        response = await websocket.receive_json()
442
        assert response["echo"]["message"] == "Hello"
443
        assert response["echo"]["count"] == 42
444
```
445

446
## Mock Objects
447

448
BlackSheep provides mock objects for testing ASGI applications and components.
449

450
### ASGI Mocks
451

452
```python { .api }
453
from blacksheep.testing import MockReceive, MockSend
454

455
async def test_asgi_mocks():
456
    # Mock ASGI components
457
    mock_receive = MockReceive()
458
    mock_send = MockSend()
459
    
460
    # Add messages to mock receive
461
    mock_receive.add_message({
462
        "type": "http.request",
463
        "body": b'{"test": "data"}',
464
        "more_body": False
465
    })
466
    
467
    # Test ASGI application directly
468
    scope = {
469
        "type": "http",
470
        "method": "POST",
471
        "path": "/test",
472
        "headers": [(b"content-type", b"application/json")]
473
    }
474
    
475
    # Call application ASGI interface
476
    await app(scope, mock_receive, mock_send)
477
    
478
    # Check sent messages
479
    sent_messages = mock_send.messages
480
    assert len(sent_messages) > 0
481
    
482
    # Verify response
483
    response_start = sent_messages[0]
484
    assert response_start["type"] == "http.response.start"
485
    assert response_start["status"] == 200
486
```
487

488
## Integration Testing
489

490
Test complete application workflows and integrations.
491

492
### Full Application Test
493

494
```python { .api }
495
import pytest
496
from blacksheep.testing import TestClient
497

498
# Complete application for integration testing
499
integration_app = Application()
500

501
# Add CORS
502
integration_app.use_cors(allow_origins="*")
503

504
# Add authentication
505
auth_strategy = integration_app.use_authentication()
506

507
# Add routes
508
@integration_app.get("/")
509
async def home():
510
    return json({"app": "integration_test", "version": "1.0"})
511

512
@integration_app.get("/health")
513
async def health_check():
514
    # Simulate health check logic
515
    return json({"status": "healthy", "timestamp": time.time()})
516

517
@integration_app.post("/api/process")
518
async def process_data(data: FromJSON[dict]):
519
    # Simulate data processing
520
    result = {
521
        "processed": True,
522
        "input_size": len(str(data.value)),
523
        "output": f"Processed: {data.value}"
524
    }
525
    return json(result)
526

527
class TestIntegrationWorkflow:
528
    
529
    @pytest.fixture
530
    def client(self):
531
        return TestClient(integration_app)
532
    
533
    @pytest.mark.asyncio
534
    async def test_complete_workflow(self, client):
535
        # Test health check
536
        response = await client.get("/health")
537
        assert response.status == 200
538
        health = await response.json()
539
        assert health["status"] == "healthy"
540
        
541
        # Test main endpoint
542
        response = await client.get("/")
543
        assert response.status == 200
544
        info = await response.json()
545
        assert info["app"] == "integration_test"
546
        
547
        # Test data processing
548
        test_data = {"items": [1, 2, 3], "name": "test"}
549
        response = await client.post("/api/process", json=test_data)
550
        assert response.status == 200
551
        
552
        result = await response.json()
553
        assert result["processed"] is True
554
        assert result["input_size"] > 0
555
    
556
    @pytest.mark.asyncio
557
    async def test_cors_headers(self, client):
558
        # Test CORS preflight
559
        response = await client.options(
560
            "/api/process",
561
            headers={"Origin": "https://example.com"}
562
        )
563
        
564
        # Check CORS headers
565
        cors_origin = response.headers.get_first(b"Access-Control-Allow-Origin")
566
        assert cors_origin == b"*"
567
    
568
    @pytest.mark.asyncio
569
    async def test_error_handling(self, client):
570
        # Test 404
571
        response = await client.get("/nonexistent")
572
        assert response.status == 404
573
        
574
        # Test invalid JSON
575
        response = await client.post(
576
            "/api/process",
577
            content=TextContent("invalid json"),
578
            headers={"Content-Type": "application/json"}
579
        )
580
        # Should handle parsing error appropriately
581
```
582

583
## Performance Testing
584

585
Basic performance testing with the test client.
586

587
### Load Testing
588

589
```python { .api }
590
import asyncio
591
import time
592
from statistics import mean, stdev
593

594
async def performance_test():
595
    client = TestClient(app)
596
    
597
    # Warm up
598
    for _ in range(10):
599
        await client.get("/")
600
    
601
    # Performance test
602
    iterations = 100
603
    start_times = []
604
    
605
    for i in range(iterations):
606
        start = time.time()
607
        response = await client.get("/")
608
        end = time.time()
609
        
610
        assert response.status == 200
611
        start_times.append((end - start) * 1000)  # Convert to ms
612
    
613
    # Calculate statistics
614
    avg_time = mean(start_times)
615
    std_dev = stdev(start_times) if len(start_times) > 1 else 0
616
    min_time = min(start_times)
617
    max_time = max(start_times)
618
    
619
    print(f"Performance Results ({iterations} iterations):")
620
    print(f"  Average: {avg_time:.2f}ms")
621
    print(f"  Std Dev: {std_dev:.2f}ms")
622
    print(f"  Min: {min_time:.2f}ms")
623
    print(f"  Max: {max_time:.2f}ms")
624
    
625
    # Assert performance threshold
626
    assert avg_time < 50  # Under 50ms average
627

628
# Concurrent request testing
629
async def concurrent_test():
630
    client = TestClient(app)
631
    
632
    async def make_request():
633
        response = await client.get("/")
634
        return response.status
635
    
636
    # Run 20 concurrent requests
637
    tasks = [make_request() for _ in range(20)]
638
    results = await asyncio.gather(*tasks)
639
    
640
    # All should succeed
641
    assert all(status == 200 for status in results)
642
    print(f"Concurrent test: {len(results)} requests completed successfully")
643

644
asyncio.run(performance_test())
645
asyncio.run(concurrent_test())
646
```
647

648
## Test Configuration
649

650
Configure testing environment and fixtures.
651

652
### Test Configuration
653

654
```python { .api }
655
import pytest
656
import os
657
from blacksheep import Application
658
from blacksheep.testing import TestClient
659

660
# Test configuration
661
@pytest.fixture(scope="session")
662
def test_config():
663
    return {
664
        "DEBUG": True,
665
        "TESTING": True,
666
        "DATABASE_URL": "sqlite:///:memory:",
667
        "SECRET_KEY": "test-secret-key"
668
    }
669

670
@pytest.fixture
671
def app_with_config(test_config):
672
    """Create application with test configuration"""
673
    app = Application(debug=test_config["DEBUG"])
674
    
675
    # Configure for testing
676
    app.configuration = test_config
677
    
678
    # Add test routes
679
    @app.get("/config")
680
    async def get_config():
681
        return json({"debug": app.debug})
682
    
683
    return app
684

685
@pytest.fixture
686
def client(app_with_config):
687
    """Test client with configured application"""
688
    return TestClient(app_with_config)
689

690
# Environment-specific tests
691
@pytest.mark.asyncio
692
async def test_debug_mode(client):
693
    response = await client.get("/config")
694
    data = await response.json()
695
    assert data["debug"] is True
696

697
# Skip tests based on conditions
698
@pytest.mark.skipif(os.getenv("CI") is None, reason="Only run in CI")
699
async def test_ci_specific():
700
    # Test that only runs in CI environment
701
    pass
702
```
703

704
BlackSheep's testing utilities provide comprehensive support for testing all aspects of your web application, from simple unit tests to complex integration scenarios, ensuring your application works correctly and performs well.