[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.

11 files changed, 234 insertions, 181 deletions

diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py
index b2d446a..bb2f991 100644
--- a/Lib/asyncio/base_events.py
+++ b/Lib/asyncio/base_events.py
@@ -350,7 +350,7 @@ class Server(events.AbstractServer):
         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 @@ class BaseEventLoop(events.AbstractEventLoop):
         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 @@ class BaseEventLoop(events.AbstractEventLoop):
             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 @@ class BaseEventLoop(events.AbstractEventLoop):
             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)
diff --git a/Lib/asyncio/runners.py b/Lib/asyncio/runners.py
index 268635d..6920acb 100644
--- a/Lib/asyncio/runners.py
+++ b/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():
diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
index c9506b1..820304e 100644
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -193,8 +193,8 @@ class Process:
             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)
 
diff --git a/Lib/asyncio/tasks.py b/Lib/asyncio/tasks.py
index d7e0575..22ed328 100644
--- a/Lib/asyncio/tasks.py
+++ b/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()
diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py
index 3efa669..56fcc08 100644
--- a/Lib/asyncio/unix_events.py
+++ b/Lib/asyncio/unix_events.py
@@ -323,7 +323,7 @@ class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
             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
 
diff --git a/Lib/test/test_asyncgen.py b/Lib/test/test_asyncgen.py
index 8d2ad22..1599738 100644
--- a/Lib/test/test_asyncgen.py
+++ b/Lib/test/test_asyncgen.py
@@ -1077,6 +1077,84 @@ class AsyncGenAsyncioTest(unittest.TestCase):
 
         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):
diff --git a/Lib/test/test_asyncio/__init__.py b/Lib/test/test_asyncio/__init__.py
index 4a7a868..5535b88 100644
--- a/Lib/test/test_asyncio/__init__.py
+++ b/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)
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index 7114c2f..72189bf 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -205,8 +205,8 @@ class MySubprocessProtocol(asyncio.SubprocessProtocol):
         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,20 +1736,19 @@ class SubprocessTestsMixin:
         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)
-            self.assertEqual('CONNECTED', proto.state)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
+        self.assertEqual('CONNECTED', proto.state)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'Python The Winner')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
-            self.assertEqual(b'Python The Winner', proto.data[1])
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'Python The Winner')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
+        self.assertEqual(b'Python The Winner', proto.data[1])
 
     def test_subprocess_interactive(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo.py')
@@ -1758,51 +1757,48 @@ class SubprocessTestsMixin:
                         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)
-            self.assertEqual('CONNECTED', proto.state)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
+        self.assertEqual('CONNECTED', proto.state)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'Python ')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            proto.got_data[1].clear()
-            self.assertEqual(b'Python ', proto.data[1])
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'Python ')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        proto.got_data[1].clear()
+        self.assertEqual(b'Python ', proto.data[1])
 
-            stdin.write(b'The Winner')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            self.assertEqual(b'Python The Winner', proto.data[1])
+        stdin.write(b'The Winner')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        self.assertEqual(b'Python The Winner', proto.data[1])
 
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        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')
-            transp, proto = self.loop.run_until_complete(connect)
-            self.assertIsInstance(proto, MySubprocessProtocol)
-            self.loop.run_until_complete(proto.connected)
+        connect = self.loop.subprocess_shell(
+                        functools.partial(MySubprocessProtocol, self.loop),
+                        'echo Python')
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.get_pipe_transport(0).close()
-            self.loop.run_until_complete(proto.completed)
-            self.assertEqual(0, proto.returncode)
-            self.assertTrue(all(f.done() for f in proto.disconnects.values()))
-            self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
-            self.assertEqual(proto.data[2], b'')
-            transp.close()
+        transp.get_pipe_transport(0).close()
+        self.loop.run_until_complete(proto.completed)
+        self.assertEqual(0, proto.returncode)
+        self.assertTrue(all(f.done() for f in proto.disconnects.values()))
+        self.assertEqual(proto.data[1].rstrip(b'\r\n'), b'Python')
+        self.assertEqual(proto.data[2], b'')
+        transp.close()
 
     def test_subprocess_exitcode(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)
+        transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.loop.run_until_complete(proto.completed)
         self.assertEqual(7, proto.returncode)
@@ -1812,8 +1808,7 @@ class SubprocessTestsMixin:
         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)
+        transp, proto = self.loop.run_until_complete(connect)
         self.assertIsInstance(proto, MySubprocessProtocol)
         self.assertIsNone(transp.get_pipe_transport(0))
         self.assertIsNone(transp.get_pipe_transport(1))
@@ -1829,15 +1824,14 @@ class SubprocessTestsMixin:
                         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)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.kill()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
