11

我正在测试一个可能会死锁的异步函数。我尝试添加一个固定装置以限制该功能在引发故障之前仅运行 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

--

编辑：米哈伊尔的解决方案工作正常。不过，我找不到将其合并到固定装置中的方法。

4

2 回答 2

6

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

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)）。

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

于 2019-04-15T16:24:21.297 回答
0

有一种方法可以使用固定装置进行超时，只需将以下钩子添加到conftest.py.

  • 任何带有前缀的fixture必须返回测试可以运行timeout的秒数( int, )。float
  • 选择最接近的夹具 wrt 范围。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

于 2021-11-24T12:12:45.550 回答