0
# Testing
1

2
Comprehensive testing infrastructure for aiohttp applications including test servers, test clients, pytest integration, and utilities for both client and server testing. Provides complete testing capabilities with proper async support and resource management.
3

4
## Capabilities
5

6
### Test Server Infrastructure
7

8
Test server implementations for testing web applications and request handlers in controlled environments.
9

10
```python { .api }
11
class BaseTestServer:
12
    def __init__(self, *, scheme='http', host='127.0.0.1', port=None, **kwargs):
13
        """
14
        Base test server.
15
        
16
        Parameters:
17
        - scheme (str): URL scheme (http/https)
18
        - host (str): Server host
19
        - port (int): Server port (auto-assigned if None)
20
        """
21
    
22
    async def start_server(self, **kwargs):
23
        """Start test server."""
24
    
25
    async def close(self):
26
        """Close test server."""
27
    
28
    @property
29
    def host(self):
30
        """Server host."""
31
    
32
    @property
33
    def port(self):
34
        """Server port."""
35
    
36
    @property 
37
    def scheme(self):
38
        """URL scheme."""
39
    
40
    def make_url(self, path):
41
        """Create URL for path."""
42

43
class TestServer(BaseTestServer):
44
    def __init__(self, app, *, port=None, **kwargs):
45
        """
46
        Test server for web applications.
47
        
48
        Parameters:
49
        - app: Web application instance
50
        - port (int): Server port
51
        """
52

53
class RawTestServer(BaseTestServer):
54
    def __init__(self, handler, *, port=None, **kwargs):
55
        """
56
        Raw test server for request handlers.
57
        
58
        Parameters:
59
        - handler: Request handler function
60
        - port (int): Server port
61
        """
62
```
63

64
### Test Client
65

66
HTTP client specifically designed for testing with automatic server integration and simplified request methods.
67

68
```python { .api }
69
class TestClient:
70
    def __init__(self, server, *, cookie_jar=None, **kwargs):
71
        """
72
        Test HTTP client.
73
        
74
        Parameters:
75
        - server: Test server instance
76
        - cookie_jar: Cookie jar for client
77
        """
78
    
79
    async def start_server(self):
80
        """Start associated test server."""
81
    
82
    async def close(self):
83
        """Close client and server."""
84
    
85
    def make_url(self, path):
86
        """Create URL for path."""
87
    
88
    async def request(self, method, path, **kwargs):
89
        """Make HTTP request to test server."""
90
    
91
    async def get(self, path, **kwargs):
92
        """Make GET request."""
93
    
94
    async def post(self, path, **kwargs):
95
        """Make POST request."""
96
    
97
    async def put(self, path, **kwargs):
98
        """Make PUT request."""
99
    
100
    async def patch(self, path, **kwargs):
101
        """Make PATCH request."""
102
    
103
    async def delete(self, path, **kwargs):
104
        """Make DELETE request."""
105
    
106
    async def head(self, path, **kwargs):
107
        """Make HEAD request."""
108
    
109
    async def options(self, path, **kwargs):
110
        """Make OPTIONS request."""
111
    
112
    async def ws_connect(self, path, **kwargs):
113
        """Establish WebSocket connection."""
114
    
115
    @property
116
    def session(self):
117
        """Underlying HTTP session."""
118
    
119
    @property
120
    def host(self):
121
        """Server host."""
122
    
123
    @property
124
    def port(self):
125
        """Server port."""
126
```
127

128
### Test Case Base Class
129

130
Base test case class providing utilities and setup for aiohttp application testing.
131

132
```python { .api }
133
class AioHTTPTestCase:
134
    """Base test case for aiohttp applications."""
135
    
136
    async def get_application(self):
137
        """
138
        Create application for testing.
139
        Must be implemented by subclasses.
140
        
141
        Returns:
142
        Application: Web application instance
143
        """
144
    
145
    async def setUpAsync(self):
146
        """Async setup method."""
147
    
148
    async def tearDownAsync(self):
149
        """Async teardown method."""
150
    
151
    async def get_server(self, app):
152
        """
153
        Create test server for application.
154
        
155
        Parameters:
156
        - app: Web application
157
        
158
        Returns:
159
        TestServer: Test server instance
160
        """
161
    
162
    async def get_client(self, server):
163
        """
164
        Create test client for server.
165
        
166
        Parameters:
167
        - server: Test server
168
        
169
        Returns:  
170
        TestClient: Test client instance
171
        """
172
    
173
    @property
174
    def client(self):
175
        """Test client instance."""
176
    
177
    @property
178
    def server(self):
179
        """Test server instance."""
180
    
181
    @property
182
    def app(self):
183
        """Web application instance."""
184
```
185

