← Back to branch summary

~matsubara/ubuntu/trusty/python-urllib3/bug-1412545

« back to all changes in this revision

Viewing changes to test/test_connectionpool.py

  • Committer: Package Import Robot
  • Author(s): Barry Warsaw
  • Date: 2012-11-01 15:31:36 UTC
  • mfrom: (1.1.2)
  • Revision ID: package-import@ubuntu.com-20121101153136-w2fm02askveqxxkx
Tags: 1.5-0ubuntu1
* New upstream release.  Remaining changes:
  - 01_do-not-use-embedded-python-six.patch: refreshed
  - 02_require-cert-verification.patch: refreshed
  - 03-fix-appenginge.patch: removed

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/test_connectionpool.py
1
1
import unittest
2
2
 
3
3
from urllib3.connectionpool import connection_from_url, HTTPConnectionPool
4
 
from urllib3.util import get_host, make_headers
5
 
from urllib3.exceptions import EmptyPoolError, LocationParseError
 
4
from urllib3.packages.ssl_match_hostname import CertificateError
 
5
from urllib3.exceptions import (
 
6
    ClosedPoolError,
 
7
    EmptyPoolError,
 
8
    HostChangedError,
 
9
    MaxRetryError,
 
10
    SSLError,
 
11
    TimeoutError,
 
12
)
 
13
 
 
14
from socket import timeout as SocketTimeout
 
15
from ssl import SSLError as BaseSSLError
 
16
 
 
17
try:   # Python 3
 
18
    from queue import Empty
 
19
    from http.client import HTTPException
 
20
except ImportError:
 
21
    from Queue import Empty
 
22
    from httplib import HTTPException
6
23
 
7
24
 
8
25
class TestConnectionPool(unittest.TestCase):
9
 
    def test_get_host(self):
10
 
        url_host_map = {
11
 
            'http://google.com/mail': ('http', 'google.com', None),
12
 
            'http://google.com/mail/': ('http', 'google.com', None),
13
 
            'google.com/mail': ('http', 'google.com', None),
14
 
            'http://google.com/': ('http', 'google.com', None),
15
 
            'http://google.com': ('http', 'google.com', None),
16
 
            'http://www.google.com': ('http', 'www.google.com', None),
17
 
            'http://mail.google.com': ('http', 'mail.google.com', None),
18
 
            'http://google.com:8000/mail/': ('http', 'google.com', 8000),
19
 
            'http://google.com:8000': ('http', 'google.com', 8000),
20
 
            'https://google.com': ('https', 'google.com', None),
21
 
            'https://google.com:8000': ('https', 'google.com', 8000),
22
 
            'http://user:password@127.0.0.1:1234': ('http', '127.0.0.1', 1234),
23
 
        }
24
 
        for url, expected_host in url_host_map.items():
25
 
            returned_host = get_host(url)
26
 
            self.assertEquals(returned_host, expected_host)
27
 
 
28
26
    def test_same_host(self):
