diff options
Diffstat (limited to 'test/with_dummyserver/test_proxy_poolmanager.py')
| -rw-r--r-- | test/with_dummyserver/test_proxy_poolmanager.py | 42 |
1 files changed, 40 insertions, 2 deletions
diff --git a/test/with_dummyserver/test_proxy_poolmanager.py b/test/with_dummyserver/test_proxy_poolmanager.py index df300fe..c593f2d 100644 --- a/test/with_dummyserver/test_proxy_poolmanager.py +++ b/test/with_dummyserver/test_proxy_poolmanager.py @@ -4,11 +4,12 @@ import unittest from nose.tools import timed -from dummyserver.testcase import HTTPDummyProxyTestCase +from dummyserver.testcase import HTTPDummyProxyTestCase, IPv6HTTPDummyProxyTestCase from dummyserver.server import ( DEFAULT_CA, DEFAULT_CA_BAD, get_unreachable_address) from .. import TARPIT_HOST +from urllib3._collections import HTTPHeaderDict from urllib3.poolmanager import proxy_from_url, ProxyManager from urllib3.exceptions import ( MaxRetryError, SSLError, ProxyError, ConnectTimeoutError) @@ -48,7 +49,7 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase): def test_proxy_conn_fail(self): host, port = get_unreachable_address() - http = proxy_from_url('http://%s:%s/' % (host, port), retries=1) + http = proxy_from_url('http://%s:%s/' % (host, port), retries=1, timeout=0.05) self.assertRaises(MaxRetryError, http.request, 'GET', '%s/' % self.https_url) self.assertRaises(MaxRetryError, http.request, 'GET', @@ -223,6 +224,22 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase): self.assertEqual(returned_headers.get('Host'), '%s:%s'%(self.https_host,self.https_port)) + def test_headerdict(self): + default_headers = HTTPHeaderDict(a='b') + proxy_headers = HTTPHeaderDict() + proxy_headers.add('foo', 'bar') + + http = proxy_from_url( + self.proxy_url, + headers=default_headers, + proxy_headers=proxy_headers) + + request_headers = HTTPHeaderDict(baz='quux') + r = http.request('GET', '%s/headers' % self.http_url, headers=request_headers) + returned_headers = json.loads(r.data.decode()) + self.assertEqual(returned_headers.get('Foo'), 'bar') + self.assertEqual(returned_headers.get('Baz'), 'quux') + def test_proxy_pooling(self): http = proxy_from_url(self.proxy_url) @@ -283,5 +300,26 @@ class TestHTTPProxyManager(HTTPDummyProxyTestCase): except MaxRetryError as e: assert isinstance(e.reason, ConnectTimeoutError) + +class TestIPv6HTTPProxyManager(IPv6HTTPDummyProxyTestCase): + + def setUp(self): + self.http_url = 'http://%s:%d' % (self.http_host, self.http_port) + self.http_url_alt = 'http://%s:%d' % (self.http_host_alt, + self.http_port) + self.https_url = 'https://%s:%d' % (self.https_host, self.https_port) + self.https_url_alt = 'https://%s:%d' % (self.https_host_alt, + self.https_port) + self.proxy_url = 'http://[%s]:%d' % (self.proxy_host, self.proxy_port) + + def test_basic_ipv6_proxy(self): + http = proxy_from_url(self.proxy_url) + + r = http.request('GET', '%s/' % self.http_url) + self.assertEqual(r.status, 200) + + r = http.request('GET', '%s/' % self.https_url) + self.assertEqual(r.status, 200) + if __name__ == '__main__': unittest.main() |