diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 24450bf7..37c63b24 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -159,6 +159,49 @@ def pytest_runtest_setup(item): ) +class EventLoopClockAdvancer: + """ + A helper object that when called will advance the event loop's time. If the + call is awaited, the caller task will wait an iteration for the update to + wake up any awaiting handlers. + """ + + __slots__ = ("offset", "loop", "sleep_duration", "_base_time") + + def __init__(self, loop, sleep_duration=1e-6): + self.offset = 0.0 + self._base_time = loop.time + self.loop = loop + self.sleep_duration = sleep_duration + + # incorporate offset timing into the event loop + self.loop.time = self.time + + def time(self): + """ + Return the time according to the event loop's clock. The time is + adjusted by an offset. + """ + return self._base_time() + self.offset + + async def __call__(self, seconds): + """ + Advance time by a given offset in seconds. Returns an awaitable + that will complete after all tasks scheduled for after advancement + of time are proceeding. + """ + # sleep so that the loop does everything currently waiting + await asyncio.sleep(self.sleep_duration) + + if seconds > 0: + # advance the clock by the given offset + self.offset += seconds + + # Once the clock is adjusted, new tasks may have just been + # scheduled for running in the next pass through the event loop + await asyncio.sleep(self.sleep_duration) + + @pytest.yield_fixture def event_loop(request): """Create an instance of the default event loop for each test case.""" @@ -195,3 +238,8 @@ def factory(): return port return factory + + +@pytest.fixture +def advance_time(event_loop, request): + return EventLoopClockAdvancer(event_loop) diff --git a/tests/test_simple_35.py b/tests/test_simple_35.py index 1e4d697c..00ea9aca 100644 --- a/tests/test_simple_35.py +++ b/tests/test_simple_35.py @@ -86,3 +86,75 @@ async def test_asyncio_marker_method(self, event_loop): def test_async_close_loop(event_loop): event_loop.close() return 'ok' + + +@pytest.mark.asyncio +async def test_advance_time_fixture(event_loop, advance_time): + """ + Test the `advance_time` fixture using a sleep timer + """ + # A task is created that will sleep some number of seconds + SLEEP_TIME = 10 + + # create the task + task = event_loop.create_task(asyncio.sleep(SLEEP_TIME)) + assert not task.done() + + # start the task + await advance_time(0) + assert not task.done() + + # process the timeout + await advance_time(SLEEP_TIME) + assert task.done() + + +@pytest.mark.asyncio +async def test_advance_time_fixture_call_later(event_loop, advance_time): + """ + Test the `advance_time` fixture using loop.call_later + """ + # A task is created that will sleep some number of seconds + SLEEP_TIME = 10 + result = [] + + # create a simple callback that adds a value to result + def callback(): + result.append(True) + + # create the task + event_loop.call_later(SLEEP_TIME, callback) + + # start the task + await advance_time(0) + assert not result + + # process the timeout + await advance_time(SLEEP_TIME) + assert result + + +@pytest.mark.asyncio +async def test_advance_time_fixture_coroutine(event_loop, advance_time): + """ + Test the `advance_time` fixture using loop.call_later + """ + # A task is created that will sleep some number of seconds + SLEEP_TIME = 10 + result = [] + + # create a simple callback that adds a value to result + async def callback(): + await asyncio.sleep(SLEEP_TIME) + result.append(True) + + # create the task + task = event_loop.create_task(callback()) + + # start the task + await advance_time(0) + assert not task.done() + + # process the timeout + await advance_time(SLEEP_TIME) + assert task.done() and result