异步代码怎么测试？pytest-asyncio 和 AsyncMock 怎么用？

简化版

异步代码测试的第一个障碍是：pytest 默认不认识协程函数——你写 async def test_x(): ...，pytest 只是调用它得到一个协程对象然后丢掉，测试体一行都不会执行，却显示「通过」（只会有一条 coroutine was never awaited 的警告，很容易被忽略）。解决办法是装 pytest-asyncio，然后二选一：给每个异步测试加 @pytest.mark.asyncio 装饰器（strict 模式，默认），或者在配置里写 asyncio_mode = "auto" 让所有 async def test_* 自动被识别（推荐，省去满屏的装饰器）。第二个障碍是 mock：unittest.mock.MagicMock() 的返回值不能被 await，所以从 Python 3.8 起有了 AsyncMock——patch() 在自动检测到被替换的对象是协程函数时会自动使用 AsyncMock，它的返回值是可等待的，还提供 assert_awaited_once() 这类断言。第三个要点是「异步 fixture」：pytest-asyncio 支持 async def 的 fixture，但要注意fixture 的作用域必须和事件循环的作用域匹配——一个 session 级的异步 fixture（比如数据库连接池）如果跑在函数级的 loop 上，第二个测试就会报 attached to a different loop。最后是测异步特有的行为：超时用 asyncio.wait_for、取消用 task.cancel() 后断言清理是否执行、并发竞态用 asyncio.gather 加断言最终一致性。核心记忆：装 pytest-asyncio 并开 auto 模式；mock 异步函数用 AsyncMock；注意 fixture 和 loop 的作用域匹配。

详细版

异步测试的三件套：

问题	工具	关键点
pytest 不认协程测试	`pytest-asyncio`	`asyncio_mode = "auto"`
mock 异步函数	`AsyncMock`（3.8+）	`patch` 会自动选用
异步 fixture	`@pytest_asyncio.fixture`	作用域要和 loop 匹配
测超时/取消	`wait_for` / `task.cancel()`	断言清理逻辑执行了
替代方案	`anyio.pytest_plugin`	同时支持 asyncio 和 trio

# pyproject.toml / pytest.ini
# [tool.pytest.ini_options]
# asyncio_mode = "auto"                     # ★推荐：不用写装饰器★
# asyncio_default_fixture_loop_scope = "function"   # ★新版本要显式指定，否则警告★

import pytest, asyncio
from unittest.mock import AsyncMock, patch

# ① ★基本异步测试★
async def test_fetch():                      # auto 模式下直接写
    result = await fetch("http://x")
    assert result == "ok"

@pytest.mark.asyncio                         # strict 模式下必须加
async def test_fetch_strict():
    ...

# ② ★异步 fixture★
import pytest_asyncio

@pytest_asyncio.fixture                      # ★auto 模式下 @pytest.fixture 也行★
async def client():
    c = await create_client()
    yield c                                  # ★yield 前是 setup，后是 teardown★
    await c.close()

async def test_with_client(client):
    assert await client.ping()

# ③ ★mock 异步函数：AsyncMock★
async def test_with_mock():
    with patch("mymod.fetch", new_callable=AsyncMock) as m:
        m.return_value = {"id": 1}           # ★await 后得到的值★
        result = await get_user(1)
        m.assert_awaited_once_with(1)        # ★异步专用断言★
        assert result["id"] == 1

# ★patch 会自动检测：如果目标是 async def，自动用 AsyncMock★
async def test_autospec():
    with patch("mymod.fetch") as m:          # ★fetch 是协程函数 → m 是 AsyncMock★
        m.return_value = "v"
        assert await mymod.fetch() == "v"

# ④ mock 抛异常 / 多次返回不同值
m = AsyncMock(side_effect=[1, 2, TimeoutError])
assert await m() == 1
assert await m() == 2
with pytest.raises(TimeoutError):
    await m()

# ⑤ ★测超时行为★
async def test_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_op(), timeout=0.1)

# ⑥ ★测取消时的清理★
async def test_cancel_cleanup():
    cleaned = False
    async def work():
        nonlocal cleaned
        try:
            await asyncio.sleep(10)
        finally:
            cleaned = True                    # ★验证 finally 执行了★
    task = asyncio.create_task(work())
    await asyncio.sleep(0)                    # ★让任务真正开始★
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert cleaned

# ⑦ ★测并发正确性★
async def test_concurrent_counter():
    counter = Counter()
    await asyncio.gather(*(counter.incr() for _ in range(1000)))
    assert counter.value == 1000              # ★断言最终一致性★

# ⑧ 常用断言
m.assert_awaited()            # 被 await 过
m.assert_awaited_once()
m.assert_awaited_with(1, 2)
m.await_count                 # ★次数★
m.await_args_list             # 每次的参数

