如何使用 pytest 测试异步函数？

Question

@pytest.fixture
def d_service():
    c = DService()
    return c

# @pytest.mark.asyncio  # tried it too
async def test_get_file_list(d_service):
    files = await d_service.get_file_list('')
    print(files)

然而，却出现了以下错误？

已收集 0 项 / 1 错误

===================================== 错误============== =====================
________________ 收集测试/e2e_tests/test_d.py 时出错 _________________
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:617: 在 __call__ 中
    返回 self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:222: 在 _hookexec 中
    返回 self._inner_hookexec(钩子、方法、kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:216：在
    第一个结果=hook.spec_opts.get('第一个结果'),
..\..\..\..\..\anaconda3\lib\site-packages\_pytest\python.py:171：在 pytest_pycollect_makeitem 中
    res = 结果.get_result()
..\..\..\..\..\anaconda3\lib\site-packages\anyio\pytest_plugin.py:98：在 pytest_pycollect_makeitem 中
    标记=collector.get_closest_marker('anyio')
E AttributeError：“模块”对象没有属性“get_closest_marker”
!!!!!!!!!!!!!!!!!!!!! 已中断：收集期间出现 1 个错误!!!!!!!!!!!!!!!!!!
=========================== 2.53 秒内出现 1 个错误 ================== =========

---

我安装了以下软件包。错误消失，但测试被跳过。

pip install pytest-asyncio  

（基础）PS>pytest -s 测试\e2e_tests\test_d.py
=================================================== =================================================== ==============测试会话开始================================== =================================================== ==================================
平台 win32 -- Python 3.6.4、pytest-6.2.5、py-1.11.0、pluggy-1.0.0
根目录：C:\Users\X01324908\source\rds\research_data_science\sftp\file_handler
插件：anyio-3.3.4、asyncio-0.16.0
已收集 1 件

测试\e2e_tests\test_d.py

=================================================== =================================================== ================ 警告摘要 ================================== =================================================== ===================================
测试/e2e_tests/test_d.py::test_get_file_list
  c：\ users \ x01324908 \ anaconda3 \ lib \ site-packages \ _pytest \ python.py：172：PytestUnhandledCoroutineWarning：异步def函数不受本机支持，已被跳过。
  您需要为您的异步框架安装合适的插件，例如：
    - 安尼奥
    - pytest异步
    - pytest-tornasync
    - pytest三重奏
    - pytest 扭曲
    warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))

-- 文档：https://docs.pytest.org/en/stable/warnings.html
=============

Answer 1

这对我有用，请尝试：

import asyncio
import pytest

pytest_plugins = ('pytest_asyncio',)

@pytest.mark.asyncio
async def test_simple():
    await asyncio.sleep(0.5)

的输出pytest -v确认通过：

collected 1 item
test_async.py::test_simple PASSED

我已经安装了：

pytest                        6.2.5
pytest-asyncio                0.16.0
# anyio not installed

Answer 2

您可以通过在名为以下内​​容的文件夹中添加一个文件来pytest-asyncio自动检测async def test_*函数作为正确的测试：tests/pytest.ini

# pytest.ini
[pytest]
asyncio_mode=auto

这样，您甚至不需要装饰/标记您的async def测试，如文档的模式部分中所述。

Answer 3

自制解决方案

\n

在事件循环中运行测试协程的装饰器：

\n

import asyncio\nimport inspect\n\n\ndef asyncio_run(async_func):\n\n    def wrapper(*args, **kwargs):\n        return asyncio.run(async_func(*args, **kwargs))\n    \n    wrapper.__signature__ = inspect.signature(async_func)  # without this, fixtures are not injected\n\n    return wrapper\n\n\n@asyncio_run\nasync def test_get_file_list(d_service):\n    files = await d_service.get_file_list(\'\')\n    print(files)\n\n

\n

个人建议

\n

尽可能避免异步测试。尝试使大部分代码纯\xe2\x80\x94 那么你的大部分测试将是快速且可靠/确定性的（异步代码可能不是这种情况）。搜索“功能核心，命令式外壳”了解更多信息。

\n