在 pytest 中执行 异步（async）单元测试非常简单，只需安装 pytest-asyncio 插件，并使用 async def 定义测试函数即可。以下是完整指南：

---

一、安装依赖

pip install pytest pytest-asyncio

注意：pytest-asyncio 是官方推荐的异步测试支持插件。

---

二、编写异步测试用例

基本语法：直接使用 async def

# test_async_example.py
import asyncio
import pytest

async def async_add(a, b):
    await asyncio.sleep(0.01)  # 模拟异步操作
    return a + b

@pytest.mark.asyncio
async def test_async_add():
    result = await async_add(2, 3)
    assert result == 5

关键点：

• 测试函数必须是 async def
• 使用 @pytest.mark.asyncio 标记（必需）
• 在函数内可直接 await 异步函数

---

三、运行测试

pytest test_async_example.py -v

输出示例：

test_async_example.py::test_async_add PASSED 

 pytest 会自动识别 @pytest.mark.asyncio 并在事件循环中运行测试。

---

四、高级用法

1. Fixture 支持异步

@pytest.fixture
async def db_connection():
    conn = await create_db_connection()
    yield conn
    await conn.close()

@pytest.mark.asyncio
async def test_db_query(db_connection):
    result = await db_connection.fetch("SELECT 1")
    assert result == [1]

 注意：异步 fixture 必须用 async def，且调用它的测试函数也必须是 @pytest.mark.asyncio。

---

2. 跳过或参数化异步测试

@pytest.mark.asyncio
@pytest.mark.parametrize("a,b,expected", [(1,2,3), (4,5,9)])
async def test_add_parametrized(a, b, expected):
    assert await async_add(a, b) == expected

@pytest.mark.asyncio
@pytest.mark.skipif(not HAS_NETWORK, reason="No network")
async def test_network_call():
    ...

---

3. 测试异常

@pytest.mark.asyncio
async def test_async_exception():
    with pytest.raises(ValueError):
        await async_divide(1, 0)

---

4. 超时控制（防 hang）

@pytest.mark.asyncio
@pytest.mark.timeout(5)  # 需要 pytest-timeout 插件
async def test_slow_operation():
    await slow_async_task()

安装：pip install pytest-timeout

---

五、配置（可选）

方式1：在 pytest.ini 中启用自动标记（推荐）

# pytest.ini
[tool:pytest]
markers =
    asyncio: mark test as async
asyncio_mode = auto

设置 asyncio_mode = auto 后，无需手动加 @pytest.mark.asyncio，pytest 会自动检测 async def 测试函数。

强烈推荐此方式，减少样板代码。

方式2：命令行启用

pytest --asyncio-mode=auto

---

六、常见问题与解决方案

问题	原因	解决方案
RuntimeError: no running event loop	忘记加 @pytest.mark.asyncio	添加标记 或 设置 asyncio_mode=auto
异步 fixture 不工作	fixture 未用 async def	确保 fixture 是 async def
多个测试共享事件循环冲突	默认每个测试新建 loop	使用 @pytest.fixture(scope="session") 共享 loop（高级）

---

七、完整示例项目结构

project/
├── my_module.py
├── test/
│   ├── __init__.py
│   └── test_my_module.py
└── pytest.ini

pytest.ini:

[tool:pytest]
asyncio_mode = auto

test/test_my_module.py:

import asyncio
import pytest

async def fetch_data(url: str) -> str:
    await asyncio.sleep(0.1)
    return f"Data from {url}"

async def test_fetch_data():
    result = await fetch_data("https://example.com")
    assert "example.com" in result

@pytest.fixture
async def mock_server():
    server = await start_mock_server()
    yield server
    await server.stop()

async def test_with_fixture(mock_server):
    data = await fetch_data(mock_server.url)
    assert data == "mocked"

运行：

pytest test/ -v

---

八、替代方案（不推荐）

虽然有其他方式（如手动创建事件循环），但 pytest-asyncio 是标准且最简洁的方式：

❌ 不推荐的手动方式：

def test_old_way():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(async_add(2, 3))
    assert result == 5

问题：无法与 fixture 集成，不支持现代 asyncio 特性。

---

总结

在 pytest 中测试 async 代码只需三步：

1. pip install pytest-asyncio
2. 写 async def test_xxx()
3. （可选）配置 asyncio_mode = auto 避免手动标记

这样你就能像写同步测试一样轻松地测试异步代码，享受完整的 pytest 生态（fixture、parametrize、coverage 等）支持。