Как тайм-аут асинхронного теста в 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

Удобный способ ограничить функцию (или блок кода) тайм-аутом - использовать модуль 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

Мне просто понравился подход Куимби к маркировке тестов тайм-аутами. Вот моя попытка улучшить его, используя отметки pytest :

      # 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)

Нетрудно включить это в asyncioотметить на pytest-asyncio, поэтому мы можем получить такой синтаксис:

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

РЕДАКТИРОВАТЬ: похоже, что для этого уже есть PR .

Answer 3

Есть способ использовать фикстуры для тайм-аута, просто нужно добавить следующий хук в conftest.py.

• Любое приспособление с префиксом timeoutдолжен вернуть количество секунд ( int, float) тест может быть запущен.
• Выбирается ближайшее приспособление относительно прицела. 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 4

Вместо использования приспособления я решил это следующим образом, используя декоратор:

      def timeout(delay):
    def decorator(func):
        @wraps(func)
        async def new_func(*args, **kwargs):
            async with asyncio.timeout(delay):
                return await func(*args, **kwargs)

        return new_func

    return decorator
    

@pytest.mark.asyncio
@timeout(3)
async def test_forever_fails():
    await asyncio.Future()

Требуется Python 3.11.

Или я верюtrioпредоставляет что-то подобное для более ранних версий Python.