-            transp.close()
+        transp.kill()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
+        transp.close()
 
     def test_subprocess_terminate(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo.py')
@@ -1846,15 +1840,14 @@ class SubprocessTestsMixin:
                         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)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            transp.terminate()
-            self.loop.run_until_complete(proto.completed)
-            self.check_terminated(proto.returncode)
-            transp.close()
+        transp.terminate()
+        self.loop.run_until_complete(proto.completed)
+        self.check_terminated(proto.returncode)
+        transp.close()
 
     @unittest.skipIf(sys.platform == 'win32', "Don't have SIGHUP")
     def test_subprocess_send_signal(self):
@@ -1869,15 +1862,14 @@ class SubprocessTestsMixin:
                             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)
+            transp, proto = self.loop.run_until_complete(connect)
+            self.assertIsInstance(proto, MySubprocessProtocol)
+            self.loop.run_until_complete(proto.connected)
 
-                transp.send_signal(signal.SIGHUP)
-                self.loop.run_until_complete(proto.completed)
-                self.assertEqual(-signal.SIGHUP, proto.returncode)
-                transp.close()
+            transp.send_signal(signal.SIGHUP)
+            self.loop.run_until_complete(proto.completed)
+            self.assertEqual(-signal.SIGHUP, proto.returncode)
+            transp.close()
         finally:
             signal.signal(signal.SIGHUP, old_handler)
 
@@ -1888,20 +1880,19 @@ class SubprocessTestsMixin:
                         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)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            stdin.write(b'test')
+        stdin = transp.get_pipe_transport(0)
+        stdin.write(b'test')
 
-            self.loop.run_until_complete(proto.completed)
+        self.loop.run_until_complete(proto.completed)
 
-            transp.close()
-            self.assertEqual(b'OUT:test', proto.data[1])
-            self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
-            self.assertEqual(0, proto.returncode)
+        transp.close()
+        self.assertEqual(b'OUT:test', proto.data[1])
+        self.assertTrue(proto.data[2].startswith(b'ERR:test'), proto.data[2])
+        self.assertEqual(0, proto.returncode)
 
     def test_subprocess_stderr_redirect_to_stdout(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo2.py')
@@ -1910,23 +1901,22 @@ class SubprocessTestsMixin:
                         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)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            self.assertIsNotNone(transp.get_pipe_transport(1))
-            self.assertIsNone(transp.get_pipe_transport(2))
+        stdin = transp.get_pipe_transport(0)
+        self.assertIsNotNone(transp.get_pipe_transport(1))
+        self.assertIsNone(transp.get_pipe_transport(2))
 
-            stdin.write(b'test')
-            self.loop.run_until_complete(proto.completed)
-            self.assertTrue(proto.data[1].startswith(b'OUT:testERR:test'),
-                            proto.data[1])
-            self.assertEqual(b'', proto.data[2])
+        stdin.write(b'test')
+        self.loop.run_until_complete(proto.completed)
+        self.assertTrue(proto.data[1].startswith(b'OUT:testERR:test'),
+                        proto.data[1])
+        self.assertEqual(b'', proto.data[2])
 
-            transp.close()
-            self.assertEqual(0, proto.returncode)
+        transp.close()
+        self.assertEqual(0, proto.returncode)
 
     def test_subprocess_close_client_stream(self):
         prog = os.path.join(os.path.dirname(__file__), 'echo3.py')
@@ -1934,33 +1924,32 @@ class SubprocessTestsMixin:
         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)
+        transp, proto = self.loop.run_until_complete(connect)
+        self.assertIsInstance(proto, MySubprocessProtocol)
+        self.loop.run_until_complete(proto.connected)
 
-            stdin = transp.get_pipe_transport(0)
-            stdout = transp.get_pipe_transport(1)
-            stdin.write(b'test')
-            self.loop.run_until_complete(proto.got_data[1].wait())
-            self.assertEqual(b'OUT:test', proto.data[1])
+        stdin = transp.get_pipe_transport(0)
+        stdout = transp.get_pipe_transport(1)
+        stdin.write(b'test')
+        self.loop.run_until_complete(proto.got_data[1].wait())
+        self.assertEqual(b'OUT:test', proto.data[1])
 
