如何正确Mock和测试Async异步方法调用 解决Pytest运行时警告
根因分析
普通Mock对象仅支持模拟同步方法,当你用它替换异步方法poll后,事件循环调用poll时返回的协程对象没有被await处理,就会触发RuntimeWarning。你当前的测试仅验证了方法被调用,没有验证异步方法的实际执行状态,逻辑上存在缺陷。
正确实现方案
方案1:使用AsyncMock(推荐)
Python 3.8及以上版本自带的AsyncMock是专门用于模拟异步方法的工具,会自动处理await逻辑,不会产生未等待协程的警告。同时需要修正原有代码中事件循环调度异步方法的错误写法,避免业务代码本身也出现同类警告。
修改后的完整代码如下:
import asyncio from unittest.mock import patch, AsyncMock class Server: def __init__(self, test_mode=False): self.loop = asyncio.get_event_loop() self.test_mode = test_mode async def poll(self): print("Polling...") # 修正:调用异步方法需用create_task封装为任务再传给call_later self.loop.call_later(0.5, asyncio.create_task, self.poll()) def run(self): # 修正:调用异步方法需用create_task封装为任务再传给call_soon self.loop.call_soon(asyncio.create_task, self.poll()) if self.test_mode: self.loop.call_soon(self.loop.stop) self.loop.run_forever() @patch.object(Server, "poll", new_callable=AsyncMock) def test_that_poll_is_called(mock_poll: AsyncMock): Server(test_mode=True).run() # 使用异步专属断言,验证方法确实被等待执行 mock_poll.assert_awaited_once()
方案2:兼容旧版本Python的实现
如果使用的Python版本低于3.8没有AsyncMock,可以手动构造一个空协程赋值给mock的返回值,规避警告:
import asyncio from mock import patch async def empty_coroutine(): pass # 原有Server类代码不变 @patch.object(Server, "poll") def test_that_poll_is_called(mock_poll): mock_poll.return_value = empty_coroutine() Server(test_mode=True).run() mock_poll.assert_called_once()
注意事项
- 异步方法的mock验证优先使用
assert_awaited_once、assert_awaited_with这类AsyncMock专属的断言方法,比普通的assert_called_*系列断言更严谨,能确保异步方法被正确等待执行。 - 事件循环的
call_soon、call_later方法仅支持接收同步回调,直接传入异步方法会返回未被等待的协程,必须用asyncio.create_task封装后再传入。
内容的提问来源于stack exchange,提问作者Subhash
相关产品推荐
相关产品推荐

