0

我正在尝试使用 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 时，由于该函数不是 asyncio 协程，因此我一直RuntimeError: There is no current event loop in thread 'MainThread'.无法@pytest.mark.asyncio用于此测试。launcher我在这里做错了什么？

4

1 回答 1

0

有关显式创建事件循环的信息，请参见此处：

RuntimeError：异步+ apscheduler中的线程中没有当前事件循环

请在此处特别查看 get_event_loop() 的文档：

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

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

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

于 2021-09-29T08:59:37.463 回答