Add type hints to test.with_dummyserver.test_proxy_poolmanager
MODIFY
dummyserver/testcase.py
dummyserver/testcase.py
import socket
import threading
from contextlib import contextmanager
-from typing import Any, Callable, ClassVar, Generator, Iterable, Optional
+from typing import Any, Callable, ClassVar, Dict, Generator, Iterable, Optional
import pytest
from tornado import httpserver, ioloop, web
@@ -179,23 +179,32 @@ class HTTPSDummyServerTestCase(HTTPDummyServerTestCase):
class HTTPDummyProxyTestCase:
io_loop: ClassVar[ioloop.IOLoop]
- http_host = "localhost"
- http_host_alt = "127.0.0.1"
+ http_host: ClassVar[str] = "localhost"
+ http_host_alt: ClassVar[str] = "127.0.0.1"
http_server: ClassVar[httpserver.HTTPServer]
http_port: ClassVar[int]
+ http_url: ClassVar[str]
+ http_url_alt: ClassVar[str]
- https_host = "localhost"
- https_host_alt = "127.0.0.1"
- https_certs = DEFAULT_CERTS
+ https_host: ClassVar[str] = "localhost"
+ https_host_alt: ClassVar[str] = "127.0.0.1"
+ https_certs: ClassVar[Dict[str, Any]] = DEFAULT_CERTS
https_server: ClassVar[httpserver.HTTPServer]
https_port: ClassVar[int]
+ https_url: ClassVar[str]
+ https_url_alt: ClassVar[str]
- proxy_host = "localhost"
- proxy_host_alt = "127.0.0.1"
+ proxy_host: ClassVar[str] = "localhost"
+ proxy_host_alt: ClassVar[str] = "127.0.0.1"
proxy_server: ClassVar[httpserver.HTTPServer]
proxy_port: ClassVar[int]
+ proxy_url: ClassVar[str]
https_proxy_server: ClassVar[httpserver.HTTPServer]
https_proxy_port: ClassVar[int]
+ https_proxy_url: ClassVar[str]
+
+ certs_dir: ClassVar[str] = ""
+ bad_ca_path: ClassVar[str] = ""
server_thread: ClassVar[threading.Thread]

MODIFY
noxfile.py
noxfile.py
@@ -172,6 +172,7 @@ def mypy(session: nox.Session) -> None:
"test/with_dummyserver/test_integration.py",
"test/with_dummyserver/test_no_ssl.py",
"test/with_dummyserver/test_poolmanager.py",
+ "test/with_dummyserver/test_proxy_poolmanager.py",
"test/conftest.py",
"test/contrib/test_pyopenssl_dependencies.py",
"test/contrib/test_socks.py",