186
### Testing Utilities
187

188
Utility functions for creating mocked objects and managing test environments.
189

190
```python { .api }
191
def unused_port():
192
    """
193
    Get unused TCP port.
194
    
195
    Returns:
196
    int: Available port number
197
    """
198

199
def make_mocked_request(
200
    method,
201
    path,
202
    headers=None,
203
    *,
204
    version=None,
205
    closing=False,
206
    app=None,
207
    writer=None,
208
    protocol=None,
209
    transport=None,
210
    payload=None,
211
    sslcontext=None,
212
    client_max_size=1024**2,
213
    loop=None,
214
    match_info=None
215
):
216
    """
217
    Create mocked request for testing.
218
    
219
    Parameters:
220
    - method (str): HTTP method
221
    - path (str): Request path
222
    - headers (dict): Request headers
223
    - version: HTTP version
224
    - closing (bool): Connection closing
225
    - app: Application instance
226
    - writer: Response writer
227
    - protocol: Request protocol
228
    - transport: Network transport
229
    - payload: Request payload
230
    - sslcontext: SSL context
231
    - client_max_size (int): Max client payload size
232
    - loop: Event loop
233
    - match_info: URL match info
234
    
235
    Returns:
236
    Request: Mocked request object
237
    """
238

239
def make_mocked_coro(return_value=None, raise_exception=None):
240
    """
241
    Create mocked coroutine.
242
    
243
    Parameters:
244
    - return_value: Value to return from coroutine
245
    - raise_exception: Exception to raise from coroutine
246
    
247
    Returns:
248
    Mock: Mocked coroutine function
249
    """
250

251
def setup_test_loop(loop_factory=None):
252
    """
253
    Setup event loop for testing.
254
    
255
    Parameters:
256
    - loop_factory: Factory function for creating loop
257
    
258
    Returns:
259
    Event loop instance
260
    """
261

262
def teardown_test_loop(loop, fast=False):
263
    """
264
    Teardown test event loop.
265
    
266
    Parameters:
267
    - loop: Event loop to teardown
268
    - fast (bool): Fast teardown mode
269
    """
270

271
def loop_context(loop_factory=None, fast=False):
272
    """
273
    Context manager for test event loop.
274
    
275
    Parameters:
276
    - loop_factory: Factory function for creating loop
277
    - fast (bool): Fast teardown mode
278
    
279
    Returns:
280
    Context manager yielding event loop
281
    """
282

283
def unittest_run_loop(func):
284
    """
285
    Decorator to run test function in event loop.
286
    
287
    Parameters:
288
    - func: Test function to run
289
    
290
    Returns:
291
    Wrapped test function
292
    """
293
```
294

295
### Pytest Integration
296

297
Pytest fixtures and utilities for seamless integration with pytest testing framework.
298

299
```python { .api }
300
# Pytest fixtures (available when using aiohttp.pytest_plugin)
301

302
def loop():
303
    """
304
    Event loop fixture.
305
    
306
    Yields:
307
    Event loop for test
308
    """
309

310
def unused_port():
311
    """
312
    Unused port fixture.
313
    
314
    Returns:
315
    int: Available port number
316
    """
317

318
def aiohttp_client():
319
    """
320
    Test client factory fixture.
321
    
322
    Returns:
323
    Function: Factory function for creating test clients
324
    """
325

326
def aiohttp_server():
327
    """
328
    Test server factory fixture.
329
    
330
    Returns:
331
    Function: Factory function for creating test servers
332
    """
333

334
def aiohttp_raw_server():
335
    """
336
    Raw test server factory fixture.
337
    
338
    Returns:
339
    Function: Factory function for creating raw test servers
340
    """
341
```
342

