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
|
import unittest
import json
from dummyserver.testcase import (HTTPDummyServerTestCase,
IPv6HTTPDummyServerTestCase)
from urllib3.poolmanager import PoolManager
from urllib3.connectionpool import port_by_scheme
from urllib3.exceptions import MaxRetryError, SSLError
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_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_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):
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()
|