We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Create a mechanism to test different HTTP protocols (HTTP/1.1 and HTTP/2) with the same test case
Currently our test suite only has to worry about HTTP/1.1, but with HTTP/2 support coming we're going to need to run tests both in HTTP/1.1 and HTTP/2 modes to avoid duplicating the entire test suite for the high-level tests like for `PoolManager`.

My proposal for the mechanism would be to control Hypercorn's ALPN to only offer `http/1.1` or `h2` depending on the test suite and then either run tests twice using a fixture or allowing external control somehow (environment variable?) and then running a separate job on CI? Open to other ideas too!

--- 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/socketserver.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 logging
10 import os
11 import socket
12 import ssl
13 import sys
14 import threading
15 import typing
16 import warnings
17 
18 import trustme
19 from cryptography.hazmat.backends import default_backend
20 from cryptography.hazmat.primitives import serialization
21 
22 from urllib3.exceptions import HTTPWarning
23 from urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version
24 
25 if typing.TYPE_CHECKING:
26     from typing_extensions import ParamSpec
27 
28     P = ParamSpec("P")
29 
30 log = logging.getLogger(__name__)
31 
32 CERTS_PATH = os.path.join(os.path.dirname(__file__), "certs")
33 DEFAULT_CERTS: dict[str, typing.Any] = {
34     "certfile": os.path.join(CERTS_PATH, "server.crt"),
35     "keyfile": os.path.join(CERTS_PATH, "server.key"),
36     "cert_reqs": ssl.CERT_OPTIONAL,
37     "ca_certs": os.path.join(CERTS_PATH, "cacert.pem"),
38     "alpn_protocols": ALPN_PROTOCOLS,
39 }
40 DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
41 DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")
42 
43 
44 def _resolves_to_ipv6(host: str) -> bool:
45     """Returns True if the system resolves host to an IPv6 address by default."""
46     resolves_to_ipv6 = False
47     try:
48         for res in socket.getaddrinfo(host, None, socket.AF_UNSPEC):
49             af, _, _, _, _ = res
50             if af == socket.AF_INET6:
51                 resolves_to_ipv6 = True
52     except socket.gaierror:
53         pass
54 
55     return resolves_to_ipv6
56 
57 
58 def _has_ipv6(host: str) -> bool:
59     """Returns True if the system can bind an IPv6 address."""
60     sock = None
61     has_ipv6 = False
62 
63     if socket.has_ipv6:
64         # has_ipv6 returns true if cPython was compiled with IPv6 support.
65         # It does not tell us if the system has IPv6 support enabled. To
66         # determine that we must bind to an IPv6 address.
67         # https://github.com/urllib3/urllib3/pull/611
68         # https://bugs.python.org/issue658327
69         try:
70             sock = socket.socket(socket.AF_INET6)
71             sock.bind((host, 0))
72             has_ipv6 = _resolves_to_ipv6("localhost")
73         except Exception:
74             pass
75 
76     if sock:
77         sock.close()
78     return has_ipv6
79 
80 
81 # Some systems may have IPv6 support but DNS may not be configured
82 # properly. We can not count that localhost will resolve to ::1 on all
83 # systems. See https://github.com/urllib3/urllib3/pull/611 and
84 # https://bugs.python.org/issue18792
85 HAS_IPV6_AND_DNS = _has_ipv6("localhost")
86 HAS_IPV6 = _has_ipv6("::1")
87 
88 
89 # Different types of servers we have:
90 
91 
92 class NoIPv6Warning(HTTPWarning):
93     "IPv6 is not available"
94 
95 
96 class SocketServerThread(threading.Thread):
97     """
98     :param socket_handler: Callable which receives a socket argument for one
99         request.
100     :param ready_event: Event which gets set when the socket handler is
101         ready to receive requests.
102     """
103 
104     USE_IPV6 = HAS_IPV6_AND_DNS
105 
106     def __init__(
107         self,
108         socket_handler: typing.Callable[[socket.socket], None],
109         host: str = "localhost",
110         ready_event: threading.Event | None = None,
111     ) -> None:
112         super().__init__()
113         self.daemon = True
114 
115         self.socket_handler = socket_handler
116         self.host = host
117         self.ready_event = ready_event
118 
119     def _start_server(self) -> None:
120         if self.USE_IPV6:
121             sock = socket.socket(socket.AF_INET6)
122         else:
123             warnings.warn("No IPv6 support. Falling back to IPv4.", NoIPv6Warning)
124             sock = socket.socket(socket.AF_INET)
125         if sys.platform != "win32":
126             sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
127 
128         with sock:
129             sock.bind((self.host, 0))
130             self.port = sock.getsockname()[1]
131 
132             # Once listen() returns, the server socket is ready
133             sock.listen(1)
134 
135             if self.ready_event:
136                 self.ready_event.set()
137 
138             self.socket_handler(sock)
139 
140     def run(self) -> None:
141         self._start_server()
142 
143 
144 def ssl_options_to_context(  # type: ignore[no-untyped-def]
145     keyfile=None,
146     certfile=None,
147     server_side=None,
148     cert_reqs=None,
149     ssl_version: str | int | None = None,
150     ca_certs=None,
151     do_handshake_on_connect=None,
152     suppress_ragged_eofs=None,
153     ciphers=None,
154     alpn_protocols=None,
155 ) -> ssl.SSLContext:
156     """Return an equivalent SSLContext based on ssl.wrap_socket args."""
157     ssl_version = resolve_ssl_version(ssl_version)
158     cert_none = resolve_cert_reqs("CERT_NONE")
159     if cert_reqs is None:
160         cert_reqs = cert_none
161     else:
162         cert_reqs = resolve_cert_reqs(cert_reqs)
163 
164     ctx = ssl.SSLContext(ssl_version)
165     ctx.load_cert_chain(certfile, keyfile)
166     ctx.verify_mode = cert_reqs
167     if ctx.verify_mode != cert_none:
168         ctx.load_verify_locations(cafile=ca_certs)
169     if alpn_protocols and hasattr(ctx, "set_alpn_protocols"):
170         try:
171             ctx.set_alpn_protocols(alpn_protocols)
172         except NotImplementedError:
173             pass
174     return ctx
175 
176 
177 def get_unreachable_address() -> tuple[str, int]:
178     # reserved as per rfc2606
179     return ("something.invalid", 54321)
180 
181 
182 def encrypt_key_pem(private_key_pem: trustme.Blob, password: bytes) -> trustme.Blob:
183     private_key = serialization.load_pem_private_key(
184         private_key_pem.bytes(), password=None, backend=default_backend()
185     )
186     encrypted_key = private_key.private_bytes(
187         serialization.Encoding.PEM,
188         serialization.PrivateFormat.TraditionalOpenSSL,
189         serialization.BestAvailableEncryption(password),
190     )
191     return trustme.Blob(encrypted_key)
192
```
--- 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/socketserver.py b/dummyserver/socketserver.py
--- a/dummyserver/socketserver.py
+++ b/dummyserver/socketserver.py
@@ -20,7 +20,7 @@
 from cryptography.hazmat.primitives import serialization
 
 from urllib3.exceptions import HTTPWarning
-from urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version
+from urllib3.util import resolve_cert_reqs, resolve_ssl_version
 
 if typing.TYPE_CHECKING:
     from typing_extensions import ParamSpec
@@ -35,7 +35,7 @@
     "keyfile": os.path.join(CERTS_PATH, "server.key"),
     "cert_reqs": ssl.CERT_OPTIONAL,
     "ca_certs": os.path.join(CERTS_PATH, "cacert.pem"),
-    "alpn_protocols": ALPN_PROTOCOLS,
+    "alpn_protocols": ["h2", "http/1.1"],
 }
 DEFAULT_CA = os.path.join(CERTS_PATH, "cacert.pem")
 DEFAULT_CA_KEY = os.path.join(CERTS_PATH, "cacert.key")

{"golden_diff": "diff --git a/dummyserver/socketserver.py b/dummyserver/socketserver.py\n--- a/dummyserver/socketserver.py\n+++ b/dummyserver/socketserver.py\n@@ -20,7 +20,7 @@\n from cryptography.hazmat.primitives import serialization\n \n from urllib3.exceptions import HTTPWarning\n-from urllib3.util import ALPN_PROTOCOLS, resolve_cert_reqs, resolve_ssl_version\n+from urllib3.util import resolve_cert_reqs, resolve_ssl_version\n \n if typing.TYPE_CHECKING:\n     from typing_extensions import ParamSpec\n@@ -35,7 +35,7 @@\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+    \"alpn_protocols\": [\"h2\", \"http/1.1\"],\n }\n DEFAULT_CA = os.path.join(CERTS_PATH, \"cacert.pem\")\n DEFAULT_CA_KEY = os.path.join(CERTS_PATH, \"cacert.key\")\n", "issue": "Create a mechanism to test different HTTP protocols (HTTP/1.1 and HTTP/2) with the same test case\nCurrently our test suite only has to worry about HTTP/1.1, but with HTTP/2 support coming we're going to need to run tests both in HTTP/1.1 and HTTP/2 modes to avoid duplicating the entire test suite for the high-level tests like for `PoolManager`.\r\n\r\nMy proposal for the mechanism would be to control Hypercorn's ALPN to only offer `http/1.1` or `h2` depending on the test suite and then either run tests twice using a fixture or allowing external control somehow (environment variable?) and then running a separate job on CI? Open to other ideas too!\n", "before_files": [{"content": "#!/usr/bin/env python\n\n\"\"\"\nDummy server used for unit testing.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport logging\nimport os\nimport socket\nimport ssl\nimport sys\nimport threading\nimport typing\nimport warnings\n\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\n        with sock:\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\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 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", "path": "dummyserver/socketserver.py"}], "after_files": [{"content": "#!/usr/bin/env python\n\n\"\"\"\nDummy server used for unit testing.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport logging\nimport os\nimport socket\nimport ssl\nimport sys\nimport threading\nimport typing\nimport warnings\n\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 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\": [\"h2\", \"http/1.1\"],\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\n        with sock:\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\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 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", "path": "dummyserver/socketserver.py"}]}