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

--

编辑：米哈伊尔的解决方案工作正常。但是，我无法找到将它合并到灯具中的方法。

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)）创建装饰器并不难。

---

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