⚠️ 三个必须避开的坑：① 没装 pytest-asyncio 时，async def test_x() 会「静默通过」——pytest 调用它得到一个协程对象后直接丢弃，测试体一行都没执行，你却看到绿色的 PASSED。唯一的线索是 RuntimeWarning: coroutine 'test_x' was never awaited，很容易被淹没。在 CI 里配置 filterwarnings = error 能把它变成失败，避免「假绿」。② MagicMock() 的返回值不能被 await——用它去替换一个协程函数，测试会报 TypeError: object MagicMock can't be used in 'await' expression。要用 AsyncMock（3.8+）；好消息是 patch() 会自动检测目标是不是协程函数并自动选用 AsyncMock，所以大多数时候直接 patch("mod.fetch") 就对了，只有手动构造 mock 对象时才需要显式写 AsyncMock()。③ fixture 的作用域必须和事件循环的作用域匹配：pytest-asyncio 默认每个测试函数一个全新的事件循环，所以一个 scope="session" 的异步 fixture（数据库连接池、HTTP 会话）会在第一个测试里绑定到那个 loop，第二个测试用新 loop 时就报 got Future attached to a different loop。解决办法是把 loop 的作用域也提升（新版用 asyncio_default_fixture_loop_scope 或 @pytest.mark.asyncio(loop_scope="session")）。

完整版教学

一、为什么异步测试需要特殊支持

★ 问题的根源：pytest 的测试函数是"调用它，看有没有抛异常"
  def test_x():
      assert 1 == 1          # ★调用即执行★

  async def test_y():
      assert 1 == 2          # ★调用只得到协程对象，什么都没执行★
  → pytest 拿到一个协程对象，发现"没抛异常" → ★报告 PASSED★
  → ★这是最危险的情况：断言根本没跑，测试却是绿的★

  唯一的线索：
    RuntimeWarning: coroutine 'test_y' was never awaited
  ★ 在满屏输出里极易被忽略
  ✓ ★CI 里配 filterwarnings = error★ → 变成失败，杜绝"假绿"

★ 需要解决的三件事：
  ① ★谁来运行协程★：需要一个事件循环来 await 测试函数
  ② ★fixture 也可能是异步的★：async def fixture 要能被正确 setup/teardown
  ③ ★loop 的生命周期★：每个测试一个新 loop？还是共享？

★ 两个主流方案：
  ★pytest-asyncio★（最常用）
    - 专注 asyncio
    - 两种模式：strict（默认，要装饰器）/ auto（自动识别）
  ★anyio 的 pytest 插件★
    - 同时支持 ★asyncio 和 trio★
    - 用 @pytest.mark.anyio 或 anyio_backend fixture
    - 如果你的库要同时支持两种后端 → 用它

  还有：asynctest（★已废弃，被 AsyncMock 取代★）、
        unittest.IsolatedAsyncioTestCase（★标准库自带，3.8+★）

★ 标准库的方案（不用 pytest 时）：
  import unittest
  class TestX(unittest.IsolatedAsyncioTestCase):     # ★3.8+★
      async def asyncSetUp(self):
          self.client = await create_client()
      async def test_fetch(self):
          self.assertEqual(await self.client.get(), "ok")
      async def asyncTearDown(self):
          await self.client.close()
  ★ 每个测试方法一个独立的事件循环

问题的根源在于 pytest 的测试函数执行模型是「调用它，看有没有抛异常」——而调用一个 async def 只会得到协程对象，测试体一行都不执行，pytest 发现没抛异常就报告 PASSED。这是最危险的情况：断言根本没跑，测试却是绿的，唯一的线索是那条容易被忽略的 coroutine was never awaited 警告——所以 CI 里应该配 filterwarnings = error 把它变成失败。要解决的是三件事：谁来运行协程、异步 fixture 怎么 setup/teardown、loop 的生命周期怎么管。两个主流方案：pytest-asyncio（最常用，专注 asyncio，有 strict/auto 两种模式）和 anyio 的 pytest 插件（同时支持 asyncio 和 trio，适合需要兼容两种后端的库）。不用 pytest 时，标准库自带 unittest.IsolatedAsyncioTestCase（3.8+），它提供 asyncSetUp/asyncTearDown 并为每个测试方法创建独立的事件循环。

二、pytest-asyncio 的模式与配置

★ 两种模式：
  ① ★strict（默认）★：每个异步测试都要显式标记
     @pytest.mark.asyncio
     async def test_x(): ...
     ✓ 优点：明确，不会误判别的插件的 async 测试
     ✗ 缺点：★满屏装饰器★

  ② ★auto（推荐）★：所有 async def test_* 自动被当作异步测试
     # pyproject.toml
     [tool.pytest.ini_options]
     asyncio_mode = "auto"
     ✓ 直接写 async def test_x()，不用装饰器
     ✓ 异步 fixture 也自动识别（不用 @pytest_asyncio.fixture）
     ✗ 如果同时用 anyio/trio 的插件可能冲突

