如何使用 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 时出错 _________________
..\..\..\..\.. naconda3\lib\site-packages\pluggy\__init__.py:617: 在 __call__ 中
    返回 self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
..\..\..\..\.. naconda3\lib\site-packages\pluggy\__init__.py:222: 在 _hookexec 中
    返回 self._inner_hookexec(钩子、方法、kwargs)
..\..\..\..\.. naconda3\lib\site-packages\pluggy\__init__.py:216: 在 
    第一个结果=hook.spec_opts.get('第一个结果'),
..\..\..\..\.. naconda3\lib\site-packages\_pytest\python.py:171：在 pytest_pycollect_makeitem 中
    res = 结果.get_result()
..\..\..\..\.. naconda3\lib\site-packages nyio\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 测试 2e_tests est_d.py
=================================================== =================================================== ==============测试会话开始================================== =================================================== =================================
平台 win32 -- Python 3.6.4、pytest-6.2.5、py-1.11.0、pluggy-1.0.0
根目录：C:\Users\X01324908\source
ds
esearch_data_science\sftp ile_handler
插件：anyio-3.3.4、asyncio-0.16.0
已收集 1 件

测试 2e_tests est_d.py s

=================================================== =================================================== ================ 警告摘要 ================================== =================================================== ===================================
测试/e2e_tests/test_d.py::test_get_file_list
  c:\users\x01324908 naconda3\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

自制解决方案

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

import asyncio
import inspect


def asyncio_run(async_func):

    def wrapper(*args, **kwargs):
        return asyncio.run(async_func(*args, **kwargs))
    
    wrapper.__signature__ = inspect.signature(async_func)  # without this, fixtures are not injected

    return wrapper


@asyncio_run
async def test_get_file_list(d_service):
    files = await d_service.get_file_list('')
    print(files)

个人建议

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

编辑： 您可以忽略与测试相关的“功能核心，命令式外壳”，尽管这对于一般 IMO 的设计仍然是有用的建议。 对于测试，您只需抽象效果类型。哦，这是Python吗？只需使用更好的语言，例如 Haskell 或 Scala。