Python 单元测试：如何让 AsyncMock 阻塞直到某个点

Question

我有一些想要测试的异步Python代码，其中包括类似的内容

async def task(self):
    while True:
        await self.thing.change()
        ...
        self.run_callbacks()

在我的测试文件中，我在 setUp() 函数中启动任务，如下所示：

def setUp(self):
   ...
   self.loop = asyncio.get_event_loop()
   self.task = self.loop.create_task(self.object_under_test.task())

然后想要验证回调是否运行：

def test_callback(self):
    callback = mock.MagicMock()
    self.object_under_test.add_callback('cb1', callback)
    # ???
    callback.assert_called_once()

我陷入困境的是如何在任务中模拟

thing.change()

。它需要阻塞，直到我告诉它（在？？？），此时它会返回。如果我在启动任务后模拟它，那么它不会有任何区别，因为任务已经在等待未模拟的函数返回。如果我在开始任务之前模拟它，那么它总是会返回一些东西而不是阻塞。

有什么建议我可以实现这一目标吗？

Answer 1

我已经弄清楚了如何使用 asyncio.Event 对象进行同步并作为模拟的副作用：

# my_task.py
import asyncio

async def async_task():
    while True:
        result = await trigger()
        await asyncio.sleep(0)

async def trigger():
    # Some asynchronous function you want to mock
    await asyncio.sleep(1)
    print('triggered')
    return "Trigger Completed"

# test_my_task.py
import asyncio
import unittest
from unittest.mock import patch, AsyncMock
import my_task

class TestAsyncTask(unittest.TestCase):

    def setUp(self):
        self.trigger_event = asyncio.Event()
        async def mocked_trigger():
            await self.trigger_event.wait()
            return "Mocked Trigger Completed"
        my_task.trigger = AsyncMock(side_effect=mocked_trigger)

    def test_async_task_with_mocked_trigger(self):
        loop = asyncio.get_event_loop()

        task = loop.create_task(my_task.async_task())

        loop.run_until_complete(asyncio.sleep(1))
        self.trigger_event.set()
        loop.run_until_complete(asyncio.sleep(0))

        self.assertGreaterEqual(my_task.trigger.await_count, 1)

if __name__ == "__main__":
    unittest.main()