refactor clock_event_loop fixture to become default event_loop

- uses a custom policy meta class that uses the existing policy to
  modify the new loop it creates.
- The new loop it creates is modified to provide `advance_time`
  coroutine for testing
- Extends pytest-dev#96 to make ClockEventLoop the default policy for testing

Author: derekbrokeit

pytest_asyncio/plugin.py

@@ -193,64 +193,62 @@ def pytest_runtest_setup(item):
 # to marked test functions
 _markers_2_fixtures = {
     'asyncio': 'event_loop',
-    'asyncio_clock': 'clock_event_loop',
 }
 
 
-@pytest.yield_fixture
-def event_loop(request):
-    """Create an instance of the default event loop for each test case."""
-    loop = asyncio.get_event_loop_policy().new_event_loop()
-    yield loop
-    loop.close()
-
-
-def _clock_event_loop_class():
+def _event_loop_policy():
     """
     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__):
+    class ClockEventLoopPolicy(asyncio.get_event_loop_policy().__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
+        def new_event_loop(self):
+            parent_loop = super().new_event_loop()
+            parent_loop.close()
+
+            class ClockEventLoop(parent_loop.__class__):
+                def __init__(self, *args, **kwargs):
+                    super().__init__(*args, **kwargs)
+                    self._clockoffset = 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._clockoffset
+
+                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._clockoffset += 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()
+    return ClockEventLoopPolicy()
 
 
 @pytest.yield_fixture
-def clock_event_loop(request):
+def 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)
+    # reset the event loop policy: modify existing policy to give time advancement
+    asyncio.set_event_loop_policy(_event_loop_policy())
+    loop = asyncio.get_event_loop_policy().new_event_loop()
     yield loop
     loop.close()
 

tests/test_simple_35.py

@@ -88,39 +88,22 @@ def test_async_close_loop(event_loop):
     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):
+@pytest.mark.asyncio
+async def test_event_loop_advance_time(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))
+    task = event_loop.create_task(asyncio.sleep(SLEEP_TIME))
     assert not task.done()
 
     # start the task
-    await clock_event_loop.advance_time(0)
+    await event_loop.advance_time(0)
     assert not task.done()
 
     # process the timeout
-    await clock_event_loop.advance_time(SLEEP_TIME)
+    await event_loop.advance_time(SLEEP_TIME)
     assert task.done()