pytest-asyncio 基础使用笔记

小田测测看

发布于 2026-06-17 17:01:32

450

一、安装与基本配置

1. 安装

pip install pytest-asyncio

2. pytest 配置（可选）

在pyproject.toml或pytest.ini中添加：

[pytest]
asyncio_mode = auto  # 自动选择事件循环策略（auto/fixture/legacy）

二、测试异步函数

1. 测试普通异步函数

# test_example.py
import pytest

asyncdeffetch_data():
    return"data"

@pytest.mark.asyncio
asyncdeftest_fetch_data():
    result = await fetch_data()
    assert result == "data"

# 测试异步生成器
asyncdefasync_generator():
    yield1
    yield2

@pytest.mark.asyncio
asyncdeftest_async_generator():
    results = [i asyncfor i in async_generator()]
    assert results == [1, 2]

2. 基础异步 fixture

@pytest.fixture
async def client():
    # 异步初始化
    async with HttpClient() as client:
        yield client  # 提供给测试函数使用
        # 自动清理（无需手动关闭）

@pytest.mark.asyncio
async def test_client(client):
    response = await client.get("https://www.baidu.com")
    assert response.status == 200

3. 处理异常

async def raise_error():
    raise ValueError("Invalid input")

@pytest.mark.asyncio
async def test_raise_error():
    with pytest.raises(ValueError) as exc_info:
        await raise_error()
    assert str(exc_info.value) == "Invalid input"

4.模拟异步依赖

# 使用`unittest.mock`
from unittest.mock import MagicMock

@pytest.mark.asyncio
asyncdeftest_mocked_dependency():
    # 模拟异步函数
    async_func = MagicMock(return_value="mocked")
    result = await async_func()
    assert result == "mocked"

# 模拟异步上下文管理器
from unittest.mock import AsyncMock

@pytest.mark.asyncio
asyncdeftest_async_context_manager():
    mock = AsyncMock()
    mock.__aenter__.return_value = "mocked"
    
    asyncwith mock as result:
        assert result == "mocked"

三、其他

1. 测试异步事件循环

@pytest.mark.asyncio
async def test_event_loop(event_loop):
    # 使用 pytest-asyncio 内置的 event_loop fixture
    future = event_loop.create_future()
    event_loop.call_later(0.1, future.set_result, "done")
    result = await future
    assert result == "done"

2. 参数化异步测试

@pytest.mark.asyncio
@pytest.mark.parametrize("input, expected", [
    (1, 2),
    (3, 6),
])
async def test_async_param(input, expected):
    result = await multiply_by_two(input)
    assert result == expected

四、与其他库配合

1. FastAPI 异步测试

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/async")
asyncdefread_async():
    return {"hello": "world"}

@pytest.mark.asyncio
asyncdeftest_fastapi_app():
    client = TestClient(app)
    response = client.get("/async")
    assert response.status_code == 200

2. SQLAlchemy 异步 ORM

from sqlalchemy.ext.asyncio import create_async_engine

@pytest.fixture
async def db_session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncSession(engine) as session:
        yield session
        await session.rollback()

参考文档