[PR #9741/acc2912 backport][3.11] Fix race getting an unused port when there are multiple test runners (#9744)

Author: bdraco

tests/conftest.py

@@ -15,7 +15,7 @@
 
 from aiohttp.client_proto import ResponseHandler
 from aiohttp.http import WS_KEY
-from aiohttp.test_utils import loop_context
+from aiohttp.test_utils import get_unused_port_socket, loop_context
 
 try:
     import trustme
@@ -249,3 +249,18 @@ def enable_cleanup_closed() -> Generator[None, None, None]:
     """
     with mock.patch("aiohttp.connector.NEEDS_CLEANUP_CLOSED", True):
         yield
+
+
+@pytest.fixture
+def unused_port_socket() -> Generator[socket.socket, None, None]:
+    """Return a socket that is unused on the current host.
+
+    Unlike aiohttp_used_port, the socket is yielded so there is no
+    race condition between checking if the port is in use and
+    binding to it later in the test.
+    """
+    s = get_unused_port_socket("127.0.0.1")
+    try:
+        yield s
+    finally:
+        s.close()

tests/test_run_app.py

@@ -9,7 +9,16 @@
 import subprocess
 import sys
 import time
-from typing import AsyncIterator, Callable, NoReturn, Set
+from typing import (
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Coroutine,
+    NoReturn,
+    Optional,
+    Set,
+    Tuple,
+)
 from unittest import mock
 from uuid import uuid4
 
@@ -440,8 +449,10 @@ def test_run_app_https(patched_loop) -> None:
     )
 
 
-def test_run_app_nondefault_host_port(patched_loop, aiohttp_unused_port) -> None:
-    port = aiohttp_unused_port()
+def test_run_app_nondefault_host_port(
+    patched_loop: asyncio.AbstractEventLoop, unused_port_socket: socket.socket
+) -> None:
+    port = unused_port_socket.getsockname()[1]
     host = "127.0.0.1"
 
     app = web.Application()
@@ -454,7 +465,24 @@ def test_run_app_nondefault_host_port(patched_loop, aiohttp_unused_port) -> None
     )
 
 
-def test_run_app_multiple_hosts(patched_loop) -> None:
+def test_run_app_with_sock(
+    patched_loop: asyncio.AbstractEventLoop, unused_port_socket: socket.socket
+) -> None:
+    sock = unused_port_socket
+    app = web.Application()
+    web.run_app(
+        app,
+        sock=sock,
+        print=stopper(patched_loop),
+        loop=patched_loop,
+    )
+
+    patched_loop.create_server.assert_called_with(  # type: ignore[attr-defined]
+        mock.ANY, sock=sock, ssl=None, backlog=128
+    )
+
+
+def test_run_app_multiple_hosts(patched_loop: asyncio.AbstractEventLoop) -> None:
     hosts = ("127.0.0.1", "127.0.0.2")
 
     app = web.Application()
@@ -931,8 +959,16 @@ async def stop(self, request: web.Request) -> web.Response:
         asyncio.get_running_loop().call_soon(self.raiser)
         return web.Response()
 
-    def run_app(self, port: int, timeout: int, task, extra_test=None) -> asyncio.Task:
+    def run_app(
+        self,
+        sock: socket.socket,
+        timeout: int,
+        task: Callable[[], Coroutine[None, None, None]],
+        extra_test: Optional[Callable[[ClientSession], Awaitable[None]]] = None,
+    ) -> Tuple["asyncio.Task[None]", int]:
         num_connections = -1
+        t = test_task = None
+        port = sock.getsockname()[1]
 
         class DictRecordClear(dict):
             def clear(self):
@@ -956,15 +992,15 @@ async def test() -> None:
                     try:
                         with pytest.raises(asyncio.TimeoutError):
                             async with sess.get(
-                                f"http://localhost:{port}/",
+                                f"http://127.0.0.1:{port}/",
                                 timeout=ClientTimeout(total=0.1),
                             ):
                                 pass
                     except ClientConnectorError:
                         await asyncio.sleep(0.5)
                     else:
                         break
-                async with sess.get(f"http://localhost:{port}/stop"):
+                async with sess.get(f"http://127.0.0.1:{port}/stop"):
                     pass
 
                 if extra_test:
@@ -989,50 +1025,46 @@ async def handler(request: web.Request) -> web.Response:
         app.router.add_get("/stop", self.stop)
 
         with mock.patch("aiohttp.web_app.Server", ServerWithRecordClear):
-            web.run_app(app, port=port, shutdown_timeout=timeout)
+            web.run_app(app, sock=sock, shutdown_timeout=timeout)
         assert test_task.exception() is None
         return t, num_connections
 
-    def test_shutdown_wait_for_handler(
-        self, aiohttp_unused_port: Callable[[], int]
-    ) -> None:
-        port = aiohttp_unused_port()
+    def test_shutdown_wait_for_handler(self, unused_port_socket: socket.socket) -> None:
+        sock = unused_port_socket
         finished = False
 
         async def task():
             nonlocal finished
             await asyncio.sleep(2)
             finished = True
 
-        t, connection_count = self.run_app(port, 3, task)
+        t, connection_count = self.run_app(sock, 3, task)
 
         assert finished is True
         assert t.done()
         assert not t.cancelled()
         assert connection_count == 0
 
-    def test_shutdown_timeout_handler(
-        self, aiohttp_unused_port: Callable[[], int]
-    ) -> None:
-        port = aiohttp_unused_port()
+    def test_shutdown_timeout_handler(self, unused_port_socket: socket.socket) -> None:
+        sock = unused_port_socket
         finished = False
 
         async def task():
             nonlocal finished
             await asyncio.sleep(2)
             finished = True
 
-        t, connection_count = self.run_app(port, 1, task)
+        t, connection_count = self.run_app(sock, 1, task)
 
         assert finished is False
         assert t.done()
         assert t.cancelled()
         assert connection_count == 1
 
     def test_shutdown_timeout_not_reached(
-        self, aiohttp_unused_port: Callable[[], int]
+        self, unused_port_socket: socket.socket
     ) -> None:
-        port = aiohttp_unused_port()
+        sock = unused_port_socket
         finished = False
 
         async def task():
@@ -1041,7 +1073,8 @@ async def task():
             finished = True
 
         start_time = time.time()
-        t, connection_count = self.run_app(port, 15, task)
+
+        t, connection_count = self.run_app(sock, 15, task)
 
         assert finished is True
         assert t.done()
@@ -1050,9 +1083,10 @@ async def task():
         assert time.time() - start_time < 10
 
     def test_shutdown_new_conn_rejected(
-        self, aiohttp_unused_port: Callable[[], int]
+        self, unused_port_socket: socket.socket
     ) -> None:
-        port = aiohttp_unused_port()
+        sock = unused_port_socket
+        port = sock.getsockname()[1]
         finished = False
 
         async def task() -> None:
@@ -1066,33 +1100,34 @@ async def test(sess: ClientSession) -> None:
             with pytest.raises(ClientConnectorError):
                 # Use a new session to try and open a new connection.
                 async with ClientSession() as sess:
-                    async with sess.get(f"http://localhost:{port}/"):
+                    async with sess.get(f"http://127.0.0.1:{port}/"):
                         pass
             assert finished is False
 
-        t, connection_count = self.run_app(port, 10, task, test)
+        t, connection_count = self.run_app(sock, 10, task, test)
 
         assert finished is True
         assert t.done()
         assert connection_count == 0
 
     def test_shutdown_pending_handler_responds(
-        self, aiohttp_unused_port: Callable[[], int]
+        self, unused_port_socket: socket.socket
     ) -> None:
-        port = aiohttp_unused_port()
+        sock = unused_port_socket
+        port = sock.getsockname()[1]
         finished = False
 
         async def test() -> None:
-            async def test_resp(sess):
-                async with sess.get(f"http://localhost:{port}/") as resp:
+            async def test_resp(sess: ClientSession) -> None:
+                async with sess.get(f"http://127.0.0.1:{port}/") as resp:
                     assert await resp.text() == "FOO"
 
             await asyncio.sleep(1)
             async with ClientSession() as sess:
                 t = asyncio.create_task(test_resp(sess))
                 await asyncio.sleep(1)
                 # Handler is in-progress while we trigger server shutdown.
-                async with sess.get(f"http://localhost:{port}/stop"):
+                async with sess.get(f"http://127.0.0.1:{port}/stop"):
                     pass
 
                 assert finished is False
@@ -1117,19 +1152,22 @@ async def handler(request: web.Request) -> web.Response:
         app.router.add_get("/", handler)
         app.router.add_get("/stop", self.stop)
 
-        web.run_app(app, port=port, shutdown_timeout=5)
+        web.run_app(app, sock=sock, shutdown_timeout=5)
+        assert t is not None
         assert t.exception() is None
         assert finished is True
 
     def test_shutdown_close_idle_keepalive(
-        self, aiohttp_unused_port: Callable[[], int]
+        self, unused_port_socket: socket.socket
     ) -> None:
-        port = aiohttp_unused_port()
+        sock = unused_port_socket
+        port = sock.getsockname()[1]
+        t = None
 
         async def test() -> None:
             await asyncio.sleep(1)
             async with ClientSession() as sess:
-                async with sess.get(f"http://localhost:{port}/stop"):
+                async with sess.get(f"http://127.0.0.1:{port}/stop"):
                     pass
 
                 # Hold on to keep-alive connection.
@@ -1148,15 +1186,14 @@ async def run_test(app: web.Application) -> None:
         app.cleanup_ctx.append(run_test)
         app.router.add_get("/stop", self.stop)
 
-        web.run_app(app, port=port, shutdown_timeout=10)
+        web.run_app(app, sock=sock, shutdown_timeout=10)
         # If connection closed, then test() will be cancelled in cleanup_ctx.
         # If not, then shutdown_timeout will allow it to sleep until complete.
         assert t.cancelled()
 
-    def test_shutdown_close_websockets(
-        self, aiohttp_unused_port: Callable[[], int]
-    ) -> None:
-        port = aiohttp_unused_port()
+    def test_shutdown_close_websockets(self, unused_port_socket: socket.socket) -> None:
+        sock = unused_port_socket
+        port = sock.getsockname()[1]
         WS = web.AppKey("ws", Set[web.WebSocketResponse])
         client_finished = server_finished = False
 
@@ -1177,8 +1214,8 @@ async def close_websockets(app: web.Application) -> None:
         async def test() -> None:
             await asyncio.sleep(1)
             async with ClientSession() as sess:
-                async with sess.ws_connect(f"http://localhost:{port}/ws") as ws:
-                    async with sess.get(f"http://localhost:{port}/stop"):
+                async with sess.ws_connect(f"http://127.0.0.1:{port}/ws") as ws:
+                    async with sess.get(f"http://127.0.0.1:{port}/stop"):
                         pass
 
                     async for msg in ws:
@@ -1203,22 +1240,23 @@ async def run_test(app: web.Application) -> None:
         app.router.add_get("/stop", self.stop)
 
         start = time.time()
-        web.run_app(app, port=port, shutdown_timeout=10)
+        web.run_app(app, sock=sock, shutdown_timeout=10)
         assert time.time() - start < 5
         assert client_finished
         assert server_finished
 
     def test_shutdown_handler_cancellation_suppressed(
-        self, aiohttp_unused_port: Callable[[], int]
+        self, unused_port_socket: socket.socket
     ) -> None:
-        port = aiohttp_unused_port()
+        sock = unused_port_socket
+        port = sock.getsockname()[1]
         actions = []
 
         async def test() -> None:
             async def test_resp(sess):
                 t = ClientTimeout(total=0.4)
                 with pytest.raises(asyncio.TimeoutError):
-                    async with sess.get(f"http://localhost:{port}/", timeout=t) as resp:
+                    async with sess.get(f"http://127.0.0.1:{port}/", timeout=t) as resp:
                         assert await resp.text() == "FOO"
                 actions.append("CANCELLED")
 
@@ -1227,7 +1265,7 @@ async def test_resp(sess):
                 await asyncio.sleep(0.5)
                 # Handler is in-progress while we trigger server shutdown.
                 actions.append("PRESTOP")
-                async with sess.get(f"http://localhost:{port}/stop"):
+                async with sess.get(f"http://127.0.0.1:{port}/stop"):
                     pass
 
                 actions.append("STOPPING")
@@ -1255,6 +1293,6 @@ async def handler(request: web.Request) -> web.Response:
         app.router.add_get("/", handler)
         app.router.add_get("/stop", self.stop)
 
-        web.run_app(app, port=port, shutdown_timeout=2, handler_cancellation=True)
+        web.run_app(app, sock=sock, shutdown_timeout=2, handler_cancellation=True)
         assert t.exception() is None
         assert actions == ["CANCELLED", "SUPPRESSED", "PRESTOP", "STOPPING", "DONE"]

tests/test_test_utils.py

@@ -375,9 +375,16 @@ async def test_client_context_manager_response(method, app, loop) -> None:
                 assert "Hello, world" in text
 
 
-async def test_custom_port(loop, app, aiohttp_unused_port) -> None:
-    port = aiohttp_unused_port()
-    client = _TestClient(TestServer(app, loop=loop, port=port), loop=loop)
+async def test_custom_port(
+    loop: asyncio.AbstractEventLoop,
+    app: web.Application,
+    unused_port_socket: socket,
+) -> None:
+    sock = unused_port_socket
+    port = sock.getsockname()[1]
+    client = _TestClient(
+        TestServer(app, port=port, socket_factory=lambda *args, **kwargs: sock)
+    )
     await client.start_server()
 
     assert client.server.port == port