Pytest 一个调用异步协程的函数答案

【问题标题】：Pytest a function which invokes asyncio coroutinesPytest 一个调用异步协程的函数
【发布时间】：2021-09-28 07:09:57
【问题描述】：

我正在尝试使用 mock 和 pytest-asyncio 编写单元测试用例。我有一个使用asyncio.run 启动异步协程的正常功能。 [使用python3.7]

import asyncio

async def sample_async(arg2):
     # do something
     proc = await asyncio.create_subprocess_shell(arg2)
     # some more asyncio calls
     rc = proc.returncode
     return rc

def launcher(arg1, arg2):
    if arg1 == "no_asyncio":
        # do something
        print("No asyncio")
        return 0
    else:
        return_code = asyncio.run(sample_async(arg2))
        # do something
        return return_code

我能够为 asyncio 函数 sample_async 编写单元测试，但不能为 launcher 编写单元测试。这是我尝试过的：

class AsyncMock(MagicMock):
    async def __call__(self, *args, **kwargs):
        return super(AsyncMock, self).__call__(*args, **kwargs)


@patch("asyncio.create_subprocess_shell", new_callable=AsyncMock)
def test_launcher(async_shell):
    arg1 = "async"
    arg2 = "/bin/bash ls"
    sample.launcher(arg1, arg2)
    async_shell.assert_called_once()

当我尝试运行 pytest 时，我不断收到 RuntimeError: There is no current event loop in thread 'MainThread'. 我无法在此测试中使用 @pytest.mark.asyncio，因为函数 launcher 不是异步协程。我在这里做错了什么？

【问题讨论】：

标签： python-asyncio pytest-asyncio

【解决方案1】：

查看这里关于显式创建事件循环：

RuntimeError: There is no current event loop in thread in async + apscheduler

请参阅此处的 get_event_loop() 文档：

https://docs.python.org/3/library/asyncio-eventloop.html

如果当前OS线程中没有设置当前事件循环，OS线程是main，并且还没有调用set_event_loop()，asyncio会创建一个新的事件循环并设置为当前。

很可能，因为您的操作系统线程不是主线程，您需要对循环进行一些手动初始化。确保您清楚地了解事件循环以及这些函数在其生命周期中的作用，将循环视为具有方法和其他对象（任务）的对象。正如文档所说，主操作系统线程为您做了一些“魔术”，而不仅仅是“幕后”。

【讨论】：