343
## Usage Examples
344

345
### Basic Application Testing
346

347
```python
348
import pytest
349
from aiohttp import web
350
import aiohttp
351
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
352

353
class TestWebApp(AioHTTPTestCase):
354
    async def get_application(self):
355
        """Create test application."""
356
        async def hello(request):
357
            return web.Response(text='Hello, World!')
358
        
359
        async def json_handler(request):
360
            return web.json_response({'message': 'Hello, JSON!'})
361
        
362
        app = web.Application()
363
        app.router.add_get('/', hello)
364
        app.router.add_get('/json', json_handler)
365
        return app
366
    
367
    @unittest_run_loop
368
    async def test_hello(self):
369
        """Test hello endpoint."""
370
        resp = await self.client.request('GET', '/')
371
        self.assertEqual(resp.status, 200)
372
        text = await resp.text()
373
        self.assertEqual(text, 'Hello, World!')
374
    
375
    @unittest_run_loop
376
    async def test_json(self):
377
        """Test JSON endpoint."""
378
        resp = await self.client.request('GET', '/json')
379
        self.assertEqual(resp.status, 200)
380
        data = await resp.json()
381
        self.assertEqual(data['message'], 'Hello, JSON!')
382
```
383

384
### Pytest-based Testing
385

386
```python
387
import pytest
388
from aiohttp import web
389

390
async def hello_handler(request):
391
    name = request.query.get('name', 'World')
392
    return web.Response(text=f'Hello, {name}!')
393

394
async def create_app():
395
    app = web.Application()
396
    app.router.add_get('/hello', hello_handler)
397
    return app
398

399
@pytest.fixture
400
async def client(aiohttp_client):
401
    """Create test client."""
402
    app = await create_app()
403
    return await aiohttp_client(app)
404

405
async def test_hello_default(client):
406
    """Test hello with default name."""
407
    resp = await client.get('/hello')
408
    assert resp.status == 200
409
    text = await resp.text()
410
    assert text == 'Hello, World!'
411

412
async def test_hello_custom_name(client):
413
    """Test hello with custom name.""" 
414
    resp = await client.get('/hello?name=Alice')
415
    assert resp.status == 200
416
    text = await resp.text()
417
    assert text == 'Hello, Alice!'
418

419
async def test_hello_post_method(client):
420
    """Test unsupported POST method."""
421
    resp = await client.post('/hello')
422
    assert resp.status == 405  # Method Not Allowed
423
```
424

425
### WebSocket Testing
426

427
```python
428
import pytest
429
from aiohttp import web, WSMsgType
430
import json
431

432
async def websocket_handler(request):
433
    ws = web.WebSocketResponse()
434
    await ws.prepare(request)
435
    
436
    async for msg in ws:
437
        if msg.type == WSMsgType.TEXT:
438
            data = json.loads(msg.data)
439
            if data['type'] == 'echo':
440
                await ws.send_json({
441
                    'type': 'echo_response',
442
                    'message': data['message']
443
                })
444
        elif msg.type == WSMsgType.ERROR:
445
            print(f'WebSocket error: {ws.exception()}')
446
    
447
    return ws
448

449
@pytest.fixture
450
async def websocket_client(aiohttp_client):
451
    """Create WebSocket test client."""
452
    app = web.Application()
453
    app.router.add_get('/ws', websocket_handler)
454
    client = await aiohttp_client(app)
455
    return client
456

457
async def test_websocket_echo(websocket_client):
458
    """Test WebSocket echo functionality."""
459
    async with websocket_client.ws_connect('/ws') as ws:
460
        # Send echo request
461
        await ws.send_json({
462
            'type': 'echo',
463
            'message': 'Hello, WebSocket!'
464
        })
465
        
466
        # Receive echo response
467
        msg = await ws.receive()
468
        assert msg.type == WSMsgType.TEXT
469
        
470
        data = json.loads(msg.data)
471
        assert data['type'] == 'echo_response'
472
        assert data['message'] == 'Hello, WebSocket!'
473
```
474

475
### API Testing with Database
476

