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

pytest_asyncio/plugin.py

@@ -193,6 +193,7 @@ def pytest_runtest_setup(item):
 # to marked test functions
 _markers_2_fixtures = {
  'asyncio': 'event_loop',
+ 'asyncio_clock': 'clock_event_loop',
 }
 
 
@@ -204,6 +205,56 @@ def event_loop(request):
  loop.close()
 
 
+def _clock_event_loop_class():
+ """
+ Create a new class for ClockEventLoop based on the current
+ class-type produced by `asyncio.new_event_loop()`. This is important
+ for instances in which the enent-loop-policy has been changed.
+ """
+ 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
+
+ def advance_time(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.
+ '''
+ 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))
+
+ return ClockEventLoop
+
+
+@pytest.yield_fixture
+def clock_event_loop(request):
+ """Create an instance of the default event loop for each test case."""
+ loop = _clock_event_loop_class()()
+ 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,41 @@ 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
+ """
+ assert hasattr(asyncio.get_event_loop(), 'advance_time')
+
+
+def test_clock_loop_loop_fixture(clock_event_loop):
+ """
+ Test that the clock_event_loop fixture returns a proper instance of the loop
+ """
+ assert hasattr(asyncio.get_event_loop(), 'advance_time')
+ 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
+ """
+ # A task is created that will sleep some number of seconds
+ SLEEP_TIME = 10
+
+ # create the task
+ task = clock_event_loop.create_task(asyncio.sleep(SLEEP_TIME))
+ 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(SLEEP_TIME)
+ assert task.done()