如何使用 fixture 使 pytest 中的异步测试超时？

Question

我正在测试一个可能会死锁的异步函数。我试图添加一个固定装置以在引发故障之前将功能限制为仅运行 5 秒，但到目前为止它还没有奏效。

设置：

pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0

代码：

import asyncio
import pytest

@pytest.fixture
def my_fixture():
  # attempt to start a timer that will stop the test somehow
  asyncio.ensure_future(time_limit())
  yield 'eggs'


async def time_limit():
  await asyncio.sleep(5)
  print('time limit reached')     # this isn't printed
  raise AssertionError


@pytest.mark.asyncio
async def test(my_fixture):
  assert my_fixture == 'eggs'
  await asyncio.sleep(10)
  print('this should not print')  # this is printed
  assert 0

--

编辑：Mikhail 的解决方案工作正常。不过，我找不到将它合并到夹具中的方法。

Answer 1

使用超时限制函数（或代码块）的简便方法是使用 async-timeout 模块。您可以在测试函数中使用它，或者，例如，创建一个装饰器。与固定装置不同，它允许为每个测试指定具体时间：

import asyncio
import pytest
from async_timeout import timeout


def with_timeout(t):
    def wrapper(corofunc):
        async def run(*args, **kwargs):
            with timeout(t):
                return await corofunc(*args, **kwargs)
        return run       
    return wrapper


@pytest.mark.asyncio
@with_timeout(2)
async def test_sleep_1():
    await asyncio.sleep(1)
    assert 1 == 1


@pytest.mark.asyncio
@with_timeout(2)
async def test_sleep_3():
    await asyncio.sleep(3)
    assert 1 == 1

具体时间创建装饰器并不难(with_timeout_5 = partial(with_timeout, 5))。

我不知道如何创建纹理（如果你真的需要夹具），但上面的代码可以提供起点。也不确定是否有更好地实现目标的通用方法。

Answer 2

有一种方法可以使用 fixtures 来超时，只需要在 conftest.py.

中添加以下钩子即可

任何以 timeout 为前缀的 fixture 必须 return 几秒（int, float）测试才能运行.
选择最近的灯具 w.r.t 范围。 autouse 灯具的优先级低于明确选择的灯具。后一个是首选。不幸的是，函数参数列表中的顺序无关紧要。
如果没有这样的夹具，则测试不受限制，将照常运行无限期地进行。
测试也必须用 pytest.mark.asyncio 标记，但无论如何都是需要的。

# Add to conftest.py
import asyncio

import pytest

_TIMEOUT_FIXTURE_PREFIX = "timeout"


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_setup(item: pytest.Item):
    """Wrap all tests marked with pytest.mark.asyncio with their specified timeout.

    Must run as early as possible.

    Parameters
    ----------
    item : pytest.Item
        Test to wrap
    """
    yield
    orig_obj = item.obj
    timeouts = [n for n in item.funcargs if n.startswith(_TIMEOUT_FIXTURE_PREFIX)]
    # Picks the closest timeout fixture if there are multiple
    tname = None if len(timeouts) == 0 else timeouts[-1]

    # Only pick marked functions
    if item.get_closest_marker("asyncio") is not None and tname is not None:

        async def new_obj(*args, **kwargs):
            """Timed wrapper around the test function."""
            try:
                return await asyncio.wait_for(
                    orig_obj(*args, **kwargs), timeout=item.funcargs[tname]
                )
            except Exception as e:
                pytest.fail(f"Test {item.name} did not finish in time.")

        item.obj = new_obj

示例：

@pytest.fixture
def timeout_2s():
    return 2


@pytest.fixture(scope="module", autouse=True)
def timeout_5s():
    # You can do whatever you need here, just return/yield a number
    return 5


async def test_timeout_1():
    # Uses timeout_5s fixture by default
    await aio.sleep(0)  # Passes
    return 1


async def test_timeout_2(timeout_2s):
    # Uses timeout_2s because it is closest
    await aio.sleep(5)  # Timeouts

警告

可能和其他一些插件不兼容，我只用pytest-asyncio测试过，如果item被一些hook重新定义肯定不行。

如何使用 fixture 使 pytest 中的异步测试超时？

How to timeout an async test in pytest with fixture?

python

pytest

python-asyncio

pytest-asyncio