add advance_time fixture and test (closes #83, #95, #96 #110)

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)

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