Python Pytest 异步测试完全指南：pytest-asyncio 实战 2026

编程语言

异步测试的痛点

当你用 asyncio 写了一个漂亮的异步服务，信心满满地准备写测试时，却发现——普通的 pytest 测试函数根本跑不了 async def。你尝试手动 asyncio.run()，fixture 里的异步初始化又成了问题；你想 Mock 一个异步函数，unittest.mock.patch 却告诉你协程对象不是可调用的；你写了参数化测试，结果事件循环冲突导致用例全部报错。异步测试的复杂度远超同步测试，配置不当甚至会让测试套件完全无法运行。本文将系统讲解 pytest-asyncio 的配置、模式和最佳实践，帮你彻底解决 Python 异步测试难题。

核心概念

概念 说明
pytest-asyncio pytest 的异步测试插件，自动为标记的测试函数创建事件循环
@pytest.mark.asyncio 标记测试函数为异步，pytest-asyncio 会自动在事件循环中执行
async fixture 使用 async def 定义的 fixture，支持异步资源初始化和清理
event_loop fixture pytest-asyncio 内置 fixture，提供测试用的事件循环实例
asyncio_mode 配置项，控制 pytest-asyncio 的行为模式（auto/strict）
AsyncMock unittest.mock 提供的异步 Mock 对象，用于替换异步函数
scope fixture 作用域，控制异步 fixture 的生命周期（function/session）

问题分析：异步测试的 5 大挑战

  1. 事件循环管理混乱：每个测试需要独立的事件循环，循环复用或冲突会导致状态污染和不可预期的失败
  2. 异步 Fixture 难以编写：数据库连接、HTTP 客户端等异步资源的初始化和清理需要特殊处理
  3. 异步 Mock 行为异常：普通 Mock 无法正确模拟协程函数，返回值是协程对象而非实际结果
  4. 测试执行顺序不确定：异步测试并发执行时，共享状态的测试可能因执行顺序不同而失败
  5. 参数化与异步冲突：@pytest.mark.parametrize 与 @pytest.mark.asyncio 组合使用时标记顺序和作用域容易出错

模式一：pytest-asyncio 基础配置

安装与配置

pip install pytest pytest-asyncio

pyproject.toml 配置

[tool.pytest.ini_options]
asyncio_mode = "auto"

基础异步测试

# test_basic.py
import asyncio
import pytest

# 方式一：auto 模式，无需手动标记
async def test_auto_mode():
    """asyncio_mode = "auto" 时，所有 async def 测试自动识别"""
    await asyncio.sleep(0.1)
    result = await fetch_data()
    assert result == "expected"

# 方式二：strict 模式，需要显式标记
@pytest.mark.asyncio
async def test_explicit_mark():
    """asyncio_mode = "strict" 时，必须使用 @pytest.mark.asyncio"""
    await asyncio.sleep(0.1)
    assert True

async def fetch_data():
    await asyncio.sleep(0.01)
    return "expected"

# 方式三：指定事件循环策略
@pytest.mark.asyncio(loop_scope="session")
async def test_session_loop():
    """使用 session 级别的事件循环，跨测试共享"""
    await asyncio.sleep(0.01)
    assert True

conftest.py 全局配置

# conftest.py
import asyncio
import pytest

@pytest.fixture(scope="session")
def event_loop_policy():
    """自定义事件循环策略"""
    return asyncio.DefaultEventLoopPolicy()

@pytest.fixture(scope="session")
def loop_scope():
    """设置事件循环作用域为 session"""
    return "session"

模式二：Async Fixture 与依赖注入

基础异步 Fixture

# conftest.py
import asyncio
import pytest
from typing import AsyncGenerator

@pytest.fixture
async def db_connection() -> AsyncGenerator:
    """异步数据库连接 fixture"""
    conn = await create_connection("postgresql://localhost/test")
    yield conn
    await conn.close()

