We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix `OSError: [Errno 98] Address already in use` in `bind_sockets` call in CI
We often get the following traceback in GitHub Actions:

```
==================================== ERRORS ====================================
______________ ERROR at setup of TestRetryAfter.test_retry_after _______________
Traceback (most recent call last):
  File "/home/runner/work/urllib3/urllib3/dummyserver/testcase.py", line 141, in setup_class
    cls._start_server()
  File "/home/runner/work/urllib3/urllib3/dummyserver/testcase.py", line 129, in _start_server
    app, cls.io_loop, cls.certs, cls.scheme, cls.host
  File "/home/runner/work/urllib3/urllib3/dummyserver/server.py", line 182, in run_tornado_app
    sockets = tornado.netutil.bind_sockets(None, address=host)
  File "/home/runner/work/urllib3/urllib3/.nox/test-3-7/lib/python3.7/site-packages/tornado/netutil.py", line 161, in bind_sockets
    sock.bind(sockaddr)
OSError: [Errno 98] Address already in use
```

I think we hit https://github.com/tornadoweb/tornado/issues/1860. When we request a socket with host `localhost` and port zero (`None` in Python), then Tornado gets a free IPv4 port and requests that same port in IPv6. But that port could easily be taken with IPv6, especially in crowded CI environments.

From Tornado's maintainer in issue above:

> Yeah, putting the whole of bind_sockets in a retry loop seems reasonable, but it's a bit of a special case (we'd only want to retry when port 0 is requested). I'd accept a PR to do that, but my advice would be that whenever you use port 0 you bind to a specific IP address (i.e. either 127.0.0.1 or ::1) instead of a hostname. That will limit you to a single network interface too and avoid this problem.

The problem with requiring 127.0.0.1 is that we have tests where we use both `localhost` and `127.0.0.1` in order to get two hosts.

