From 0c183b9d52b45bac22a2ff9db0e6348b655f4ab2 Mon Sep 17 00:00:00 2001 From: SVN-Git Migration Date: Thu, 8 Oct 2015 13:19:30 -0700 Subject: Imported Upstream version 1.2.2 --- test/__init__.py | 0 test/benchmark.py | 77 +++++++++++++++++++++++++ test/test_collections.py | 111 ++++++++++++++++++++++++++++++++++++ test/test_connectionpool.py | 136 ++++++++++++++++++++++++++++++++++++++++++++ test/test_poolmanager.py | 47 +++++++++++++++ test/test_response.py | 68 ++++++++++++++++++++++ 6 files changed, 439 insertions(+) create mode 100644 test/__init__.py create mode 100644 test/benchmark.py create mode 100644 test/test_collections.py create mode 100644 test/test_connectionpool.py create mode 100644 test/test_poolmanager.py create mode 100644 test/test_response.py (limited to 'test') diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/benchmark.py b/test/benchmark.py new file mode 100644 index 0000000..e7049c4 --- /dev/null +++ b/test/benchmark.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python + +""" +Really simple rudimentary benchmark to compare ConnectionPool versus standard +urllib to demonstrate the usefulness of connection re-using. +""" +from __future__ import print_function + +import sys +import time +import urllib + +sys.path.append('../') +import urllib3 + + +# URLs to download. Doesn't matter as long as they're from the same host, so we +# can take advantage of connection re-using. +TO_DOWNLOAD = [ + 'http://code.google.com/apis/apps/', + 'http://code.google.com/apis/base/', + 'http://code.google.com/apis/blogger/', + 'http://code.google.com/apis/calendar/', + 'http://code.google.com/apis/codesearch/', + 'http://code.google.com/apis/contact/', + 'http://code.google.com/apis/books/', + 'http://code.google.com/apis/documents/', + 'http://code.google.com/apis/finance/', + 'http://code.google.com/apis/health/', + 'http://code.google.com/apis/notebook/', + 'http://code.google.com/apis/picasaweb/', + 'http://code.google.com/apis/spreadsheets/', + 'http://code.google.com/apis/webmastertools/', + 'http://code.google.com/apis/youtube/', +] + + +def urllib_get(url_list): + assert url_list + for url in url_list: + now = time.time() + r = urllib.urlopen(url) + elapsed = time.time() - now + print("Got in %0.3f: %s" % (elapsed, url)) + + +def pool_get(url_list): + assert url_list + pool = urllib3.connection_from_url(url_list[0]) + for url in url_list: + now = time.time() + r = pool.get_url(url) + elapsed = time.time() - now + print("Got in %0.3fs: %s" % (elapsed, url)) + + +if __name__ == '__main__': + print("Running pool_get ...") + now = time.time() + pool_get(TO_DOWNLOAD) + pool_elapsed = time.time() - now + + print("Running urllib_get ...") + now = time.time() + urllib_get(TO_DOWNLOAD) + urllib_elapsed = time.time() - now + + print("Completed pool_get in %0.3fs" % pool_elapsed) + print("Completed urllib_get in %0.3fs" % urllib_elapsed) + + +""" +Example results: + +Completed pool_get in 1.163s +Completed urllib_get in 2.318s +""" diff --git a/test/test_collections.py b/test/test_collections.py new file mode 100644 index 0000000..f8275e0 --- /dev/null +++ b/test/test_collections.py @@ -0,0 +1,111 @@ +import unittest + +from urllib3._collections import RecentlyUsedContainer as Container +from urllib3.packages import six +xrange = six.moves.xrange + +class TestLRUContainer(unittest.TestCase): + def test_maxsize(self): + d = Container(5) + + for i in xrange(5): + d[i] = str(i) + + self.assertEqual(len(d), 5) + + for i in xrange(5): + self.assertEqual(d[i], str(i)) + + d[i+1] = str(i+1) + + self.assertEqual(len(d), 5) + self.assertFalse(0 in d) + self.assertTrue(i+1 in d) + + def test_expire(self): + d = Container(5) + + for i in xrange(5): + d[i] = str(i) + + for i in xrange(5): + d.get(0) + + # Add one more entry + d[5] = '5' + + # Check state + self.assertEqual(list(d.keys()), [0, 2, 3, 4, 5]) + + def test_pruning(self): + d = Container(5) + + for i in xrange(5): + d[i] = str(i) + + # Contend 2 entries for the most-used slot to balloon the heap + for i in xrange(100): + d.get(i % 2) + + self.assertTrue(len(d.access_log) <= d.CLEANUP_FACTOR * d._maxsize) + + def test_same_key(self): + d = Container(5) + + for i in xrange(10): + d['foo'] = i + + self.assertEqual(list(d.keys()), ['foo']) + + d._prune_invalidated_entries() + + self.assertEqual(len(d.access_log), 1) + + def test_access_ordering(self): + d = Container(5) + + for i in xrange(10): + d[i] = True + + self.assertEqual(d._get_ordered_access_keys(), [9,8,7,6,5]) + + new_order = [7,8,6,9,5] + for k in reversed(new_order): + d[k] + + self.assertEqual(d._get_ordered_access_keys(), new_order) + + def test_delete(self): + d = Container(5) + + for i in xrange(5): + d[i] = True + + del d[0] + self.assertFalse(0 in d) + + d.pop(1) + self.assertFalse(1 in d) + + d.pop(1, None) + + def test_get(self): + d = Container(5) + + for i in xrange(5): + d[i] = True + + r = d.get(4) + self.assertEqual(r, True) + + r = d.get(5) + self.assertEqual(r, None) + + r = d.get(5, 42) + self.assertEqual(r, 42) + + self.assertRaises(KeyError, lambda: d[5]) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_connectionpool.py b/test/test_connectionpool.py new file mode 100644 index 0000000..4281d42 --- /dev/null +++ b/test/test_connectionpool.py @@ -0,0 +1,136 @@ +import unittest + +from urllib3.connectionpool import ( + connection_from_url, + get_host, + HTTPConnectionPool, + make_headers) + +from urllib3.exceptions import EmptyPoolError, LocationParseError + + +class TestConnectionPool(unittest.TestCase): + def test_get_host(self): + url_host_map = { + 'http://google.com/mail': ('http', 'google.com', None), + 'http://google.com/mail/': ('http', 'google.com', None), + 'google.com/mail': ('http', 'google.com', None), + 'http://google.com/': ('http', 'google.com', None), + 'http://google.com': ('http', 'google.com', None), + 'http://www.google.com': ('http', 'www.google.com', None), + 'http://mail.google.com': ('http', 'mail.google.com', None), + 'http://google.com:8000/mail/': ('http', 'google.com', 8000), + 'http://google.com:8000': ('http', 'google.com', 8000), + 'https://google.com': ('https', 'google.com', None), + 'https://google.com:8000': ('https', 'google.com', 8000), + 'http://user:password@127.0.0.1:1234': ('http', '127.0.0.1', 1234), + } + for url, expected_host in url_host_map.items(): + returned_host = get_host(url) + self.assertEquals(returned_host, expected_host) + + def test_same_host(self): + same_host = [ + ('http://google.com/', '/'), + ('http://google.com/', 'http://google.com/'), + ('http://google.com/', 'http://google.com'), + ('http://google.com/', 'http://google.com/abra/cadabra'), + ('http://google.com:42/', 'http://google.com:42/abracadabra'), + ] + + for a, b in same_host: + c = connection_from_url(a) + self.assertTrue(c.is_same_host(b), "%s =? %s" % (a, b)) + + not_same_host = [ + ('https://google.com/', 'http://google.com/'), + ('http://google.com/', 'https://google.com/'), + ('http://yahoo.com/', 'http://google.com/'), + ('http://google.com:42', 'https://google.com/abracadabra'), + ('http://google.com', 'https://google.net/'), + ] + + for a, b in not_same_host: + c = connection_from_url(a) + self.assertFalse(c.is_same_host(b), "%s =? %s" % (a, b)) + + def test_invalid_host(self): + # TODO: Add more tests + invalid_host = [ + 'http://google.com:foo', + ] + + for location in invalid_host: + self.assertRaises(LocationParseError, get_host, location) + + + def test_make_headers(self): + self.assertEqual( + make_headers(accept_encoding=True), + {'accept-encoding': 'gzip,deflate'}) + + self.assertEqual( + make_headers(accept_encoding='foo,bar'), + {'accept-encoding': 'foo,bar'}) + + self.assertEqual( + make_headers(accept_encoding=['foo', 'bar']), + {'accept-encoding': 'foo,bar'}) + + self.assertEqual( + make_headers(accept_encoding=True, user_agent='banana'), + {'accept-encoding': 'gzip,deflate', 'user-agent': 'banana'}) + + self.assertEqual( + make_headers(user_agent='banana'), + {'user-agent': 'banana'}) + + self.assertEqual( + make_headers(keep_alive=True), + {'connection': 'keep-alive'}) + + self.assertEqual( + make_headers(basic_auth='foo:bar'), + {'authorization': 'Basic Zm9vOmJhcg=='}) + + def test_max_connections(self): + pool = HTTPConnectionPool(host='localhost', maxsize=1, block=True) + + pool._get_conn(timeout=0.01) + + try: + pool._get_conn(timeout=0.01) + self.fail("Managed to get a connection without EmptyPoolError") + except EmptyPoolError: + pass + + try: + pool.get_url('/', pool_timeout=0.01) + self.fail("Managed to get a connection without EmptyPoolError") + except EmptyPoolError: + pass + + self.assertEqual(pool.num_connections, 1) + + def test_pool_edgecases(self): + pool = HTTPConnectionPool(host='localhost', maxsize=1, block=False) + + conn1 = pool._get_conn() + conn2 = pool._get_conn() # New because block=False + + pool._put_conn(conn1) + pool._put_conn(conn2) # Should be discarded + + self.assertEqual(conn1, pool._get_conn()) + self.assertNotEqual(conn2, pool._get_conn()) + + self.assertEqual(pool.num_connections, 3) + + def test_exception_str(self): + self.assertEqual( + str(EmptyPoolError(HTTPConnectionPool(host='localhost'), "Test.")), + "HTTPConnectionPool(host='localhost', port=None): Test.") + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_poolmanager.py b/test/test_poolmanager.py new file mode 100644 index 0000000..12722f7 --- /dev/null +++ b/test/test_poolmanager.py @@ -0,0 +1,47 @@ +import unittest + +from urllib3.poolmanager import PoolManager +from urllib3 import connection_from_url + + +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) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/test_response.py b/test/test_response.py new file mode 100644 index 0000000..0ef379c --- /dev/null +++ b/test/test_response.py @@ -0,0 +1,68 @@ +import unittest +import zlib + +from io import BytesIO + +from urllib3.response import HTTPResponse + +class TestLegacyResponse(unittest.TestCase): + def test_getheaders(self): + headers = {'host': 'example.com'} + r = HTTPResponse(headers=headers) + self.assertEqual(r.getheaders(), headers) + + def test_getheader(self): + headers = {'host': 'example.com'} + r = HTTPResponse(headers=headers) + self.assertEqual(r.getheader('host'), 'example.com') + + +class TestResponse(unittest.TestCase): + def test_cache_content(self): + r = HTTPResponse('foo') + self.assertEqual(r.data, 'foo') + self.assertEqual(r._body, 'foo') + + def test_default(self): + r = HTTPResponse() + self.assertEqual(r.data, None) + + def test_none(self): + r = HTTPResponse(None) + self.assertEqual(r.data, None) + + def test_preload(self): + fp = BytesIO(b'foo') + + r = HTTPResponse(fp, preload_content=True) + + self.assertEqual(fp.tell(), len(b'foo')) + self.assertEqual(r.data, b'foo') + + def test_no_preload(self): + fp = BytesIO(b'foo') + + r = HTTPResponse(fp, preload_content=False) + + self.assertEqual(fp.tell(), 0) + self.assertEqual(r.data, b'foo') + self.assertEqual(fp.tell(), len(b'foo')) + + def test_decode_bad_data(self): + fp = BytesIO(b'\x00' * 10) + self.assertRaises(zlib.error, HTTPResponse, fp, headers={ + 'content-encoding': 'deflate' + }) + + def test_decode_deflate(self): + import zlib + data = zlib.compress(b'foo') + + fp = BytesIO(data) + r = HTTPResponse(fp, headers={'content-encoding': 'deflate'}) + + self.assertEqual(r.data, b'foo') + + +if __name__ == '__main__': + unittest.main() -- cgit v1.2.3