★ 事件循环的作用域（★新版本的重点变化★）：
  pytest-asyncio 0.23 起引入了更细的 loop 作用域控制：
    asyncio_default_fixture_loop_scope = "function"   # ★不设会有 DeprecationWarning★
    # 可选：function / class / module / package / session

  单个测试指定：
    @pytest.mark.asyncio(loop_scope="session")
    async def test_x(): ...

  ★ 为什么重要：
    默认每个测试函数一个新 loop
    → ★session 级的异步 fixture 会跨 loop 使用 → 报错★
    → 需要把 loop 作用域提到和 fixture 一致

★ 老写法的变迁（★踩坑重灾区★）：
  ✗ 旧：自定义 event_loop fixture 来改作用域
     @pytest.fixture(scope="session")
     def event_loop():
         loop = asyncio.new_event_loop()
         yield loop
         loop.close()
     → ★0.23 起被弃用★，会有警告；新版本要用 loop_scope

  ✗ 旧：@pytest.mark.asyncio(scope="session")   → 改名为 loop_scope
  ✓ 现在：配置文件里设 asyncio_default_fixture_loop_scope
          + 需要时用 @pytest.mark.asyncio(loop_scope=...)

★ 完整的配置示例：
  # pyproject.toml
  [tool.pytest.ini_options]
  asyncio_mode = "auto"
  asyncio_default_fixture_loop_scope = "function"
  filterwarnings = [
      "error",                                    # ★警告变错误★
      "ignore::DeprecationWarning:some_old_lib",  # 白名单
  ]
  addopts = "-ra --strict-markers"

★ 检查是否生效：
  写一个故意失败的异步测试：
    async def test_should_fail():
        assert False
  → ★必须是 FAILED★；如果是 PASSED，说明协程根本没被执行

pytest-asyncio 有两种模式：strict（默认） 要求每个异步测试加 @pytest.mark.asyncio 装饰器，auto（推荐） 让所有 async def test_* 自动被识别（异步 fixture 也自动识别，省去满屏装饰器）。事件循环的作用域是新版本的重点变化：0.23 起引入了 asyncio_default_fixture_loop_scope 配置（不设会有 DeprecationWarning）和 @pytest.mark.asyncio(loop_scope=...) 标记，用来控制「几个测试共用一个 loop」——这很重要，因为默认每个测试函数一个新 loop，而 session 级的异步 fixture 会跨 loop 使用导致报错。老写法（自定义 event_loop fixture 来改作用域、@pytest.mark.asyncio(scope=...)）已被弃用，是升级时的踩坑重灾区。最后有个实用的自检方法：写一个 assert False 的异步测试，确认它真的是 FAILED——如果显示 PASSED，说明协程根本没被执行。

三、异步 fixture 与作用域匹配

★ 异步 fixture 的写法：
  @pytest_asyncio.fixture              # ★strict 模式必须用这个★
  async def db():
      conn = await create_conn()
      yield conn                        # ★setup 到此，之后是 teardown★
      await conn.close()

  # auto 模式下 @pytest.fixture 也能识别 async def
  @pytest.fixture
  async def db(): ...

★ ★作用域不匹配是最常见的报错★：
  @pytest_asyncio.fixture(scope="session")
  async def pool():
      return await create_pool()        # ★在第一个测试的 loop 上创建★

  async def test_a(pool): ...           # loop A：正常
  async def test_b(pool): ...           # ★loop B：报错！★
  → RuntimeError: Task got Future attached to a different loop
  → 或 got Future <Future pending> attached to a different loop

  ★ 原因：asyncio 的很多对象（Lock、Queue、连接、Future）
    ★绑定在创建它们的那个 loop 上★，不能跨 loop 使用

  ✓ 三种解法：
    ① ★把 loop 作用域也提到 session★
       # pyproject.toml
       asyncio_default_fixture_loop_scope = "session"
       # 或标记：@pytest.mark.asyncio(loop_scope="session")
    ② ★把 fixture 降为 function 作用域★（每个测试新建，简单但慢）
    ③ 用同步 fixture 保存"配置"，在测试里再创建异步资源

★ 作用域搭配的原则：
  ★异步资源的 fixture 作用域 ≤ 事件循环的作用域★
  ┌──────────────┬──────────────┬──────────┐
  │ fixture 作用域│ loop 作用域   │ 结果      │
  ├──────────────┼──────────────┼──────────┤
  │ function     │ function     │ ✅        │
  │ session      │ session      │ ✅        │
  │ ★session★   │ ★function★   │ ❌ ★报错★ │
  │ function     │ session      │ ✅（浪费）│
  └──────────────┴──────────────┴──────────┘

