foo

pytest_asyncio/plugin.py

@@ -166,6 +166,26 @@ def pytest_runtest_setup(item):
 }
 
 
+class EventLoopClockAdvancer:
+    __slots__ = ("offset", "_base_time",)
+    def __init__(self, loop):
+        self.offset = 0.0
+        self._base_time = loop.time
+        loop.time = self.time
+
+    def time(self):
+        return self._base_time() + self.offset
+
+    def __call__(self, seconds):
+        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
+        return self.create_task(asyncio.sleep(0))
+
+
 @pytest.yield_fixture
 def event_loop(request):
     """Create an instance of the default event loop for each test case."""
@@ -198,3 +218,8 @@ def factory():
 
         return port
     return factory
+
+
+@pytest.fixture
+def advance_time(event_loop):
+    return EventLoopClockAdvancer(event_loop)

tests/test_simple_35.py

@@ -86,3 +86,24 @@ 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(advance_time):
+    """
+    Test the `advance_time` fixture
+    """
+    # 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()