Add type hints to connection modules
MODIFY
noxfile.py
noxfile.py
@@ -9,6 +9,7 @@ import nox
# ignored.
TYPED_FILES = {
"src/urllib3/contrib/__init__.py",
+ "src/urllib3/connection.py",
"src/urllib3/exceptions.py",
"src/urllib3/_collections.py",
"src/urllib3/fields.py",
@@ -16,6 +17,7 @@ TYPED_FILES = {
"src/urllib3/packages/__init__.py",
"src/urllib3/packages/ssl_match_hostname/__init__.py",
"src/urllib3/packages/ssl_match_hostname/_implementation.py",
+ "src/urllib3/util/connection.py",
"src/urllib3/util/queue.py",
"src/urllib3/util/response.py",
"src/urllib3/util/url.py",

MODIFY
src/urllib3/connection.py
src/urllib3/connection.py
@@ -3,10 +3,29 @@ import logging
import os
import re
import socket
+import sys
import warnings
+from copy import copy
from http.client import HTTPConnection as _HTTPConnection
from http.client import HTTPException # noqa: F401
from socket import timeout as SocketTimeout
+from typing import (
+ IO,
+ TYPE_CHECKING,
+ Any,
+ Callable,
+ Dict,
+ Iterable,
+ Mapping,
+ NamedTuple,
+ Optional,
+ Tuple,
+ Union,
+ cast,
+)
+
+if TYPE_CHECKING:
+ from typing_extensions import Literal
from .util.proxy import create_proxy_ssl_context
from .util.util import to_str
@@ -16,9 +35,9 @@ try: # Compiled with SSL?
BaseSSLError = ssl.SSLError
except (ImportError, AttributeError): # Platform-specific: No SSL.
- ssl = None
+ ssl = None # type: ignore
- class BaseSSLError(BaseException):
+ class BaseSSLError(BaseException): # type: ignore
pass
@@ -55,6 +74,14 @@ RECENT_DATE = datetime.date(2020, 7, 1)
_CONTAINS_CONTROL_CHAR_RE = re.compile(r"[^-!#$%&'*+.^_`|~0-9a-zA-Z]")
+HTTPBody = Union[bytes, IO[Any], Iterable[bytes], str]
+
+
+class ProxyConfig(NamedTuple):
+ ssl_context: "ssl.SSLContext"
+ use_forwarding_for_https: bool
+
+
class HTTPConnection(_HTTPConnection):
"""
Based on :class:`http.client.HTTPConnection` but provides an extra constructor
@@ -80,31 +107,62 @@ class HTTPConnection(_HTTPConnection):
Or you may want to disable the defaults by passing an empty list (e.g., ``[]``).
"""
- default_port = port_by_scheme["http"]
+ default_port: int = port_by_scheme["http"]
#: Disable Nagle's algorithm by default.
#: ``[(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]``
- default_socket_options = [(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)]
+ default_socket_options: connection.SocketOptions = [
+ (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+ ]
#: Whether this connection verifies the host's certificate.
- is_verified = False
+ is_verified: bool = False
- def __init__(self, *args, **kw):
+ source_address: Optional[Tuple[str, int]]
+ socket_options: Optional[connection.SocketOptions]
+ _tunnel_host: Optional[str]
+ _tunnel: Callable[["HTTPConnection"], None]
+
+ def __init__(
+ self,
+ host: str,
+ port: Optional[int] = None,
+ timeout: Optional[float] = connection.SOCKET_GLOBAL_DEFAULT_TIMEOUT,
+ source_address: Optional[Tuple[str, int]] = None,
+ blocksize: int = 8192,
+ socket_options: Optional[connection.SocketOptions] = default_socket_options,
+ proxy: Optional[str] = None,
+ proxy_config: Optional[ProxyConfig] = None,
+ ) -> None:
# Pre-set source_address.
- self.source_address = kw.get("source_address")
+ self.source_address = source_address
- #: The socket options provided by the user. If no options are
- #: provided, we use the default options.
- self.socket_options = kw.pop("socket_options", self.default_socket_options)
+ self.socket_options = socket_options
# Proxy options provided by the user.
- self.proxy = kw.pop("proxy", None)
- self.proxy_config = kw.pop("proxy_config", None)
-
- super().__init__(*args, **kw)
+ self.proxy = proxy
+ self.proxy_config = proxy_config
+
+ if sys.version_info >= (3, 7):
+ super().__init__(
+ host=host,
+ port=port,
+ timeout=timeout,
+ source_address=source_address,
+ blocksize=blocksize,
+ )
+ else:
+ super().__init__(
+ host=host, port=port, timeout=timeout, source_address=source_address
+ )
- @property
- def host(self):
+ # https://github.com/python/mypy/issues/4125
+ # Mypy treats this as LSP violation, which is considered a bug.
+ # If `host` is made a property it violates LSP, because a writeable attribute is overriden with a read-only one.
+ # However, there is also a `host` setter so LSP is not violated.
+ # Potentailly, a `@host.deleter` might be needed depending on how this issue will be fixed.
+ @property # type: ignore
+ def host(self) -> str: # type: ignore
"""
Getter method to remove any trailing dots that indicate the hostname is an FQDN.
@@ -123,7 +181,7 @@ class HTTPConnection(_HTTPConnection):
return self._dns_host.rstrip(".")
@host.setter
- def host(self, value):
+ def host(self, value: str) -> None:
"""
Setter for the `host` property.
@@ -132,21 +190,18 @@ class HTTPConnection(_HTTPConnection):
"""
self._dns_host = value
- def _new_conn(self):
+ def _new_conn(self) -> socket.socket:
"""Establish a socket connection and set nodelay settings on it.
:return: New socket connection.
"""
- extra_kw = {}
- if self.source_address:
- extra_kw["source_address"] = self.source_address
-
- if self.socket_options:
- extra_kw["socket_options"] = self.socket_options
try:
conn = connection.create_connection(
- (self._dns_host, self.port), self.timeout, **extra_kw
+ (self._dns_host, self.port),
+ self.timeout,
+ source_address=self.source_address,
+ socket_options=self.socket_options,
)
except SocketTimeout:
@@ -156,14 +211,14 @@ class HTTPConnection(_HTTPConnection):
)
except OSError as e:
- raise NewConnectionError(self, f"Failed to establish a new connection: {e}")
+ raise NewConnectionError(self, f"Failed to establish a new connection: {e}") # type: ignore
return conn
- def _is_using_tunnel(self):
+ def _is_using_tunnel(self) -> Optional[str]:
return self._tunnel_host
- def _prepare_conn(self, conn):
+ def _prepare_conn(self, conn: socket.socket) -> None:
self.sock = conn
if self._is_using_tunnel():
# TODO: Fix tunnel so it doesn't depend on self.sock state.
@@ -171,11 +226,17 @@ class HTTPConnection(_HTTPConnection):
# Mark this connection as not reusable
self.auto_open = 0
- def connect(self):
+ def connect(self) -> None:
conn = self._new_conn()
self._prepare_conn(conn)
- def putrequest(self, method, url, *args, **kwargs):
+ def putrequest(
+ self,
+ method: str,
+ url: str,
+ skip_host: bool = False,
+ skip_accept_encoding: bool = False,
+ ) -> None:
""""""
# Empty docstring because the indentation of CPython's implementation
# is broken but we don't want this method in our documentation.
@@ -185,9 +246,11 @@ class HTTPConnection(_HTTPConnection):
f"Method cannot contain non-token characters {method!r} (found at least {match.group()!r})"
)
- return super().putrequest(method, url, *args, **kwargs)
+ return super().putrequest(
+ method, url, skip_host=skip_host, skip_accept_encoding=skip_accept_encoding
+ )
- def putheader(self, header, *values):
+ def putheader(self, header: str, *values: str) -> None:
""""""
if not any(isinstance(v, str) and v == SKIP_HEADER for v in values):
super().putheader(header, *values)
@@ -198,22 +261,39 @@ class HTTPConnection(_HTTPConnection):
)
)
- def request(self, method, url, body=None, headers=None):
+ # `request` method's signature intentionally violates LSP.
+ # urllib3's API is different from `http.client.HTTPConnection` and the subclassing is only incidental.
+ def request( # type: ignore
+ self,
+ method: str,
+ url: str,
+ body: Optional[HTTPBody] = None,
+ headers: Optional[Mapping[str, str]] = None,
+ ) -> None:
if headers is None:
headers = {}
else:
# Avoid modifying the headers passed into .request()
- headers = headers.copy()
+ headers = copy(headers)
if "user-agent" not in (to_str(k.lower()) for k in headers):
- headers["User-Agent"] = _get_default_user_agent()
+ updated_headers = {"User-Agent": _get_default_user_agent()}
+ updated_headers.update(headers)
+ headers = updated_headers
super().request(method, url, body=body, headers=headers)
- def request_chunked(self, method, url, body=None, headers=None):
+ def request_chunked(
+ self,
+ method: str,
+ url: str,
+ body: Union[None, HTTPBody, Tuple[Union[bytes, str]]] = None,
+ headers: Optional[Mapping[str, str]] = None,
+ ) -> None:
"""
Alternative to the common request method, which sends the
body with chunked encoding and not as one block
"""
- headers = headers or {}
+ if headers is None:
+ headers = {}
header_keys = {to_str(k.lower()) for k in headers}
skip_accept_encoding = "accept-encoding" in header_keys
skip_host = "host" in header_keys
@@ -247,6 +327,12 @@ class HTTPConnection(_HTTPConnection):
self.send(b"0\r\n\r\n")
+_PCTRTT = Tuple[Tuple[str, str], ...]
+_PCTRTTT = Tuple[_PCTRTT, ...]
+_PeerCertRetDictType = Dict[str, Union[str, _PCTRTTT, _PCTRTT]]
+_PeerCertRetType = Union[_PeerCertRetDictType, bytes, None]
+
+
class HTTPSConnection(HTTPConnection):
"""
Many of the parameters to this constructor are passed to the underlying SSL
@@ -255,28 +341,41 @@ class HTTPSConnection(HTTPConnection):
default_port = port_by_scheme["https"]
- cert_reqs = None
- ca_certs = None
- ca_cert_dir = None
- ca_cert_data = None
- ssl_version = None
- assert_fingerprint = None
- tls_in_tls_required = False
+ cert_reqs: Optional[int] = None
+ ca_certs: Optional[str] = None
+ ca_cert_dir: Optional[str] = None
+ ca_cert_data: Union[None, str, bytes] = None
+ ssl_version: Optional[int] = None
+ assert_fingerprint: Optional[str] = None
+ tls_in_tls_required: bool = False
def __init__(
self,
+ host: str,
+ port: Optional[int] = None,
+ key_file: Optional[str] = None,
+ cert_file: Optional[str] = None,
+ key_password: Optional[str] = None,
+ timeout: Optional[float] = connection.SOCKET_GLOBAL_DEFAULT_TIMEOUT,
+ ssl_context: Optional["ssl.SSLContext"] = None,
+ server_hostname: Optional[str] = None,
+ source_address: Optional[Tuple[str, int]] = None,
+ blocksize: int = 8192,
+ socket_options: Optional[connection.SocketOptions] = None,
+ proxy: Optional[str] = None,
+ proxy_config: Optional[ProxyConfig] = None,
+ ) -> None:
+
+ super().__init__(
host,
- port=None,
- key_file=None,
- cert_file=None,
- key_password=None,
- timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
- ssl_context=None,
- server_hostname=None,
- **kw,
- ):
-
- super().__init__(host, port, timeout=timeout, **kw)
+ port=port,
+ timeout=timeout,
+ source_address=source_address,
+ blocksize=blocksize,
+ socket_options=socket_options,
+ proxy=proxy,
+ proxy_config=proxy_config,
+ )
self.key_file = key_file
self.cert_file = cert_file
@@ -286,16 +385,16 @@ class HTTPSConnection(HTTPConnection):
def set_cert(
self,
- key_file=None,
- cert_file=None,
- cert_reqs=None,
- key_password=None,
- ca_certs=None,
- assert_hostname=None,
- assert_fingerprint=None,
- ca_cert_dir=None,
- ca_cert_data=None,
- ):
+ key_file: Optional[str] = None,
+ cert_file: Optional[str] = None,
+ cert_reqs: Optional[int] = None,
+ key_password: Optional[str] = None,
+ ca_certs: Optional[str] = None,
+ assert_hostname: Union[None, str, "Literal[False]"] = None,
+ assert_fingerprint: Optional[str] = None,
+ ca_cert_dir: Optional[str] = None,
+ ca_cert_data: Union[None, str, bytes] = None,
+ ) -> None:
"""
This method should only be called once, before the connection is used.
"""
@@ -317,10 +416,10 @@ class HTTPSConnection(HTTPConnection):
self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
self.ca_cert_data = ca_cert_data
- def connect(self):
+ def connect(self) -> None:
# Add certificate verification
conn = self._new_conn()
- hostname = self.host
+ hostname: str = self.host
tls_in_tls = False
if self._is_using_tunnel():
@@ -337,7 +436,9 @@ class HTTPSConnection(HTTPConnection):
self.auto_open = 0
# Override the host with the one we're requesting data from.
- hostname = self._tunnel_host
+ hostname = cast(
+ str, self._tunnel_host
+ ) # self._tunnel_host is not None, because self._is_using_tunnel() returned a truthy value.
server_hostname = hostname
if self.server_hostname is not None:
@@ -433,17 +534,20 @@ class HTTPSConnection(HTTPConnection):
cert = self.sock.getpeercert()
_match_hostname(cert, self.assert_hostname or server_hostname)
- self.is_verified = (
- context.verify_mode == ssl.CERT_REQUIRED
- or self.assert_fingerprint is not None
+ self.is_verified = context.verify_mode == ssl.CERT_REQUIRED or bool(
+ self.assert_fingerprint
)
- def _connect_tls_proxy(self, hostname, conn):
+ def _connect_tls_proxy(
+ self, hostname: Optional[str], conn: socket.socket
+ ) -> "ssl.SSLSocket":
"""
Establish a TLS connection to the proxy using the provided SSL context.
"""
- proxy_config = self.proxy_config
+ proxy_config = cast(
+ ProxyConfig, self.proxy_config
+ ) # `_connect_tls_proxy` is called when self._is_using_tunnel() is truthy.
ssl_context = proxy_config.ssl_context
try:
@@ -482,7 +586,7 @@ class HTTPSConnection(HTTPConnection):
)
-def _match_hostname(cert, asserted_hostname):
+def _match_hostname(cert: _PeerCertRetType, asserted_hostname: str) -> None:
try:
match_hostname(cert, asserted_hostname)
except CertificateError as e:
@@ -493,11 +597,11 @@ def _match_hostname(cert, asserted_hostname):
)
# Add cert to exception and reraise so client code can inspect
# the cert when catching the exception, if they want to
- e._peer_cert = cert
+ e._peer_cert = cert # type: ignore
raise
-def _get_default_user_agent():
+def _get_default_user_agent() -> str:
return f"python-urllib3/{__version__}"
@@ -508,7 +612,7 @@ class DummyConnection:
if not ssl:
- HTTPSConnection = DummyConnection # noqa: F811
+ HTTPSConnection = DummyConnection # type: ignore # noqa: F811
VerifiedHTTPSConnection = HTTPSConnection

