Add type hints to test decorators and fixtures
MODIFY
noxfile.py
noxfile.py
@@ -159,7 +159,14 @@ def mypy(session: nox.Session) -> None:
session.install("pytest>=6.2")
session.install("nox")
session.run("mypy", "--version")
- session.run("mypy", "src/urllib3", "dummyserver", "noxfile.py")
+ session.run(
+ "mypy",
+ "src/urllib3",
+ "dummyserver",
+ "noxfile.py",
+ "test/__init__.py",
+ "test/port_helpers.py",
+ )
@nox.session

MODIFY
test/__init__.py
test/__init__.py
@@ -5,14 +5,26 @@ import platform
import socket
import sys
import warnings
+from types import ModuleType, TracebackType
+from typing import (
+ TYPE_CHECKING,
+ Any,
+ Callable,
+ Dict,
+ List,
+ Optional,
+ Type,
+ TypeVar,
+ cast,
+)
import pytest
try:
try:
- import brotlicffi as brotli
+ import brotlicffi as brotli # type: ignore[import]
except ImportError:
- import brotli
+ import brotli # type: ignore[import]
except ImportError:
brotli = None
@@ -25,7 +37,17 @@ from urllib3.util import ssl_
try:
import urllib3.contrib.pyopenssl as pyopenssl
except ImportError:
- pyopenssl = None
+ pyopenssl = None # type: ignore[assignment]
+
+if TYPE_CHECKING:
+ import ssl
+
+ from typing_extensions import Literal
+
+
+_RT = TypeVar("_RT") # return type
+_TestFuncT = TypeVar("_TestFuncT", bound=Callable[..., Any])
+
# We need a host that will not immediately close the connection with a TCP
# Reset.
@@ -56,7 +78,7 @@ if os.environ.get("CI") or os.environ.get("GITHUB_ACTIONS") == "true":
LONG_TIMEOUT = 0.5
-def _can_resolve(host):
+def _can_resolve(host: str) -> bool:
""" Returns True if the system can resolve host to an address. """
try:
socket.getaddrinfo(host, None, socket.AF_UNSPEC)
@@ -65,10 +87,10 @@ def _can_resolve(host):
return False
-def has_alpn(ctx_cls=None):
+def has_alpn(ctx_cls: Optional[Type["ssl.SSLContext"]] = None) -> bool:
""" Detect if ALPN support is enabled. """
ctx_cls = ctx_cls or util.SSLContext
- ctx = ctx_cls(protocol=ssl_.PROTOCOL_TLS)
+ ctx = ctx_cls(protocol=ssl_.PROTOCOL_TLS) # type: ignore[misc]
try:
if hasattr(ctx, "set_alpn_protocols"):
ctx.set_alpn_protocols(ssl_.ALPN_PROTOCOLS)
@@ -84,84 +106,78 @@ def has_alpn(ctx_cls=None):
RESOLVES_LOCALHOST_FQDN = _can_resolve("localhost.")
-def clear_warnings(cls=HTTPWarning):
+def clear_warnings(cls: Type[Warning] = HTTPWarning) -> None:
new_filters = []
- for f in warnings.filters:
+ for f in warnings.filters: # type: ignore[attr-defined]
if issubclass(f[2], cls):
continue
new_filters.append(f)
- warnings.filters[:] = new_filters
+ warnings.filters[:] = new_filters # type: ignore[attr-defined]
-def setUp():
+def setUp() -> None:
clear_warnings()
warnings.simplefilter("ignore", HTTPWarning)
-def notWindows(test):
+def notWindows() -> Callable[[_TestFuncT], _TestFuncT]:
"""Skips this test on Windows"""
-
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
- msg = f"{test.__name__} does not run on Windows"
- if platform.system() == "Windows":
- pytest.skip(msg)
- return test(*args, **kwargs)
-
- return wrapper
+ return pytest.mark.skipif(
+ platform.system() == "Windows",
+ reason="Test does not run on Windows",
+ )
-def onlyBrotli():
+def onlyBrotli() -> Callable[[_TestFuncT], _TestFuncT]:
return pytest.mark.skipif(
brotli is None, reason="only run if brotli library is present"
)
-def notBrotli():
+def notBrotli() -> Callable[[_TestFuncT], _TestFuncT]:
return pytest.mark.skipif(
brotli is not None, reason="only run if a brotli library is absent"
)
-def onlySecureTransport(test):
- """Runs this test when SecureTransport is in use."""
-
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
- msg = f"{test.__name__} only runs with SecureTransport"
- if not ssl_.IS_SECURETRANSPORT:
- pytest.skip(msg)
- return test(*args, **kwargs)
+# Hack to make pytest evaluate a condition at test runtime instead of collection time.
+def lazy_condition(condition: Callable[[], bool]) -> bool:
+ class LazyCondition:
+ def __bool__(self) -> bool:
+ return condition()
- return wrapper
+ return cast(bool, LazyCondition())
-def notSecureTransport(test):
- """Skips this test when SecureTransport is in use."""
+def onlySecureTransport() -> Callable[[_TestFuncT], _TestFuncT]:
+ """Runs this test when SecureTransport is in use."""
+ return pytest.mark.skipif(
+ lazy_condition(lambda: not ssl_.IS_SECURETRANSPORT),
+ reason="Test only runs with SecureTransport",
+ )
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
- msg = f"{test.__name__} does not run with SecureTransport"
- if ssl_.IS_SECURETRANSPORT:
- pytest.skip(msg)
- return test(*args, **kwargs)
- return wrapper
+def notSecureTransport() -> Callable[[_TestFuncT], _TestFuncT]:
+ """Skips this test when SecureTransport is in use."""
+ return pytest.mark.skipif(
+ lazy_condition(lambda: ssl_.IS_SECURETRANSPORT),
+ reason="Test does not run with SecureTransport",
+ )
_requires_network_has_route = None
-def requires_network(test):
+def requires_network() -> Callable[[_TestFuncT], _TestFuncT]:
"""Helps you skip tests that require the network"""
- def _is_unreachable_err(err):
+ def _is_unreachable_err(err: Exception) -> bool:
return getattr(err, "errno", None) in (
errno.ENETUNREACH,
errno.EHOSTUNREACH, # For OSX
)
- def _has_route():
+ def _has_route() -> bool:
try:
sock = socket.create_connection((TARPIT_HOST, 80), 0.0001)
sock.close()
@@ -174,50 +190,35 @@ def requires_network(test):
else:
raise
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
global _requires_network_has_route
if _requires_network_has_route is None:
_requires_network_has_route = _has_route()
- if _requires_network_has_route:
- return test(*args, **kwargs)
- else:
- msg = f"Can't run {test.__name__} because the network is unreachable"
- pytest.skip(msg)
-
- return wrapper
+ return pytest.mark.skipif(
+ not _requires_network_has_route,
+ reason="Can't run the test because the network is unreachable",
+ )
-def requires_ssl_context_keyfile_password(test):
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
- if ssl_.IS_SECURETRANSPORT:
- pytest.skip(
- "%s requires password parameter for "
- "SSLContext.load_cert_chain()" % test.__name__
+def requires_ssl_context_keyfile_password() -> Callable[[_TestFuncT], _TestFuncT]:
+ return pytest.mark.skipif(
+ lazy_condition(lambda: ssl_.IS_SECURETRANSPORT),
+ reason="Test requires password parameter for SSLContext.load_cert_chain()",
)
- return test(*args, **kwargs)
-
- return wrapper
-def resolvesLocalhostFQDN(test):
+def resolvesLocalhostFQDN() -> Callable[[_TestFuncT], _TestFuncT]:
"""Test requires successful resolving of 'localhost.'"""
-
- @functools.wraps(test)
- def wrapper(*args, **kwargs):
- if not RESOLVES_LOCALHOST_FQDN:
- pytest.skip("Can't resolve localhost.")
- return test(*args, **kwargs)
-
- return wrapper
+ return pytest.mark.skipif(
+ not RESOLVES_LOCALHOST_FQDN,
+ reason="Can't resolve localhost.",
+ )
-def withPyOpenSSL(test):
+def withPyOpenSSL(test: Callable[..., _RT]) -> Callable[..., _RT]:
@functools.wraps(test)
- def wrapper(*args, **kwargs):
+ def wrapper(*args: Any, **kwargs: Any) -> _RT:
if not pyopenssl:
pytest.skip("pyopenssl not available, skipping test.")
return test(*args, **kwargs)
@@ -231,35 +232,40 @@ def withPyOpenSSL(test):
class _ListHandler(logging.Handler):
- def __init__(self):
+ def __init__(self) -> None:
super().__init__()
- self.records = []
+ self.records: List[logging.LogRecord] = []
- def emit(self, record):
+ def emit(self, record: logging.LogRecord) -> None:
self.records.append(record)
class LogRecorder:
- def __init__(self, target=logging.root):
+ def __init__(self, target: logging.Logger = logging.root) -> None:
super().__init__()
self._target = target
self._handler = _ListHandler()
@property
- def records(self):
+ def records(self) -> List[logging.LogRecord]:
return self._handler.records
- def install(self):
+ def install(self) -> None:
self._target.addHandler(self._handler)
- def uninstall(self):
+ def uninstall(self) -> None:
self._target.removeHandler(self._handler)
- def __enter__(self):
+ def __enter__(self) -> List[logging.LogRecord]:
self.install()
return self.records
- def __exit__(self, exc_type, exc_value, traceback):
+ def __exit__(
+ self,
+ exc_type: Optional[Type[BaseException]],
+ exc_value: Optional[BaseException],
+ traceback: Optional[TracebackType],
+ ) -> "Literal[False]":
self.uninstall()
return False
@@ -272,15 +278,17 @@ class ImportBlocker:
specified cannot be imported, even if they are a builtin.
"""
- def __init__(self, *namestoblock):
+ def __init__(self, *namestoblock: str) -> None:
self.namestoblock = namestoblock
- def find_module(self, fullname, path=None):
+ def find_module(
+ self, fullname: str, path: Optional[str] = None
+ ) -> Optional["ImportBlocker"]:
if fullname in self.namestoblock:
return self
return None
- def load_module(self, fullname):
+ def load_module(self, fullname: str) -> None:
raise ImportError(f"import of {fullname} is blocked")
@@ -292,19 +300,22 @@ class ModuleStash:
modules
"""
- def __init__(self, namespace, modules=sys.modules):
+ def __init__(
+ self, namespace: str, modules: Dict[str, ModuleType] = sys.modules
+ ) -> None:
self.namespace = namespace
self.modules = modules
- self._data = {}
+ self._data: Dict[str, ModuleType] = {}
- def stash(self):
- self._data[self.namespace] = self.modules.pop(self.namespace, None)
+ def stash(self) -> None:
+ if self.namespace in self.modules:
+ self._data[self.namespace] = self.modules.pop(self.namespace)
for module in list(self.modules.keys()):
if module.startswith(self.namespace + "."):
self._data[module] = self.modules.pop(module)
- def pop(self):
+ def pop(self) -> None:
self.modules.pop(self.namespace, None)
for module in list(self.modules.keys()):

MODIFY
test/port_helpers.py
test/port_helpers.py
@@ -9,7 +9,10 @@ HOST = "127.0.0.1"
HOSTv6 = "::1"
-def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM):
+def find_unused_port(
+ family: socket.AddressFamily = socket.AF_INET,
+ socktype: socket.SocketKind = socket.SOCK_STREAM,
+) -> int:
"""Returns an unused port that should be suitable for binding. This is
achieved by creating a temporary socket with the same family and type as
the 'sock' parameter (default is AF_INET, SOCK_STREAM), and binding it to
@@ -70,7 +73,7 @@ def find_unused_port(family=socket.AF_INET, socktype=socket.SOCK_STREAM):
return port
-def bind_port(sock, host=HOST):
+def bind_port(sock: socket.socket, host: str = HOST) -> int:
"""Bind the socket to a free port and return the port number. Relies on
ephemeral ports in order to ensure we are using an unbound port. This is
important as many tests may be running simultaneously, especially in a
@@ -109,4 +112,5 @@ def bind_port(sock, host=HOST):
sock.bind((host, 0))
port = sock.getsockname()[1]
+ assert isinstance(port, int)
return port