★ 典型的 fixture 组织：
  # conftest.py
  @pytest_asyncio.fixture(scope="session")
  async def engine():                    # ★重资源：整个测试会话共用★
      e = create_async_engine(TEST_DB_URL)
      yield e
      await e.dispose()

  @pytest_asyncio.fixture                # ★function 级：每个测试一个事务★
  async def session(engine):
      async with engine.connect() as conn:
          txn = await conn.begin()
          async with AsyncSession(bind=conn) as s:
              yield s
          await txn.rollback()           # ★回滚，保证测试隔离★

★ 注意 teardown 的异常：
  yield 之后的代码在 teardown 阶段执行
  → ★如果测试失败，teardown 仍然会执行★
  → teardown 里的 await 如果超时/报错，会掩盖真正的失败原因
  ✓ teardown 里加 try/except 或超时保护

异步 fixture 用 @pytest_asyncio.fixture（auto 模式下 @pytest.fixture 也能识别），yield 前是 setup、后是 teardown。最常见的报错是作用域不匹配：一个 scope="session" 的异步 fixture 在第一个测试的 loop 上创建了资源，第二个测试用新 loop 时就报 got Future attached to a different loop——因为 asyncio 的很多对象（Lock、Queue、连接、Future）绑定在创建它们的那个 loop 上。原则是**「异步资源的 fixture 作用域 ≤ 事件循环的作用域」，解法要么把 loop 作用域也提到 session（asyncio_default_fixture_loop_scope = "session"），要么把 fixture 降为 function 级。典型的组织方式是：session 级的 engine（重资源共用）+ function 级的 session（每个测试一个事务，结束时回滚保证隔离）。还有个易忽略的点：teardown 在测试失败时仍会执行，如果它自己超时或报错会掩盖真正的失败原因**，所以要加保护。

四、mock 异步函数

★ AsyncMock（Python 3.8+）：
  from unittest.mock import AsyncMock, MagicMock

  m = MagicMock()
  await m()                    # ✗ TypeError: object MagicMock can't be used
                               #   in 'await' expression
  m = AsyncMock()
  await m()                    # ✓ 返回 AsyncMock 的默认值
  m.return_value = 42
  assert await m() == 42       # ★return_value 是 await 之后的值★

★ patch 会自动选择（★最省心★）：
  # mymod.py
  async def fetch(url): ...
  def parse(data): ...

  with patch("mymod.fetch") as mf, patch("mymod.parse") as mp:
      # ★mf 自动是 AsyncMock（因为 fetch 是协程函数）★
      # ★mp 是普通 MagicMock★
      mf.return_value = "raw"
      mp.return_value = {"k": "v"}
  ★ 依据：patch 检查目标是否 asyncio.iscoroutinefunction

★ 手动指定：
  patch("mymod.fetch", new_callable=AsyncMock)
  patch.object(client, "get", AsyncMock(return_value="x"))
  ★ 用 autospec=True 能同时保证签名正确：
    patch("mymod.fetch", autospec=True)   # ★参数不匹配会报错★

★ 异步专用断言：
  m.assert_awaited()                 # 至少被 await 一次
  m.assert_awaited_once()
  m.assert_awaited_with(1, 2)
  m.assert_awaited_once_with(url="x")
  m.assert_any_await(...)
  m.await_count / m.await_args / m.await_args_list
  ★ 注意区别：called/assert_called_* 只表示"被调用（创建了协程）"，
    ★awaited/assert_awaited_* 才表示"真的被 await 了"★
  → 测"是否真的执行了"要用 awaited 系列

★ side_effect 的用法：
  m = AsyncMock(side_effect=[1, 2, 3])          # 依次返回
  m = AsyncMock(side_effect=TimeoutError)        # ★抛异常★
  m = AsyncMock(side_effect=lambda x: x * 2)     # ★同步函数也行★
  async def fake(x): return x * 2
  m = AsyncMock(side_effect=fake)                # ★异步函数★

★ mock 异步上下文管理器 / 迭代器（★容易卡住的地方★）：
  # async with
  cm = AsyncMock()
  cm.__aenter__.return_value = "resource"
  cm.__aexit__.return_value = False
  # ★MagicMock 不支持 __aenter__，要用 AsyncMock 或 MagicMock 手动加★

  # async for
  m = MagicMock()
  m.__aiter__.return_value = iter([1, 2, 3])     # ★3.8+ 支持★
  # 或者用真实的异步生成器代替 mock（★往往更简单★）
  async def fake_stream():
      for i in [1, 2, 3]:
          yield i

