aboutsummaryrefslogtreecommitdiff
path: root/test/test_poolmanager.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_poolmanager.py')
-rw-r--r--test/test_poolmanager.py92
1 files changed, 92 insertions, 0 deletions
diff --git a/test/test_poolmanager.py b/test/test_poolmanager.py
new file mode 100644
index 0000000..6195d51
--- /dev/null
+++ b/test/test_poolmanager.py
@@ -0,0 +1,92 @@
+import unittest
+
+from urllib3.poolmanager import PoolManager
+from urllib3 import connection_from_url
+from urllib3.exceptions import (
+ ClosedPoolError,
+ LocationValueError,
+)
+
+
+class TestPoolManager(unittest.TestCase):
+ 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)
+
+ def test_many_urls(self):
+ urls = [
+ "http://localhost:8081/foo",
+ "http://www.google.com/mail",
+ "http://localhost:8081/bar",
+ "https://www.google.com/",
+ "https://www.google.com/mail",
+ "http://yahoo.com",
+ "http://bing.com",
+ "http://yahoo.com/",
+ ]
+
+ connections = set()
+
+ p = PoolManager(10)
+
+ for url in urls:
+ conn = p.connection_from_url(url)
+ connections.add(conn)
+
+ self.assertEqual(len(connections), 5)
+
+ def test_manager_clear(self):
+ p = PoolManager(5)
+
+ conn_pool = p.connection_from_url('http://google.com')
+ self.assertEqual(len(p.pools), 1)
+
+ conn = conn_pool._get_conn()
+
+ p.clear()
+ self.assertEqual(len(p.pools), 0)
+
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
+
+ conn_pool._put_conn(conn)
+
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
+
+ self.assertEqual(len(p.pools), 0)
+
+
+ def test_nohost(self):
+ p = PoolManager(5)
+ self.assertRaises(LocationValueError, p.connection_from_url, 'http://@')
+ self.assertRaises(LocationValueError, p.connection_from_url, None)
+
+ def test_contextmanager(self):
+ with PoolManager(1) as p:
+ conn_pool = p.connection_from_url('http://google.com')
+ self.assertEqual(len(p.pools), 1)
+ conn = conn_pool._get_conn()
+
+ self.assertEqual(len(p.pools), 0)
+
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
+
+ conn_pool._put_conn(conn)
+
+ self.assertRaises(ClosedPoolError, conn_pool._get_conn)
+
+ self.assertEqual(len(p.pools), 0)
+
+
+if __name__ == '__main__':
+ unittest.main()