Add ClockEventLoop class with fixture and test (close #95)

pytest_asyncio/plugin.py

@@ -189,10 +189,51 @@ def pytest_runtest_setup(item):
             )
 
 
+class ClockEventLoop(asyncio.new_event_loop().__class__):
+    """
+    A custom event loop that explicitly advances time when requested. Otherwise,
+    this event loop advances time as expected.
+    """
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._offset = 0
+
+    def time(self):
+        """
+        Return the time according the event loop's clock.
+
+        This time is adjusted by the stored offset that allows for advancement
+        with `advance_time`.
+        """
+        return super().time() + self._offset
+
+    async def advance_time(self, seconds):
+        '''
+        Advance time by a given offset in seconds.
+        '''
+        if seconds < 0:
+            # cannot go backwards in time, so return immediately
+            return
+
+        # advance the clock by the given offset
+        self._offset += seconds
+
+        # ensure waiting callbacks are run before advancing the clock
+        await asyncio.sleep(0, loop=self)
+
+        if seconds > 0:
+            # Once the clock is adjusted, new tasks may have just been scheduled for running
+            # in the next pass through the event loop and advance again for the task
+            # that calls `advance_time`
+            await asyncio.sleep(0, loop=self)
+            await asyncio.sleep(0, loop=self)
+
+
 # maps marker to the name of the event loop fixture that will be available
 # to marked test functions
 _markers_2_fixtures = {
     'asyncio': 'event_loop',
+    'asyncio_clock': 'clock_event_loop',
 }
 
 
@@ -204,6 +245,15 @@ def event_loop(request):
     loop.close()
 
 
+@pytest.yield_fixture
+def clock_event_loop(request):
+    """Create an instance of the default event loop for each test case."""
+    loop = ClockEventLoop()
+    asyncio.get_event_loop_policy().set_event_loop(loop)
+    yield loop
+    loop.close()
+
+
 def _unused_tcp_port():
     """Find an unused localhost TCP port from 1024-65535 and return it."""
     with contextlib.closing(socket.socket()) as sock:

tests/test_simple_35.py

@@ -86,3 +86,43 @@ async def test_asyncio_marker_method(self, event_loop):
 def test_async_close_loop(event_loop):
     event_loop.close()
     return 'ok'
+
+
+@pytest.mark.asyncio_clock
+async def test_mark_asyncio_clock():
+    """
+    Test that coroutines marked with asyncio_clock are run with a ClockEventLoop
+    """
+    import pytest_asyncio
+    assert isinstance(asyncio.get_event_loop(), pytest_asyncio.plugin.ClockEventLoop)
+
+
+def test_clock_loop_loop_fixture(clock_event_loop):
+    """
+    Test that the clock_event_loop fixture returns a proper instance of the loop
+    """
+    import pytest_asyncio
+    assert isinstance(asyncio.get_event_loop(), pytest_asyncio.plugin.ClockEventLoop)
+    clock_event_loop.close()
+    return 'ok'
+
+
+@pytest.mark.asyncio_clock
+async def test_clock_loop_advance_time(clock_event_loop):
+    """
+    Test the sliding time event loop fixture
+    """
+    async def short_nap():
+        await asyncio.sleep(1)
+
+    # create the task
+    task = clock_event_loop.create_task(short_nap())
+    assert not task.done()
+
+    # start the task
+    await clock_event_loop.advance_time(0)
+    assert not task.done()
+
+    # process the timeout
+    await clock_event_loop.advance_time(1)
+    assert task.done()