★ 什么时候不该 mock：
  ✗ mock 掉整个异步库 → ★测的是 mock 不是你的代码★
  ✓ 优先用：真实的内存实现（如 aiosqlite）、
            测试容器（testcontainers）、
            本地假服务器（aiohttp 的 test_utils）
  ✓ mock 只用于：★外部依赖、慢操作、难以构造的错误路径★

AsyncMock（3.8+）解决的是「MagicMock 的返回值不能被 await」的问题。最省心的是 patch() 会自动检测目标是不是协程函数并自动选用 AsyncMock（依据 asyncio.iscoroutinefunction），所以直接 patch("mymod.fetch") 就对了；手动构造时才需要显式写 AsyncMock() 或 new_callable=AsyncMock（加 autospec=True 还能校验参数签名）。异步专用断言要分清：assert_called_* 只表示「被调用（创建了协程）」，assert_awaited_* 才表示「真的被 await 了」——测「是否真的执行」必须用后者。两个容易卡住的地方：mock 异步上下文管理器（要设置 __aenter__/__aexit__）和异步迭代器（设置 __aiter__，但往往直接写一个真实的异步生成器更简单）。最后一个原则：mock 只用于外部依赖、慢操作和难以构造的错误路径，把整个异步库都 mock 掉就变成「测 mock 而不是测你的代码」了。

五、测试异步特有的行为

★ 测超时：
  async def test_timeout():
      with pytest.raises(asyncio.TimeoutError):
          await asyncio.wait_for(never_returns(), timeout=0.05)
  ★ 用很小的超时值让测试跑得快
  ★ 3.11+ 也可以测 asyncio.timeout 上下文管理器

★ 测取消与清理（★重要★）：
  async def test_cleanup_on_cancel():
      released = []
      async def work():
          try:
              await asyncio.sleep(10)
          except asyncio.CancelledError:
              released.append("cleanup")
              raise                          # ★必须重新抛出★
      t = asyncio.create_task(work())
      await asyncio.sleep(0)                 # ★让任务真正开始执行★
      t.cancel()
      with pytest.raises(asyncio.CancelledError):
          await t
      assert released == ["cleanup"]         # ★验证清理逻辑执行了★

  ★ await asyncio.sleep(0) 的作用：
    create_task 只是"排队"，★协程还没开始跑★
    sleep(0) 让出一次控制权 → 任务被调度执行到第一个 await
    → ★不加这一行，cancel 可能在任务还没开始时就发生（清理逻辑不会执行）★

★ 测并发正确性：
  async def test_no_race():
      counter = SafeCounter()
      await asyncio.gather(*(counter.incr() for _ in range(1000)))
      assert counter.value == 1000           # ★最终一致性★

  ★ 想放大竞态：在关键点插入 await asyncio.sleep(0)
    （强制让出控制权，模拟"最坏的交错"）

★ 控制时间（★别在测试里真的 sleep★）：
  ✗ await asyncio.sleep(60)      # ★测试跑一分钟★
  ✓ 方案 A：把时间做成可注入的参数（sleep_fn=asyncio.sleep）
  ✓ 方案 B：mock 掉 asyncio.sleep
     with patch("asyncio.sleep", new_callable=AsyncMock):
         await retry_with_backoff(...)       # ★退避不再真的等★
  ✓ 方案 C：用小的时间值（0.01 秒）
  ✓ 方案 D：用 time-machine / freezegun（★对 asyncio.sleep 支持有限★）

★ 测异步生成器 / 异步上下文管理器：
  async def test_agen():
      items = [x async for x in stream()]    # ★异步推导式★
      assert items == [1, 2, 3]

  async def test_acm():
      async with resource() as r:
          assert r.ready
      assert r.closed                        # ★验证 __aexit__ 执行了★

★ 测试后检查有没有遗留任务（★发现任务泄漏★）：
  @pytest.fixture(autouse=True)
  async def no_leaked_tasks():
      before = asyncio.all_tasks()
      yield
      leaked = asyncio.all_tasks() - before - {asyncio.current_task()}
      assert not leaked, f"泄漏了任务: {[t.get_name() for t in leaked]}"
  ★ 这个 fixture 能在测试阶段就发现"忘了 cancel"的问题

测试异步特有的行为有几个要点。测取消时有个关键细节：create_task 之后要 await asyncio.sleep(0) 让任务真正开始执行——因为 create_task 只是把任务排进队列，协程还没跑；不加这一行，cancel() 可能在任务尚未开始时就发生，清理逻辑根本不会执行，测试就失去了意义。测并发正确性用 gather 大量并发后断言最终一致性，想放大竞态可以在关键点插入 await asyncio.sleep(0) 强制让出控制权。绝不要在测试里真的 sleep 很久——把时间做成可注入的参数、mock 掉 asyncio.sleep、或直接用很小的时间值。最后推荐一个非常实用的 autouse fixture：对比测试前后的 asyncio.all_tasks()，断言没有遗留任务——它能在测试阶段就发现「忘了 cancel」这类任务泄漏问题。

