Cherry-pick some pytest-asyncio fixes (#11247)

Author: Dreamsorcerer

tests/conftest.py

@@ -243,6 +243,11 @@ def assert_sock_fits(sock_path: str) -> None:
                 return
 
 
+@pytest.fixture
+async def event_loop(loop: asyncio.AbstractEventLoop) -> asyncio.AbstractEventLoop:
+    return asyncio.get_running_loop()
+
+
 @pytest.fixture
 def selector_loop() -> Iterator[asyncio.AbstractEventLoop]:
     factory = asyncio.SelectorEventLoop

tests/isolated/check_for_client_response_leak.py

@@ -1,10 +1,11 @@
 import asyncio
 import contextlib
 import gc
+import socket
 import sys
 
 from aiohttp import ClientError, ClientSession, web
-from aiohttp.test_utils import get_unused_port_socket
+from aiohttp.test_utils import REUSE_ADDRESS
 
 gc.set_debug(gc.DEBUG_LEAK)
 
@@ -18,30 +19,32 @@ async def stream_handler(request: web.Request) -> web.Response:
         return web.Response()
 
     app.router.add_get("/stream", stream_handler)
-    sock = get_unused_port_socket("127.0.0.1")
-    port = sock.getsockname()[1]
-
-    runner = web.AppRunner(app)
-    await runner.setup()
-    site = web.SockSite(runner, sock)
-    await site.start()
-
-    session = ClientSession()
-
-    async def fetch_stream(url: str) -> None:
-        """Fetch a stream and read a few bytes from it."""
-        with contextlib.suppress(ClientError):
-            await session.get(url)
-
-    client_task = asyncio.create_task(fetch_stream(f"http://localhost:{port}/stream"))
-    await client_task
-    gc.collect()
-    client_response_present = any(
-        type(obj).__name__ == "ClientResponse" for obj in gc.garbage
-    )
-    await session.close()
-    await runner.cleanup()
-    sys.exit(1 if client_response_present else 0)
+    with socket.create_server(("127.0.0.1", 0), reuse_port=REUSE_ADDRESS) as sock:
+        port = sock.getsockname()[1]
+
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.SockSite(runner, sock)
+        await site.start()
+
+        session = ClientSession()
+
+        async def fetch_stream(url: str) -> None:
+            """Fetch a stream and read a few bytes from it."""
+            with contextlib.suppress(ClientError):
+                await session.get(url)
+
+        client_task = asyncio.create_task(
+            fetch_stream(f"http://localhost:{port}/stream")
+        )
+        await client_task
+        gc.collect()
+        client_response_present = any(
+            type(obj).__name__ == "ClientResponse" for obj in gc.garbage
+        )
+        await session.close()
+        await runner.cleanup()
+        sys.exit(1 if client_response_present else 0)
 
 
 asyncio.run(main())

tests/isolated/check_for_request_leak.py

@@ -1,10 +1,11 @@
 import asyncio
 import gc
+import socket
 import sys
 from typing import NoReturn
 
 from aiohttp import ClientSession, web
-from aiohttp.test_utils import get_unused_port_socket
+from aiohttp.test_utils import REUSE_ADDRESS
 
 gc.set_debug(gc.DEBUG_LEAK)
 
@@ -17,24 +18,24 @@ async def handler(request: web.Request) -> NoReturn:
         assert False
 
     app.router.add_route("GET", "/json", handler)
-    sock = get_unused_port_socket("127.0.0.1")
-    port = sock.getsockname()[1]
-
-    runner = web.AppRunner(app)
-    await runner.setup()
-    site = web.SockSite(runner, sock)
-    await site.start()
-
-    async with ClientSession() as session:
-        async with session.get(f"http://127.0.0.1:{port}/json") as resp:
-            await resp.read()
-
-    # Give time for the cancelled task to be collected
-    await asyncio.sleep(0.5)
-    gc.collect()
-    request_present = any(type(obj).__name__ == "Request" for obj in gc.garbage)
-    await session.close()
-    await runner.cleanup()
+    with socket.create_server(("127.0.0.1", 0), reuse_port=REUSE_ADDRESS) as sock:
+        port = sock.getsockname()[1]
+
+        runner = web.AppRunner(app)
+        await runner.setup()
+        site = web.SockSite(runner, sock)
+        await site.start()
+
+        async with ClientSession() as session:
+            async with session.get(f"http://127.0.0.1:{port}/json") as resp:
+                await resp.read()
+
+        # Give time for the cancelled task to be collected
+        await asyncio.sleep(0.5)
+        gc.collect()
+        request_present = any(type(obj).__name__ == "Request" for obj in gc.garbage)
+        await session.close()
+        await runner.cleanup()
     sys.exit(1 if request_present else 0)