477
```python
478
import pytest
479
from aiohttp import web
480
import aiohttp
481
import asyncio
482
import sqlite3
483
from contextlib import asynccontextmanager
484

485
class UserService:
486
    def __init__(self, db_path=':memory:'):
487
        self.db_path = db_path
488
        self.connection = None
489
    
490
    async def initialize(self):
491
        """Initialize database."""
492
        # In real applications, use aiopg, aiomysql, etc.
493
        self.connection = sqlite3.connect(self.db_path)
494
        self.connection.execute('''
495
            CREATE TABLE IF NOT EXISTS users (
496
                id INTEGER PRIMARY KEY,
497
                name TEXT NOT NULL,
498
                email TEXT UNIQUE NOT NULL
499
            )
500
        ''')
501
        self.connection.commit()
502
    
503
    async def create_user(self, name, email):
504
        """Create new user."""
505
        cursor = self.connection.execute(
506
            'INSERT INTO users (name, email) VALUES (?, ?)',
507
            (name, email)
508
        )
509
        self.connection.commit()
510
        return cursor.lastrowid
511
    
512
    async def get_user(self, user_id):
513
        """Get user by ID."""
514
        cursor = self.connection.execute(
515
            'SELECT id, name, email FROM users WHERE id = ?',
516
            (user_id,)
517
        )
518
        row = cursor.fetchone()
519
        if row:
520
            return {'id': row[0], 'name': row[1], 'email': row[2]}
521
        return None
522
    
523
    async def close(self):
524
        """Close database connection."""
525
        if self.connection:
526
            self.connection.close()
527

528
async def create_user_handler(request):
529
    """Create new user."""
530
    data = await request.json()
531
    user_service = request.app['user_service']
532
    
533
    try:
534
        user_id = await user_service.create_user(
535
            data['name'], 
536
            data['email']
537
        )
538
        return web.json_response({
539
            'id': user_id,
540
            'name': data['name'],
541
            'email': data['email']
542
        }, status=201)
543
    
544
    except sqlite3.IntegrityError:
545
        raise web.HTTPConflict(text='Email already exists')
546

547
async def get_user_handler(request):
548
    """Get user by ID."""
549
    user_id = int(request.match_info['user_id'])
550
    user_service = request.app['user_service']
551
    
552
    user = await user_service.get_user(user_id)
553
    if user:
554
        return web.json_response(user)
555
    else:
556
        raise web.HTTPNotFound()
557

558
async def create_app():
559
    """Create application with user service."""
560
    app = web.Application()
561
    
562
    # Initialize user service
563
    user_service = UserService()
564
    await user_service.initialize()
565
    app['user_service'] = user_service
566
    
567
    # Add routes
568
    app.router.add_post('/users', create_user_handler)
569
    app.router.add_get('/users/{user_id}', get_user_handler)
570
    
571
    # Cleanup
572
    async def cleanup_user_service(app):
573
        await app['user_service'].close()
574
    
575
    app.on_cleanup.append(cleanup_user_service)
576
    
577
    return app
578

579
@pytest.fixture
580
async def client(aiohttp_client):
581
    """Create test client with database."""
582
    app = await create_app()
583
    return await aiohttp_client(app)
584

585
async def test_create_user(client):
586
    """Test user creation."""
587
    user_data = {
588
        'name': 'John Doe',
589
        'email': 'john@example.com'
590
    }
591
    
592
    resp = await client.post('/users', json=user_data)
593
    assert resp.status == 201
594
    
595
    data = await resp.json()
596
    assert data['name'] == 'John Doe'
597
    assert data['email'] == 'john@example.com'
598
    assert 'id' in data
599

600
async def test_get_user(client):
601
    """Test user retrieval."""
602
    # First create a user
603
    user_data = {
604
        'name': 'Jane Doe',
605
        'email': 'jane@example.com'
606
    }
607
    
608
    create_resp = await client.post('/users', json=user_data)
609
    created_user = await create_resp.json()
610
    user_id = created_user['id']
611
    
612
    # Then retrieve the user
613
    get_resp = await client.get(f'/users/{user_id}')
614
    assert get_resp.status == 200
615
    
616
    user = await get_resp.json()
617
    assert user['id'] == user_id
618
    assert user['name'] == 'Jane Doe'
619
    assert user['email'] == 'jane@example.com'
620

621
async def test_get_nonexistent_user(client):
622
    """Test getting non-existent user."""
623
    resp = await client.get('/users/999')
624
    assert resp.status == 404
625

626
async def test_duplicate_email(client):
627
    """Test creating user with duplicate email."""
628
    user_data = {
629
        'name': 'User One',
630
        'email': 'duplicate@example.com'
631
    }
632
    
633
    # Create first user
634
    resp1 = await client.post('/users', json=user_data)
635
    assert resp1.status == 201
636
    
637
    # Try to create user with same email
638
    user_data['name'] = 'User Two'
639
    resp2 = await client.post('/users', json=user_data)
640
    assert resp2.status == 409  # Conflict
641
```
642