六、常见坑与实践清单

★ 坑一：测试"假绿"★
  没装 pytest-asyncio / 忘了标记 → 协程没执行却 PASSED
  ✓ CI 里 filterwarnings = error
  ✓ 定期写个 assert False 的异步测试验证配置有效

★ 坑二：跨 loop 使用对象★
  RuntimeError: ... attached to a different loop
  → ★fixture 作用域 > loop 作用域★
  ✓ 统一作用域，或把资源降为 function 级

★ 坑三：全局状态污染★
  异步代码里常见的全局：连接池单例、ContextVar、缓存
  → 测试之间互相影响
  ✓ 用 fixture 重置；ContextVar 用 token + reset
  ✓ 每个测试独立的事件循环有帮助，但★全局变量不会自动重置★

★ 坑四：测试里真的发网络请求★
  → 慢、不稳定、CI 里可能没网
  ✓ ★aiohttp 的 test_utils / aioresponses / respx（httpx）★
  ✓ 或起一个本地的测试服务器（fixture 管理生命周期）

★ 坑五：event_loop fixture 的旧写法★
  自定义 event_loop fixture ★在新版 pytest-asyncio 已弃用★
  ✓ 用 asyncio_default_fixture_loop_scope / loop_scope 标记

★ 坑六：测试之间的任务泄漏★
  上一个测试留下的后台任务在下一个测试里还在跑
  ✓ ★autouse fixture 检查 all_tasks★
  ✓ 或在 teardown 里 cancel 所有非当前任务

★ 坑七：mock 的 await 断言用错★
  assert_called_once() 只说明"协程被创建了"
  ★assert_awaited_once() 才说明"真的执行了"★

★ 实践清单：
  □ ★asyncio_mode = "auto"★（省装饰器）
  □ ★asyncio_default_fixture_loop_scope 显式设置★（避免警告）
  □ ★filterwarnings = error★（防假绿 + 抓 never awaited）
  □ 异步 fixture 的作用域 ≤ loop 作用域
  □ mock 异步函数用 ★AsyncMock★（patch 通常自动）
  □ 断言用 ★assert_awaited_*★ 而不是 assert_called_*
  □ ★测取消前先 await asyncio.sleep(0)★
  □ 不在测试里真 sleep（★mock 掉或用小值★）
  □ ★autouse fixture 检查任务泄漏★
  □ 外部依赖用假服务器而不是 mock 整个客户端
  □ 数据库测试用★事务 + 回滚★保证隔离

★ 一句话总结：
  ★"异步测试的三个关键：让协程真的被执行（pytest-asyncio + 防假绿）、
    让 mock 可以被 await（AsyncMock + assert_awaited）、
    让 fixture 和事件循环的作用域对齐（跨 loop 就报错）。"★

七个坑里最危险的是**「测试假绿」（协程没执行却 PASSED，用 filterwarnings = error 防住）和跨 loop 使用对象**（fixture 作用域大于 loop 作用域）。其余几个：全局状态污染（连接池单例、ContextVar、缓存不会随 loop 重建而重置，要用 fixture 显式重置）、测试里真的发网络请求（用 aioresponses/respx 或本地测试服务器）、event_loop fixture 的旧写法已弃用、测试之间的任务泄漏、以及 assert_called_once 和 assert_awaited_once 用混。实践清单里最关键的五条：开 auto 模式、显式设置 loop 作用域、filterwarnings = error 防假绿、测取消前先 await asyncio.sleep(0)、用 autouse fixture 检查任务泄漏。一句话总结：异步测试的三个关键是「让协程真的被执行」「让 mock 可以被 await」「让 fixture 和事件循环的作用域对齐」。

记忆钩子：「★异步测试最危险的坑是『假绿』★——pytest 默认不认识协程函数，async def test_x() 只是被调用后得到一个协程对象就丢弃，★测试体一行都没执行却显示 PASSED★，唯一线索是容易被忽略的 coroutine was never awaited 警告 → ★CI 里配 filterwarnings = error 杜绝它★。解法是装 ★pytest-asyncio★ 并开 ★asyncio_mode = ‘auto’★（省去满屏 @pytest.mark.asyncio），标准库的替代是 ★unittest.IsolatedAsyncioTestCase（3.8+）★，需要同时支持 trio 就用 anyio 的插件。★第二个关键是 mock：MagicMock 的返回值不能被 await★（会 TypeError），要用 ★AsyncMock（3.8+）★——好消息是 ★patch() 会自动检测目标是不是协程函数并自动选用★；断言要分清 ★assert_called_* 只说明『协程被创建』，assert_awaited_* 才说明『真的被 await 执行了』★。★第三个关键是 fixture 与 loop 的作用域必须对齐★：asyncio 的 Lock/Queue/连接/Future ★都绑定在创建它们的那个 loop 上★，所以 session 级的异步 fixture 配 function 级的 loop 会报 ★got Future attached to a different loop★——原则是『异步资源 fixture 的作用域 ≤ loop 作用域』，新版用 ★asyncio_default_fixture_loop_scope 配置或 @pytest.mark.asyncio(loop_scope=)★（★自定义 event_loop fixture 的老写法已弃用★）。测异步特有行为的三个技巧：★测取消前要先 await asyncio.sleep(0) 让任务真正开始★（否则 cancel 时协程还没跑、清理逻辑不会执行）、★别在测试里真 sleep★（mock 掉 asyncio.sleep 或用 0.01 这类小值）、★用 autouse fixture 对比前后的 asyncio.all_tasks() 断言没有任务泄漏★。」