@pytest.fixture
async def http_client() -> AsyncGenerator:
    """异步 HTTP 客户端 fixture"""
    import httpx
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        yield client

async def create_connection(dsn: str):
    """模拟异步数据库连接"""
    await asyncio.sleep(0.01)
    return {"dsn": dsn, "status": "connected"}

# test_fixtures.py
@pytest.mark.asyncio
async def test_with_db(db_connection):
    """使用异步 fixture 的测试"""
    assert db_connection["status"] == "connected"
    result = await db_query(db_connection, "SELECT 1")
    assert result is not None

@pytest.mark.asyncio
async def test_with_http_client(http_client):
    """使用异步 HTTP 客户端 fixture 的测试"""
    response = await http_client.get("/api/health")
    assert response.status_code == 200

async def db_query(conn, sql):
    await asyncio.sleep(0.01)
    return {"sql": sql, "result": True}

Fixture 依赖链

# conftest.py
@pytest.fixture
async def app():
    """应用实例 fixture"""
    app = await create_app()
    yield app
    await app.shutdown()

@pytest.fixture
async def db_connection(app):
    """依赖 app fixture 的数据库连接"""
    conn = await app.get_db()
    yield conn
    await conn.close()

@pytest.fixture
async def user_service(db_connection):
    """依赖 db_connection 的用户服务"""
    return UserService(db_connection)

# test_service.py
@pytest.mark.asyncio
async def test_create_user(user_service):
    user = await user_service.create("test@example.com", "password123")
    assert user.email == "test@example.com"

class UserService:
    def __init__(self, db):
        self.db = db

    async def create(self, email, password):
        await asyncio.sleep(0.01)
        return type("User", (), {"email": email})()

async def create_app():
    await asyncio.sleep(0.01)
    return type("App", (), {
        "get_db": lambda self: create_connection("postgresql://localhost/test"),
        "shutdown": lambda self: asyncio.sleep(0.01)
    })()

模式三：Mock 异步函数与外部 API

AsyncMock 基础

# test_mock.py
import asyncio
from unittest.mock import AsyncMock, patch, MagicMock
import pytest