So I'd be in favor of putting the `bind_sockets` in a retry loop.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dummyserver/server.py`
Content:
```
1 #!/usr/bin/env python
2 
3 """
4 Dummy server used for unit testing.
5 """
6 
7 from __future__ import annotations
8 
9 import asyncio
10 import concurrent.futures
11 import contextlib
12 import logging
13 import os
14 import socket
15 import ssl
16 import sys
17 import threading
18 import typing
19 import warnings
20 from collections.abc import Coroutine, Generator
21 from datetime import datetime
22 
23 import tornado.httpserver
24 import tornado.ioloop
25 import tornado.netutil
26 import tornado.web
27 import trustme
28 from cryptography.hazmat.backends import default_backend
29 from cryptography.hazmat.primitives import serialization
30 
31 from urllib3.exceptions import HTTPWarning
32 from urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version
33 
34 if typing.TYPE_CHECKING:
35     from typing_extensions import ParamSpec
36 
37     P = ParamSpec("P")
38 
39 log = logging.getLogger(__name__)
40 
41 CERTS_PATH = os.path.join(os.path.dirname(__file__), "certs")
42 DEFAULT_CERTS: dict[str, typing.Any] = {
43     "certfile": os.path.join(CERTS_PATH, "server.crt"),
44     "keyfile": os.path.join(CERTS_PATH, "server.key"),
45     "cert_reqs": ssl.CERT_OPTIONAL,
46     "ca_certs": os.path.join(CERTS_PATH, "cacert.pem"),
47     "alpn_protocols": ALPN_PROTOCOLS,
48 }
49 DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
50 DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")
51 
52 
53 def _resolves_to_ipv6(host: str) -> bool:
54     """Returns True if the system resolves host to an IPv6 address by default."""
55     resolves_to_ipv6 = False
56     try:
57         for res in socket.getaddrinfo(host, None, socket.AF_UNSPEC):
58             af, _, _, _, _ = res
59             if af == socket.AF_INET6:
60                 resolves_to_ipv6 = True
61     except socket.gaierror:
62         pass
63 
64     return resolves_to_ipv6
65 
66 
67 def _has_ipv6(host: str) -> bool:
68     """Returns True if the system can bind an IPv6 address."""
69     sock = None
70     has_ipv6 = False
71 
72     if socket.has_ipv6:
73         # has_ipv6 returns true if cPython was compiled with IPv6 support.
74         # It does not tell us if the system has IPv6 support enabled. To
75         # determine that we must bind to an IPv6 address.
76         # https://github.com/urllib3/urllib3/pull/611
77         # https://bugs.python.org/issue658327
78         try:
79             sock = socket.socket(socket.AF_INET6)
80             sock.bind((host, 0))
81             has_ipv6 = _resolves_to_ipv6("localhost")
82         except Exception:
83             pass
84 
85     if sock:
86         sock.close()
87     return has_ipv6
88 
89 
90 # Some systems may have IPv6 support but DNS may not be configured
91 # properly. We can not count that localhost will resolve to ::1 on all
92 # systems. See https://github.com/urllib3/urllib3/pull/611 and
93 # https://bugs.python.org/issue18792
94 HAS_IPV6_AND_DNS = _has_ipv6("localhost")
95 HAS_IPV6 = _has_ipv6("::1")
96 
97 
98 # Different types of servers we have:
99 
100 
101 class NoIPv6Warning(HTTPWarning):
102     "IPv6 is not available"
103 
104 
105 class SocketServerThread(threading.Thread):
106     """
107     :param socket_handler: Callable which receives a socket argument for one
108         request.
109     :param ready_event: Event which gets set when the socket handler is
110         ready to receive requests.
111     """
112 
113     USE_IPV6 = HAS_IPV6_AND_DNS
114 
115     def __init__(
116         self,
117         socket_handler: typing.Callable[[socket.socket], None],
118         host: str = "localhost",
119         ready_event: threading.Event | None = None,
120     ) -> None:
121         super().__init__()
122         self.daemon = True
123 
124         self.socket_handler = socket_handler
125         self.host = host
126         self.ready_event = ready_event
127 
128     def _start_server(self) -> None:
129         if self.USE_IPV6:
130             sock = socket.socket(socket.AF_INET6)
131         else:
132             warnings.warn("No IPv6 support. Falling back to IPv4.", NoIPv6Warning)
133             sock = socket.socket(socket.AF_INET)
134         if sys.platform != "win32":
135             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
136         sock.bind((self.host, 0))
137         self.port = sock.getsockname()[1]
138 
139         # Once listen() returns, the server socket is ready
140         sock.listen(1)
141 
142         if self.ready_event:
143             self.ready_event.set()
144 
145         self.socket_handler(sock)
146         sock.close()
147 
148     def run(self) -> None:
149         self._start_server()
150 
151 
152 def ssl_options_to_context(  # type: ignore[no-untyped-def]
153     keyfile=None,
154     certfile=None,
155     server_side=None,
156     cert_reqs=None,
157     ssl_version: str | int | None = None,
158     ca_certs=None,
159     do_handshake_on_connect=None,
160     suppress_ragged_eofs=None,
161     ciphers=None,
162     alpn_protocols=None,
163 ) -> ssl.SSLContext:
164     """Return an equivalent SSLContext based on ssl.wrap_socket args."""
165     ssl_version = resolve_ssl_version(ssl_version)
166     cert_none = resolve_cert_reqs("CERT_NONE")
167     if cert_reqs is None:
168         cert_reqs = cert_none
169     else:
170         cert_reqs = resolve_cert_reqs(cert_reqs)
171 
172     ctx = ssl.SSLContext(ssl_version)
173     ctx.load_cert_chain(certfile, keyfile)
174     ctx.verify_mode = cert_reqs
175     if ctx.verify_mode != cert_none:
176         ctx.load_verify_locations(cafile=ca_certs)
177     if alpn_protocols and hasattr(ctx, "set_alpn_protocols"):
178         try:
179             ctx.set_alpn_protocols(alpn_protocols)
180         except NotImplementedError:
181             pass
182     return ctx
183 
184 
185 def run_tornado_app(
186     app: tornado.web.Application,
187     certs: dict[str, typing.Any] | None,
188     scheme: str,
189     host: str,
190 ) -> tuple[tornado.httpserver.HTTPServer, int]:
191     # We can't use fromtimestamp(0) because of CPython issue 29097, so we'll
192     # just construct the datetime object directly.
193     app.last_req = datetime(1970, 1, 1)  # type: ignore[attr-defined]
194 
195     if scheme == "https":
196         assert certs is not None
197         ssl_opts = ssl_options_to_context(**certs)
198         http_server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_opts)
199     else:
200         http_server = tornado.httpserver.HTTPServer(app)
201 
202     sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]
203     port = sockets[0].getsockname()[1]
204     http_server.add_sockets(sockets)
205     return http_server, port
206 
207 
208 def get_unreachable_address() -> tuple[str, int]:
209     # reserved as per rfc2606
210     return ("something.invalid", 54321)
211 
212 
213 def encrypt_key_pem(private_key_pem: trustme.Blob, password: bytes) -> trustme.Blob:
214     private_key = serialization.load_pem_private_key(
215         private_key_pem.bytes(), password=None, backend=default_backend()
216     )
217     encrypted_key = private_key.private_bytes(
218         serialization.Encoding.PEM,
219         serialization.PrivateFormat.TraditionalOpenSSL,
220         serialization.BestAvailableEncryption(password),
221     )
222     return trustme.Blob(encrypted_key)
223 
224 
225 R = typing.TypeVar("R")
226 
227 
228 def _run_and_close_tornado(
229     async_fn: typing.Callable[P, Coroutine[typing.Any, typing.Any, R]],
230     *args: P.args,
231     **kwargs: P.kwargs,
232 ) -> R:
233     tornado_loop = None
234 
235     async def inner_fn() -> R:
236         nonlocal tornado_loop
237         tornado_loop = tornado.ioloop.IOLoop.current()
238         return await async_fn(*args, **kwargs)
239 
240     try:
241         return asyncio.run(inner_fn())
242     finally:
243         tornado_loop.close(all_fds=True)  # type: ignore[union-attr]
244 
245 
246 @contextlib.contextmanager
247 def run_loop_in_thread() -> Generator[tornado.ioloop.IOLoop, None, None]:
248     loop_started: concurrent.futures.Future[
249         tuple[tornado.ioloop.IOLoop, asyncio.Event]
250     ] = concurrent.futures.Future()
251     with concurrent.futures.ThreadPoolExecutor(
252         1, thread_name_prefix="test IOLoop"
253     ) as tpe:
254 
255         async def run() -> None:
256             io_loop = tornado.ioloop.IOLoop.current()
257             stop_event = asyncio.Event()
258             loop_started.set_result((io_loop, stop_event))
259             await stop_event.wait()
260 
261         # run asyncio.run in a thread and collect exceptions from *either*
262         # the loop failing to start, or failing to close
263         ran = tpe.submit(_run_and_close_tornado, run)  # type: ignore[arg-type]
264         for f in concurrent.futures.as_completed((loop_started, ran)):  # type: ignore[misc]
265             if f is loop_started:
266                 io_loop, stop_event = loop_started.result()
267                 try:
268                     yield io_loop
269                 finally:
270                     io_loop.add_callback(stop_event.set)
271 
272             elif f is ran:
273                 # if this is the first iteration the loop failed to start
274                 # if it's the second iteration the loop has finished or
275                 # the loop failed to close and we need to raise the exception
276                 ran.result()
277                 return
278 
279 
280 def main() -> int:
281     # For debugging dummyserver itself - python -m dummyserver.server
282     from .handlers import TestingApp
283 
284     host = "127.0.0.1"
285 
286     async def amain() -> int:
287         app = tornado.web.Application([(r".*", TestingApp)])
288         server, port = run_tornado_app(app, None, "http", host)
289 
290         print(f"Listening on http://{host}:{port}")
291         await asyncio.Event().wait()
292         return 0
293 
294     return asyncio.run(amain())
295 
296 
297 if __name__ == "__main__":
298     sys.exit(main())
299
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dummyserver/server.py b/dummyserver/server.py
--- a/dummyserver/server.py
+++ b/dummyserver/server.py
@@ -9,6 +9,7 @@
 import asyncio
 import concurrent.futures
 import contextlib
+import errno
 import logging
 import os
 import socket
@@ -199,7 +200,24 @@
     else:
         http_server = tornado.httpserver.HTTPServer(app)
 
-    sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]
+    # When we request a socket with host localhost and port zero (None in Python), then
+    # Tornado gets a free IPv4 port and requests that same port in IPv6. But that port
+    # could easily be taken with IPv6, especially in crowded CI environments. For this
+    # reason we put bind_sockets in a retry loop. Full details:
+    #  * https://github.com/urllib3/urllib3/issues/2171
+    #  * https://github.com/tornadoweb/tornado/issues/1860
+    for i in range(10):
+        try:
+            sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]
+        except OSError as e:
+            if e.errno == errno.EADDRINUSE:
+                # TODO this should be a warning if there's a way for pytest to print it
+                print(
+                    f"Retrying bind_sockets({host}) after EADDRINUSE", file=sys.stderr
+                )
+                continue
+        break
+
     port = sockets[0].getsockname()[1]
     http_server.add_sockets(sockets)
     return http_server, port

{"golden_diff": "diff --git a/dummyserver/server.py b/dummyserver/server.py\n--- a/dummyserver/server.py\n+++ b/dummyserver/server.py\n@@ -9,6 +9,7 @@\n import asyncio\n import concurrent.futures\n import contextlib\n+import errno\n import logging\n import os\n import socket\n@@ -199,7 +200,24 @@\n     else:\n         http_server = tornado.httpserver.HTTPServer(app)\n \n-    sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]\n+    # When we request a socket with host localhost and port zero (None in Python), then\n+    # Tornado gets a free IPv4 port and requests that same port in IPv6. But that port\n+    # could easily be taken with IPv6, especially in crowded CI environments. For this\n+    # reason we put bind_sockets in a retry loop. Full details:\n+    #  * https://github.com/urllib3/urllib3/issues/2171\n+    #  * https://github.com/tornadoweb/tornado/issues/1860\n+    for i in range(10):\n+        try:\n+            sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]\n+        except OSError as e:\n+            if e.errno == errno.EADDRINUSE:\n+                # TODO this should be a warning if there's a way for pytest to print it\n+                print(\n+                    f\"Retrying bind_sockets({host}) after EADDRINUSE\", file=sys.stderr\n+                )\n+                continue\n+        break\n+\n     port = sockets[0].getsockname()[1]\n     http_server.add_sockets(sockets)\n     return http_server, port\n", "issue": "Fix `OSError: [Errno 98] Address already in use` in `bind_sockets` call in CI\nWe often get the following traceback in GitHub Actions:\r\n\r\n```\r\n==================================== ERRORS ====================================\r\n______________ ERROR at setup of TestRetryAfter.test_retry_after _______________\r\nTraceback (most recent call last):\r\n  File \"/home/runner/work/urllib3/urllib3/dummyserver/testcase.py\", line 141, in setup_class\r\n    cls._start_server()\r\n  File \"/home/runner/work/urllib3/urllib3/dummyserver/testcase.py\", line 129, in _start_server\r\n    app, cls.io_loop, cls.certs, cls.scheme, cls.host\r\n  File \"/home/runner/work/urllib3/urllib3/dummyserver/server.py\", line 182, in run_tornado_app\r\n    sockets = tornado.netutil.bind_sockets(None, address=host)\r\n  File \"/home/runner/work/urllib3/urllib3/.nox/test-3-7/lib/python3.7/site-packages/tornado/netutil.py\", line 161, in bind_sockets\r\n    sock.bind(sockaddr)\r\nOSError: [Errno 98] Address already in use\r\n```\r\n\r\nI think we hit https://github.com/tornadoweb/tornado/issues/1860. When we request a socket with host `localhost` and port zero (`None` in Python), then Tornado gets a free IPv4 port and requests that same port in IPv6. But that port could easily be taken with IPv6, especially in crowded CI environments.\r\n\r\nFrom Tornado's maintainer in issue above:\r\n\r\n> Yeah, putting the whole of bind_sockets in a retry loop seems reasonable, but it's a bit of a special case (we'd only want to retry when port 0 is requested). I'd accept a PR to do that, but my advice would be that whenever you use port 0 you bind to a specific IP address (i.e. either 127.0.0.1 or ::1) instead of a hostname. That will limit you to a single network interface too and avoid this problem.\r\n\r\nThe problem with requiring 127.0.0.1 is that we have tests where we use both `localhost` and `127.0.0.1` in order to get two hosts.\r\n\r\nSo I'd be in favor of putting the `bind_sockets` in a retry loop.\n", "before_files": [{"content": "#!/usr/bin/env python\n\n\"\"\"\nDummy server used for unit testing.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport concurrent.futures\nimport contextlib\nimport logging\nimport os\nimport socket\nimport ssl\nimport sys\nimport threading\nimport typing\nimport warnings\nfrom collections.abc import Coroutine, Generator\nfrom datetime import datetime\n\nimport tornado.httpserver\nimport tornado.ioloop\nimport tornado.netutil\nimport tornado.web\nimport trustme\nfrom cryptography.hazmat.backends import default_backend\nfrom cryptography.hazmat.primitives import serialization\n\nfrom urllib3.exceptions import HTTPWarning\nfrom urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version\n\nif typing.TYPE_CHECKING:\n    from typing_extensions import ParamSpec\n\n    P = ParamSpec(\"P\")\n\nlog = logging.getLogger(__name__)\n\nCERTS_PATH = os.path.join(os.path.dirname(__file__), \"certs\")\nDEFAULT_CERTS: dict[str, typing.Any] = {\n    \"certfile\": os.path.join(CERTS_PATH, \"server.crt\"),\n    \"keyfile\": os.path.join(CERTS_PATH, \"server.key\"),\n    \"cert_reqs\": ssl.CERT_OPTIONAL,\n    \"ca_certs\": os.path.join(CERTS_PATH, \"cacert.pem\"),\n    \"alpn_protocols\": ALPN_PROTOCOLS,\n}\nDEFAULT_CA = os.path.join(CERTS_PATH, \"cacert.pem\")\nDEFAULT_CA_KEY = os.path.join(CERTS_PATH, \"cacert.key\")\n\n\ndef _resolves_to_ipv6(host: str) -> bool:\n    \"\"\"Returns True if the system resolves host to an IPv6 address by default.\"\"\"\n    resolves_to_ipv6 = False\n    try:\n        for res in socket.getaddrinfo(host, None, socket.AF_UNSPEC):\n            af, _, _, _, _ = res\n            if af == socket.AF_INET6:\n                resolves_to_ipv6 = True\n    except socket.gaierror:\n        pass\n\n    return resolves_to_ipv6\n\n\ndef _has_ipv6(host: str) -> bool:\n    \"\"\"Returns True if the system can bind an IPv6 address.\"\"\"\n    sock = None\n    has_ipv6 = False\n\n    if socket.has_ipv6:\n        # has_ipv6 returns true if cPython was compiled with IPv6 support.\n        # It does not tell us if the system has IPv6 support enabled. To\n        # determine that we must bind to an IPv6 address.\n        # https://github.com/urllib3/urllib3/pull/611\n        # https://bugs.python.org/issue658327\n        try:\n            sock = socket.socket(socket.AF_INET6)\n            sock.bind((host, 0))\n            has_ipv6 = _resolves_to_ipv6(\"localhost\")\n        except Exception:\n            pass\n\n    if sock:\n        sock.close()\n    return has_ipv6\n\n\n# Some systems may have IPv6 support but DNS may not be configured\n# properly. We can not count that localhost will resolve to ::1 on all\n# systems. See https://github.com/urllib3/urllib3/pull/611 and\n# https://bugs.python.org/issue18792\nHAS_IPV6_AND_DNS = _has_ipv6(\"localhost\")\nHAS_IPV6 = _has_ipv6(\"::1\")\n\n\n# Different types of servers we have:\n\n\nclass NoIPv6Warning(HTTPWarning):\n    \"IPv6 is not available\"\n\n\nclass SocketServerThread(threading.Thread):\n    \"\"\"\n    :param socket_handler: Callable which receives a socket argument for one\n        request.\n    :param ready_event: Event which gets set when the socket handler is\n        ready to receive requests.\n    \"\"\"\n\n    USE_IPV6 = HAS_IPV6_AND_DNS\n\n    def __init__(\n        self,\n        socket_handler: typing.Callable[[socket.socket], None],\n        host: str = \"localhost\",\n        ready_event: threading.Event | None = None,\n    ) -> None:\n        super().__init__()\n        self.daemon = True\n\n        self.socket_handler = socket_handler\n        self.host = host\n        self.ready_event = ready_event\n\n    def _start_server(self) -> None:\n        if self.USE_IPV6:\n            sock = socket.socket(socket.AF_INET6)\n        else:\n            warnings.warn(\"No IPv6 support. Falling back to IPv4.\", NoIPv6Warning)\n            sock = socket.socket(socket.AF_INET)\n        if sys.platform != \"win32\":\n            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n        sock.bind((self.host, 0))\n        self.port = sock.getsockname()[1]\n\n        # Once listen() returns, the server socket is ready\n        sock.listen(1)\n\n        if self.ready_event:\n            self.ready_event.set()\n\n        self.socket_handler(sock)\n        sock.close()\n\n    def run(self) -> None:\n        self._start_server()\n\n\ndef ssl_options_to_context(  # type: ignore[no-untyped-def]\n    keyfile=None,\n    certfile=None,\n    server_side=None,\n    cert_reqs=None,\n    ssl_version: str | int | None = None,\n    ca_certs=None,\n    do_handshake_on_connect=None,\n    suppress_ragged_eofs=None,\n    ciphers=None,\n    alpn_protocols=None,\n) -> ssl.SSLContext:\n    \"\"\"Return an equivalent SSLContext based on ssl.wrap_socket args.\"\"\"\n    ssl_version = resolve_ssl_version(ssl_version)\n    cert_none = resolve_cert_reqs(\"CERT_NONE\")\n    if cert_reqs is None:\n        cert_reqs = cert_none\n    else:\n        cert_reqs = resolve_cert_reqs(cert_reqs)\n\n    ctx = ssl.SSLContext(ssl_version)\n    ctx.load_cert_chain(certfile, keyfile)\n    ctx.verify_mode = cert_reqs\n    if ctx.verify_mode != cert_none:\n        ctx.load_verify_locations(cafile=ca_certs)\n    if alpn_protocols and hasattr(ctx, \"set_alpn_protocols\"):\n        try:\n            ctx.set_alpn_protocols(alpn_protocols)\n        except NotImplementedError:\n            pass\n    return ctx\n\n\ndef run_tornado_app(\n    app: tornado.web.Application,\n    certs: dict[str, typing.Any] | None,\n    scheme: str,\n    host: str,\n) -> tuple[tornado.httpserver.HTTPServer, int]:\n    # We can't use fromtimestamp(0) because of CPython issue 29097, so we'll\n    # just construct the datetime object directly.\n    app.last_req = datetime(1970, 1, 1)  # type: ignore[attr-defined]\n\n    if scheme == \"https\":\n        assert certs is not None\n        ssl_opts = ssl_options_to_context(**certs)\n        http_server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_opts)\n    else:\n        http_server = tornado.httpserver.HTTPServer(app)\n\n    sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]\n    port = sockets[0].getsockname()[1]\n    http_server.add_sockets(sockets)\n    return http_server, port\n\n\ndef get_unreachable_address() -> tuple[str, int]:\n    # reserved as per rfc2606\n    return (\"something.invalid\", 54321)\n\n\ndef encrypt_key_pem(private_key_pem: trustme.Blob, password: bytes) -> trustme.Blob:\n    private_key = serialization.load_pem_private_key(\n        private_key_pem.bytes(), password=None, backend=default_backend()\n    )\n    encrypted_key = private_key.private_bytes(\n        serialization.Encoding.PEM,\n        serialization.PrivateFormat.TraditionalOpenSSL,\n        serialization.BestAvailableEncryption(password),\n    )\n    return trustme.Blob(encrypted_key)\n\n\nR = typing.TypeVar(\"R\")\n\n\ndef _run_and_close_tornado(\n    async_fn: typing.Callable[P, Coroutine[typing.Any, typing.Any, R]],\n    *args: P.args,\n    **kwargs: P.kwargs,\n) -> R:\n    tornado_loop = None\n\n    async def inner_fn() -> R:\n        nonlocal tornado_loop\n        tornado_loop = tornado.ioloop.IOLoop.current()\n        return await async_fn(*args, **kwargs)\n\n    try:\n        return asyncio.run(inner_fn())\n    finally:\n        tornado_loop.close(all_fds=True)  # type: ignore[union-attr]\n\n\n@contextlib.contextmanager\ndef run_loop_in_thread() -> Generator[tornado.ioloop.IOLoop, None, None]:\n    loop_started: concurrent.futures.Future[\n        tuple[tornado.ioloop.IOLoop, asyncio.Event]\n    ] = concurrent.futures.Future()\n    with concurrent.futures.ThreadPoolExecutor(\n        1, thread_name_prefix=\"test IOLoop\"\n    ) as tpe:\n\n        async def run() -> None:\n            io_loop = tornado.ioloop.IOLoop.current()\n            stop_event = asyncio.Event()\n            loop_started.set_result((io_loop, stop_event))\n            await stop_event.wait()\n\n        # run asyncio.run in a thread and collect exceptions from *either*\n        # the loop failing to start, or failing to close\n        ran = tpe.submit(_run_and_close_tornado, run)  # type: ignore[arg-type]\n        for f in concurrent.futures.as_completed((loop_started, ran)):  # type: ignore[misc]\n            if f is loop_started:\n                io_loop, stop_event = loop_started.result()\n                try:\n                    yield io_loop\n                finally:\n                    io_loop.add_callback(stop_event.set)\n\n            elif f is ran:\n                # if this is the first iteration the loop failed to start\n                # if it's the second iteration the loop has finished or\n                # the loop failed to close and we need to raise the exception\n                ran.result()\n                return\n\n\ndef main() -> int:\n    # For debugging dummyserver itself - python -m dummyserver.server\n    from .handlers import TestingApp\n\n    host = \"127.0.0.1\"\n\n    async def amain() -> int:\n        app = tornado.web.Application([(r\".*\", TestingApp)])\n        server, port = run_tornado_app(app, None, \"http\", host)\n\n        print(f\"Listening on http://{host}:{port}\")\n        await asyncio.Event().wait()\n        return 0\n\n    return asyncio.run(amain())\n\n\nif __name__ == \"__main__\":\n    sys.exit(main())\n", "path": "dummyserver/server.py"}], "after_files": [{"content": "#!/usr/bin/env python\n\n\"\"\"\nDummy server used for unit testing.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport asyncio\nimport concurrent.futures\nimport contextlib\nimport errno\nimport logging\nimport os\nimport socket\nimport ssl\nimport sys\nimport threading\nimport typing\nimport warnings\nfrom collections.abc import Coroutine, Generator\nfrom datetime import datetime\n\nimport tornado.httpserver\nimport tornado.ioloop\nimport tornado.netutil\nimport tornado.web\nimport trustme\nfrom cryptography.hazmat.backends import default_backend\nfrom cryptography.hazmat.primitives import serialization\n\nfrom urllib3.exceptions import HTTPWarning\nfrom urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version\n\nif typing.TYPE_CHECKING:\n    from typing_extensions import ParamSpec\n\n    P = ParamSpec(\"P\")\n\nlog = logging.getLogger(__name__)\n\nCERTS_PATH = os.path.join(os.path.dirname(__file__), \"certs\")\nDEFAULT_CERTS: dict[str, typing.Any] = {\n    \"certfile\": os.path.join(CERTS_PATH, \"server.crt\"),\n    \"keyfile\": os.path.join(CERTS_PATH, \"server.key\"),\n    \"cert_reqs\": ssl.CERT_OPTIONAL,\n    \"ca_certs\": os.path.join(CERTS_PATH, \"cacert.pem\"),\n    \"alpn_protocols\": ALPN_PROTOCOLS,\n}\nDEFAULT_CA = os.path.join(CERTS_PATH, \"cacert.pem\")\nDEFAULT_CA_KEY = os.path.join(CERTS_PATH, \"cacert.key\")\n\n\ndef _resolves_to_ipv6(host: str) -> bool:\n    \"\"\"Returns True if the system resolves host to an IPv6 address by default.\"\"\"\n    resolves_to_ipv6 = False\n    try:\n        for res in socket.getaddrinfo(host, None, socket.AF_UNSPEC):\n            af, _, _, _, _ = res\n            if af == socket.AF_INET6:\n                resolves_to_ipv6 = True\n    except socket.gaierror:\n        pass\n\n    return resolves_to_ipv6\n\n\ndef _has_ipv6(host: str) -> bool:\n    \"\"\"Returns True if the system can bind an IPv6 address.\"\"\"\n    sock = None\n    has_ipv6 = False\n\n    if socket.has_ipv6:\n        # has_ipv6 returns true if cPython was compiled with IPv6 support.\n        # It does not tell us if the system has IPv6 support enabled. To\n        # determine that we must bind to an IPv6 address.\n        # https://github.com/urllib3/urllib3/pull/611\n        # https://bugs.python.org/issue658327\n        try:\n            sock = socket.socket(socket.AF_INET6)\n            sock.bind((host, 0))\n            has_ipv6 = _resolves_to_ipv6(\"localhost\")\n        except Exception:\n            pass\n\n    if sock:\n        sock.close()\n    return has_ipv6\n\n\n# Some systems may have IPv6 support but DNS may not be configured\n# properly. We can not count that localhost will resolve to ::1 on all\n# systems. See https://github.com/urllib3/urllib3/pull/611 and\n# https://bugs.python.org/issue18792\nHAS_IPV6_AND_DNS = _has_ipv6(\"localhost\")\nHAS_IPV6 = _has_ipv6(\"::1\")\n\n\n# Different types of servers we have:\n\n\nclass NoIPv6Warning(HTTPWarning):\n    \"IPv6 is not available\"\n\n\nclass SocketServerThread(threading.Thread):\n    \"\"\"\n    :param socket_handler: Callable which receives a socket argument for one\n        request.\n    :param ready_event: Event which gets set when the socket handler is\n        ready to receive requests.\n    \"\"\"\n\n    USE_IPV6 = HAS_IPV6_AND_DNS\n\n    def __init__(\n        self,\n        socket_handler: typing.Callable[[socket.socket], None],\n        host: str = \"localhost\",\n        ready_event: threading.Event | None = None,\n    ) -> None:\n        super().__init__()\n        self.daemon = True\n\n        self.socket_handler = socket_handler\n        self.host = host\n        self.ready_event = ready_event\n\n    def _start_server(self) -> None:\n        if self.USE_IPV6:\n            sock = socket.socket(socket.AF_INET6)\n        else:\n            warnings.warn(\"No IPv6 support. Falling back to IPv4.\", NoIPv6Warning)\n            sock = socket.socket(socket.AF_INET)\n        if sys.platform != \"win32\":\n            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)\n        sock.bind((self.host, 0))\n        self.port = sock.getsockname()[1]\n\n        # Once listen() returns, the server socket is ready\n        sock.listen(1)\n\n        if self.ready_event:\n            self.ready_event.set()\n\n        self.socket_handler(sock)\n        sock.close()\n\n    def run(self) -> None:\n        self._start_server()\n\n\ndef ssl_options_to_context(  # type: ignore[no-untyped-def]\n    keyfile=None,\n    certfile=None,\n    server_side=None,\n    cert_reqs=None,\n    ssl_version: str | int | None = None,\n    ca_certs=None,\n    do_handshake_on_connect=None,\n    suppress_ragged_eofs=None,\n    ciphers=None,\n    alpn_protocols=None,\n) -> ssl.SSLContext:\n    \"\"\"Return an equivalent SSLContext based on ssl.wrap_socket args.\"\"\"\n    ssl_version = resolve_ssl_version(ssl_version)\n    cert_none = resolve_cert_reqs(\"CERT_NONE\")\n    if cert_reqs is None:\n        cert_reqs = cert_none\n    else:\n        cert_reqs = resolve_cert_reqs(cert_reqs)\n\n    ctx = ssl.SSLContext(ssl_version)\n    ctx.load_cert_chain(certfile, keyfile)\n    ctx.verify_mode = cert_reqs\n    if ctx.verify_mode != cert_none:\n        ctx.load_verify_locations(cafile=ca_certs)\n    if alpn_protocols and hasattr(ctx, \"set_alpn_protocols\"):\n        try:\n            ctx.set_alpn_protocols(alpn_protocols)\n        except NotImplementedError:\n            pass\n    return ctx\n\n\ndef run_tornado_app(\n    app: tornado.web.Application,\n    certs: dict[str, typing.Any] | None,\n    scheme: str,\n    host: str,\n) -> tuple[tornado.httpserver.HTTPServer, int]:\n    # We can't use fromtimestamp(0) because of CPython issue 29097, so we'll\n    # just construct the datetime object directly.\n    app.last_req = datetime(1970, 1, 1)  # type: ignore[attr-defined]\n\n    if scheme == \"https\":\n        assert certs is not None\n        ssl_opts = ssl_options_to_context(**certs)\n        http_server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_opts)\n    else:\n        http_server = tornado.httpserver.HTTPServer(app)\n\n    # When we request a socket with host localhost and port zero (None in Python), then\n    # Tornado gets a free IPv4 port and requests that same port in IPv6. But that port\n    # could easily be taken with IPv6, especially in crowded CI environments. For this\n    # reason we put bind_sockets in a retry loop. Full details:\n    #  * https://github.com/urllib3/urllib3/issues/2171\n    #  * https://github.com/tornadoweb/tornado/issues/1860\n    for i in range(10):\n        try:\n            sockets = tornado.netutil.bind_sockets(None, address=host)  # type: ignore[arg-type]\n        except OSError as e:\n            if e.errno == errno.EADDRINUSE:\n                # TODO this should be a warning if there's a way for pytest to print it\n                print(\n                    f\"Retrying bind_sockets({host}) after EADDRINUSE\", file=sys.stderr\n                )\n                continue\n        break\n\n    port = sockets[0].getsockname()[1]\n    http_server.add_sockets(sockets)\n    return http_server, port\n\n\ndef get_unreachable_address() -> tuple[str, int]:\n    # reserved as per rfc2606\n    return (\"something.invalid\", 54321)\n\n\ndef encrypt_key_pem(private_key_pem: trustme.Blob, password: bytes) -> trustme.Blob:\n    private_key = serialization.load_pem_private_key(\n        private_key_pem.bytes(), password=None, backend=default_backend()\n    )\n    encrypted_key = private_key.private_bytes(\n        serialization.Encoding.PEM,\n        serialization.PrivateFormat.TraditionalOpenSSL,\n        serialization.BestAvailableEncryption(password),\n    )\n    return trustme.Blob(encrypted_key)\n\n\nR = typing.TypeVar(\"R\")\n\n\ndef _run_and_close_tornado(\n    async_fn: typing.Callable[P, Coroutine[typing.Any, typing.Any, R]],\n    *args: P.args,\n    **kwargs: P.kwargs,\n) -> R:\n    tornado_loop = None\n\n    async def inner_fn() -> R:\n        nonlocal tornado_loop\n        tornado_loop = tornado.ioloop.IOLoop.current()\n        return await async_fn(*args, **kwargs)\n\n    try:\n        return asyncio.run(inner_fn())\n    finally:\n        tornado_loop.close(all_fds=True)  # type: ignore[union-attr]\n\n\n@contextlib.contextmanager\ndef run_loop_in_thread() -> Generator[tornado.ioloop.IOLoop, None, None]:\n    loop_started: concurrent.futures.Future[\n        tuple[tornado.ioloop.IOLoop, asyncio.Event]\n    ] = concurrent.futures.Future()\n    with concurrent.futures.ThreadPoolExecutor(\n        1, thread_name_prefix=\"test IOLoop\"\n    ) as tpe:\n\n        async def run() -> None:\n            io_loop = tornado.ioloop.IOLoop.current()\n            stop_event = asyncio.Event()\n            loop_started.set_result((io_loop, stop_event))\n            await stop_event.wait()\n\n        # run asyncio.run in a thread and collect exceptions from *either*\n        # the loop failing to start, or failing to close\n        ran = tpe.submit(_run_and_close_tornado, run)  # type: ignore[arg-type]\n        for f in concurrent.futures.as_completed((loop_started, ran)):  # type: ignore[misc]\n            if f is loop_started:\n                io_loop, stop_event = loop_started.result()\n                try:\n                    yield io_loop\n                finally:\n                    io_loop.add_callback(stop_event.set)\n\n            elif f is ran:\n                # if this is the first iteration the loop failed to start\n                # if it's the second iteration the loop has finished or\n                # the loop failed to close and we need to raise the exception\n                ran.result()\n                return\n\n\ndef main() -> int:\n    # For debugging dummyserver itself - python -m dummyserver.server\n    from .handlers import TestingApp\n\n    host = \"127.0.0.1\"\n\n    async def amain() -> int:\n        app = tornado.web.Application([(r\".*\", TestingApp)])\n        server, port = run_tornado_app(app, None, \"http\", host)\n\n        print(f\"Listening on http://{host}:{port}\")\n        await asyncio.Event().wait()\n        return 0\n\n    return asyncio.run(amain())\n\n\nif __name__ == \"__main__\":\n    sys.exit(main())\n", "path": "dummyserver/server.py"}]}