Access the event loop directly via asyncio due to pytest's fixture de…

…precation

Signed-off-by: Sergey Vasilyev <[email protected]>

requirements.txt

@@ -17,7 +17,7 @@ pre-commit
 pyngrok
 pytest>=6.0.0
 pytest-aiohttp
-pytest-asyncio<0.22  # until the "event_loop" deprecation is solved
+pytest-asyncio
 pytest-cov
 pytest-mock
 pytest-timeout

tests/conftest.py

@@ -703,8 +703,13 @@ def assert_logs_fn(patterns, prohibited=[], strict=False):
 #
 # Helpers for asyncio checks.
 #
+@pytest.fixture()
+async def loop():
+    yield asyncio.get_running_loop()
+
+
 @pytest.fixture(autouse=True)
-def _no_asyncio_pending_tasks(event_loop):
+def _no_asyncio_pending_tasks(loop: asyncio.AbstractEventLoop):
     """
     Ensure there are no unattended asyncio tasks after the test.
 
@@ -725,7 +730,7 @@ def _no_asyncio_pending_tasks(event_loop):
 
     # Let the pytest-asyncio's async2sync wrapper to finish all callbacks. Otherwise, it raises:
     #   <Task pending name='Task-2' coro=<<async_generator_athrow without __name__>()>>
-    event_loop.run_until_complete(asyncio.sleep(0))
+    loop.run_until_complete(asyncio.sleep(0))
 
     # Detect all leftover tasks.
     after = _get_all_tasks()

tests/logging/conftest.py

@@ -11,10 +11,10 @@ def _caplog_all_levels(caplog):
 
 
 @pytest.fixture(autouse=True)
-def event_queue_loop(event_loop):
-    token = event_queue_loop_var.set(event_loop)
+def event_queue_loop(loop):  # must be sync-def
+    token = event_queue_loop_var.set(loop)
     try:
-        yield event_loop
+        yield loop
     finally:
         event_queue_loop_var.reset(token)
 

tests/posting/conftest.py

@@ -6,10 +6,10 @@
 
 
 @pytest.fixture()
-def event_queue_loop(event_loop):
-    token = event_queue_loop_var.set(event_loop)
+def event_queue_loop(loop):  # must be sync-def
+    token = event_queue_loop_var.set(loop)
     try:
-        yield event_loop
+        yield loop
     finally:
         event_queue_loop_var.reset(token)
 

tests/posting/test_threadsafety.py

@@ -49,14 +49,14 @@
 
 
 @pytest.fixture()
-def awakener(event_loop):
+def awakener():
     handles = []
 
     def noop():
         pass
 
     def awaken_fn(delay, fn=noop):
-        handle = event_loop.call_later(delay, fn)
+        handle = asyncio.get_running_loop().call_later(delay, fn)
         handles.append(handle)
 
     try:

tests/primitives/test_conditions.py

@@ -18,7 +18,7 @@ async def test_no_triggering():
         await asyncio.wait([task])
 
 
-async def test_triggering(event_loop, timer):
+async def test_triggering(timer):
     source = asyncio.Condition()
     target = asyncio.Condition()
     task = asyncio.create_task(condition_chain(source, target))
@@ -28,7 +28,8 @@ async def delayed_trigger():
             async with source:
                 source.notify_all()
 
-        event_loop.call_later(0.1, asyncio.create_task, delayed_trigger())
+        loop = asyncio.get_running_loop()
+        loop.call_later(0.1, asyncio.create_task, delayed_trigger())
 
         with timer:
             async with target:

tests/primitives/test_containers.py

@@ -11,19 +11,20 @@ async def test_empty_by_default():
         await asyncio.wait_for(container.wait(), timeout=0.1)
 
 
-async def test_does_not_wake_up_when_reset(event_loop, timer):
+async def test_does_not_wake_up_when_reset(timer):
     container = Container()
 
     async def reset_it():
         await container.reset()
 
-    event_loop.call_later(0.05, asyncio.create_task, reset_it())
+    loop = asyncio.get_running_loop()
+    loop.call_later(0.05, asyncio.create_task, reset_it())
 
     with pytest.raises(asyncio.TimeoutError):
         await asyncio.wait_for(container.wait(), timeout=0.1)
 
 
-async def test_wakes_up_when_preset(event_loop, timer):
+async def test_wakes_up_when_preset(timer):
     container = Container()
     await container.set(123)
 
@@ -34,13 +35,14 @@ async def test_wakes_up_when_preset(event_loop, timer):
     assert result == 123
 
 
-async def test_wakes_up_when_set(event_loop, timer):
+async def test_wakes_up_when_set(timer):
     container = Container()
 
     async def set_it():
         await container.set(123)
 
-    event_loop.call_later(0.1, asyncio.create_task, set_it())
+    loop = asyncio.get_running_loop()
+    loop.call_later(0.1, asyncio.create_task, set_it())
 
     with timer:
         result = await container.wait()
@@ -49,14 +51,15 @@ async def set_it():
     assert result == 123
 
 
-async def test_iterates_when_set(event_loop, timer):
+async def test_iterates_when_set(timer):
     container = Container()
 
     async def set_it(v):
         await container.set(v)
 
-    event_loop.call_later(0.1, asyncio.create_task, set_it(123))
-    event_loop.call_later(0.2, asyncio.create_task, set_it(234))
+    loop = asyncio.get_running_loop()
+    loop.call_later(0.1, asyncio.create_task, set_it(123))
+    loop.call_later(0.2, asyncio.create_task, set_it(234))
 
     values = []
     with timer:
@@ -69,7 +72,7 @@ async def set_it(v):
     assert values == [123, 234]
 
 
-async def test_iterates_when_preset(event_loop, timer):
+async def test_iterates_when_preset(timer):
     container = Container()
     await container.set(123)
 

tests/reactor/conftest.py

@@ -41,7 +41,7 @@ def watcher_limited(mocker, settings):
 
 
 @pytest.fixture()
-def watcher_in_background(settings, resource, event_loop, worker_spy, stream):
+async def watcher_in_background(settings, resource, worker_spy, stream):
 
     # Prevent remembering the streaming objects in the mocks.
     async def do_nothing(*args, **kwargs):
@@ -57,7 +57,7 @@ async def do_nothing(*args, **kwargs):
         settings=settings,
         processor=do_nothing,
     )
-    task = event_loop.create_task(coro)
+    task = asyncio.create_task(coro)
 
     try:
         # Go for a test.
@@ -66,6 +66,6 @@ async def do_nothing(*args, **kwargs):
         # Terminate the watcher to cleanup the loop.
         task.cancel()
         try:
-            event_loop.run_until_complete(task)
+            await task
         except asyncio.CancelledError:
             pass  # cancellations are expected at this point

tests/reactor/test_queueing.py

@@ -123,7 +123,7 @@ async def test_watchevent_demultiplexing(worker_mock, timer, resource, processor
 ])
 @pytest.mark.usefixtures('watcher_limited')
 async def test_watchevent_batching(settings, resource, processor, timer,
-                                   stream, events, uids, vals, event_loop):
+                                   stream, events, uids, vals):
     """ Verify that only the last event per uid is actually handled. """
 
     # Override the default timeouts to make the tests faster.