MODIFY
src/urllib3/connectionpool.py
src/urllib3/connectionpool.py
@@ -1032,7 +1032,7 @@ class HTTPSConnectionPool(HTTPConnectionPool):
)
-def connection_from_url(url: str, **kw: Any) -> ConnectionPool:
+def connection_from_url(url: str, **kw: Any) -> HTTPConnectionPool:
"""
Given a url, return an :class:`.ConnectionPool` instance of its host.

MODIFY
src/urllib3/poolmanager.py
src/urllib3/poolmanager.py
@@ -547,8 +547,10 @@ class ProxyManager(PoolManager):
) -> None:
if isinstance(proxy_url, HTTPConnectionPool):
- proxy_url = f"{proxy_url.scheme}://{proxy_url.host}:{proxy_url.port}"
- proxy = parse_url(proxy_url)
+ str_proxy_url = f"{proxy_url.scheme}://{proxy_url.host}:{proxy_url.port}"
+ else:
+ str_proxy_url = proxy_url
+ proxy = parse_url(str_proxy_url)
if proxy.scheme not in ("http", "https"):
raise ProxySchemeUnknown(proxy.scheme)

MODIFY
test/with_dummyserver/test_proxy_poolmanager.py
test/with_dummyserver/test_proxy_poolmanager.py
@@ -9,8 +9,10 @@ import trustme
from dummyserver.server import DEFAULT_CA, HAS_IPV6, get_unreachable_address
from dummyserver.testcase import HTTPDummyProxyTestCase, IPv6HTTPDummyProxyTestCase
+from urllib3 import HTTPResponse
from urllib3._collections import HTTPHeaderDict
-from urllib3.connectionpool import VerifiedHTTPSConnection, connection_from_url
+from urllib3.connection import VerifiedHTTPSConnection
+from urllib3.connectionpool import connection_from_url
from urllib3.exceptions import (
ConnectTimeoutError,
HTTPSProxyError,
@@ -31,7 +33,7 @@ pytestmark = pytest.mark.flaky
class TestHTTPProxyManager(HTTPDummyProxyTestCase):
@classmethod
- def setup_class(cls):
+ def setup_class(cls) -> None:
super().setup_class()
cls.http_url = f"http://{cls.http_host}:{int(cls.http_port)}"
cls.http_url_alt = f"http://{cls.http_host_alt}:{int(cls.http_port)}"
@@ -48,11 +50,11 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
bad_ca.cert_pem.write_to_path(cls.bad_ca_path)
@classmethod
- def teardown_class(cls):
+ def teardown_class(cls) -> None:
super().teardown_class()
shutil.rmtree(cls.certs_dir)
- def test_basic_proxy(self):
+ def test_basic_proxy(self) -> None:
with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
r = http.request("GET", f"{self.http_url}/")
assert r.status == 200
@@ -60,7 +62,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
r = http.request("GET", f"{self.https_url}/")
assert r.status == 200
- def test_https_proxy(self):
+ def test_https_proxy(self) -> None:
with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
r = https.request("GET", f"{self.https_url}/")
assert r.status == 200
@@ -68,7 +70,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
r = https.request("GET", f"{self.http_url}/")
assert r.status == 200
- def test_https_proxy_with_proxy_ssl_context(self):
+ def test_https_proxy_with_proxy_ssl_context(self) -> None:
proxy_ssl_context = create_urllib3_context()
proxy_ssl_context.load_verify_locations(DEFAULT_CA)
with proxy_from_url(
@@ -83,7 +85,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
assert r.status == 200
@withPyOpenSSL
- def test_https_proxy_pyopenssl_not_supported(self):
+ def test_https_proxy_pyopenssl_not_supported(self) -> None:
with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
r = https.request("GET", f"{self.http_url}/")
assert r.status == 200
@@ -94,7 +96,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
https.request("GET", f"{self.https_url}/")
@onlySecureTransport()
- def test_https_proxy_securetransport_not_supported(self):
+ def test_https_proxy_securetransport_not_supported(self) -> None:
with proxy_from_url(self.https_proxy_url, ca_certs=DEFAULT_CA) as https:
r = https.request("GET", f"{self.http_url}/")
assert r.status == 200
@@ -104,7 +106,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
):
https.request("GET", f"{self.https_url}/")
- def test_https_proxy_forwarding_for_https(self):
+ def test_https_proxy_forwarding_for_https(self) -> None:
with proxy_from_url(
self.https_proxy_url,
ca_certs=DEFAULT_CA,
@@ -116,7 +118,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
r = https.request("GET", f"{self.https_url}/")
assert r.status == 200
- def test_nagle_proxy(self):
+ def test_nagle_proxy(self) -> None:
""" Test that proxy connections do not have TCP_NODELAY turned on """
with ProxyManager(self.proxy_url) as http:
hc2 = http.connection_from_host(self.http_host, self.http_port)
@@ -133,7 +135,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
finally:
conn.close()
- def test_proxy_conn_fail(self):
+ def test_proxy_conn_fail(self) -> None:
host, port = get_unreachable_address()
with proxy_from_url(
f"http://{host}:{port}/", retries=1, timeout=LONG_TIMEOUT
@@ -147,7 +149,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
http.request("GET", f"{self.http_url}/")
assert type(e.value.reason) == ProxyError
- def test_https_conn_failed(self):
+ def test_https_conn_failed(self) -> None:
"""
Simulates a misconfiguration that is common for users. The test attempts
to establish a TLS connection to a non-TLS proxy
@@ -160,9 +162,9 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
http.request("GET", f"{self.https_url}/")
assert type(e.value.reason) == HTTPSProxyError
- def test_oldapi(self):
+ def test_oldapi(self) -> None:
with ProxyManager(
- connection_from_url(self.proxy_url), ca_certs=DEFAULT_CA
+ connection_from_url(self.proxy_url), ca_certs=DEFAULT_CA # type: ignore[arg-type]
) as http:
r = http.request("GET", f"{self.http_url}/")
assert r.status == 200
@@ -170,7 +172,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
r = http.request("GET", f"{self.https_url}/")
assert r.status == 200
- def test_proxy_verified(self):
+ def test_proxy_verified(self) -> None:
with proxy_from_url(
self.proxy_url, cert_reqs="REQUIRED", ca_certs=self.bad_ca_path
) as http:
@@ -204,7 +206,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
https_fail_pool.request("GET", "/", retries=0)
assert isinstance(e.value.reason, SSLError)
- def test_redirect(self):
+ def test_redirect(self) -> None:
with proxy_from_url(self.proxy_url) as http:
r = http.request(
"GET",
@@ -224,7 +226,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
assert r.status == 200
assert r.data == b"Dummy server!"
- def test_cross_host_redirect(self):
+ def test_cross_host_redirect(self) -> None:
with proxy_from_url(self.proxy_url) as http:
cross_host_location = f"{self.http_url_alt}/echo?a=b"
with pytest.raises(MaxRetryError):
@@ -241,9 +243,11 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
fields={"target": f"{self.http_url_alt}/echo?a=b"},
retries=1,
)
+ assert isinstance(r, HTTPResponse)
+ assert r._pool is not None
assert r._pool.host != self.http_host_alt
- def test_cross_protocol_redirect(self):
+ def test_cross_protocol_redirect(self) -> None:
with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
cross_protocol_location = f"{self.https_url}/echo?a=b"
with pytest.raises(MaxRetryError):
@@ -260,9 +264,11 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
fields={"target": f"{self.https_url}/echo?a=b"},
retries=1,
)
+ assert isinstance(r, HTTPResponse)
+ assert r._pool is not None
assert r._pool.host == self.https_host
- def test_headers(self):
+ def test_headers(self) -> None:
with proxy_from_url(
self.proxy_url,
headers={"Foo": "bar"},
@@ -338,7 +344,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
)
- def test_https_headers(self):
+ def test_https_headers(self) -> None:
with proxy_from_url(
self.https_proxy_url,
headers={"Foo": "bar"},
@@ -371,7 +377,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
)
- def test_https_headers_forwarding_for_https(self):
+ def test_https_headers_forwarding_for_https(self) -> None:
with proxy_from_url(
self.https_proxy_url,
headers={"Foo": "bar"},
@@ -388,7 +394,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
returned_headers.get("Host") == f"{self.https_host}:{self.https_port}"
)
- def test_headerdict(self):
+ def test_headerdict(self) -> None:
default_headers = HTTPHeaderDict(a="b")
proxy_headers = HTTPHeaderDict()
proxy_headers.add("foo", "bar")
@@ -402,7 +408,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
assert returned_headers.get("Foo") == "bar"
assert returned_headers.get("Baz") == "quux"
- def test_proxy_pooling(self):
+ def test_proxy_pooling(self) -> None:
with proxy_from_url(self.proxy_url, cert_reqs="NONE") as http:
for x in range(2):
http.urlopen("GET", self.http_url)
@@ -420,7 +426,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
http.urlopen("GET", self.https_url_alt)
assert len(http.pools) == 3
- def test_proxy_pooling_ext(self):
+ def test_proxy_pooling_ext(self) -> None:
with proxy_from_url(self.proxy_url) as http:
hc1 = http.connection_from_url(self.http_url)
hc2 = http.connection_from_host(self.http_host, self.http_port)
@@ -444,7 +450,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
@pytest.mark.timeout(0.5)
@requires_network()
- def test_https_proxy_timeout(self):
+ def test_https_proxy_timeout(self) -> None:
with proxy_from_url(f"https://{TARPIT_HOST}") as https:
with pytest.raises(MaxRetryError) as e:
https.request("GET", self.http_url, timeout=SHORT_TIMEOUT)
@@ -452,13 +458,13 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
@pytest.mark.timeout(0.5)
@requires_network()
- def test_https_proxy_pool_timeout(self):
+ def test_https_proxy_pool_timeout(self) -> None:
with proxy_from_url(f"https://{TARPIT_HOST}", timeout=SHORT_TIMEOUT) as https:
with pytest.raises(MaxRetryError) as e:
https.request("GET", self.http_url)
assert type(e.value.reason) == ConnectTimeoutError
- def test_scheme_host_case_insensitive(self):
+ def test_scheme_host_case_insensitive(self) -> None:
"""Assert that upper-case schemes and hosts are normalized."""
with proxy_from_url(self.proxy_url.upper(), ca_certs=DEFAULT_CA) as http:
r = http.request("GET", f"{self.http_url.upper()}/")
@@ -484,7 +490,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
),
],
)
- def test_invalid_schema(self, url, error_msg):
+ def test_invalid_schema(self, url: str, error_msg: str) -> None:
with pytest.raises(ProxySchemeUnknown, match=error_msg):
proxy_from_url(url)
@@ -492,7 +498,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase):
@pytest.mark.skipif(not HAS_IPV6, reason="Only runs on IPv6 systems")
class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase):
@classmethod
- def setup_class(cls):
+ def setup_class(cls) -> None:
HTTPDummyProxyTestCase.setup_class()
cls.http_url = f"http://{cls.http_host}:{int(cls.http_port)}"
cls.http_url_alt = f"http://{cls.http_host_alt}:{int(cls.http_port)}"
@@ -500,7 +506,7 @@ class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase):
cls.https_url_alt = f"https://{cls.https_host_alt}:{int(cls.https_port)}"
cls.proxy_url = f"http://[{cls.proxy_host}]:{int(cls.proxy_port)}"
- def test_basic_ipv6_proxy(self):
+ def test_basic_ipv6_proxy(self) -> None:
with proxy_from_url(self.proxy_url, ca_certs=DEFAULT_CA) as http:
r = http.request("GET", f"{self.http_url}/")
assert r.status == 200

Add type hints to test.with_dummyserver.test_proxy_poolmanager