aiohttp 自动化测试实战指南

aiohttp 是一个基于 asyncio 的异步 HTTP 客户端/服务器框架，非常适合高性能的网络应用。以下是 aiohttp 自动化测试的全面实战指南。

一、测试环境搭建

1. 安装依赖

pip install pytest pytest-asyncio aiohttp aioresponses

2. 项目结构

project/
├── app.py          # aiohttp 应用
├── test/
│   ├── conftest.py # 测试配置
│   ├── test_api.py # API 测试
│   └── test_client.py # 客户端测试
└── requirements.txt

二、测试 aiohttp 服务端

1. 基础测试示例

app.py:

from aiohttp import web

async def hello(request):
    return web.json_response({'message': 'Hello world'})

app = web.Application()
app.router.add_get('/', hello)

test/test_api.py:

from aiohttp.test_utils import TestClient, unittest_run_loop
from aiohttp import web
import pytest
from app import app

class TestHelloView:
    @pytest.fixture
    async def client(self, aiohttp_client):
        return await aiohttp_client(app)

    @unittest_run_loop
    async def test_hello(self, client):
        resp = await client.get('/')
        assert resp.status == 200
        data = await resp.json()
        assert data == {'message': 'Hello world'}

2. 测试带数据库的服务

扩展 app.py:

async def get_user(request):
    user_id = int(request.match_info['id'])
    async with request.app['db'].acquire() as conn:
        cursor = await conn.execute(
            "SELECT id, name FROM users WHERE id = $1", user_id)
        user = await cursor.fetchone()
    if not user:
        return web.json_response({'error': 'Not found'}, status=404)
    return web.json_response(dict(user))

# 初始化时添加数据库连接
async def init_db(app):
    app['db'] = await asyncpg.create_pool(dsn='postgresql://user:pass@localhost/db')

app.on_startup.append(init_db)

测试代码:

import asyncpg
from unittest.mock import AsyncMock

class TestUserView:
    @pytest.fixture
    async def mock_db(self):
        mock_pool = AsyncMock(spec=asyncpg.Pool)
        mock_conn = AsyncMock()
        mock_pool.acquire.return_value.__aenter__.return_value = mock_conn
        return mock_pool

    @pytest.fixture
    async def client(self, aiohttp_client, mock_db):
        app = web.Application()
        app['db'] = mock_db
        app.router.add_get('/users/{id}', get_user)
        return await aiohttp_client(app)

    async def test_get_user(self, client, mock_db):
        mock_conn = mock_db.acquire.return_value.__aenter__.return_value
        mock_conn.execute.return_value.fetchone.return_value = (1, 'John')
        
        resp = await client.get('/users/1')
        assert resp.status == 200
        data = await resp.json()
        assert data == {'id': 1, 'name': 'John'}

三、测试 aiohttp 客户端

1. 使用 aioresponses 模拟外部 API

测试代码:

import aiohttp
import pytest
from aioresponses import aioresponses

async def fetch_data(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

@pytest.mark.asyncio
async def test_fetch_data():
    with aioresponses() as m:
        m.get('https://api.example.com/data', 
              payload={'result': 'ok'},
              status=200)
        
        result = await fetch_data('https://api.example.com/data')
        assert result == {'result': 'ok'}

2. 测试带重试的客户端

客户端代码:

async def fetch_with_retry(url, retries=3, timeout=5):
    last_error = None
    async with aiohttp.ClientSession() as session:
        for attempt in range(retries):
            try:
                async with session.get(url, timeout=timeout) as resp:
                    resp.raise_for_status()
                    return await resp.json()
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                last_error = e
                await asyncio.sleep(1 * (attempt + 1))
        raise last_error

测试代码:

@pytest.mark.asyncio
async def test_fetch_with_retry_success():
    with aioresponses() as m:
        m.get('https://api.example.com/data',
              payload={'result': 'ok'})
        
        result = await fetch_with_retry('https://api.example.com/data')
        assert result == {'result': 'ok'}

@pytest.mark.asyncio
async def test_fetch_with_retry_failure():
    with aioresponses() as m:
        m.get('https://api.example.com/data',
              exception=aiohttp.ClientError("Connection error"))
        
        with pytest.raises(aiohttp.ClientError):
            await fetch_with_retry('https://api.example.com/data', retries=1)

四、高级测试技巧

1. 测试 WebSocket

服务端代码:

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    async for msg in ws:
        if msg.type == aiohttp.WSMsgType.TEXT:
            await ws.send_str(f"ECHO: {msg.data}")
        elif msg.type == aiohttp.WSMsgType.ERROR:
            print('ws connection closed with exception %s' % ws.exception())
    return ws

测试代码:

class TestWebSocket:
    @pytest.fixture
    async def client(self, aiohttp_client):
        app = web.Application()
        app.router.add_get('/ws', websocket_handler)
        return await aiohttp_client(app)

    @unittest_run_loop
    async def test_websocket(self, client):
        async with client.ws_connect('/ws') as ws:
            await ws.send_str("hello")
            resp = await ws.receive()
            assert resp.data == "ECHO: hello"

2. 测试中间件

中间件代码:

async def auth_middleware(app, handler):
    async def middleware(request):
        if not request.headers.get('Authorization'):
            raise web.HTTPUnauthorized()
        return await handler(request)
    return middleware

测试代码:

async def test_auth_middleware(aiohttp_client):
    app = web.Application(middlewares=[auth_middleware])
    app.router.add_get('/', hello)
    
    client = await aiohttp_client(app)
    
    # 测试未授权
    resp = await client.get('/')
    assert resp.status == 401
    
    # 测试已授权
    resp = await client.get('/', headers={'Authorization': 'Bearer token'})
    assert resp.status == 200

五、测试配置最佳实践

1. conftest.py 配置

import pytest
from aiohttp.test_utils import TestClient
from app import app

@pytest.fixture
async def client(aiohttp_client):
    return await aiohttp_client(app)

@pytest.fixture
def mock_db(mocker):
    mock = mocker.patch('asyncpg.create_pool')
    mock.return_value.__aenter__.return_value = mocker.AsyncMock()
    return mock

2. 测试覆盖率配置

# pytest.ini
[pytest]
asyncio_mode = auto
testpaths = tests
python_files = test_*.py
addopts = --cov=app --cov-report=term-missing

六、CI/CD 集成示例

GitHub Actions 配置

name: Python Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:13
        env:
          POSTGRES_PASSWORD: postgres
        ports:
          - 5432:5432
        options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5

    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt
        pip install pytest pytest-asyncio pytest-cov pytest-mock asyncpg aioresponses
    - name: Run tests
      env:
        DATABASE_URL: postgresql://postgres:postgres@localhost:5432/postgres
      run: |
        pytest --cov=app --cov-report=xml
    - name: Upload coverage
      uses: codecov/codecov-action@v1

七、常见问题解决方案

1. 处理测试中的事件循环

@pytest.fixture
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

2. 测试超时设置

@pytest.mark.asyncio
@pytest.mark.timeout  # 每个测试用例5秒超时
async def test_slow_operation():
    await asyncio.sleep(4)  # 正常通过
    # await asyncio.sleep(6)  # 会超时失败

3. 测试文件上传

async def test_upload_file(client):
    data = FormData()
    data.add_field('file', 
                  b'file content', 
                  filename='test.txt',
                  content_type='text/plain')
    
    resp = await client.post('/upload', data=data)
    assert resp.status == 200

通过以上实战方法，您可以全面覆盖 aiohttp 应用的自动化测试需求，包括服务端 API、客户端请求、WebSocket、中间件等各个方面。记得根据实际项目需求调整测试策略和覆盖范围。