如何用夹具在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

--

编辑:米哈伊尔的解决方案很好。不过，我找不到办法把它集成到一个固定装置中。

Answer 1

限制函数(或代码块)超时的方便方法是使用异步超时模块。您可以在测试函数中使用它，或者，例如，创建一个装饰器。与夹具不同，它允许为每个测试指定具体的时间：

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

有一种用于超时的方法，只需在conftest.py中添加以下钩子即可。

• 任何以timeout为前缀的夹具都必须返回测试可以运行的若干秒(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被某些钩子重新定义，它肯定不能工作。

Answer 3

我只是喜欢用超时标记测试的昆比法。下面是我使用脓痕改进它的尝试

# tests/conftest.py
import asyncio


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_pyfunc_call(pyfuncitem: pytest.Function):
    """
    Wrap all tests marked with pytest.mark.async_timeout with their specified timeout.
    """
    orig_obj = pyfuncitem.obj

    if marker := pyfuncitem.get_closest_marker("async_timeout"):

        async def new_obj(*args, **kwargs):
            """Timed wrapper around the test function."""
            try:
                return await asyncio.wait_for(orig_obj(*args, **kwargs), timeout=marker.args[0])
            except (asyncio.CancelledError, asyncio.TimeoutError):
                pytest.fail(f"Test {pyfuncitem.name} did not finish in time.")

        pyfuncitem.obj = new_obj

    yield


def pytest_configure(config: pytest.Config):
    config.addinivalue_line("markers", "async_timeout(timeout): cancels the test execution after the specified amount of seconds")

用法：

@pytest.mark.asyncio
@pytest.mark.async_timeout(10)
async def potentially_hanging_function():
    await asyncio.sleep(20)

应该不难将其包含到pytest-asyncio中，因此我们可以获得如下语法：

@pytest.mark.asyncio(timeout=10)
async def potentially_hanging_function():
    await asyncio.sleep(20)

编辑:看起来像已经有公关了。