-            stdout.close()
-            self.loop.run_until_complete(proto.disconnects[1])
-            stdin.write(b'xxx')
-            self.loop.run_until_complete(proto.got_data[2].wait())
-            if sys.platform != 'win32':
-                self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
-            else:
-                # After closing the read-end of a pipe, writing to the
-                # write-end using os.write() fails with errno==EINVAL and
-                # GetLastError()==ERROR_INVALID_NAME on Windows!?!  (Using
-                # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
-                self.assertEqual(b'ERR:OSError', proto.data[2])
-            with test_utils.disable_logger():
-                transp.close()
-            self.loop.run_until_complete(proto.completed)
-            self.check_killed(proto.returncode)
+        stdout.close()
+        self.loop.run_until_complete(proto.disconnects[1])
+        stdin.write(b'xxx')
+        self.loop.run_until_complete(proto.got_data[2].wait())
+        if sys.platform != 'win32':
+            self.assertEqual(b'ERR:BrokenPipeError', proto.data[2])
+        else:
+            # After closing the read-end of a pipe, writing to the
+            # write-end using os.write() fails with errno==EINVAL and
+            # GetLastError()==ERROR_INVALID_NAME on Windows!?!  (Using
+            # WriteFile() we get ERROR_BROKEN_PIPE as expected.)
+            self.assertEqual(b'ERR:OSError', proto.data[2])
+        with test_utils.disable_logger():
+            transp.close()
+        self.loop.run_until_complete(proto.completed)
+        self.check_killed(proto.returncode)
 
     def test_subprocess_wait_no_same_group(self):
         # start the new process in a new session
diff --git a/Lib/test/test_asyncio/test_queues.py b/Lib/test/test_asyncio/test_queues.py
index 5c9aaa8..81e888f 100644
--- a/Lib/test/test_asyncio/test_queues.py
+++ b/Lib/test/test_asyncio/test_queues.py
@@ -301,11 +301,12 @@ class QueueGetTests(_QueueTestBase):
         with self.assertWarns(DeprecationWarning):
             q = asyncio.Queue(queue_size, loop=self.loop)
 
-        self.loop.run_until_complete(
-            asyncio.gather(producer(q, producer_num_items),
-                           consumer(q, producer_num_items),
-                           loop=self.loop),
-            )
+        with self.assertWarns(DeprecationWarning):
+            self.loop.run_until_complete(
+                asyncio.gather(producer(q, producer_num_items),
+                               consumer(q, producer_num_items),
+                               loop=self.loop),
+                )
 
     def test_cancelled_getters_not_being_held_in_self_getters(self):
         def a_generator():
@@ -555,8 +556,9 @@ class QueuePutTests(_QueueTestBase):
         t1 = putter(1)
         t2 = putter(2)
         t3 = putter(3)
-        self.loop.run_until_complete(
-            asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop))
+        with self.assertWarns(DeprecationWarning):
+            self.loop.run_until_complete(
+                asyncio.gather(getter(), t0, t1, t2, t3, loop=self.loop))
 
     def test_cancelled_puts_not_being_held_in_self_putters(self):
         def a_generator():
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py
index d2d7803..5e14b62 100644
--- a/Lib/test/test_asyncio/test_tasks.py
+++ b/Lib/test/test_asyncio/test_tasks.py
@@ -1606,8 +1606,9 @@ class BaseTaskTests:
             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,7 +3349,8 @@ class FutureGatherTests(GatherTestsBase, test_utils.TestCase):
         with self.assertRaises(ValueError):
             asyncio.gather(fut1, fut2)
         with self.assertRaises(ValueError):
-            asyncio.gather(fut1, loop=self.other_loop)
+            with self.assertWarns(DeprecationWarning):
+                asyncio.gather(fut1, loop=self.other_loop)
 
     def test_constructor_homogenous_futures(self):
         children = [self.other_loop.create_future() for i in range(3)]
@@ -3356,7 +3358,8 @@ class FutureGatherTests(GatherTestsBase, test_utils.TestCase):
         self.assertIs(fut._loop, self.other_loop)
         self._run_loop(self.other_loop)
         self.assertFalse(fut.done())
-        fut = asyncio.gather(*children, loop=self.other_loop)
+        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)
         self.assertFalse(fut.done())
@@ -3429,7 +3432,8 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
         self.set_event_loop(self.other_loop, cleanup=False)
         gen3 = coro()
         gen4 = coro()
-        fut2 = asyncio.gather(gen3, gen4, loop=self.other_loop)
+        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,7 +3443,8 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
             def coro(s):
                 return s
         c = coro('abc')
-        fut = asyncio.gather(c, c, coro('def'), c, loop=self.one_loop)
+        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 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
 
         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 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase):
         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 @@ class CompatibilityTests(test_utils.TestCase):
                 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)
diff --git a/Misc/NEWS.d/next/Library/2021-09-04-13-10-25.bpo-45097.5J4IC-.rst b/Misc/NEWS.d/next/Library/2021-09-04-13-10-25.bpo-45097.5J4IC-.rst
new file mode 100644
index 0000000..1788300
--- /dev/null
+++ b/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.