Python异步测试用Pytest-Asyncio吗

本文目录导读：

1. 为什么选择 pytest-asyncio？
2. 基本用法
3. 高级配置
4. 实际项目示例
5. 最佳实践
6. 常见问题

是的，pytest-asyncio 是目前 Python 异步测试的主流选择。

为什么选择 pytest-asyncio？

1. 官方推荐：pytest 生态中最成熟的异步测试插件
2. 简单易用：只需添加一个 @pytest.mark.asyncio 装饰器
3. 功能完善：支持 async/await、asyncio.gather() 等所有异步操作

基本用法

安装

pip install pytest-asyncio

基础测试

import pytest
# 测试异步函数
@pytest.mark.asyncio
async def test_async_function():
    result = await some_async_function()
    assert result == expected_value
# 测试异步上下文管理器
@pytest.mark.asyncio
async def test_async_context_manager():
    async with AsyncClient() as client:
        response = await client.get("/api/data")
        assert response.status_code == 200
# 多个异步操作
@pytest.mark.asyncio
async def test_multiple_async_ops():
    results = await asyncio.gather(
        async_op1(),
        async_op2(),
        async_op3()
    )
    assert all(results)

Fixture 中使用异步

import pytest
import pytest_asyncio
# 异步 fixture
@pytest_asyncio.fixture
async def async_fixture():
    # setup
    resource = await create_async_resource()
    yield resource
    # teardown
    await cleanup_async_resource(resource)
@pytest.mark.asyncio
async def test_with_async_fixture(async_fixture):
    result = await async_fixture.process()
    assert result.status == "success"

高级配置

全局启用（推荐）

在 pyproject.toml 或 pytest.ini 中配置：

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

# pytest.ini
[pytest]
asyncio_mode = auto

这样就不需要每个测试都加 @pytest.mark.asyncio：

# 不需要装饰器
async def test_auto_async():
    result = await some_async_function()
    assert result == expected_value

自定义事件循环

@pytest.fixture(scope="session")
def event_loop():
    """创建自定义事件循环"""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

超时控制

@pytest.mark.asyncio
@pytest.mark.timeout(5)  # 5秒超时
async def test_with_timeout():
    result = await long_running_operation()
    assert result is not None

实际项目示例

测试 aiohttp

import pytest
from aiohttp import web
@pytest.mark.asyncio
async def test_aiohttp_client(aiohttp_client, loop):
    app = web.Application()
    app.router.add_get('/', lambda r: web.Response(text='Hello'))
    client = await aiohttp_client(app)
    resp = await client.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert text == 'Hello'

测试数据库操作

import pytest
from asyncpg import create_pool
@pytest.mark.asyncio
async def test_database():
    pool = await create_pool(
        user='user',
        database='test_db',
        host='localhost'
    )
    async with pool.acquire() as conn:
        result = await conn.fetch('SELECT * FROM users')
        assert len(result) > 0
    await pool.close()

测试 WebSocket

@pytest.mark.asyncio
async def test_websocket():
    async with websockets.connect('ws://localhost:8000/ws') as ws:
        await ws.send('Hello')
        response = await ws.recv()
        assert response == 'World'

最佳实践

1. 使用 asyncio_mode = "auto"：减少样板代码
2. Fixture 作用域：根据需求设置 scope (function/class/module/session)
3. 资源管理：使用 async with 确保资源正确释放
4. 并发测试：可以用 asyncio.gather() 测试并发场景
5. 错误处理：测试异步异常也需要 async with pytest.raises()

常见问题

事件循环冲突

# 解决：创建新的 event loop
@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

Fixture 不是异步的

# 使用 pytest-asyncio 的 fixture
@pytest_asyncio.fixture  # 注意这里是 pytest_asyncio，不是 pytest
async def my_fixture():
    yield await async_resource()

pytest-asyncio 是 Python 异步测试的首选方案，功能完善、生态成熟，与 pytest 完美集成，对于大多数异步测试需求,它都能很好地满足。