async def fetch_user(user_id: int):
    """被测函数：从外部 API 获取用户"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

@pytest.mark.asyncio
async def test_fetch_user_mocked():
    """使用 AsyncMock 替换异步函数"""
    mock_get = AsyncMock(return_value={"id": 1, "name": "张三"})

    with patch("httpx.AsyncClient.get", mock_get):
        result = await fetch_user(1)

    assert result == {"id": 1, "name": "张三"}
    mock_get.assert_called_once()

@pytest.mark.asyncio
async def test_fetch_user_side_effect():
    """使用 side_effect 模拟多次调用的不同返回值"""
    mock_get = AsyncMock(side_effect=[
        {"id": 1, "name": "张三"},
        {"id": 2, "name": "李四"},
    ])

    with patch("httpx.AsyncClient.get", mock_get):
        result1 = await fetch_user(1)
        result2 = await fetch_user(2)

    assert result1["name"] == "张三"
    assert result2["name"] == "李四"

Mock 异步上下文管理器

@pytest.mark.asyncio
async def test_mock_async_context_manager():
    """Mock 异步上下文管理器"""
    mock_client = AsyncMock()
    mock_client.get.return_value = MagicMock(
        status_code=200,
        json=lambda: {"status": "ok"}
    )
    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
    mock_client.__aexit__ = AsyncMock(return_value=False)

    with patch("httpx.AsyncClient", return_value=mock_client):
        async with httpx.AsyncClient() as client:
            response = await client.get("/health")
            assert response.status_code == 200

import httpx

Mock 异步异常

@pytest.mark.asyncio
async def test_fetch_user_timeout():
    """测试异步超时异常处理"""
    mock_get = AsyncMock(side_effect=httpx.TimeoutException("Request timeout"))

    with patch("httpx.AsyncClient.get", mock_get):
        with pytest.raises(httpx.TimeoutException):
            await fetch_user(1)

@pytest.mark.asyncio
async def test_fetch_user_retry():
    """测试异步重试逻辑"""
    call_count = 0

    async def failing_then_success(*args, **kwargs):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise httpx.TimeoutException("timeout")
        return MagicMock(status_code=200, json=lambda: {"id": 1})

    mock_get = AsyncMock(side_effect=failing_then_success)

    with patch("httpx.AsyncClient.get", mock_get):
        result = await fetch_with_retry(1, max_retries=3)
        assert result["id"] == 1

async def fetch_with_retry(user_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(f"https://api.example.com/users/{user_id}")
                return response.json()
        except httpx.TimeoutException:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(0.1 * (attempt + 1))

模式四：测试异步上下文管理器与生成器

异步上下文管理器测试

# test_context_manager.py
import asyncio
import pytest
from contextlib import asynccontextmanager

class AsyncDatabasePool:
    def __init__(self, dsn):
        self.dsn = dsn
        self._connections = []

    async def __aenter__(self):
        await asyncio.sleep(0.01)
        self._pool = {"status": "open", "size": 10}
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await asyncio.sleep(0.01)
        self._pool = None

    async def acquire(self):
        await asyncio.sleep(0.01)
        if self._pool is None:
            raise RuntimeError("Pool is closed")
        return {"id": len(self._connections), "pool": self.dsn}

@pytest.mark.asyncio
async def test_async_context_manager():
    """测试异步上下文管理器的进入和退出"""
    pool = AsyncDatabasePool("postgresql://localhost/test")
    async with pool as p:
        assert p._pool["status"] == "open"
        conn = await p.acquire()
        assert conn["id"] == 0

    assert pool._pool is None

@pytest.mark.asyncio
async def test_async_context_manager_exception():
    """测试异步上下文管理器异常时的清理"""
    pool = AsyncDatabasePool("postgresql://localhost/test")
    with pytest.raises(ValueError):
        async with pool as p:
            raise ValueError("test error")

    assert pool._pool is None

异步生成器测试

# test_async_generator.py
async def async_stream_data(count: int):
    """异步生成器：模拟流式数据"""
    for i in range(count):
        await asyncio.sleep(0.01)
        yield {"index": i, "data": f"item_{i}"}

@pytest.mark.asyncio
async def test_async_generator():
    """测试异步生成器的输出"""
    results = []
    async for item in async_stream_data(3):
        results.append(item)

    assert len(results) == 3
    assert results[0] == {"index": 0, "data": "item_0"}
    assert results[2] == {"index": 2, "data": "item_2"}

@pytest.mark.asyncio
async def test_async_generator_with_anext():
    """使用 anext 测试异步生成器"""
    gen = async_stream_data(2)
    first = await anext(gen)
    assert first["index"] == 0
    second = await anext(gen)
    assert second["index"] == 1

    with pytest.raises(StopAsyncIteration):
        await anext(gen)

模式五：参数化异步测试与组织

参数化异步测试

# test_parametrize.py
import pytest

async def compute(x: int, y: int) -> int:
    await asyncio.sleep(0.01)
    return x + y

@pytest.mark.asyncio
@pytest.mark.parametrize("x, y, expected", [
    (1, 2, 3),
    (10, 20, 30),
    (-1, 1, 0),
    (100, 200, 300),
])
async def test_compute_parametrize(x, y, expected):
    """参数化异步测试"""
    result = await compute(x, y)
    assert result == expected

@pytest.mark.asyncio
@pytest.mark.parametrize("input_data", [
    {"name": "张三", "age": 25},
    {"name": "李四", "age": 30},
    {"name": "王五", "age": 35},
], ids=["zhangsan", "lisi", "wangwu"])
async def test_create_user_parametrize(input_data):
    """参数化异步测试：自定义 ID"""
    result = await create_user(input_data)
    assert result["name"] == input_data["name"]

async def create_user(data):
    await asyncio.sleep(0.01)
    return {**data, "id": 1}

测试类组织

# test_user_api.py
import pytest

@pytest.mark.asyncio
class TestUserAPI:
    """异步测试类组织"""

    async def test_list_users(self):
        users = await list_users()
        assert len(users) > 0

    async def test_get_user(self):
        user = await get_user(1)
        assert user["id"] == 1

    async def test_create_user(self):
        user = await create_user({"name": "test"})
        assert user["name"] == "test"

    @pytest.mark.parametrize("user_id", [1, 2, 3])
    async def test_get_user_parametrize(self, user_id):
        user = await get_user(user_id)
        assert user["id"] == user_id

async def list_users():
    await asyncio.sleep(0.01)
    return [{"id": 1}, {"id": 2}]

async def get_user(user_id):
    await asyncio.sleep(0.01)
    return {"id": user_id}

async def create_user(data):
    await asyncio.sleep(0.01)
    return {**data, "id": 99}

陷阱指南

陷阱 1：忘记标记异步测试

❌ 错误写法：

# strict 模式下，忘记 @pytest.mark.asyncio
async def test_something():
    result = await fetch_data()
    assert result is not None
# 报错：coroutine was never awaited

✅ 正确写法：

@pytest.mark.asyncio
async def test_something():
    result = await fetch_data()
    assert result is not None

陷阱 2：在同步 Fixture 中返回协程

❌ 错误写法：

@pytest.fixture
def db_connection():
    return create_connection("postgresql://localhost/test")
    # 返回的是协程对象，不是连接

✅ 正确写法：

@pytest.fixture
async def db_connection():
    conn = await create_connection("postgresql://localhost/test")
    yield conn
    await conn.close()

陷阱 3：Mock 异步函数使用普通 Mock

❌ 错误写法：

from unittest.mock import MagicMock

mock_fetch = MagicMock(return_value={"id": 1})
# 调用 mock_fetch() 返回 MagicMock，不是协程
result = await mock_fetch()  # TypeError

✅ 正确写法：

from unittest.mock import AsyncMock

mock_fetch = AsyncMock(return_value={"id": 1})
result = await mock_fetch()
assert result == {"id": 1}

陷阱 4：在异步测试中使用 time.sleep

❌ 错误写法：

@pytest.mark.asyncio
async def test_with_delay():
    time.sleep(1)  # 阻塞事件循环！
    result = await fetch_data()
    assert result is not None

✅ 正确写法：

@pytest.mark.asyncio
async def test_with_delay():
    await asyncio.sleep(1)  # 非阻塞等待
    result = await fetch_data()
    assert result is not None

陷阱 5：参数化标记顺序错误

❌ 错误写法：

@pytest.mark.asyncio
@pytest.mark.parametrize("x", [1, 2, 3])
# parametrize 在 asyncio 之后，可能导致标记不生效
async def test_value(x):
    assert x > 0

✅ 正确写法：

@pytest.mark.parametrize("x", [1, 2, 3])
@pytest.mark.asyncio
# parametrize 在外层，asyncio 在内层
async def test_value(x):
    assert x > 0

错误排查表

序号 报错信息 原因 解决方法
1 coroutine was never awaited 异步函数未在事件循环中执行 添加 @pytest.mark.asyncio 或设置 asyncio_mode = "auto"
2 RuntimeError: no running event loop 在同步上下文中调用了 await 确保测试函数是 async def 并正确标记
3 Fixture "xxx" is async but test is not 同步测试使用了异步 fixture 将测试改为 async def 并添加 @pytest.mark.asyncio
4 TypeError: object MagicMock can't be used in await expression 使用普通 Mock 替代异步函数 改用 AsyncMock 替换异步函数
5 DeprecationWarning: The 'event_loop' fixture is deprecated 使用了旧版 event_loop fixture 升级 pytest-asyncio 到 0.23+，使用 loop_scope 参数
6 assert False on asyncio_mode="strict" strict 模式下未标记异步测试 添加 @pytest.mark.asyncio 或切换为 auto 模式
7 RuntimeError: Event loop is closed 测试结束后事件循环被关闭 检查 fixture scope，避免 session 级 fixture 使用 function 级循环
8 pluggy.PluggyTeardownError 异步 fixture 清理失败 确保 yield 后的清理代码也是异步的，使用 async with
9 TimeoutError in async test 异步测试超时 使用 @pytest.mark.timeout 或 asyncio.wait_for 控制超时
10 AssertionError: expected coroutine, got MagicMock Mock 返回值类型不匹配 使用 AsyncMock(return_value=...) 确保返回协程

高级优化

1. 使用 session 级事件循环提升性能

# conftest.py
@pytest.fixture(scope="session")
def loop_scope():
    return "session"

多个测试共享一个事件循环，避免重复创建和销毁的开销，特别适合有大量异步测试的项目。

2. 使用 aioresponses Mock HTTP 请求

from aioresponses import aioresponses

@pytest.mark.asyncio
async def test_with_aioresponses():
    with aioresponses() as m:
        m.get("https://api.example.com/users/1", payload={"id": 1, "name": "张三"})
        result = await fetch_user(1)
        assert result["name"] == "张三"

3. 自定义异步测试超时

@pytest.mark.asyncio
async def test_with_custom_timeout():
    async with asyncio.timeout(2.0):
        result = await slow_operation()
        assert result is not None

async def slow_operation():
    await asyncio.sleep(0.5)
    return "done"

4. 并行执行异步测试

pip install pytest-xdist
pytest -n auto  # 自动并行执行

结合 pytest-xdist 可以让异步测试在多个 worker 中并行运行，大幅缩短测试时间。

5. 使用 hypothesis 进行异步属性测试

from hypothesis import given, strategies as st

@pytest.mark.asyncio
@given(st.integers(min_value=0, max_value=100))
async def test_async_property(x):
    result = await compute(x, x)
    assert result == x * 2

async def compute(a, b):
    await asyncio.sleep(0.001)
    return a + b

方案对比

特性 pytest-asyncio asyncio.run() unittest.IsolatedAsyncioTestCase 手动事件循环
配置复杂度 低（安装即用） 无需配置 中（需继承类） 高（手动管理）
Fixture 支持 完整支持 不支持 有限支持 不支持
参数化测试 完整支持 需手动包装 有限支持 需手动包装
Mock 支持 AsyncMock 完美配合 需手动处理 需手动处理 需手动处理
测试隔离 自动隔离 手动隔离 自动隔离 手动隔离
并行执行 支持 xdist 不支持 不支持 不支持
社区生态 丰富 Python 内置 Python 内置 无
推荐场景 生产级异步测试 简单脚本验证 unittest 项目迁移 不推荐

工具推荐

在 Python pytest 异步测试实践中，以下 工具库 工具可以帮到你：

  • JSON 格式化 — 格式化异步 API 返回的 JSON 数据，快速排查响应结构问题
  • Hash 计算 — 为异步缓存生成键名，实现请求去重和缓存验证
  • cURL 转代码 — 将 cURL 命令转换为 Python 异步 HTTP 请求代码

pytest-asyncio 是 Python 异步测试的事实标准。掌握其配置模式、async fixture 编写、AsyncMock 使用和参数化测试组织，你就能构建稳定、高效的生产级异步测试套件。记住：异步测试的核心是正确管理事件循环生命周期，避免同步阻塞，善用 AsyncMock。

本站提供浏览器本地工具，免注册即可试用 →

#Python测试#pytest异步#asyncio测试#pytest-asyncio#2026#编程语言