Python单元测试：如何让AsyncMock阻塞至指定时机

你需要的是一个可手动触发返回的Mock，让thing.change()在测试启动后保持阻塞，直到你主动触发它继续执行。这里有两种简单可行的方案：

方案一：用asyncio.Event控制阻塞时机 #

利用asyncio.Event的wait()方法作为Mock的副作用，让change()一直等待事件触发：

1. 在setUp中提前配置Mock和控制事件：

def setUp(self):
    ...
    self.loop = asyncio.get_event_loop()
    # 创建一个事件，用来控制change()的返回时机
    self.change_trigger = asyncio.Event()
    # Mock thing.change方法，让它等待事件触发
    self.object_under_test.thing.change = mock.AsyncMock(side_effect=self.change_trigger.wait)
    # 启动任务，此时任务会卡在await self.thing.change()
    self.task = self.loop.create_task(self.object_under_test.task())

2. 在测试方法中触发事件并验证回调：

def test_callback(self):
    callback = mock.MagicMock()
    self.object_under_test.add_callback('cb1', callback)
    
    # 触发change()返回，让任务继续执行
    self.change_trigger.set()
    # 让事件循环处理完当前任务的剩余逻辑（执行回调）
    self.loop.run_until_complete(asyncio.sleep(0))
    
    callback.assert_called_once()

3. 别忘了在tearDown中取消任务，避免影响其他测试：

def tearDown(self):
    self.task.cancel()
    self.loop.run_until_complete(self.task)

方案二：用asyncio.Future控制返回 #

通过创建一个未完成的Future，让Mock返回它，直到你手动设置Future的结果：

1. 在setUp中配置：

def setUp(self):
    ...
    self.loop = asyncio.get_event_loop()
    # 创建一个未完成的Future
    self.change_future = self.loop.create_future()
    # Mock change方法返回这个Future
    self.object_under_test.thing.change = mock.AsyncMock(return_value=self.change_future)
    self.task = self.loop.create_task(self.object_under_test.task())

2. 测试方法中触发并验证：

def test_callback(self):
    callback = mock.MagicMock()
    self.object_under_test.add_callback('cb1', callback)
    
    # 设置Future完成，让change()返回
    self.change_future.set_result(None)
    self.loop.run_until_complete(asyncio.sleep(0))
    
    callback.assert_called_once()

原理说明 #

这两种方案的核心都是提前设置一个阻塞的Mock，让任务启动后卡在await self.thing.change()处，直到测试代码主动触发继续信号。调用asyncio.sleep(0)是为了让事件循环腾出时间，执行任务中await之后的回调逻辑，确保断言前回调已经被执行。

内容的提问来源于stack exchange，提问作者askvictor