0
# Testing Framework
1
2
Comprehensive testing tools for HTTP routes, WebSocket connections, and CLI commands with full async support, test fixtures, and extensive assertion capabilities.
3
4
## Capabilities
5
6
### HTTP Test Client
7
8
Async test client for testing HTTP routes with full request/response simulation.
9
10
```python { .api }
11
class QuartClient:
12
"""
13
HTTP test client for testing Quart applications.
14
15
Provides methods for making HTTP requests to test endpoints
16
with full async support and response inspection.
17
"""
18
19
async def get(
20
self,
21
path: str,
22
query_string: dict | str | None = None,
23
headers: dict | None = None,
24
**kwargs
25
):
26
"""
27
Make GET request to test endpoint.
28
29
Args:
30
path: Request path
31
query_string: Query parameters as dict or string
32
headers: Request headers
33
**kwargs: Additional request arguments
34
35
Returns:
36
Response object for inspection
37
"""
38
39
async def post(
40
self,
41
path: str,
42
data: bytes | str | dict | None = None,
43
json: dict | None = None,
44
form: dict | None = None,
45
files: dict | None = None,
46
headers: dict | None = None,
47
**kwargs
48
):
49
"""
50
Make POST request to test endpoint.
51
52
Args:
53
path: Request path
54
data: Raw request body data
55
json: JSON data (sets appropriate content-type)
56
form: Form data
57
files: File uploads
58
headers: Request headers
59
**kwargs: Additional request arguments
60
61
Returns:
62
Response object for inspection
63
"""
64
65
async def put(self, path: str, **kwargs):
66
"""Make PUT request to test endpoint."""
67
68
async def delete(self, path: str, **kwargs):
69
"""Make DELETE request to test endpoint."""
70
71
async def patch(self, path: str, **kwargs):
72
"""Make PATCH request to test endpoint."""
73
74
async def head(self, path: str, **kwargs):
75
"""Make HEAD request to test endpoint."""
76
77
async def options(self, path: str, **kwargs):
78
"""Make OPTIONS request to test endpoint."""
79
80
async def websocket(self, path: str, **kwargs):
81
"""
82
Create WebSocket test connection.
83
84
Args:
85
path: WebSocket path
86
**kwargs: Additional WebSocket arguments
87
88
Returns:
89
WebSocket test connection (async context manager)
90
"""
91
```
92
93
### CLI Test Runner
94
95
Test runner for CLI commands with argument parsing and output capture.
96
97
```python { .api }
98
class QuartCliRunner:
99
"""
100
CLI test runner for testing Quart CLI commands.
101
102
Provides interface for invoking CLI commands in test environment
103
with output capture and result inspection.
104
"""
105
106
def invoke(
107
self,
108
cli,
109
args: list[str] | None = None,
110
input: str | bytes | None = None,
111
env: dict | None = None,
112
catch_exceptions: bool = True,
113
**kwargs
114
):
115
"""
116
Invoke CLI command in test environment.
117
118
Args:
119
cli: Click command or group to invoke
120
args: Command line arguments
121
input: Standard input data
122
env: Environment variables
123
catch_exceptions: Whether to catch exceptions
124
**kwargs: Additional invoke arguments
125
126
Returns:
127
Click Result object with exit_code, output, and exception
128
"""
129
```
130
131
### Test Application Wrapper
132
133
Enhanced application instance for testing with additional test utilities.
134
135
```python { .api }
136
class TestApp(Quart):
137
"""
138
Test application wrapper with additional testing utilities.
139
140
Extends Quart application with test-specific methods and
141
simplified test client/runner creation.
142
"""
143
144
def test_client(self, **kwargs) -> QuartClient:
145
"""
146
Create HTTP test client for this application.
147
148
Args:
149
**kwargs: Additional client configuration
150
151
Returns:
152
QuartClient instance for testing HTTP endpoints
153
"""
154
155
def test_cli_runner(self, **kwargs) -> QuartCliRunner:
156
"""
157
Create CLI test runner for this application.
158
159
Args:
160
**kwargs: Additional runner configuration
161
162
Returns:
163
QuartCliRunner instance for testing CLI commands
164
"""
165
```
166
167
### Testing Utility Functions
168
169
Helper functions for creating test fixtures and mock data.
170
171
```python { .api }
172
def make_test_body_with_headers(
173
data: bytes | str | None = None,
174
form: dict | None = None,
175
files: dict | None = None
176
) -> tuple[bytes, dict]:
177
"""
178
Create test request body with appropriate headers.
179
180
Args:
181
data: Raw body data
182
form: Form data
183
files: File uploads
184
185
Returns:
186
Tuple of (body_bytes, headers_dict)
187
"""
188
189
def make_test_headers_path_and_query_string(
190
path: str,
191
headers: dict | None = None,
192
query_string: dict | str | None = None
193
) -> tuple[dict, str, bytes]:
194
"""
195
Create test headers, path, and query string.
196
197
Args:
198
path: Request path
199
headers: Request headers
200
query_string: Query parameters
201
202
Returns:
203
Tuple of (headers_dict, path_str, query_bytes)
204
"""
205
206
def make_test_scope(
207
method: str = "GET",
208
path: str = "/",
209
query_string: bytes = b"",
210
headers: list | None = None,
211
**kwargs
212
) -> dict:
213
"""
214
Create ASGI test scope dictionary.
215
216
Args:
217
method: HTTP method
218
path: Request path
219
query_string: Query string bytes
220
headers: Request headers as list of tuples
221
**kwargs: Additional ASGI scope fields
222
223
Returns:
224
ASGI scope dictionary
225
"""
226
227
async def no_op_push() -> None:
228
"""No-operation push promise function for testing."""
229
230
# Testing constants
231
sentinel: object
232
"""Sentinel object for testing placeholder values."""
233
```
234
235
### Testing Exceptions
236
237
Specialized exceptions for WebSocket testing scenarios.
238
239
```python { .api }
240
class WebsocketResponseError(Exception):
241
"""
242
Exception raised when WebSocket test encounters response error.
243
244
Used to indicate protocol violations or unexpected responses
245
during WebSocket testing.
246
"""
247
```
248
249
### Usage Examples
250
251
#### Basic HTTP Route Testing
252
253
```python
254
import pytest
255
from quart import Quart, jsonify, request
256
257
# Create test application
258
@pytest.fixture
259
async def app():
260
app = Quart(__name__)
261
262
@app.route('/')
263
async def index():
264
return 'Hello, World!'
265
266
@app.route('/json')
267
async def json_endpoint():
268
return jsonify({'message': 'Hello, JSON!'})
269
270
@app.route('/user/<int:user_id>')
271
async def get_user(user_id):
272
return jsonify({'id': user_id, 'name': f'User {user_id}'})
273
274
@app.route('/create', methods=['POST'])
275
async def create_item():
276
data = await request.get_json()
277
return jsonify({'created': data, 'id': 123}), 201
278
279
return app
280
281
@pytest.fixture
282
async def client(app):
283
return app.test_client()
284
285
# Test basic GET request
286
async def test_index(client):
287
response = await client.get('/')
288
assert response.status_code == 200
289
assert await response.get_data() == b'Hello, World!'
290
291
# Test JSON response
292
async def test_json_endpoint(client):
293
response = await client.get('/json')
294
assert response.status_code == 200
295
296
json_data = await response.get_json()
297
assert json_data == {'message': 'Hello, JSON!'}
298
299
# Test URL parameters
300
async def test_user_endpoint(client):
301
response = await client.get('/user/42')
302
assert response.status_code == 200
303
304
json_data = await response.get_json()
305
assert json_data['id'] == 42
306
assert json_data['name'] == 'User 42'
307
308
# Test POST with JSON data
309
async def test_create_item(client):
310
test_data = {'name': 'Test Item', 'value': 100}
311
312
response = await client.post('/create', json=test_data)
313
assert response.status_code == 201
314
315
json_data = await response.get_json()
316
assert json_data['created'] == test_data
317
assert json_data['id'] == 123
318
```
319
320
#### Advanced HTTP Testing
321
322
```python
323
import pytest
324
from quart import Quart, request, session, jsonify
325
326
@pytest.fixture
327
async def app():
328
app = Quart(__name__)
329
app.secret_key = 'test-secret-key'
330
331
@app.route('/upload', methods=['POST'])
332
async def upload_file():
333
files = await request.files
334
uploaded_file = files.get('document')
335
336
if uploaded_file and uploaded_file.filename:
337
content = await uploaded_file.read()
338
return jsonify({
339
'filename': uploaded_file.filename,
340
'size': len(content),
341
'content_type': uploaded_file.content_type
342
})
343
344
return jsonify({'error': 'No file uploaded'}), 400
345
346
@app.route('/form', methods=['POST'])
347
async def process_form():
348
form_data = await request.form
349
return jsonify(dict(form_data))
350
351
@app.route('/session/set/<key>/<value>')
352
async def set_session(key, value):
353
session[key] = value
354
return jsonify({'set': {key: value}})
355
356
@app.route('/session/get/<key>')
357
async def get_session(key):
358
value = session.get(key)
359
return jsonify({key: value})
360
361
@app.route('/headers')
362
async def check_headers():
363
return jsonify({
364
'user_agent': request.headers.get('User-Agent'),
365
'custom_header': request.headers.get('X-Custom-Header')
366
})
367
368
return app
369
370
# Test file upload
371
async def test_file_upload(client):
372
file_data = b'This is test file content'
373
374
response = await client.post('/upload', files={
375
'document': (io.BytesIO(file_data), 'test.txt', 'text/plain')
376
})
377
378
assert response.status_code == 200
379
json_data = await response.get_json()
380
assert json_data['filename'] == 'test.txt'
381
assert json_data['size'] == len(file_data)
382
assert json_data['content_type'] == 'text/plain'
383
384
# Test form data
385
async def test_form_submission(client):
386
form_data = {'username': 'testuser', 'email': 'test@example.com'}
387
388
response = await client.post('/form', form=form_data)
389
assert response.status_code == 200
390
391
json_data = await response.get_json()
392
assert json_data == form_data
393
394
# Test session handling
395
async def test_session(client):
396
# Set session value
397
response = await client.get('/session/set/user_id/123')
398
assert response.status_code == 200
399
400
# Get session value
401
response = await client.get('/session/get/user_id')
402
assert response.status_code == 200
403
404
json_data = await response.get_json()
405
assert json_data['user_id'] == '123'
406
407
# Test custom headers
408
async def test_custom_headers(client):
409
response = await client.get('/headers', headers={
410
'User-Agent': 'Test Client 1.0',
411
'X-Custom-Header': 'CustomValue'
412
})
413
414
assert response.status_code == 200
415
json_data = await response.get_json()
416
assert json_data['user_agent'] == 'Test Client 1.0'
417
assert json_data['custom_header'] == 'CustomValue'
418
419
# Test query parameters
420
async def test_query_parameters(client):
421
response = await client.get('/search', query_string={
422
'q': 'test query',
423
'page': 2,
424
'limit': 50
425
})
426
427
# Or using string format
428
response = await client.get('/search?q=test+query&page=2&limit=50')
429
```
430
431
#### WebSocket Testing
432
433
```python
434
import pytest
435
from quart import Quart, websocket
436
import json
437
438
@pytest.fixture
439
async def app():
440
app = Quart(__name__)
441
442
@app.websocket('/ws/echo')
443
async def echo_websocket():
444
await websocket.accept()
445
446
while True:
447
try:
448
message = await websocket.receive()
449
await websocket.send(f'Echo: {message}')
450
except ConnectionClosed:
451
break
452
453
@app.websocket('/ws/json')
454
async def json_websocket():
455
await websocket.accept()
456
457
while True:
458
try:
459
data = await websocket.receive_json()
460
response = {
461
'type': 'response',
462
'original': data,
463
'timestamp': time.time()
464
}
465
await websocket.send_json(response)
466
except ConnectionClosed:
467
break
468
469
@app.websocket('/ws/auth')
470
async def auth_websocket():
471
# Check authentication
472
token = websocket.args.get('token')
473
if token != 'valid_token':
474
await websocket.close(code=1008, reason='Authentication failed')
475
return
476
477
await websocket.accept()
478
await websocket.send('Authenticated successfully')
479
480
while True:
481
try:
482
message = await websocket.receive()
483
await websocket.send(f'Authenticated echo: {message}')
484
except ConnectionClosed:
485
break
486
487
return app
488
489
# Test basic WebSocket echo
490
async def test_websocket_echo(client):
491
async with client.websocket('/ws/echo') as ws:
492
await ws.send('Hello, WebSocket!')
493
response = await ws.receive()
494
assert response == 'Echo: Hello, WebSocket!'
495
496
# Test JSON WebSocket communication
497
async def test_websocket_json(client):
498
async with client.websocket('/ws/json') as ws:
499
test_data = {'message': 'Hello', 'value': 42}
500
await ws.send_json(test_data)
501
502
response = await ws.receive_json()
503
assert response['type'] == 'response'
504
assert response['original'] == test_data
505
assert 'timestamp' in response
506
507
# Test WebSocket authentication
508
async def test_websocket_auth_success(client):
509
async with client.websocket('/ws/auth', query_string={'token': 'valid_token'}) as ws:
510
welcome = await ws.receive()
511
assert welcome == 'Authenticated successfully'
512
513
await ws.send('Test message')
514
response = await ws.receive()
515
assert response == 'Authenticated echo: Test message'
516
517
async def test_websocket_auth_failure(client):
518
# Test authentication failure
519
with pytest.raises(WebsocketResponseError):
520
async with client.websocket('/ws/auth', query_string={'token': 'invalid_token'}) as ws:
521
# Should raise exception due to authentication failure
522
pass
523
524
# Test WebSocket connection handling
525
async def test_websocket_multiple_messages(client):
526
async with client.websocket('/ws/echo') as ws:
527
messages = ['Message 1', 'Message 2', 'Message 3']
528
529
for message in messages:
530
await ws.send(message)
531
response = await ws.receive()
532
assert response == f'Echo: {message}'
533
```
534
535
#### CLI Command Testing
536
537
```python
538
import pytest
539
import click
540
from quart import Quart
541
542
@pytest.fixture
543
async def app():
544
app = Quart(__name__)
545
546
@app.cli.command()
547
@click.argument('name')
548
def greet(name):
549
"""Greet someone."""
550
click.echo(f'Hello, {name}!')
551
552
@app.cli.command()
553
@click.option('--count', default=1, help='Number of greetings')
554
@click.argument('name')
555
def multi_greet(count, name):
556
"""Greet someone multiple times."""
557
for i in range(count):
558
click.echo(f'{i+1}. Hello, {name}!')
559
560
@app.cli.command()
561
def database_init():
562
"""Initialize database."""
563
click.echo('Initializing database...')
564
# Simulate database initialization
565
click.echo('Database initialized successfully!')
566
567
return app
568
569
# Test basic CLI command
570
async def test_greet_command(app):
571
runner = app.test_cli_runner()
572
573
result = runner.invoke(app.cli, ['greet', 'World'])
574
assert result.exit_code == 0
575
assert 'Hello, World!' in result.output
576
577
# Test CLI command with options
578
async def test_multi_greet_command(app):
579
runner = app.test_cli_runner()
580
581
result = runner.invoke(app.cli, ['multi-greet', '--count', '3', 'Alice'])
582
assert result.exit_code == 0
583
assert '1. Hello, Alice!' in result.output
584
assert '2. Hello, Alice!' in result.output
585
assert '3. Hello, Alice!' in result.output
586
587
# Test CLI command without arguments
588
async def test_database_init_command(app):
589
runner = app.test_cli_runner()
590
591
result = runner.invoke(app.cli, ['database-init'])
592
assert result.exit_code == 0
593
assert 'Initializing database...' in result.output
594
assert 'Database initialized successfully!' in result.output
595
596
# Test CLI command error handling
597
async def test_cli_error_handling(app):
598
runner = app.test_cli_runner()
599
600
# Test missing required argument
601
result = runner.invoke(app.cli, ['greet'])
602
assert result.exit_code != 0
603
assert 'Error' in result.output
604
```
605
606
#### Integration Testing
607
608
```python
609
import pytest
610
from quart import Quart, render_template, request, jsonify
611
612
@pytest.fixture
613
async def app():
614
app = Quart(__name__)
615
app.config['TESTING'] = True
616
617
# Mock database
618
users_db = {
619
1: {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
620
2: {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
621
}
622
623
@app.route('/api/users')
624
async def list_users():
625
return jsonify(list(users_db.values()))
626
627
@app.route('/api/users/<int:user_id>')
628
async def get_user(user_id):
629
user = users_db.get(user_id)
630
if not user:
631
return jsonify({'error': 'User not found'}), 404
632
return jsonify(user)
633
634
@app.route('/api/users', methods=['POST'])
635
async def create_user():
636
data = await request.get_json()
637
638
# Validate required fields
639
if not data or 'name' not in data or 'email' not in data:
640
return jsonify({'error': 'Name and email required'}), 400
641
642
# Create new user
643
new_id = max(users_db.keys()) + 1 if users_db else 1
644
user = {
645
'id': new_id,
646
'name': data['name'],
647
'email': data['email']
648
}
649
users_db[new_id] = user
650
651
return jsonify(user), 201
652
653
@app.route('/api/users/<int:user_id>', methods=['DELETE'])
654
async def delete_user(user_id):
655
if user_id not in users_db:
656
return jsonify({'error': 'User not found'}), 404
657
658
del users_db[user_id]
659
return '', 204
660
661
return app
662
663
# Test complete CRUD operations
664
async def test_user_crud_operations(client):
665
# List users (should have initial data)
666
response = await client.get('/api/users')
667
assert response.status_code == 200
668
users = await response.get_json()
669
assert len(users) == 2
670
671
# Get specific user
672
response = await client.get('/api/users/1')
673
assert response.status_code == 200
674
user = await response.get_json()
675
assert user['name'] == 'Alice'
676
677
# Create new user
678
new_user = {'name': 'Charlie', 'email': 'charlie@example.com'}
679
response = await client.post('/api/users', json=new_user)
680
assert response.status_code == 201
681
created_user = await response.get_json()
682
assert created_user['name'] == 'Charlie'
683
assert created_user['id'] == 3
684
685
# Verify user was created
686
response = await client.get('/api/users')
687
users = await response.get_json()
688
assert len(users) == 3
689
690
# Delete user
691
response = await client.delete('/api/users/3')
692
assert response.status_code == 204
693
694
# Verify user was deleted
695
response = await client.get('/api/users/3')
696
assert response.status_code == 404
697
698
# Test error conditions
699
async def test_error_conditions(client):
700
# Test getting non-existent user
701
response = await client.get('/api/users/999')
702
assert response.status_code == 404
703
error = await response.get_json()
704
assert 'error' in error
705
706
# Test creating user with missing data
707
response = await client.post('/api/users', json={'name': 'Incomplete'})
708
assert response.status_code == 400
709
710
# Test deleting non-existent user
711
response = await client.delete('/api/users/999')
712
assert response.status_code == 404
713
```