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

Author: derekbrokeit

pytest_asyncio/plugin.py

@@ -159,6 +159,23 @@ def pytest_runtest_setup(item):
             item.fixturenames.append(fixture)
 
 
+class ClockEventLoop(asyncio.SelectorEventLoop):
+    _now = 0
+
+    def time(self):
+        return self._now
+
+    def advance_time(self, amount):
+        # advance the clock and run the loop
+        self._now += amount
+        self._run_once()
+
+        if amount > 0:
+            # Once advanced, new tasks may have just been scheduled for running
+            # in the next loop, advance once more to start these handlers
+            self._run_once()
+
+
 # maps marker to the name of the event loop fixture that will be available
 # to marked test functions
 _markers_2_fixtures = {
@@ -174,6 +191,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()
+
+
 @pytest.fixture
 def unused_tcp_port():
     """Find an unused localhost TCP port from 1024-65535 and return it."""

tests/test_simple_35.py

@@ -86,3 +86,22 @@ async def test_asyncio_marker_method(self, event_loop):
 def test_async_close_loop(event_loop):
     event_loop.close()
     return 'ok'
+
+
+def test_clock_loop(clock_event_loop):
+    assert clock_event_loop.time() == 0
+
+    async def foo():
+        await asyncio.sleep(1)
+
+    # create the task
+    task = clock_event_loop.create_task(foo())
+    assert not task.done()
+
+    # start the task
+    clock_event_loop.advance_time(0)
+    assert not task.done()
+
+    # process the timeout
+    clock_event_loop.advance_time(1)
+    assert task.done()