MODIFY
test/test_poolmanager.py
test/test_poolmanager.py
@@ -11,7 +11,7 @@ from urllib3.util import retry, timeout
class TestPoolManager:
- @resolvesLocalhostFQDN
+ @resolvesLocalhostFQDN()
def test_same_url(self):
# Convince ourselves that normally we don't get the same object
conn1 = connection_from_url("http://localhost:8081/foo")

MODIFY
test/with_dummyserver/test_https.py
test/with_dummyserver/test_https.py
@@ -125,7 +125,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
r = https_pool.request("GET", "/")
assert r.status == 200, r.data
- @resolvesLocalhostFQDN
+ @resolvesLocalhostFQDN()
def test_dotted_fqdn(self):
with HTTPSConnectionPool(
self.host + ".", self.port, ca_certs=DEFAULT_CA
@@ -168,7 +168,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
with pytest.raises((SSLError, ProtocolError)):
https_pool.request("GET", "/certificate", retries=False)
- @requires_ssl_context_keyfile_password
+ @requires_ssl_context_keyfile_password()
def test_client_key_password(self):
with HTTPSConnectionPool(
self.host,
@@ -182,7 +182,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
subject = json.loads(r.data.decode("utf-8"))
assert subject["organizationalUnitName"].startswith("Testing cert")
- @requires_ssl_context_keyfile_password
+ @requires_ssl_context_keyfile_password()
def test_client_encrypted_key_requires_password(self):
with HTTPSConnectionPool(
self.host,
@@ -238,7 +238,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
assert r.status == 200
assert not warn.called, warn.call_args_list
- @notSecureTransport # SecureTransport does not support cert directories
+ @notSecureTransport() # SecureTransport does not support cert directories
def test_ca_dir_verified(self, tmpdir):
# OpenSSL looks up certificates by the hash for their name, see c_rehash
# TODO infer the bytes using `cryptography.x509.Name.public_bytes`.
@@ -461,7 +461,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
)
https_pool.request("GET", "/")
- @notSecureTransport
+ @notSecureTransport()
def test_good_fingerprint_and_hostname_mismatch(self):
# This test doesn't run with SecureTransport because we don't turn off
# hostname validation without turning off all validation, which this
@@ -475,7 +475,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
)
https_pool.request("GET", "/")
- @requires_network
+ @requires_network()
def test_https_timeout(self):
timeout = Timeout(total=None, connect=SHORT_TIMEOUT)
@@ -523,7 +523,7 @@ class TestHTTPS(HTTPSDummyServerTestCase):
finally:
conn.close()
- @requires_network
+ @requires_network()
def test_enhanced_timeout(self):
with HTTPSConnectionPool(
TARPIT_HOST,

MODIFY
test/with_dummyserver/test_proxy_poolmanager.py
test/with_dummyserver/test_proxy_poolmanager.py
@@ -94,7 +94,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
):
https.request("GET", f"{self.https_url}/")
- @onlySecureTransport
+ @onlySecureTransport()
def test_https_proxy_securetransport_not_supported(self):
with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
r = https.request("GET", f"{self.http_url}/")
@@ -444,7 +444,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
assert sc3 == sc4
@pytest.mark.timeout(0.5)
- @requires_network
+ @requires_network()
def test_https_proxy_timeout(self):
with proxy_from_url(f"https://{TARPIT_HOST}") as https:
with pytest.raises(MaxRetryError) as e:
@@ -452,7 +452,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
assert type(e.value.reason) == ConnectTimeoutError
@pytest.mark.timeout(0.5)
- @requires_network
+ @requires_network()
def test_https_proxy_pool_timeout(self):
with proxy_from_url(f"https://{TARPIT_HOST}", timeout=SHORT_TIMEOUT) as https:
with pytest.raises(MaxRetryError) as e:

MODIFY
test/with_dummyserver/test_socketlevel.py
test/with_dummyserver/test_socketlevel.py
@@ -285,11 +285,11 @@ class TestClientCerts(SocketDummyServerTestCase):
done_receiving.set()
done_receiving.set()
- @requires_ssl_context_keyfile_password
+ @requires_ssl_context_keyfile_password()
def test_client_cert_with_string_password(self):
self.run_client_cert_with_password_test("letmein")
- @requires_ssl_context_keyfile_password
+ @requires_ssl_context_keyfile_password()
def test_client_cert_with_bytes_password(self):
self.run_client_cert_with_password_test(b"letmein")
@@ -340,7 +340,7 @@ class TestClientCerts(SocketDummyServerTestCase):
assert len(client_certs) == 1
- @requires_ssl_context_keyfile_password
+ @requires_ssl_context_keyfile_password()
def test_load_keyfile_with_invalid_password(self):
context = ssl_.SSLContext(ssl_.PROTOCOL_SSLv23)
@@ -1180,7 +1180,7 @@ class TestSSL(SocketDummyServerTestCase):
pool.request("GET", "/", retries=0)
assert isinstance(cm.value.reason, SSLError)
- @notSecureTransport
+ @notSecureTransport()
def test_ssl_read_timeout(self):
timed_out = Event()
@@ -1558,7 +1558,7 @@ class TestHeaders(SocketDummyServerTestCase):
request_headers = filter_non_x_headers(self.parsed_headers)
assert expected_request_headers == request_headers
- @resolvesLocalhostFQDN
+ @resolvesLocalhostFQDN()
def test_request_host_header_ignores_fqdn_dot(self):
self.start_parsing_handler()
@@ -1823,7 +1823,7 @@ class TestRetryPoolSizeDrainFail(SocketDummyServerTestCase):
class TestBrokenPipe(SocketDummyServerTestCase):
- @notWindows
+ @notWindows()
def test_ignore_broken_pipe_errors(self, monkeypatch):
# On Windows an aborted connection raises an error on
# attempts to read data out of a socket that's been closed.

Add type hints to test decorators and fixtures