Pytest-Asyncio的全面解析

以下是关于 pytest-asyncio 的全面解析，涵盖异步测试的核心技术、最佳实践和实用技巧，帮助您高效测试 Python 异步代码：

一、快速入门

1. 安装

pip install pytest-asyncio

2. 基础测试示例

import pytest

@pytest.mark.asyncio  # 标记异步测试
async def test_async_function():
    result = await async_operation()
    assert result == "expected_value"

二、核心功能详解

1. 异步 Fixture

@pytest.fixture
async def async_db():
    db = await connect_db()
    yield db
    await db.close()

@pytest.mark.asyncio
async def test_query(async_db):  # 使用异步 Fixture
    data = await async_db.fetch("SELECT 1")
    assert data == [1]

2. 多异步任务测试

@pytest.mark.asyncio
async def test_parallel_tasks():
    task1 = asyncio.create_task(async_op1())
    task2 = asyncio.create_task(async_op2())
    results = await asyncio.gather(task1, task2)
    assert results == ["result1", "result2"]

3. 超时控制

@pytest.mark.asyncio
@pytest.mark.timeout(2.0)  # 超时设置
async def test_timeout():
    await asyncio.sleep(1.5)  # 测试通过
    # await asyncio.sleep(3)   # 触发 TimeoutError

三、高级应用场景

1. 模拟异步依赖（Async Mock）

@pytest.mark.asyncio
async def test_async_mock(mocker):
    # 模拟异步函数
    mock_func = mocker.AsyncMock(return_value="mocked")
    mocker.patch("module.async_func", mock_func)
    
    result = await module.call_async_func()
    assert result == "mocked"
    mock_func.assert_awaited_once()  # 验证调用

2. 测试异步 HTTP 客户端

from aiohttp import ClientSession

@pytest.mark.asyncio
async def test_http_call():
    async with ClientSession() as session:
        resp = await session.get("https://api.example.com")
        assert resp.status == 200
        data = await resp.json()
        assert "key" in data

3. 异步上下文管理器测试

class AsyncResource:
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, *exc):
        await self.close()

@pytest.mark.asyncio
async def test_async_context():
    async with AsyncResource() as res:
        result = await res.process()
    assert result == "done"

四、最佳实践

1. 事件循环策略

# conftest.py
@pytest.fixture(scope="session")
def event_loop():
    """避免 'Event loop is closed' 错误"""
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()

2. 与同步代码混合测试

def test_sync_part():
    assert sync_func() == "sync_result"

@pytest.mark.asyncio
async def test_async_part():
    assert await async_func() == "async_result"

3. 参数化异步测试

@pytest.mark.asyncio
@pytest.mark.parametrize("input,expected", [
    (1, "one"),
    (2, "two")
])
async def test_parametrized(input, expected):
    assert await async_lookup(input) == expected

五、常见问题解决

1. 错误：Event loop is closed

解决方案：

• 使用 event_loop fixture（如上文所示）
• 或升级 pytest-asyncio>=0.20.0

2. 异步 Fixture 清理失败

@pytest.fixture
async def resource():
    res = await setup()
    try:
        yield res
    finally:  # 确保清理执行
        await cleanup(res)

3. 调试技巧

@pytest.mark.asyncio
async def test_debug():
    print("Start")  # 输出到 pytest -s
    await asyncio.sleep(0.1)
    breakpoint()    # 使用 pdb 调试

六、性能优化

1. 共享异步 Fixture

@pytest.fixture(scope="module")
async def shared_client():
    async with AsyncClient() as client:
        yield client  # 整个测试模块共享连接

2. 快速失败模式

pytest -x --asyncio-mode=strict  # 第一个失败后立即停止

3. 并行测试

pytest -n 4 --asyncio-mode=auto  # 结合 pytest-xdist

七、与其他工具集成

1. Allure 报告

@pytest.mark.asyncio
async def test_with_allure():
    with allure.step("Async step"):
        result = await async_op()
        allure.attach(str(result), name="result")

2. 覆盖率统计

pytest --cov=my_async_module --cov-report=html --asyncio-mode=auto

3. Docker 集成

FROM python:3.9
RUN pip install pytest-asyncio aiohttp
COPY tests /tests
CMD ["pytest", "-v", "--asyncio-mode=auto"]

八、架构建议

1. 测试目录结构

tests/
├── unit/
│   ├── __init__.py
│   ├── test_async_utils.py
│   └── test_sync_utils.py
├── integration/
│   └── test_async_services.py
└── e2e/
    └── test_async_workflows.py

2. CI/CD 配置示例（GitLab）

test_async:
  image: python:3.9
  script:
    - pip install pytest-asyncio
    - pytest tests/unit/test_async_*.py --asyncio-mode=strict

九、总结

• 核心价值： ✅ 原生支持 async/await 语法 ✅ 完善的异步 Fixture 生命周期管理 ✅ 与 Pytest 生态无缝集成（Mock/参数化/插件）
• 适用场景： 异步 Web 框架（FastAPI、aiohttp）数据库驱动（asyncpg、aiomysql）消息队列（aiokafka、aio-pika）任何基于 asyncio 的库测试

通过合理运用这些技术，您可以构建出高效、稳定的异步测试体系，确保协程代码的可靠性和性能。