aboutsummaryrefslogtreecommitdiff
path: root/test/test_connectionpool.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_connectionpool.py')
-rw-r--r--test/test_connectionpool.py29
1 files changed, 25 insertions, 4 deletions
diff --git a/test/test_connectionpool.py b/test/test_connectionpool.py
index a7e104a..ac1768e 100644
--- a/test/test_connectionpool.py
+++ b/test/test_connectionpool.py
@@ -1,6 +1,11 @@
import unittest
-from urllib3.connectionpool import connection_from_url, HTTPConnectionPool
+from urllib3.connectionpool import (
+ connection_from_url,
+ HTTPConnection,
+ HTTPConnectionPool,
+)
+from urllib3.util import Timeout
from urllib3.packages.ssl_match_hostname import CertificateError
from urllib3.exceptions import (
ClosedPoolError,
@@ -8,7 +13,7 @@ from urllib3.exceptions import (
HostChangedError,
MaxRetryError,
SSLError,
- TimeoutError,
+ ReadTimeoutError,
)
from socket import error as SocketError, timeout as SocketTimeout
@@ -52,6 +57,7 @@ class TestConnectionPool(unittest.TestCase):
c = connection_from_url(a)
self.assertFalse(c.is_same_host(b), "%s =? %s" % (a, b))
+
def test_max_connections(self):
pool = HTTPConnectionPool(host='localhost', maxsize=1, block=True)
@@ -108,6 +114,7 @@ class TestConnectionPool(unittest.TestCase):
"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)
@@ -122,8 +129,8 @@ class TestConnectionPool(unittest.TestCase):
self.assertEqual(pool.pool.qsize(), POOL_SIZE)
#make sure that all of the exceptions return the connection to the pool
- _test(Empty, TimeoutError)
- _test(SocketTimeout, TimeoutError)
+ _test(Empty, ReadTimeoutError)
+ _test(SocketTimeout, ReadTimeoutError)
_test(BaseSSLError, SSLError)
_test(CertificateError, SSLError)
@@ -166,6 +173,20 @@ class TestConnectionPool(unittest.TestCase):
self.assertRaises(Empty, old_pool_queue.get, block=False)
+ def test_pool_timeouts(self):
+ pool = HTTPConnectionPool(host='localhost')
+ conn = pool._new_conn()
+ self.assertEqual(conn.__class__, HTTPConnection)
+ self.assertEqual(pool.timeout.__class__, Timeout)
+ self.assertEqual(pool.timeout._read, Timeout.DEFAULT_TIMEOUT)
+ self.assertEqual(pool.timeout._connect, Timeout.DEFAULT_TIMEOUT)
+ self.assertEqual(pool.timeout.total, None)
+
+ pool = HTTPConnectionPool(host='localhost', timeout=3)
+ self.assertEqual(pool.timeout._read, 3)
+ self.assertEqual(pool.timeout._connect, 3)
+ self.assertEqual(pool.timeout.total, None)
+
if __name__ == '__main__':
unittest.main()