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.
Advertisement
Answer
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.