1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
|
import unittest
from dummyserver.testcase import HTTPSDummyServerTestCase
from nose.plugins.skip import SkipTest
try:
from google.appengine.api import urlfetch
(urlfetch)
except ImportError:
raise SkipTest("App Engine SDK not available.")
from urllib3.contrib.appengine import AppEngineManager, AppEnginePlatformError
from urllib3.exceptions import (
TimeoutError,
ProtocolError,
SSLError)
from urllib3.util.url import Url
from urllib3.util.retry import Retry
from test.with_dummyserver.test_connectionpool import (
TestConnectionPool, TestRetry)
# Prevent nose from running these test.
TestConnectionPool.__test__ = False
TestRetry.__test__ = False
# This class is used so we can re-use the tests from the connection pool.
# It proxies all requests to the manager.
class MockPool(object):
def __init__(self, host, port, manager, scheme='http'):
self.host = host
self.port = port
self.manager = manager
self.scheme = scheme
def request(self, method, url, *args, **kwargs):
url = self._absolute_url(url)
return self.manager.request(method, url, *args, **kwargs)
def urlopen(self, method, url, *args, **kwargs):
url = self._absolute_url(url)
return self.manager.urlopen(method, url, *args, **kwargs)
def _absolute_url(self, path):
return Url(
scheme=self.scheme,
host=self.host,
port=self.port,
path=path).url
# Note that this doesn't run in the sandbox, it only runs with the URLFetch
# API stub enabled. There's no need to enable the sandbox as we know for a fact
# that URLFetch is used by the connection manager.
class TestGAEConnectionManager(TestConnectionPool):
__test__ = True
# Magic class variable that tells NoseGAE to enable the URLFetch stub.
nosegae_urlfetch = True
def setUp(self):
self.manager = AppEngineManager()
self.pool = MockPool(self.host, self.port, self.manager)
# Tests specific to AppEngineManager
def test_exceptions(self):
# DeadlineExceededError -> TimeoutError
self.assertRaises(
TimeoutError,
self.pool.request,
'GET',
'/sleep?seconds=0.005',
timeout=0.001)
# InvalidURLError -> ProtocolError
self.assertRaises(
ProtocolError,
self.manager.request,
'GET',
'ftp://invalid/url')
# DownloadError -> ProtocolError
self.assertRaises(
ProtocolError,
self.manager.request,
'GET',
'http://0.0.0.0')
# ResponseTooLargeError -> AppEnginePlatformError
self.assertRaises(
AppEnginePlatformError,
self.pool.request,
'GET',
'/nbytes?length=33554433') # One byte over 32 megabtyes.
# URLFetch reports the request too large error as a InvalidURLError,
# which maps to a AppEnginePlatformError.
body = b'1' * 10485761 # One byte over 10 megabytes.
self.assertRaises(
AppEnginePlatformError,
self.manager.request,
'POST',
'/',
body=body)
# Re-used tests below this line.
# Subsumed tests
test_timeout_float = None # Covered by test_exceptions.
# Non-applicable tests
test_conn_closed = None
test_nagle = None
test_socket_options = None
test_disable_default_socket_options = None
test_defaults_are_applied = None
test_tunnel = None
test_keepalive = None
test_keepalive_close = None
test_connection_count = None
test_connection_count_bigpool = None
test_for_double_release = None
test_release_conn_parameter = None
test_stream_keepalive = None
test_cleanup_on_connection_error = None
# Tests that should likely be modified for appengine specific stuff
test_timeout = None
test_connect_timeout = None
test_connection_error_retries = None
test_total_timeout = None
test_none_total_applies_connect = None
test_timeout_success = None
test_source_address_error = None
test_bad_connect = None
test_partial_response = None
test_dns_error = None
class TestGAEConnectionManagerWithSSL(HTTPSDummyServerTestCase):
nosegae_urlfetch = True
def setUp(self):
self.manager = AppEngineManager()
self.pool = MockPool(self.host, self.port, self.manager, 'https')
def test_exceptions(self):
# SSLCertificateError -> SSLError
# SSLError is raised with dummyserver because URLFetch doesn't allow
# self-signed certs.
self.assertRaises(
SSLError,
self.pool.request,
'GET',
'/')
class TestGAERetry(TestRetry):
__test__ = True
# Magic class variable that tells NoseGAE to enable the URLFetch stub.
nosegae_urlfetch = True
def setUp(self):
self.manager = AppEngineManager()
self.pool = MockPool(self.host, self.port, self.manager)
def test_default_method_whitelist_retried(self):
""" urllib3 should retry methods in the default method whitelist """
retry = Retry(total=1, status_forcelist=[418])
# Use HEAD instead of OPTIONS, as URLFetch doesn't support OPTIONS
resp = self.pool.request(
'HEAD', '/successful_retry',
headers={'test-name': 'test_default_whitelist'},
retries=retry)
self.assertEqual(resp.status, 200)
#test_max_retry = None
#test_disabled_retry = None
if __name__ == '__main__':
unittest.main()
|