1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
import unittest
import json
from nose.plugins.skip import SkipTest
from dummyserver.server import HAS_IPV6
from dummyserver.testcase import (HTTPDummyServerTestCase,
IPv6HTTPDummyServerTestCase)
from urllib3.poolmanager import PoolManager
from urllib3.connectionpool import port_by_scheme
from urllib3.exceptions import MaxRetryError, SSLError
from urllib3.util.retry import Retry
class TestPoolManager(HTTPDummyServerTestCase):
def setUp(self):
self.base_url = 'http://%s:%d' % (self.host, self.port)
self.base_url_alt = 'http://%s:%d' % (self.host_alt, self.port)
def test_redirect(self):
http = PoolManager()
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/' % self.base_url},
redirect=False)
self.assertEqual(r.status, 303)
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/' % self.base_url})
self.assertEqual(r.status, 200)
self.assertEqual(r.data, b'Dummy server!')
def test_redirect_twice(self):
http = PoolManager()
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/redirect' % self.base_url},
redirect=False)
self.assertEqual(r.status, 303)
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/redirect?target=%s/' % (self.base_url, self.base_url)})
self.assertEqual(r.status, 200)
self.assertEqual(r.data, b'Dummy server!')
def test_redirect_to_relative_url(self):
http = PoolManager()
r = http.request('GET', '%s/redirect' % self.base_url,
fields = {'target': '/redirect'},
redirect = False)
self.assertEqual(r.status, 303)
r = http.request('GET', '%s/redirect' % self.base_url,
fields = {'target': '/redirect'})
self.assertEqual(r.status, 200)
self.assertEqual(r.data, b'Dummy server!')
def test_cross_host_redirect(self):
http = PoolManager()
cross_host_location = '%s/echo?a=b' % self.base_url_alt
try:
http.request('GET', '%s/redirect' % self.base_url,
fields={'target': cross_host_location},
timeout=0.01, retries=0)
self.fail("Request succeeded instead of raising an exception like it should.")
except MaxRetryError:
pass
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/echo?a=b' % self.base_url_alt},
timeout=0.01, retries=1)
self.assertEqual(r._pool.host, self.host_alt)
def test_too_many_redirects(self):
http = PoolManager()
try:
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/redirect?target=%s/' % (self.base_url, self.base_url)},
retries=1)
self.fail("Failed to raise MaxRetryError exception, returned %r" % r.status)
except MaxRetryError:
pass
try:
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/redirect?target=%s/' % (self.base_url, self.base_url)},
retries=Retry(total=None, redirect=1))
self.fail("Failed to raise MaxRetryError exception, returned %r" % r.status)
except MaxRetryError:
pass
def test_raise_on_redirect(self):
http = PoolManager()
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/redirect?target=%s/' % (self.base_url, self.base_url)},
retries=Retry(total=None, redirect=1, raise_on_redirect=False))
self.assertEqual(r.status, 303)
def test_missing_port(self):
# Can a URL that lacks an explicit port like ':80' succeed, or
# will all such URLs fail with an error?
http = PoolManager()
# By globally adjusting `port_by_scheme` we pretend for a moment
# that HTTP's default port is not 80, but is the port at which
# our test server happens to be listening.
port_by_scheme['http'] = self.port
try:
r = http.request('GET', 'http://%s/' % self.host, retries=0)
finally:
port_by_scheme['http'] = 80
self.assertEqual(r.status, 200)
self.assertEqual(r.data, b'Dummy server!')
def test_headers(self):
http = PoolManager(headers={'Foo': 'bar'})
r = http.request('GET', '%s/headers' % self.base_url)
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), 'bar')
r = http.request('POST', '%s/headers' % self.base_url)
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), 'bar')
r = http.request_encode_url('GET', '%s/headers' % self.base_url)
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), 'bar')
r = http.request_encode_body('POST', '%s/headers' % self.base_url)
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), 'bar')
r = http.request_encode_url('GET', '%s/headers' % self.base_url, headers={'Baz': 'quux'})
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), None)
self.assertEqual(returned_headers.get('Baz'), 'quux')
r = http.request_encode_body('GET', '%s/headers' % self.base_url, headers={'Baz': 'quux'})
returned_headers = json.loads(r.data.decode())
self.assertEqual(returned_headers.get('Foo'), None)
self.assertEqual(returned_headers.get('Baz'), 'quux')
def test_http_with_ssl_keywords(self):
http = PoolManager(ca_certs='REQUIRED')
r = http.request('GET', 'http://%s:%s/' % (self.host, self.port))
self.assertEqual(r.status, 200)
class TestIPv6PoolManager(IPv6HTTPDummyServerTestCase):
if not HAS_IPV6:
raise SkipTest("IPv6 is not supported on this system.")
def setUp(self):
self.base_url = 'http://[%s]:%d' % (self.host, self.port)
def test_ipv6(self):
http = PoolManager()
http.request('GET', self.base_url)
if __name__ == '__main__':
unittest.main()
|