如何使用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)

但是，出现了以下错误？

collected 0 items / 1 errors

=================================== ERRORS ====================================
________________ ERROR collecting tests/e2e_tests/test_d.py _________________
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:617: in __call__
    return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:222: in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:216: in 
    firstresult=hook.spec_opts.get('firstresult'),
..\..\..\..\..\anaconda3\lib\site-packages\_pytest\python.py:171: in pytest_pycollect_makeitem
    res = outcome.get_result()
..\..\..\..\..\anaconda3\lib\site-packages\anyio\pytest_plugin.py:98: in pytest_pycollect_makeitem
    marker = collector.get_closest_marker('anyio')
E   AttributeError: 'Module' object has no attribute 'get_closest_marker'
!!!!!!!!!!!!!!!!!!! Interrupted: 1 errors during collection !!!!!!!!!!!!!!!!!!!
=========================== 1 error in 2.53 seconds ===========================

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

pip install pytest-asyncio

(base) PS>pytest -s tests\e2e_tests\test_d.py
================================================================================================================== test session starts ===================================================================================================================
platform win32 -- Python 3.6.4, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: C:\Users\X01324908\source\rds\research_data_science\sftp\file_handler
plugins: anyio-3.3.4, asyncio-0.16.0
collected 1 item

tests\e2e_tests\test_d.py s

==================================================================================================================== warnings summary ====================================================================================================================
tests/e2e_tests/test_d.py::test_get_file_list
  c:\users\x01324908\anaconda3\lib\site-packages\_pytest\python.py:172: PytestUnhandledCoroutineWarning: async def functions are not natively supported and have been skipped.
  You need to install a suitable plugin for your async framework, for example:
    - anyio
    - pytest-asyncio
    - pytest-tornasync
    - pytest-trio
    - pytest-twisted
    warnings.warn(PytestUnhandledCoroutineWarning(msg.format(nodeid)))

-- Docs: 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

自制溶液

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

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)

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

How to test async function using pytest?

python

pytest

自制溶液