You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何正确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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.10 16:33:00