aboutsummaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/test_collections.py4
-rw-r--r--test/test_connectionpool.py45
-rw-r--r--test/test_exceptions.py19
-rw-r--r--test/test_filepost.py42
-rw-r--r--test/test_poolmanager.py6
-rw-r--r--test/test_proxymanager.py27
-rw-r--r--test/test_response.py48
7 files changed, 167 insertions, 24 deletions
diff --git a/test/test_collections.py b/test/test_collections.py
index 098b31a..b44c58a 100644
--- a/test/test_collections.py
+++ b/test/test_collections.py
@@ -119,9 +119,7 @@ class TestLRUContainer(unittest.TestCase):
def test_iter(self):
d = Container()
- with self.assertRaises(NotImplementedError):
- for i in d:
- self.fail("Iteration shouldn't be implemented.")
+ self.assertRaises(NotImplementedError, d.__iter__)
if __name__ == '__main__':
unittest.main()
diff --git a/test/test_connectionpool.py b/test/test_connectionpool.py
index afc3098..a7e104a 100644
--- a/test/test_connectionpool.py
+++ b/test/test_connectionpool.py
@@ -11,7 +11,7 @@ from urllib3.exceptions import (
TimeoutError,
)
-from socket import timeout as SocketTimeout
+from socket import error as SocketError, timeout as SocketTimeout
from ssl import SSLError as BaseSSLError
try: # Python 3
@@ -23,6 +23,10 @@ except ImportError:
class TestConnectionPool(unittest.TestCase):
+ """
+ Tests in this suite should exercise the ConnectionPool functionality
+ without actually making any network requests or connections.
+ """
def test_same_host(self):
same_host = [
('http://google.com/', '/'),
@@ -86,6 +90,24 @@ class TestConnectionPool(unittest.TestCase):
str(EmptyPoolError(HTTPConnectionPool(host='localhost'), "Test.")),
"HTTPConnectionPool(host='localhost', port=None): Test.")
+ def test_retry_exception_str(self):
+ self.assertEqual(
+ str(MaxRetryError(
+ HTTPConnectionPool(host='localhost'), "Test.", None)),
+ "HTTPConnectionPool(host='localhost', port=None): "
+ "Max retries exceeded with url: Test. (Caused by redirect)")
+
+ err = SocketError("Test")
+
+ # using err.__class__ here, as socket.error is an alias for OSError
+ # since Py3.3 and gets printed as this
+ self.assertEqual(
+ str(MaxRetryError(
+ HTTPConnectionPool(host='localhost'), "Test.", err)),
+ "HTTPConnectionPool(host='localhost', port=None): "
+ "Max retries exceeded with url: Test. "
+ "(Caused by {0}: Test)".format(str(err.__class__)))
+
def test_pool_size(self):
POOL_SIZE = 1
pool = HTTPConnectionPool(host='localhost', maxsize=POOL_SIZE, block=True)
@@ -95,8 +117,7 @@ class TestConnectionPool(unittest.TestCase):
def _test(exception, expect):
pool._make_request = lambda *args, **kwargs: _raise(exception)
- with self.assertRaises(expect):
- pool.request('GET', '/')
+ self.assertRaises(expect, pool.request, 'GET', '/')
self.assertEqual(pool.pool.qsize(), POOL_SIZE)
@@ -111,15 +132,15 @@ class TestConnectionPool(unittest.TestCase):
# MaxRetryError, not EmptyPoolError
# See: https://github.com/shazow/urllib3/issues/76
pool._make_request = lambda *args, **kwargs: _raise(HTTPException)
- with self.assertRaises(MaxRetryError):
- pool.request('GET', '/', retries=1, pool_timeout=0.01)
+ self.assertRaises(MaxRetryError, pool.request,
+ 'GET', '/', retries=1, pool_timeout=0.01)
self.assertEqual(pool.pool.qsize(), POOL_SIZE)
def test_assert_same_host(self):
c = connection_from_url('http://google.com:80')
- with self.assertRaises(HostChangedError):
- c.request('GET', 'http://yahoo.com:80', assert_same_host=True)
+ self.assertRaises(HostChangedError, c.request,
+ 'GET', 'http://yahoo.com:80', assert_same_host=True)
def test_pool_close(self):
pool = connection_from_url('http://google.com:80')
@@ -136,16 +157,14 @@ class TestConnectionPool(unittest.TestCase):
pool.close()
self.assertEqual(pool.pool, None)
- with self.assertRaises(ClosedPoolError):
- pool._get_conn()
+ self.assertRaises(ClosedPoolError, pool._get_conn)
pool._put_conn(conn3)
- with self.assertRaises(ClosedPoolError):
- pool._get_conn()
+ self.assertRaises(ClosedPoolError, pool._get_conn)
+
+ self.assertRaises(Empty, old_pool_queue.get, block=False)
- with self.assertRaises(Empty):
- old_pool_queue.get(block=False)
if __name__ == '__main__':
diff --git a/test/test_exceptions.py b/test/test_exceptions.py
new file mode 100644
index 0000000..3e02ca6
--- /dev/null
+++ b/test/test_exceptions.py
@@ -0,0 +1,19 @@
+import unittest
+import pickle
+
+from urllib3.exceptions import HTTPError, MaxRetryError, LocationParseError
+from urllib3.connectionpool import HTTPConnectionPool
+
+
+
+class TestPickle(unittest.TestCase):
+
+ def test_exceptions(self):
+ assert pickle.dumps(HTTPError(None))
+ assert pickle.dumps(MaxRetryError(None, None, None))
+ assert pickle.dumps(LocationParseError(None))
+
+ def test_exceptions_with_objects(self):
+ assert pickle.dumps(HTTPError('foo'))
+ assert pickle.dumps(MaxRetryError(HTTPConnectionPool('localhost'), '/', None))
+ assert pickle.dumps(LocationParseError('fake location'))
diff --git a/test/test_filepost.py b/test/test_filepost.py
index c251778..70ab100 100644
--- a/test/test_filepost.py
+++ b/test/test_filepost.py
@@ -52,19 +52,17 @@ class TestMultipartEncoding(unittest.TestCase):
self.assertEqual(encoded,
b'--' + b(BOUNDARY) + b'\r\n'
b'Content-Disposition: form-data; name="k"\r\n'
- b'Content-Type: text/plain\r\n'
b'\r\n'
b'v\r\n'
b'--' + b(BOUNDARY) + b'\r\n'
b'Content-Disposition: form-data; name="k2"\r\n'
- b'Content-Type: text/plain\r\n'
b'\r\n'
b'v2\r\n'
b'--' + b(BOUNDARY) + b'--\r\n'
, fields)
self.assertEqual(content_type,
- b'multipart/form-data; boundary=' + b(BOUNDARY))
+ 'multipart/form-data; boundary=' + str(BOUNDARY))
def test_filename(self):
@@ -82,4 +80,40 @@ class TestMultipartEncoding(unittest.TestCase):
)
self.assertEqual(content_type,
- b'multipart/form-data; boundary=' + b(BOUNDARY))
+ 'multipart/form-data; boundary=' + str(BOUNDARY))
+
+
+ def test_textplain(self):
+ fields = [('k', ('somefile.txt', b'v'))]
+
+ encoded, content_type = encode_multipart_formdata(fields, boundary=BOUNDARY)
+
+ self.assertEqual(encoded,
+ b'--' + b(BOUNDARY) + b'\r\n'
+ b'Content-Disposition: form-data; name="k"; filename="somefile.txt"\r\n'
+ b'Content-Type: text/plain\r\n'
+ b'\r\n'
+ b'v\r\n'
+ b'--' + b(BOUNDARY) + b'--\r\n'
+ )
+
+ self.assertEqual(content_type,
+ 'multipart/form-data; boundary=' + str(BOUNDARY))
+
+
+ def test_explicit(self):
+ fields = [('k', ('somefile.txt', b'v', 'image/jpeg'))]
+
+ encoded, content_type = encode_multipart_formdata(fields, boundary=BOUNDARY)
+
+ self.assertEqual(encoded,
+ b'--' + b(BOUNDARY) + b'\r\n'
+ b'Content-Disposition: form-data; name="k"; filename="somefile.txt"\r\n'
+ b'Content-Type: image/jpeg\r\n'
+ b'\r\n'
+ b'v\r\n'
+ b'--' + b(BOUNDARY) + b'--\r\n'
+ )
+
+ self.assertEqual(content_type,
+ 'multipart/form-data; boundary=' + str(BOUNDARY))
diff --git a/test/test_poolmanager.py b/test/test_poolmanager.py
index 273abf9..2faab94 100644
--- a/test/test_poolmanager.py
+++ b/test/test_poolmanager.py
@@ -54,13 +54,11 @@ class TestPoolManager(unittest.TestCase):
p.clear()
self.assertEqual(len(p.pools), 0)
- with self.assertRaises(ClosedPoolError):
- conn_pool._get_conn()
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
conn_pool._put_conn(conn)
- with self.assertRaises(ClosedPoolError):
- conn_pool._get_conn()
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
self.assertEqual(len(p.pools), 0)
diff --git a/test/test_proxymanager.py b/test/test_proxymanager.py
new file mode 100644
index 0000000..64c86e8
--- /dev/null
+++ b/test/test_proxymanager.py
@@ -0,0 +1,27 @@
+import unittest
+
+from urllib3.poolmanager import ProxyManager
+
+
+class TestProxyManager(unittest.TestCase):
+ def test_proxy_headers(self):
+ p = ProxyManager(None)
+ url = 'http://pypi.python.org/test'
+
+ # Verify default headers
+ default_headers = {'Accept': '*/*',
+ 'Host': 'pypi.python.org'}
+ headers = p._set_proxy_headers(url)
+
+ self.assertEqual(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)
+
+ self.assertEqual(headers, provided_headers)
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/test_response.py b/test/test_response.py
index 964f677..199e379 100644
--- a/test/test_response.py
+++ b/test/test_response.py
@@ -63,6 +63,54 @@ class TestResponse(unittest.TestCase):
self.assertEqual(r.data, b'foo')
+ def test_decode_deflate_case_insensitve(self):
+ import zlib
+ data = zlib.compress(b'foo')
+
+ fp = BytesIO(data)
+ r = HTTPResponse(fp, headers={'content-encoding': 'DeFlAtE'})
+
+ self.assertEqual(r.data, b'foo')
+
+ def test_chunked_decoding_deflate(self):
+ import zlib
+ data = zlib.compress(b'foo')
+
+ fp = BytesIO(data)
+ r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
+ preload_content=False)
+
+ self.assertEqual(r.read(3), b'')
+ self.assertEqual(r.read(1), b'f')
+ self.assertEqual(r.read(2), b'oo')
+
+ def test_chunked_decoding_deflate2(self):
+ import zlib
+ compress = zlib.compressobj(6, zlib.DEFLATED, -zlib.MAX_WBITS)
+ data = compress.compress(b'foo')
+ data += compress.flush()
+
+ fp = BytesIO(data)
+ r = HTTPResponse(fp, headers={'content-encoding': 'deflate'},
+ preload_content=False)
+
+ self.assertEqual(r.read(1), b'')
+ self.assertEqual(r.read(1), b'f')
+ self.assertEqual(r.read(2), b'oo')
+
+ def test_chunked_decoding_gzip(self):
+ import zlib
+ compress = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS)
+ data = compress.compress(b'foo')
+ data += compress.flush()
+
+ fp = BytesIO(data)
+ r = HTTPResponse(fp, headers={'content-encoding': 'gzip'},
+ preload_content=False)
+
+ self.assertEqual(r.read(11), b'')
+ self.assertEqual(r.read(1), b'f')
+ self.assertEqual(r.read(2), b'oo')
if __name__ == '__main__':
unittest.main()
generated by cgit v1.2.3 (git 2.45.0) at 2024-11-23 10:22:32 +0000