七、常见误区与追问

误区：写了 async def test_x() 并且显示 PASSED，说明测试通过了。 如果没有装 pytest-asyncio（或者用 strict 模式却忘了加 @pytest.mark.asyncio），pytest 只是调用了这个函数、得到一个协程对象、然后丢弃——测试体里的所有断言一行都没执行，pytest 看到「没有抛异常」于是报告 PASSED。这是最危险的情况：你以为有测试保护，其实完全没有。唯一的线索是 RuntimeWarning: coroutine 'test_x' was never awaited，在满屏输出里极易被忽略。防护有两层：CI 里配置 filterwarnings = error 把警告变成失败，以及定期写一个 assert False 的异步测试来验证配置确实生效（它必须是 FAILED，如果是 PASSED 就说明协程没被执行）。
误区：用 MagicMock() 替换异步函数就行，mock 都一样。 MagicMock() 的返回值是普通对象，不能被 await——会抛 TypeError: object MagicMock can't be used in 'await' expression。Python 3.8 起提供了 AsyncMock：它的调用返回一个可等待对象，return_value 是 await 之后得到的值，还提供 assert_awaited_* 系列断言。好消息是 patch() 会自动检测——它用 asyncio.iscoroutinefunction 判断被替换的目标是不是协程函数，是的话自动使用 AsyncMock，所以大多数时候直接 patch("mod.fetch") 就正确。只有手动构造 mock 对象（AsyncMock(return_value=...)）或 patch 一个属性而非模块级函数时才需要显式指定 new_callable=AsyncMock。
误区：assert_called_once() 能验证异步函数被执行了。 它只能证明「协程对象被创建了」，不能证明「协程被 await 执行了」。因为调用一个协程函数（哪怕是 mock）只是创建对象，真正的执行发生在 await 时——如果被测代码忘了 await，assert_called_once() 依然会通过，而实际逻辑根本没跑。必须用 assert_awaited_once()（以及 assert_awaited_with、await_count、await_args_list 这一族），它们检查的是 AsyncMock 的 await 记录。这个区别在排查「忘了 await」这类 bug 时尤其重要——恰恰是这类 bug 最需要测试来兜住，而用错断言就会漏掉。
误区：把重资源（数据库连接池、HTTP 客户端）做成 session 作用域的异步 fixture 能加快测试。 想法没错，但会撞上 RuntimeError: got Future attached to a different loop。原因是 pytest-asyncio 默认为每个测试函数创建一个全新的事件循环，而 asyncio 的很多对象（Lock、Queue、Future、数据库连接、HTTP 会话）在创建时就绑定到了当时的那个 loop——第一个测试里创建的连接池到了第二个测试（新 loop）就不能用了。解决办法是让 loop 的作用域和 fixture 的作用域对齐：设置 asyncio_default_fixture_loop_scope = "session" 或给测试标记 @pytest.mark.asyncio(loop_scope="session")。原则是「异步资源 fixture 的作用域 ≤ 事件循环的作用域」。注意自定义 event_loop fixture 这个老写法在新版 pytest-asyncio 里已被弃用。
误区：测试取消行为时，create_task 之后直接 cancel() 就能验证清理逻辑。 很可能什么都测不到。因为 asyncio.create_task() 只是把任务排进事件循环的队列，协程此时一行都还没执行；如果紧接着就 cancel()，任务会在尚未开始的状态下被取消——try/finally 里的清理代码根本不会运行，而你的断言可能因此失败（或者更糟：因为别的原因碰巧通过了）。正确做法是在 cancel() 之前插入 await asyncio.sleep(0)：它让出一次控制权，使任务被调度并执行到第一个 await 挂起点，这时再取消才会真正走到 except CancelledError / finally 分支。这个技巧在测试「取消时是否释放了资源」「是否重新抛出了 CancelledError」时是必需的。
追问：pytest-asyncio 的 strict 和 auto 模式该怎么选？ auto 模式适合绝大多数纯 asyncio 项目：所有 async def test_* 和 async def fixture 都会被自动识别，省去满屏的 @pytest.mark.asyncio 装饰器，新加测试时也不会忘记标记（忘记标记正是「假绿」的主要来源）。strict 模式（默认值）的价值在于「明确」：当项目里同时存在多个异步测试插件时（比如同时用 pytest-asyncio 和 anyio 的插件，或者部分测试用 trio），auto 模式会「抢走」所有异步测试导致冲突，此时必须用 strict 模式显式标记各自的归属。所以决策很简单：单一 asyncio 后端 → auto；混用多种异步框架 → strict。无论选哪种，都建议配上 filterwarnings = error 作为兜底，防止漏标记时静默通过。
追问：异步代码里的外部依赖（HTTP、数据库）应该 mock 还是用真实的？ 分层考虑。单元测试层：mock 掉外部调用是合理的，但要 mock 在正确的边界上——mock 你自己的「客户端封装层」而不是 aiohttp 本身，否则测的是 mock 的行为而非你的代码。更好的选择是用专门的假服务器/拦截库：HTTP 用 aioresponses（aiohttp）或 respx（httpx）在传输层拦截、或用 aiohttp.test_utils 起一个真实的本地服务器；数据库用 aiosqlite 这类内存实现。集成测试层：用真实的依赖 + 测试容器（testcontainers 启动真实的 PostgreSQL/Redis），配合「每个测试一个事务、结束时回滚」保证隔离——这是数据库测试的标准做法，既真实又快。原则是：mock 只用于「外部依赖、慢操作、难以构造的错误路径（超时、连接重置）」，能用真实的轻量实现就别 mock。
追问：怎么在测试里发现「任务泄漏」这类问题？ 用一个 autouse 的 fixture 对比测试前后的任务集合：before = asyncio.all_tasks() → yield → 测试结束后计算 asyncio.all_tasks() - before - {current_task}，断言这个差集为空，否则报出泄漏任务的名字。这个 fixture 能在测试阶段就抓住「后台 worker 忘了 cancel」「fire-and-forget 的任务没有结束条件」这类问题——而这些问题在生产环境往往要跑几小时才暴露成内存增长。配套的两个做法：给所有 create_task 加 name=（这样报错信息能直接指出是哪个任务），以及在 fixture 的 teardown 里主动 cancel 残留任务并 gather(return_exceptions=True)，避免它们污染下一个测试。此外，pytest-asyncio 每个测试用独立 loop 的默认行为本身也有帮助（loop 关闭时会警告 Task was destroyed but it is pending!），但全局变量和单例不会随 loop 重建而重置，仍需显式处理。