MODIFY
src/urllib3/poolmanager.py
src/urllib3/poolmanager.py
@@ -4,6 +4,7 @@ import logging
from urllib.parse import urljoin
from ._collections import RecentlyUsedContainer
+from .connection import ProxyConfig
from .connectionpool import HTTPConnectionPool, HTTPSConnectionPool, port_by_scheme
from .exceptions import (
LocationValueError,
@@ -66,9 +67,6 @@ _key_fields = (
#: All custom key schemes should include the fields in this key at a minimum.
PoolKey = collections.namedtuple("PoolKey", _key_fields)
-_proxy_config_fields = ("ssl_context", "use_forwarding_for_https")
-ProxyConfig = collections.namedtuple("ProxyConfig", _proxy_config_fields)
-
def _default_key_normalizer(key_class, request_context):
"""

MODIFY
src/urllib3/util/connection.py
src/urllib3/util/connection.py
import socket
+from typing import List, Optional, Tuple, Union
from urllib3.exceptions import LocationParseError
from .wait import wait_for_read
+SOCKET_GLOBAL_DEFAULT_TIMEOUT = socket._GLOBAL_DEFAULT_TIMEOUT # type: ignore
+SocketOptions = List[Tuple[int, int, Union[int, bytes]]]
-def is_connection_dropped(conn): # Platform-specific
+
+def is_connection_dropped(conn: socket.socket) -> bool: # Platform-specific
"""
Returns True if the connection is dropped and should be closed.
@@ -24,11 +28,11 @@ def is_connection_dropped(conn): # Platform-specific
# One additional modification is that we avoid binding to IPv6 servers
# discovered in DNS if the system doesn't have IPv6 functionality.
def create_connection(
- address,
- timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
- source_address=None,
- socket_options=None,
-):
+ address: Tuple[str, int],
+ timeout: Optional[float] = SOCKET_GLOBAL_DEFAULT_TIMEOUT,
+ source_address: Optional[Tuple[str, int]] = None,
+ socket_options: Optional[SocketOptions] = None,
+) -> socket.socket:
"""Connect to *address* and return the socket object.
Convenience function. Connect to *address* (a 2-tuple ``(host,
@@ -65,7 +69,7 @@ def create_connection(
# If provided, set socket level options before connecting.
_set_socket_options(sock, socket_options)
- if timeout is not socket._GLOBAL_DEFAULT_TIMEOUT:
+ if timeout is not SOCKET_GLOBAL_DEFAULT_TIMEOUT:
sock.settimeout(timeout)
if source_address:
sock.bind(source_address)
@@ -84,7 +88,7 @@ def create_connection(
raise OSError("getaddrinfo returns an empty list")
-def _set_socket_options(sock, options):
+def _set_socket_options(sock: socket.socket, options: Optional[SocketOptions]) -> None:
if options is None:
return
@@ -92,7 +96,7 @@ def _set_socket_options(sock, options):
sock.setsockopt(*opt)
-def allowed_gai_family():
+def allowed_gai_family() -> socket.AddressFamily:
"""This function is designed to work in the context of
getaddrinfo, where family=socket.AF_UNSPEC is the default and
will perform a DNS search for both IPv6 and IPv4 records."""
@@ -103,7 +107,7 @@ def allowed_gai_family():
return family
-def _has_ipv6(host):
+def _has_ipv6(host: str) -> bool:
""" Returns True if the system can bind an IPv6 address. """
sock = None
has_ipv6 = False

MODIFY
src/urllib3/util/proxy.py
src/urllib3/util/proxy.py
+from typing import TYPE_CHECKING, Optional, Union
+
from .ssl_ import create_urllib3_context, resolve_cert_reqs, resolve_ssl_version
+if TYPE_CHECKING:
+ import ssl
+
def connection_requires_http_tunnel(
proxy_url=None, proxy_config=None, destination_scheme=None
@@ -35,8 +40,12 @@ def connection_requires_http_tunnel(
def create_proxy_ssl_context(
- ssl_version, cert_reqs, ca_certs=None, ca_cert_dir=None, ca_cert_data=None
-):
+ ssl_version: Optional[int] = None,
+ cert_reqs: Optional[int] = None,
+ ca_certs: Optional[str] = None,
+ ca_cert_dir: Optional[str] = None,
+ ca_cert_data: Union[None, str, bytes] = None,
+) -> "ssl.SSLContext":
"""
Generates a default proxy ssl context if one hasn't been provided by the
user.

MODIFY
src/urllib3/util/ssl_.py
src/urllib3/util/ssl_.py
import hmac
import os
+import socket
import sys
import warnings
from binascii import hexlify, unhexlify
from hashlib import md5, sha1, sha256
+from typing import Optional, Union
from ..exceptions import ProxySchemeUnsupported, SNIMissingWarning, SSLError
from .url import _BRACELESS_IPV6_ADDRZ_RE, _IPV4_RE
@@ -99,7 +101,7 @@ DEFAULT_CIPHERS = ":".join(
)
-def assert_fingerprint(cert, fingerprint):
+def assert_fingerprint(cert: Optional[bytes], fingerprint: str) -> None:
"""
Checks if given fingerprint matches the supplied certificate.
@@ -109,6 +111,9 @@ def assert_fingerprint(cert, fingerprint):
Fingerprint as string of hexdigits, can be interspersed by colons.
"""
+ if cert is None:
+ raise SSLError("No certificate for the peer.")
+
fingerprint = fingerprint.replace(":", "").lower()
digest_length = len(fingerprint)
hashfunc = HASHFUNC_MAP.get(digest_length)
@@ -126,7 +131,7 @@ def assert_fingerprint(cert, fingerprint):
)
-def resolve_cert_reqs(candidate):
+def resolve_cert_reqs(candidate: Union[None, int, str]) -> int:
"""
Resolves the argument to a numeric constant, which can be passed to
the wrap_socket function/method from the ssl module.
@@ -149,7 +154,7 @@ def resolve_cert_reqs(candidate):
return candidate
-def resolve_ssl_version(candidate):
+def resolve_ssl_version(candidate: Union[None, int, str]) -> int:
"""
like resolve_cert_reqs
"""
@@ -166,8 +171,11 @@ def resolve_ssl_version(candidate):
def create_urllib3_context(
- ssl_version=None, cert_reqs=None, options=None, ciphers=None
-):
+ ssl_version: Optional[int] = None,
+ cert_reqs: Optional[int] = None,
+ options: Optional[int] = None,
+ ciphers: Optional[str] = None,
+) -> "ssl.SSLContext":
"""All arguments have the same meaning as ``ssl_wrap_socket``.
By default, this function does a lot of the same work that
@@ -257,20 +265,20 @@ def create_urllib3_context(
def ssl_wrap_socket(
- sock,
- keyfile=None,
- certfile=None,
- cert_reqs=None,
- ca_certs=None,
- server_hostname=None,
- ssl_version=None,
- ciphers=None,
- ssl_context=None,
- ca_cert_dir=None,
- key_password=None,
- ca_cert_data=None,
- tls_in_tls=False,
-):
+ sock: socket.socket,
+ keyfile: Optional[str] = None,
+ certfile: Optional[str] = None,
+ cert_reqs: Optional[int] = None,
+ ca_certs: Optional[str] = None,
+ server_hostname: Optional[str] = None,
+ ssl_version: Optional[int] = None,
+ ciphers: Optional[str] = None,
+ ssl_context: Optional["ssl.SSLContext"] = None,
+ ca_cert_dir: Optional[str] = None,
+ key_password: Optional[str] = None,
+ ca_cert_data: Union[None, str, bytes] = None,
+ tls_in_tls: bool = False,
+) -> "ssl.SSLSocket":
"""
All arguments except for server_hostname, ssl_context, and ca_cert_dir have
the same meaning as they do when using :func:`ssl.wrap_socket`.

MODIFY
src/urllib3/util/wait.py
src/urllib3/util/wait.py
import select
+import socket
from functools import partial
+from typing import Optional
__all__ = ["wait_for_read", "wait_for_write"]
@@ -92,7 +94,7 @@ def wait_for_socket(*args, **kwargs):
return wait_for_socket(*args, **kwargs)
-def wait_for_read(sock, timeout=None):
+def wait_for_read(sock: socket.socket, timeout: Optional[float] = None) -> bool:
"""Waits for reading to be available on a given socket.
Returns True if the socket is readable, or False if the timeout expired.
"""

MODIFY
test/test_ssl.py
test/test_ssl.py
@@ -2,7 +2,7 @@ from unittest import mock
import pytest
-from urllib3.exceptions import SNIMissingWarning
+from urllib3.exceptions import SNIMissingWarning, SSLError
from urllib3.util import ssl_
@@ -161,3 +161,9 @@ class TestSSL:
context.set_ciphers.assert_not_called()
else:
context.set_ciphers.assert_called_with(ssl_.DEFAULT_CIPHERS)
+
+ def test_assert_fingerprint_raises_exception_on_none_cert(self):
+ with pytest.raises(SSLError):
+ ssl_.assert_fingerprint(
+ cert=None, fingerprint="55:39:BF:70:05:12:43:FA:1F:D1:BF:4E:E8:1B:07:1D"
+ )

MODIFY
test/with_dummyserver/test_connectionpool.py
test/with_dummyserver/test_connectionpool.py
@@ -312,11 +312,14 @@ class TestConnectionPool(HTTPDummyServerTestCase):
finally:
s.close()
- def test_disable_default_socket_options(self):
- """Test that passing None disables all socket options."""
+ @pytest.mark.parametrize("socket_options", [None, []])
+ def test_disable_default_socket_options(self, socket_options):
+ """Test that passing None or empty list disables all socket options."""
# This test needs to be here in order to be run. socket.create_connection actually tries
# to connect to the host provided so we need a dummyserver to be running.
- with HTTPConnectionPool(self.host, self.port, socket_options=None) as pool:
+ with HTTPConnectionPool(
+ self.host, self.port, socket_options=socket_options
+ ) as pool:
s = pool._new_conn()._new_conn()
try:
using_nagle = s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) == 0
@@ -333,9 +336,7 @@ class TestConnectionPool(HTTPDummyServerTestCase):
conn = pool._new_conn()
try:
# Update the default socket options
- conn.default_socket_options += [
- (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
- ]
+ conn.socket_options += [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
s = conn._new_conn()
nagle_disabled = (
s.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) > 0

Add type hints to connection modules