[3.9] bpo-45097: Remove incorrect deprecation warnings in asyncio. (GH-28153)

Deprecation warnings about the loop argument were incorrectly emitted
in cases when the loop argument was used inside the asyncio library,
not from user code.

Lib/asyncio/base_events.py

@@ -350,7 +350,7 @@ async def start_serving(self):
         self._start_serving()
         # Skip one loop iteration so that all 'loop.add_reader'
         # go through.
-        await tasks.sleep(0, loop=self._loop)
+        await tasks.sleep(0)
 
     async def serve_forever(self):
         if self._serving_forever_fut is not None:
@@ -539,7 +539,7 @@ async def shutdown_asyncgens(self):
         closing_agens = list(self._asyncgens)
         self._asyncgens.clear()
 
-        results = await tasks.gather(
+        results = await tasks._gather(
             *[ag.aclose() for ag in closing_agens],
             return_exceptions=True,
             loop=self)
@@ -1457,7 +1457,7 @@ async def create_server(
             fs = [self._create_server_getaddrinfo(host, port, family=family,
                                                   flags=flags)
                   for host in hosts]
-            infos = await tasks.gather(*fs, loop=self)
+            infos = await tasks._gather(*fs, loop=self)
             infos = set(itertools.chain.from_iterable(infos))
 
             completed = False
@@ -1515,7 +1515,7 @@ async def create_server(
             server._start_serving()
             # Skip one loop iteration so that all 'loop.add_reader'
             # go through.
-            await tasks.sleep(0, loop=self)
+            await tasks.sleep(0)
 
         if self._debug:
             logger.info("%r is serving", server)

Lib/asyncio/runners.py

@@ -61,7 +61,7 @@ def _cancel_all_tasks(loop):
         task.cancel()
 
     loop.run_until_complete(
-        tasks.gather(*to_cancel, loop=loop, return_exceptions=True))
+        tasks._gather(*to_cancel, loop=loop, return_exceptions=True))
 
     for task in to_cancel:
         if task.cancelled():

Lib/asyncio/subprocess.py

@@ -193,7 +193,7 @@ async def communicate(self, input=None):
             stderr = self._read_stream(2)
         else:
             stderr = self._noop()
-        stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr,
+        stdin, stdout, stderr = await tasks._gather(stdin, stdout, stderr,
                                                     loop=self._loop)
         await self.wait()
         return (stdout, stderr)

Lib/asyncio/tasks.py

@@ -580,15 +580,16 @@ def as_completed(fs, *, loop=None, timeout=None):
     if futures.isfuture(fs) or coroutines.iscoroutine(fs):
         raise TypeError(f"expect an iterable of futures, not {type(fs).__name__}")
 
+    if loop is not None:
+        warnings.warn("The loop argument is deprecated since Python 3.8, "
+                      "and scheduled for removal in Python 3.10.",
+                      DeprecationWarning, stacklevel=2)
+
     from .queues import Queue  # Import here to avoid circular import problem.
     done = Queue(loop=loop)
 
     if loop is None:
         loop = events.get_event_loop()
-    else:
-        warnings.warn("The loop argument is deprecated since Python 3.8, "
-                      "and scheduled for removal in Python 3.10.",
-                      DeprecationWarning, stacklevel=2)
     todo = {ensure_future(f, loop=loop) for f in set(fs)}
     timeout_handle = None
 
@@ -756,6 +757,10 @@ def gather(*coros_or_futures, loop=None, return_exceptions=False):
                       "and scheduled for removal in Python 3.10.",
                       DeprecationWarning, stacklevel=2)
 
+    return _gather(*coros_or_futures, loop=loop, return_exceptions=return_exceptions)
+
+
+def _gather(*coros_or_futures, loop=None, return_exceptions=False):
     if not coros_or_futures:
         if loop is None:
             loop = events.get_event_loop()

Lib/asyncio/unix_events.py

@@ -323,7 +323,7 @@ async def create_unix_server(
             server._start_serving()
             # Skip one loop iteration so that all 'loop.add_reader'
             # go through.
-            await tasks.sleep(0, loop=self)
+            await tasks.sleep(0)
 
         return server
 

Lib/test/test_asyncgen.py

@@ -1077,6 +1077,84 @@ async def wait():
 
         self.assertEqual(finalized, 2)
 
+    def test_async_gen_asyncio_shutdown_02(self):
+        messages = []
+
+        def exception_handler(loop, context):
+            messages.append(context)
+
+        async def async_iterate():
+            yield 1
+            yield 2
+
+        it = async_iterate()
+        async def main():
+            loop = asyncio.get_running_loop()
+            loop.set_exception_handler(exception_handler)
+
+            async for i in it:
+                break
+
+        asyncio.run(main())
+
+        self.assertEqual(messages, [])
+
+    def test_async_gen_asyncio_shutdown_exception_01(self):
+        messages = []
+
+        def exception_handler(loop, context):
+            messages.append(context)
+
+        async def async_iterate():
+            try:
+                yield 1
+                yield 2
+            finally:
+                1/0
+
+        it = async_iterate()
+        async def main():
+            loop = asyncio.get_running_loop()
+            loop.set_exception_handler(exception_handler)
+
+            async for i in it:
+                break
+
+        asyncio.run(main())
+
+        message, = messages
+        self.assertEqual(message['asyncgen'], it)
+        self.assertIsInstance(message['exception'], ZeroDivisionError)
+        self.assertIn('an error occurred during closing of asynchronous generator',
+                      message['message'])
+
+    def test_async_gen_asyncio_shutdown_exception_02(self):
+        messages = []
+
+        def exception_handler(loop, context):
+            messages.append(context)
+
+        async def async_iterate():
+            try:
+                yield 1
+                yield 2
+            finally:
+                1/0
+
+        async def main():
+            loop = asyncio.get_running_loop()
+            loop.set_exception_handler(exception_handler)
+
+            async for i in async_iterate():
+                break
+
+        asyncio.run(main())
+
+        message, = messages
+        self.assertIsInstance(message['exception'], ZeroDivisionError)
+        self.assertIn('unhandled exception during asyncio.run() shutdown',
+                      message['message'])
+
     def test_async_gen_expression_01(self):
         async def arange(n):
             for i in range(n):

Lib/test/test_asyncio/__init__.py

@@ -1,38 +1,10 @@
 import os
 from test import support
-import unittest
 
 # Skip tests if we don't have concurrent.futures.
 support.import_module('concurrent.futures')
 
 
-def load_tests(loader, _, pattern):
+def load_tests(*args):
     pkg_dir = os.path.dirname(__file__)
-    suite = AsyncioTestSuite()
-    return support.load_package_tests(pkg_dir, loader, suite, pattern)
-
-
-class AsyncioTestSuite(unittest.TestSuite):
-    """A custom test suite that also runs setup/teardown for the whole package.
-
-    Normally unittest only runs setUpModule() and tearDownModule() within each
-    test module part of the test suite. Copying those functions to each file
-    would be tedious, let's run this once and for all.
-    """
-    def run(self, result, debug=False):
-        ignore = support.ignore_deprecations_from
-        tokens = {
-            ignore("asyncio.base_events", like=r".*loop argument.*"),
-            ignore("asyncio.unix_events", like=r".*loop argument.*"),
-            ignore("asyncio.futures", like=r".*loop argument.*"),
-            ignore("asyncio.runners", like=r".*loop argument.*"),
-            ignore("asyncio.subprocess", like=r".*loop argument.*"),
-            ignore("asyncio.tasks", like=r".*loop argument.*"),
-            ignore("test.test_asyncio.test_events", like=r".*loop argument.*"),
-            ignore("test.test_asyncio.test_queues", like=r".*loop argument.*"),
-            ignore("test.test_asyncio.test_tasks", like=r".*loop argument.*"),
-        }
-        try:
-            super().run(result, debug=debug)
-        finally:
-            support.clear_ignored_deprecations(*tokens)
+    return support.load_package_tests(pkg_dir, *args)

Lib/test/test_asyncio/test_events.py

@@ -205,8 +205,8 @@ def __init__(self, loop):
         self.disconnects = {fd: loop.create_future() for fd in range(3)}
         self.data = {1: b'', 2: b''}
         self.returncode = None
-        self.got_data = {1: asyncio.Event(loop=loop),
-                         2: asyncio.Event(loop=loop)}
+        self.got_data = {1: asyncio.Event(),
+                         2: asyncio.Event()}
 
     def connection_made(self, transport):
         self.transport = transport
@@ -1736,7 +1736,6 @@ def test_subprocess_exec(self):
         connect = self.loop.subprocess_exec(
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1758,7 +1757,6 @@ def test_subprocess_interactive(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1780,7 +1778,6 @@ def test_subprocess_interactive(self):
         self.check_killed(proto.returncode)
 
     def test_subprocess_shell(self):
-        with self.assertWarns(DeprecationWarning):
         connect = self.loop.subprocess_shell(
                         functools.partial(MySubprocessProtocol, self.loop),
                         'echo Python')
@@ -1801,7 +1798,6 @@ def test_subprocess_exitcode(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         'exit 7', stdin=None, stdout=None, stderr=None)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.completed)
@@ -1812,7 +1808,6 @@ def test_subprocess_close_after_finish(self):
         connect = self.loop.subprocess_shell(
                         functools.partial(MySubprocessProtocol, self.loop),
                         'exit 7', stdin=None, stdout=None, stderr=None)
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.assertIsNone(transp.get_pipe_transport(0))
@@ -1829,7 +1824,6 @@ def test_subprocess_kill(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1846,7 +1840,6 @@ def test_subprocess_terminate(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1869,7 +1862,6 @@ def test_subprocess_send_signal(self):
                             functools.partial(MySubprocessProtocol, self.loop),
                             sys.executable, prog)
 
-            with self.assertWarns(DeprecationWarning):
             transp, proto = self.loop.run_until_complete(connect)
             self.assertIsInstance(proto, MySubprocessProtocol)
             self.loop.run_until_complete(proto.connected)
@@ -1888,7 +1880,6 @@ def test_subprocess_stderr(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1910,7 +1901,6 @@ def test_subprocess_stderr_redirect_to_stdout(self):
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog, stderr=subprocess.STDOUT)
 
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)
@@ -1934,7 +1924,6 @@ def test_subprocess_close_client_stream(self):
         connect = self.loop.subprocess_exec(
                         functools.partial(MySubprocessProtocol, self.loop),
                         sys.executable, prog)
-        with self.assertWarns(DeprecationWarning):
         transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.connected)

Lib/test/test_asyncio/test_queues.py

@@ -301,6 +301,7 @@ async def producer(queue, num_items):
         with self.assertWarns(DeprecationWarning):
             q = asyncio.Queue(queue_size, loop=self.loop)
 
+        with self.assertWarns(DeprecationWarning):
             self.loop.run_until_complete(
                 asyncio.gather(producer(q, producer_num_items),
                                consumer(q, producer_num_items),
@@ -555,6 +556,7 @@ async def getter():
         t1 = putter(1)
         t2 = putter(2)
         t3 = putter(3)
+        with self.assertWarns(DeprecationWarning):
             self.loop.run_until_complete(
                 asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop))
 

Lib/test/test_asyncio/test_tasks.py

@@ -1606,8 +1606,9 @@ async def foo():
             for f in asyncio.as_completed([b, c, a], loop=loop):
                 values.append(await f)
             return values
-        with self.assertWarns(DeprecationWarning):
+        with self.assertWarns(DeprecationWarning) as w:
             res = loop.run_until_complete(self.new_task(loop, foo()))
+        self.assertEqual(w.warnings[0].filename, __file__)
         self.assertAlmostEqual(0.15, loop.time())
         self.assertTrue('a' in res[:2])
         self.assertTrue('b' in res[:2])
@@ -3348,6 +3349,7 @@ def test_constructor_heterogenous_futures(self):
         with self.assertRaises(ValueError):
             asyncio.gather(fut1, fut2)
         with self.assertRaises(ValueError):
+            with self.assertWarns(DeprecationWarning):
                 asyncio.gather(fut1, loop=self.other_loop)
 
     def test_constructor_homogenous_futures(self):
@@ -3356,6 +3358,7 @@ def test_constructor_homogenous_futures(self):
         self.assertIs(fut._loop, self.other_loop)
         self._run_loop(self.other_loop)
         self.assertFalse(fut.done())
+        with self.assertWarns(DeprecationWarning):
             fut = asyncio.gather(*children, loop=self.other_loop)
         self.assertIs(fut._loop, self.other_loop)
         self._run_loop(self.other_loop)
@@ -3429,6 +3432,7 @@ async def coro():
         self.set_event_loop(self.other_loop, cleanup=False)
         gen3 = coro()
         gen4 = coro()
+        with self.assertWarns(DeprecationWarning):
             fut2 = asyncio.gather(gen3, gen4, loop=self.other_loop)
         self.assertIs(fut2._loop, self.other_loop)
         self.other_loop.run_until_complete(fut2)
@@ -3439,6 +3443,7 @@ def test_duplicate_coroutines(self):
             def coro(s):
                 return s
         c = coro('abc')
+        with self.assertWarns(DeprecationWarning):
             fut = asyncio.gather(c, c, coro('def'), c, loop=self.one_loop)
         self._run_loop(self.one_loop)
         self.assertEqual(fut.result(), ['abc', 'abc', 'def', 'abc'])
@@ -3459,7 +3464,7 @@ async def inner():
 
         async def outer():
             nonlocal proof, gatherer
-            gatherer = asyncio.gather(child1, child2, loop=self.one_loop)
+            gatherer = asyncio.gather(child1, child2)
             await gatherer
             proof += 100
 
@@ -3486,7 +3491,7 @@ async def inner(f):
         b = self.one_loop.create_future()
 
         async def outer():
-            await asyncio.gather(inner(a), inner(b), loop=self.one_loop)
+            await asyncio.gather(inner(a), inner(b))
 
         f = asyncio.ensure_future(outer(), loop=self.one_loop)
         test_utils.run_briefly(self.one_loop)
@@ -3705,7 +3710,7 @@ def coro2():
                 return 'ok2'
 
         async def inner():
-            return await asyncio.gather(coro1(), coro2(), loop=self.loop)
+            return await asyncio.gather(coro1(), coro2())
 
         result = self.loop.run_until_complete(inner())
         self.assertEqual(['ok1', 'ok2'], result)

Misc/NEWS.d/next/Library/2021-09-04-13-10-25.bpo-45097.5J4IC-.rst

@@ -0,0 +1,2 @@
+Remove deprecation warnings about the loop argument in :mod:`asyncio`
+incorrectly emitted in cases when the user does not pass the loop argument.