如何正确模拟 aiosqlite.execute 的异步上下文管理器？

2026-11-06数据小陈

结论

问题出在 conn = AsyncMock()。调用 AsyncMock 的 execute() 时会得到 coroutine。即使 side_effect 是普通同步函数，conn.execute(...) 返回的仍是 coroutine；async with 不会先 await 它，因此会报错。

这里的 execute() 需要兼容两种调用方式：建表和插入代码会执行 await conn.execute(...)，查询代码则直接执行 async with conn.execute(...)。因此，execute() 应直接返回一个既可等待、又支持异步上下文管理协议的对象。可以把连接 mock 改为 MagicMock，再让 execute 返回这个对象。

示例

from unittest.mock import AsyncMock, MagicMock, Mock, patch

cursor = AsyncMock()
cursor.fetchone.return_value = (2,)

class AsyncExecuteResult:
    def __init__(self, cursor):
        self.cursor = cursor

    async def __aenter__(self):
        return self.cursor

    async def __aexit__(self, exc_type, exc, tb):
        return False

    def __await__(self):
        async def result():
            return self.cursor

        return result().__await__()

conn = MagicMock()
conn.execute = Mock(
    side_effect=lambda *args, **kwargs: AsyncExecuteResult(cursor)
)

async def fake_connect(*args, **kwargs):
    return conn

with patch("aiosqlite.connect", side_effect=fake_connect):
    response = client.post("/create-users", json=payload)

AsyncExecuteResult 通过 __await__ 支持建表和插入语句等待执行结果；__aenter__ 和 __aexit__ 则让查询语句可以使用 async with。conn.execute 使用普通 Mock，会直接返回这个对象，不会再把它包装成 coroutine。

cursor.fetchone.return_value 应设为 (2,)。fetchone() 由 AsyncMock 提供，调用后会返回可等待对象，等待后得到该元组。不要把返回值设成 AsyncMock(return_value=(2,))，否则等待 fetchone() 后得到的仍是 mock，而不是行数据。

注意事项

原代码在 async with 内又调用了 await cursor.close()。通常，异步上下文管理器会负责清理游标。确认所用实现的行为后，可以避免重复关闭。也可以把查询改为 cursor = await conn.execute(...)，再显式关闭游标；这样测试中的调用形式就会与原来的 async with 不同。