I am testing an async function that might get deadlocked. I tried to add a fixture to limit the function to only run for 5 seconds before raising a failure, but it hasn't worked so far.
Setup:
pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0
Code:
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
--
Edit: Mikhail's solution works fine. I can't find a way to incorporate it into a fixture, though.
Convenient way to limit function (or block of code) with timeout is to use async-timeout module. You can use it inside your test function or, for example, create a decorator. Unlike with fixture it'll allow to specify concrete time for each test:
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
It's not hard to create decorator for concrete time (with_timeout_5 = partial(with_timeout, 5)).
I don't know how to create texture (if you really need fixture), but code above can provide starting point. Also not sure if there's a common way to achieve goal better.
