本文整理汇总了Python中urllib3.poolmanager.PoolManager类的典型用法代码示例。如果您正苦于以下问题:Python PoolManager类的具体用法?Python PoolManager怎么用?Python PoolManager使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了PoolManager类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_redirect_cross_host_set_removed_headers
def test_redirect_cross_host_set_removed_headers(self):
http = PoolManager()
self.addCleanup(http.clear)
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/headers' % self.base_url_alt},
headers={'X-API-Secret': 'foo',
'Authorization': 'bar'},
retries=Retry(remove_headers_on_redirect=['X-API-Secret']))
self.assertEqual(r.status, 200)
data = json.loads(r.data.decode('utf-8'))
self.assertNotIn('X-API-Secret', data)
self.assertEqual(data['Authorization'], 'bar')
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/headers' % self.base_url_alt},
headers={'x-api-secret': 'foo',
'authorization': 'bar'},
retries=Retry(remove_headers_on_redirect=['X-API-Secret']))
self.assertEqual(r.status, 200)
data = json.loads(r.data.decode('utf-8'))
self.assertNotIn('x-api-secret', data)
self.assertNotIn('X-API-Secret', data)
self.assertEqual(data['Authorization'], 'bar')
开发者ID:jonparrott,项目名称:urllib3,代码行数:30,代码来源:test_poolmanager.py
示例2: test_raise_on_status
def test_raise_on_status(self):
http = PoolManager()
self.addCleanup(http.clear)
try:
# the default is to raise
r = http.request('GET', '%s/status' % self.base_url,
fields={'status': '500 Internal Server Error'},
retries=Retry(total=1, status_forcelist=range(500, 600)))
self.fail("Failed to raise MaxRetryError exception, returned %r" % r.status)
except MaxRetryError:
pass
try:
# raise explicitly
r = http.request('GET', '%s/status' % self.base_url,
fields={'status': '500 Internal Server Error'},
retries=Retry(total=1,
status_forcelist=range(500, 600),
raise_on_status=True))
self.fail("Failed to raise MaxRetryError exception, returned %r" % r.status)
except MaxRetryError:
pass
# don't raise
r = http.request('GET', '%s/status' % self.base_url,
fields={'status': '500 Internal Server Error'},
retries=Retry(total=1,
status_forcelist=range(500, 600),
raise_on_status=False))
self.assertEqual(r.status, 500)
开发者ID:jonparrott,项目名称:urllib3,代码行数:32,代码来源:test_poolmanager.py
示例3: test_request_survives_missing_port_number
def test_request_survives_missing_port_number(self):
# Can a URL that lacks an explicit port like ':80' succeed, or
# will all such URLs fail with an error?
def server(listener):
sock = listener.accept()[0]
read_request(sock)
sock.send('HTTP/1.1 200 OK\r\n'
'Content-Type: text/plain\r\n'
'Content-Length: 8\r\n'
'\r\n'
'Inspire.')
sock.close()
# 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.
p = PoolManager()
host, port = start_server(server)
port_by_scheme['http'] = port
try:
response = p.request('GET', 'http://%s/' % host, retries=0)
finally:
port_by_scheme['http'] = 80
self.assertEqual(response.status, 200)
self.assertEqual(response.data, 'Inspire.')
开发者ID:blep,项目名称:urllib3,代码行数:27,代码来源:test_poolmanager.py
示例4: test_merge_pool_kwargs_none
def test_merge_pool_kwargs_none(self):
"""Assert false-y values to _merge_pool_kwargs result in defaults"""
p = PoolManager(strict=True)
merged = p._merge_pool_kwargs({})
assert p.connection_pool_kw == merged
merged = p._merge_pool_kwargs(None)
assert p.connection_pool_kw == merged
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:7,代码来源:test_poolmanager.py
示例5: test_pools_keyed_with_from_host
def test_pools_keyed_with_from_host(self):
"""Assert pools are still keyed correctly with connection_from_host."""
ssl_kw = {
'key_file': '/root/totally_legit.key',
'cert_file': '/root/totally_legit.crt',
'cert_reqs': 'CERT_REQUIRED',
'ca_certs': '/root/path_to_pem',
'ssl_version': 'SSLv23_METHOD',
}
p = PoolManager(5, **ssl_kw)
conns = []
conns.append(
p.connection_from_host('example.com', 443, scheme='https')
)
for k in ssl_kw:
p.connection_pool_kw[k] = 'newval'
conns.append(
p.connection_from_host('example.com', 443, scheme='https')
)
assert all(
x is not y
for i, x in enumerate(conns)
for j, y in enumerate(conns)
if i != j
)
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:27,代码来源:test_poolmanager.py
示例6: test_raise_on_redirect
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)
开发者ID:ii0,项目名称:urllib3,代码行数:8,代码来源:test_poolmanager.py
示例7: test_assert_hostname_and_fingerprint_flag
def test_assert_hostname_and_fingerprint_flag(self):
"""Assert that pool manager can accept hostname and fingerprint flags."""
fingerprint = '92:81:FE:85:F7:0C:26:60:EC:D6:B3:BF:93:CF:F9:71:CC:07:7D:0A'
p = PoolManager(assert_hostname=True, assert_fingerprint=fingerprint)
pool = p.connection_from_url('https://example.com/')
assert 1 == len(p.pools)
assert pool.assert_hostname
assert fingerprint == pool.assert_fingerprint
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:8,代码来源:test_poolmanager.py
示例8: test_http_connection_from_host_case_insensitive
def test_http_connection_from_host_case_insensitive(self):
"""Assert scheme case is ignored when getting the https key class."""
p = PoolManager()
pool = p.connection_from_host('example.com', scheme='http')
other_pool = p.connection_from_host('EXAMPLE.COM', scheme='HTTP')
assert 1 == len(p.pools)
assert pool is other_pool
assert all(isinstance(key, PoolKey) for key in p.pools.keys())
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:9,代码来源:test_poolmanager.py
示例9: test_http_connection_from_url_case_insensitive
def test_http_connection_from_url_case_insensitive(self):
"""Assert scheme case is ignored when pooling HTTP connections."""
p = PoolManager()
pool = p.connection_from_url('http://example.com/')
other_pool = p.connection_from_url('HTTP://EXAMPLE.COM/')
assert 1 == len(p.pools)
assert pool is other_pool
assert all(isinstance(key, PoolKey) for key in p.pools.keys())
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:9,代码来源:test_poolmanager.py
示例10: test_assert_hostname_and_fingerprint_flag
def test_assert_hostname_and_fingerprint_flag(self):
"""Assert that pool manager can accept hostname and fingerprint flags."""
fingerprint = '92:81:FE:85:F7:0C:26:60:EC:D6:B3:BF:93:CF:F9:71:CC:07:7D:0A'
p = PoolManager(assert_hostname=True, assert_fingerprint=fingerprint)
self.addCleanup(p.clear)
pool = p.connection_from_url('https://example.com/')
self.assertEqual(1, len(p.pools))
self.assertTrue(pool.assert_hostname)
self.assertEqual(fingerprint, pool.assert_fingerprint)
开发者ID:cloudera,项目名称:hue,代码行数:9,代码来源:test_poolmanager.py
示例11: test_http_connection_from_host_case_insensitive
def test_http_connection_from_host_case_insensitive(self):
"""Assert scheme case is ignored when getting the https key class."""
p = PoolManager()
self.addCleanup(p.clear)
pool = p.connection_from_host('example.com', scheme='http')
other_pool = p.connection_from_host('EXAMPLE.COM', scheme='HTTP')
self.assertEqual(1, len(p.pools))
self.assertTrue(pool is other_pool)
self.assertTrue(all(isinstance(key, PoolKey) for key in p.pools.keys()))
开发者ID:cloudera,项目名称:hue,代码行数:10,代码来源:test_poolmanager.py
示例12: test_https_connection_from_url_case_insensitive
def test_https_connection_from_url_case_insensitive(self):
"""Assert scheme case is ignored when pooling HTTPS connections."""
p = PoolManager()
self.addCleanup(p.clear)
pool = p.connection_from_url('https://example.com/')
other_pool = p.connection_from_url('HTTPS://EXAMPLE.COM/')
self.assertEqual(1, len(p.pools))
self.assertTrue(pool is other_pool)
self.assertTrue(all(isinstance(key, PoolKey) for key in p.pools.keys()))
开发者ID:cloudera,项目名称:hue,代码行数:10,代码来源:test_poolmanager.py
示例13: test_raise_on_redirect
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)
开发者ID:balagopalraj,项目名称:clearlinux,代码行数:11,代码来源:test_poolmanager.py
示例14: test_http_connection_from_context_case_insensitive
def test_http_connection_from_context_case_insensitive(self):
"""Assert scheme case is ignored when getting the https key class."""
p = PoolManager()
context = {'scheme': 'http', 'host': 'example.com', 'port': '8080'}
other_context = {'scheme': 'HTTP', 'host': 'EXAMPLE.COM', 'port': '8080'}
pool = p.connection_from_context(context)
other_pool = p.connection_from_context(other_context)
assert 1 == len(p.pools)
assert pool is other_pool
assert all(isinstance(key, PoolKey) for key in p.pools.keys())
开发者ID:NickMinnellaCS96,项目名称:urllib3,代码行数:11,代码来源:test_poolmanager.py
示例15: test_redirect
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!")
开发者ID:stanvit,项目名称:urllib3,代码行数:11,代码来源:test_poolmanager.py
示例16: test_https_connection_from_context_case_insensitive
def test_https_connection_from_context_case_insensitive(self):
"""Assert scheme case is ignored when getting the https key class."""
p = PoolManager()
self.addCleanup(p.clear)
context = {'scheme': 'https', 'host': 'example.com', 'port': '443'}
other_context = {'scheme': 'HTTPS', 'host': 'EXAMPLE.COM', 'port': '443'}
pool = p.connection_from_context(context)
other_pool = p.connection_from_context(other_context)
self.assertEqual(1, len(p.pools))
self.assertTrue(pool is other_pool)
self.assertTrue(all(isinstance(key, PoolKey) for key in p.pools.keys()))
开发者ID:cloudera,项目名称:hue,代码行数:12,代码来源:test_poolmanager.py
示例17: test_redirect_cross_host_remove_headers
def test_redirect_cross_host_remove_headers(self):
http = PoolManager()
self.addCleanup(http.clear)
r = http.request('GET', '%s/redirect' % self.base_url,
fields={'target': '%s/headers' % self.base_url_alt},
headers={'Authorization': 'foo'})
self.assertEqual(r.status, 200)
data = json.loads(r.data.decode('utf-8'))
self.assertNotIn('Authorization', data)
开发者ID:christophermca,项目名称:dotfiles,代码行数:13,代码来源:test_poolmanager.py
示例18: test_redirect_twice
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!')
开发者ID:echaozh,项目名称:urllib3,代码行数:14,代码来源:test_poolmanager.py
示例19: test_redirect_to_relative_url
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!')
开发者ID:ii0,项目名称:urllib3,代码行数:14,代码来源:test_poolmanager.py
示例20: test_same_url
def test_same_url(self):
# Convince ourselves that normally we don't get the same object
conn1 = connection_from_url('http://localhost:8081/foo')
conn2 = connection_from_url('http://localhost:8081/bar')
self.assertNotEqual(conn1, conn2)
# Now try again using the PoolManager
p = PoolManager(1)
conn1 = p.connection_from_url('http://localhost:8081/foo')
conn2 = p.connection_from_url('http://localhost:8081/bar')
self.assertEqual(conn1, conn2)
开发者ID:poupas,项目名称:urllib3,代码行数:14,代码来源:test_poolmanager.py
注:本文中的urllib3.poolmanager.PoolManager类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论