| Server IP : 84.32.84.6 / Your IP : 216.73.217.6 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/gsutil/third_party/urllib3/test/ |
Upload File : |
from __future__ import annotations
import pytest
from urllib3.exceptions import MaxRetryError, NewConnectionError, ProxyError
from urllib3.poolmanager import ProxyManager
from urllib3.util.retry import Retry
from urllib3.util.url import parse_url
from .port_helpers import find_unused_port
class TestProxyManager:
@pytest.mark.parametrize("proxy_scheme", ["http", "https"])
def test_proxy_headers(self, proxy_scheme: str) -> None:
url = "http://pypi.org/project/urllib3/"
proxy_url = f"{proxy_scheme}://something:1234"
with ProxyManager(proxy_url) as p:
# Verify default headers
default_headers = {"Accept": "*/*", "Host": "pypi.org"}
headers = p._set_proxy_headers(url)
assert headers == default_headers
# Verify default headers don't overwrite provided headers
provided_headers = {
"Accept": "application/json",
"custom": "header",
"Host": "test.python.org",
}
headers = p._set_proxy_headers(url, provided_headers)
assert headers == provided_headers
# Verify proxy with nonstandard port
provided_headers = {"Accept": "application/json"}
expected_headers = provided_headers.copy()
expected_headers.update({"Host": "pypi.org:8080"})
url_with_port = "http://pypi.org:8080/project/urllib3/"
headers = p._set_proxy_headers(url_with_port, provided_headers)
assert headers == expected_headers
def test_default_port(self) -> None:
with ProxyManager("http://something") as p:
assert p.proxy is not None
assert p.proxy.port == 80
with ProxyManager("https://something") as p:
assert p.proxy is not None
assert p.proxy.port == 443
def test_invalid_scheme(self) -> None:
with pytest.raises(AssertionError):
ProxyManager("invalid://host/p")
with pytest.raises(ValueError):
ProxyManager("invalid://host/p")
def test_proxy_tunnel(self) -> None:
http_url = parse_url("http://example.com")
https_url = parse_url("https://example.com")
with ProxyManager("http://proxy:8080") as p:
assert p._proxy_requires_url_absolute_form(http_url)
assert p._proxy_requires_url_absolute_form(https_url) is False
with ProxyManager("https://proxy:8080") as p:
assert p._proxy_requires_url_absolute_form(http_url)
assert p._proxy_requires_url_absolute_form(https_url) is False
with ProxyManager("https://proxy:8080", use_forwarding_for_https=True) as p:
assert p._proxy_requires_url_absolute_form(http_url)
assert p._proxy_requires_url_absolute_form(https_url)
def test_proxy_connect_retry(self) -> None:
retry = Retry(total=None, connect=False)
port = find_unused_port()
with ProxyManager(f"http://localhost:{port}") as p:
with pytest.raises(ProxyError) as ei:
p.urlopen("HEAD", url="http://localhost/", retries=retry)
assert isinstance(ei.value.original_error, NewConnectionError)
retry = Retry(total=None, connect=2)
with ProxyManager(f"http://localhost:{port}") as p:
with pytest.raises(MaxRetryError) as ei1:
p.urlopen("HEAD", url="http://localhost/", retries=retry)
assert ei1.value.reason is not None
assert isinstance(ei1.value.reason, ProxyError)
assert isinstance(ei1.value.reason.original_error, NewConnectionError)