643
### Load Testing Utilities
644

645
```python
646
import asyncio
647
import aiohttp
648
import time
649
from concurrent.futures import ThreadPoolExecutor
650

651
async def load_test_endpoint(url, num_requests=100, concurrency=10):
652
    """
653
    Perform load testing on an endpoint.
654
    
655
    Parameters:
656
    - url (str): Endpoint URL to test
657
    - num_requests (int): Total number of requests
658
    - concurrency (int): Concurrent request limit
659
    """
660
    
661
    semaphore = asyncio.Semaphore(concurrency)
662
    results = []
663
    
664
    async def make_request(session, request_id):
665
        async with semaphore:
666
            start_time = time.time()
667
            try:
668
                async with session.get(url) as response:
669
                    status = response.status
670
                    # Consume response to complete request
671
                    await response.read()
672
                    
673
                end_time = time.time()
674
                return {
675
                    'request_id': request_id,
676
                    'status': status,
677
                    'duration': end_time - start_time,
678
                    'success': 200 <= status < 300
679
                }
680
            
681
            except Exception as e:
682
                end_time = time.time()
683
                return {
684
                    'request_id': request_id,
685
                    'status': 0,
686
                    'duration': end_time - start_time,
687
                    'success': False,
688
                    'error': str(e)
689
                }
690
    
691
    # Perform load test
692
    start_time = time.time()
693
    
694
    async with aiohttp.ClientSession() as session:
695
        tasks = [
696
            make_request(session, i) 
697
            for i in range(num_requests)
698
        ]
699
        results = await asyncio.gather(*tasks)
700
    
701
    end_time = time.time()
702
    total_duration = end_time - start_time
703
    
704
    # Calculate statistics
705
    successful_requests = [r for r in results if r['success']]
706
    failed_requests = [r for r in results if not r['success']]
707
    
708
    if successful_requests:
709
        durations = [r['duration'] for r in successful_requests]
710
        avg_duration = sum(durations) / len(durations)
711
        min_duration = min(durations)
712
        max_duration = max(durations)
713
    else:
714
        avg_duration = min_duration = max_duration = 0
715
    
716
    return {
717
        'total_requests': num_requests,
718
        'successful_requests': len(successful_requests),
719
        'failed_requests': len(failed_requests),
720
        'success_rate': len(successful_requests) / num_requests * 100,
721
        'total_duration': total_duration,
722
        'requests_per_second': num_requests / total_duration,
723
        'avg_response_time': avg_duration,
724
        'min_response_time': min_duration,
725
        'max_response_time': max_duration
726
    }
727

728
# Usage example
729
async def test_api_performance():
730
    """Test API performance under load."""
731
    results = await load_test_endpoint(
732
        'http://localhost:8080/api/test',
733
        num_requests=1000,
734
        concurrency=50
735
    )
736
    
737
    print(f"Load Test Results:")
738
    print(f"Total Requests: {results['total_requests']}")
739
    print(f"Successful: {results['successful_requests']}")
740
    print(f"Failed: {results['failed_requests']}")
741
    print(f"Success Rate: {results['success_rate']:.2f}%")
742
    print(f"Requests/sec: {results['requests_per_second']:.2f}")
743
    print(f"Avg Response Time: {results['avg_response_time']:.3f}s")
744
    print(f"Min Response Time: {results['min_response_time']:.3f}s")
745
    print(f"Max Response Time: {results['max_response_time']:.3f}s")
746

747
# Run load test
748
asyncio.run(test_api_performance())
749
```