29
27
        same_host = [
30
28
            ('http://google.com/', '/'),
50
48
            c = connection_from_url(a)
51
49
            self.assertFalse(c.is_same_host(b), "%s =? %s" % (a, b))
52
50
 
53
 
    def test_invalid_host(self):
54
 
        # TODO: Add more tests
55
 
        invalid_host = [
56
 
            'http://google.com:foo',
57
 
        ]
58
 
 
59
 
        for location in invalid_host:
60
 
            self.assertRaises(LocationParseError, get_host, location)
61
 
 
62
 
 
63
 
    def test_make_headers(self):
64
 
        self.assertEqual(
65
 
            make_headers(accept_encoding=True),
66
 
            {'accept-encoding': 'gzip,deflate'})
67
 
 
68
 
        self.assertEqual(
69
 
            make_headers(accept_encoding='foo,bar'),
70
 
            {'accept-encoding': 'foo,bar'})
71
 
 
72
 
        self.assertEqual(
73
 
            make_headers(accept_encoding=['foo', 'bar']),
74
 
            {'accept-encoding': 'foo,bar'})
75
 
 
76
 
        self.assertEqual(
77
 
            make_headers(accept_encoding=True, user_agent='banana'),
78
 
            {'accept-encoding': 'gzip,deflate', 'user-agent': 'banana'})
79
 
 
80
 
        self.assertEqual(
81
 
            make_headers(user_agent='banana'),
82
 
            {'user-agent': 'banana'})
83
 
 
84
 
        self.assertEqual(
85
 
            make_headers(keep_alive=True),
86
 
            {'connection': 'keep-alive'})
87
 
 
88
 
        self.assertEqual(
89
 
            make_headers(basic_auth='foo:bar'),
90
 
            {'authorization': 'Basic Zm9vOmJhcg=='})
91
 
 
92
51
    def test_max_connections(self):
93
52
        pool = HTTPConnectionPool(host='localhost', maxsize=1, block=True)
94
53
 
127
86
            str(EmptyPoolError(HTTPConnectionPool(host='localhost'), "Test.")),
128
87
            "HTTPConnectionPool(host='localhost', port=None): Test.")
129
88
 
 
89
    def test_pool_size(self):
 
90
        POOL_SIZE = 1
 
91
        pool = HTTPConnectionPool(host='localhost', maxsize=POOL_SIZE, block=True)
 
92
 
 
93
        def _raise(ex):
 
94
            raise ex()
 
95
 
 
96
        def _test(exception, expect):
 
97
            pool._make_request = lambda *args, **kwargs: _raise(exception)
 
98
            with self.assertRaises(expect):
 
99
                pool.request('GET', '/')
 
100
 
 
101
            self.assertEqual(pool.pool.qsize(), POOL_SIZE)
 
102
 
 
103
        #make sure that all of the exceptions return the connection to the pool
 
104
        _test(Empty, TimeoutError)
 
105
        _test(SocketTimeout, TimeoutError)
 
106
        _test(BaseSSLError, SSLError)
 
107
        _test(CertificateError, SSLError)
 
108
 
 
109
        # The pool should never be empty, and with these two exceptions being raised,
 
110
        # a retry will be triggered, but that retry will fail, eventually raising
 
111
        # MaxRetryError, not EmptyPoolError
 
112
        # See: https://github.com/shazow/urllib3/issues/76
 
113
        pool._make_request = lambda *args, **kwargs: _raise(HTTPException)
 
114
        with self.assertRaises(MaxRetryError):
 
115
            pool.request('GET', '/', retries=1, pool_timeout=0.01)
 
116
        self.assertEqual(pool.pool.qsize(), POOL_SIZE)
 
117
 
 
118
    def test_assert_same_host(self):
 
119
        c = connection_from_url('http://google.com:80')
 
120
 
 
121
        with self.assertRaises(HostChangedError):
 
122
            c.request('GET', 'http://yahoo.com:80', assert_same_host=True)
 
123
 
 
124
    def test_pool_close(self):
 
125
        pool = connection_from_url('http://google.com:80')
 
126
 
 
127
        # Populate with some connections
 
128
        conn1 = pool._get_conn()
 
129
        conn2 = pool._get_conn()
 
130
        conn3 = pool._get_conn()
 
131
        pool._put_conn(conn1)
 
132
        pool._put_conn(conn2)
 
133
 
 
134
        old_pool_queue = pool.pool
 
135
 
 
136
        pool.close()
 
137
        self.assertEqual(pool.pool, None)
 
138
 
 
139
        with self.assertRaises(ClosedPoolError):
 
140
            pool._get_conn()
 
141
 
 
142
        pool._put_conn(conn3)
 
143
 
 
144
        with self.assertRaises(ClosedPoolError):
 
145
            pool._get_conn()
 
146
 
 
147
        with self.assertRaises(Empty):
 
148
            old_pool_queue.get(block=False)
 
149
 
130
150
 
131
151
if __name__ == '__main__':
132
152
    unittest.main()