八、加强记忆

异步测试最危险的坑是「假绿」——pytest 默认不认识协程函数，async def test_x() 只是被调用后得到一个协程对象就丢弃，测试体一行都没执行却显示 PASSED，唯一线索是容易被忽略的 coroutine was never awaited 警告 → CI 里配 filterwarnings = error 杜绝它，并定期用一个 assert False 的异步测试验证配置有效。解法是装 pytest-asyncio 并开 asyncio_mode = "auto"（省去满屏 @pytest.mark.asyncio，也避免漏标记；混用多种异步框架时才用 strict）；标准库的替代是 unittest.IsolatedAsyncioTestCase（3.8+），需要同时支持 trio 就用 anyio 的插件。第二个关键是 mock：MagicMock 的返回值不能被 await（会 TypeError），要用 AsyncMock（3.8+）——好消息是 patch() 会自动检测目标是不是协程函数并自动选用；断言要分清 assert_called_* 只说明「协程被创建」，assert_awaited_* 才说明「真的被 await 执行了」（这个区别在抓「忘了 await」的 bug 时尤其重要）。第三个关键是 fixture 与 loop 的作用域必须对齐：asyncio 的 Lock/Queue/连接/Future 都绑定在创建它们的那个 loop 上，所以 session 级的异步 fixture 配 function 级的 loop 会报 got Future attached to a different loop——原则是「异步资源 fixture 的作用域 ≤ loop 作用域」，新版用 asyncio_default_fixture_loop_scope 配置或 @pytest.mark.asyncio(loop_scope=...)（自定义 event_loop fixture 的老写法已弃用）。测异步特有行为的三个技巧：测取消前要先 await asyncio.sleep(0) 让任务真正开始执行（否则 cancel 时协程还没跑、清理逻辑不会执行）、别在测试里真 sleep（mock 掉 asyncio.sleep 或用 0.01 这类小值）、用 autouse fixture 对比前后的 asyncio.all_tasks() 断言没有任务泄漏。外部依赖优先用假服务器（aioresponses/respx）或测试容器 + 事务回滚，而不是 mock 掉整个客户端。

简化版

详细版