[3.9] bpo-45097: Remove incorrect deprecation warnings in asyncio. (G…

…H-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,8 +193,8 @@ async def communicate(self, input=None):
             stderr = self._read_stream(2)
         else:
             stderr = self._noop()
-        stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr,
-                                                   loop=self._loop)
+        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)