root / kamaki / clients / connection / test.py @ 20c94832
History | View | Annotate | Download (10.2 kB)
1 |
# Copyright 2013 GRNET S.A. All rights reserved.
|
---|---|
2 |
#
|
3 |
# Redistribution and use in source and binary forms, with or
|
4 |
# without modification, are permitted provided that the following
|
5 |
# conditions are met:
|
6 |
#
|
7 |
# 1. Redistributions of source code must retain the above
|
8 |
# copyright notice, self.list of conditions and the following
|
9 |
# disclaimer.
|
10 |
#
|
11 |
# 2. Redistributions in binary form must reproduce the above
|
12 |
# copyright notice, self.list of conditions and the following
|
13 |
# disclaimer in the documentation and/or other materials
|
14 |
# provided with the distribution.
|
15 |
#
|
16 |
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
|
17 |
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
|
18 |
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
|
19 |
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
|
20 |
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
21 |
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
22 |
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
|
23 |
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
|
24 |
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
25 |
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
|
26 |
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
27 |
# POSSIBILITY OF SUCH DAMAGE.
|
28 |
#
|
29 |
# The views and conclusions contained in the software and
|
30 |
# documentation are those of the authors and should not be
|
31 |
# interpreted as representing official policies, either expressed
|
32 |
# or implied, of GRNET S.A.
|
33 |
|
34 |
from unittest import TestCase, TestSuite, makeSuite, TextTestRunner |
35 |
from mock import Mock, patch |
36 |
from random import randrange |
37 |
|
38 |
from kamaki.clients import connection |
39 |
from kamaki.clients.connection import errors, kamakicon |
40 |
|
41 |
|
42 |
class KamakiConnection(TestCase): |
43 |
v_samples = {'title': 'value', 5: 'value'} |
44 |
n_samples = {'title': None, 5: None} |
45 |
false_samples = {None: 'value', 0: 'value'} |
46 |
|
47 |
def setUp(self): |
48 |
from kamaki.clients.connection import KamakiConnection as HTTPC |
49 |
self.conn = HTTPC()
|
50 |
|
51 |
def test_set_header(self): |
52 |
cnn = self.conn
|
53 |
for k, v in self.v_samples.items(): |
54 |
cnn.set_header(k, v) |
55 |
self.assertEquals(cnn.headers[unicode(k)], unicode(v)) |
56 |
for k, v in self.n_samples.items(): |
57 |
cnn.set_header(k, v) |
58 |
self.assertEquals(cnn.headers[unicode(k)], unicode(v)) |
59 |
for k, v in self.false_samples.items(): |
60 |
self.assertRaises(AssertionError, cnn.set_header, k, v) |
61 |
self.assertEquals(len(cnn.headers), 2) |
62 |
|
63 |
def test_set_param(self): |
64 |
cnn = self.conn
|
65 |
for k, v in self.v_samples.items(): |
66 |
cnn.set_param(k, v) |
67 |
self.assertEquals(cnn.params[unicode(k)], v) |
68 |
for k, v in self.n_samples.items(): |
69 |
cnn.set_param(k, v) |
70 |
self.assertEquals(cnn.params[unicode(k)], v) |
71 |
for k, v in self.false_samples.items(): |
72 |
self.assertRaises(AssertionError, cnn.set_param, k, v) |
73 |
self.assertEquals(len(cnn.params), 2) |
74 |
|
75 |
def test_remove_header(self): |
76 |
cnn = self.conn
|
77 |
for k, v in self.v_samples.items(): |
78 |
cnn.headers[unicode(k)] = unicode(v) |
79 |
for k in self.v_samples: |
80 |
cnn.remove_header(k) |
81 |
self.assertFalse(k in cnn.headers) |
82 |
|
83 |
def test_remove_param(self): |
84 |
cnn = self.conn
|
85 |
for k, v in self.v_samples.items(): |
86 |
cnn.params[unicode(k)] = unicode(v) |
87 |
for k in self.v_samples: |
88 |
cnn.remove_param(k) |
89 |
self.assertFalse(k in cnn.params) |
90 |
|
91 |
def test_replace_headers(self): |
92 |
cnn = self.conn
|
93 |
cnn.headers = self.v_samples
|
94 |
cnn.replace_headers({1: 'one', 2: 'two'}) |
95 |
for k in self.v_samples: |
96 |
self.assertFalse(k in cnn.headers) |
97 |
|
98 |
def test_replace_params(self): |
99 |
cnn = self.conn
|
100 |
cnn.params = self.v_samples
|
101 |
cnn.replace_params({1: 'one', 2: 'two'}) |
102 |
for k in self.v_samples: |
103 |
self.assertFalse(k in cnn.params) |
104 |
|
105 |
def test_reset_headers(self): |
106 |
cnn = self.conn
|
107 |
cnn.headers = self.v_samples
|
108 |
cnn.reset_headers() |
109 |
self.assertFalse(cnn.headers)
|
110 |
|
111 |
def test_reset_params(self): |
112 |
cnn = self.conn
|
113 |
cnn.params = self.v_samples
|
114 |
cnn.reset_params() |
115 |
self.assertFalse(cnn.params)
|
116 |
|
117 |
def test_set_url(self): |
118 |
self.assertFalse(self.conn.url) |
119 |
sample_url = 'http://example.com'
|
120 |
self.conn.set_url(sample_url)
|
121 |
self.assertEquals(self.conn.url, sample_url) |
122 |
|
123 |
def test_set_path(self): |
124 |
self.assertFalse(self.conn.path) |
125 |
sample_path = '/example/local/path'
|
126 |
self.conn.set_path(sample_path)
|
127 |
self.assertEquals(self.conn.path, sample_path) |
128 |
|
129 |
def test_set_method(self): |
130 |
self.assertFalse(self.conn.method) |
131 |
sample_method = 'GET'
|
132 |
self.conn.set_method(sample_method)
|
133 |
self.assertEquals(self.conn.method, sample_method) |
134 |
|
135 |
def test_perform_request(self): |
136 |
self.assertRaises(NotImplementedError, self.conn.perform_request) |
137 |
|
138 |
|
139 |
class KamakiHTTPConnection(TestCase): |
140 |
|
141 |
def setUp(self): |
142 |
self.conn = kamakicon.KamakiHTTPConnection()
|
143 |
|
144 |
def test_perform_request(self): |
145 |
from httplib import HTTPConnection |
146 |
from objpool import http |
147 |
pr = self.conn.perform_request
|
148 |
kwargs = dict(
|
149 |
data='',
|
150 |
method='GET',
|
151 |
async_headers=dict(),
|
152 |
async_params=dict())
|
153 |
|
154 |
KCError = errors.KamakiConnectionError |
155 |
fakecon = HTTPConnection('X', 'Y') |
156 |
|
157 |
with patch.object(http, 'get_http_connection', return_value=fakecon): |
158 |
with patch.object(HTTPConnection, 'request', return_value=None): |
159 |
r = pr(**kwargs) |
160 |
self.assertTrue(isinstance(r, kamakicon.KamakiHTTPResponse)) |
161 |
|
162 |
err = IOError('IO Error') |
163 |
with patch.object(HTTPConnection, 'request', side_effect=err): |
164 |
self.assertRaises(KCError, pr, **kwargs)
|
165 |
|
166 |
err = ValueError('Cannot Establish connection') |
167 |
with patch.object(http, 'get_http_connection', side_effect=err): |
168 |
self.assertRaises(KCError, pr, **kwargs)
|
169 |
|
170 |
err = Exception('Any other error') |
171 |
with patch.object(http, 'get_http_connection', side_effect=err): |
172 |
self.assertRaises(KCError, pr, **kwargs)
|
173 |
|
174 |
|
175 |
class KamakiHTTPResponse(TestCase): |
176 |
|
177 |
class fakeResponse(object): |
178 |
sample = 'sample string'
|
179 |
getheaders = Mock(return_value={}) |
180 |
read = Mock(return_value=sample) |
181 |
status = Mock(return_value=None)
|
182 |
reason = Mock(return_value=None)
|
183 |
|
184 |
def setUp(self): |
185 |
from httplib import HTTPConnection |
186 |
self.HTC = HTTPConnection
|
187 |
self.FR = self.fakeResponse |
188 |
|
189 |
def test_text(self): |
190 |
with patch.object(self.HTC, 'getresponse', return_value=self.FR()): |
191 |
self.resp = kamakicon.KamakiHTTPResponse(self.HTC('X', 'Y')) |
192 |
self.assertEquals(self.resp.text, self.FR.sample) |
193 |
sample2 = 'some other string'
|
194 |
self.resp.text = sample2
|
195 |
self.assertNotEquals(self.resp.text, sample2) |
196 |
|
197 |
def test_json(self): |
198 |
with patch.object(self.HTC, 'getresponse', return_value=self.FR()): |
199 |
self.resp = kamakicon.KamakiHTTPResponse(self.HTC('X', 'Y')) |
200 |
self.assertRaises(errors.KamakiResponseError, self.resp.json) |
201 |
sample2 = '{"antoher":"sample", "formated":"in_json"}'
|
202 |
with patch.object(self.FR, 'read', return_value=sample2): |
203 |
self.resp = kamakicon.KamakiHTTPResponse(self.HTC('X', 'Y')) |
204 |
from json import loads |
205 |
self.assertEquals(loads(sample2), self.resp.json) |
206 |
|
207 |
def test_pool_lock(self): |
208 |
conn_num = 2000
|
209 |
for i in range(conn_num): |
210 |
kre = errors.KamakiResponseError |
211 |
with patch.object(self.HTC, 'close', return_value=True): |
212 |
self.resp = kamakicon.KamakiHTTPResponse(self.HTC('X', 'Y')) |
213 |
if randrange(10): |
214 |
with patch.object(
|
215 |
self.HTC,
|
216 |
'getresponse',
|
217 |
return_value=self.FR()):
|
218 |
self.assertEquals(self.resp.text, self.FR.sample) |
219 |
else:
|
220 |
with patch.object(
|
221 |
self.HTC,
|
222 |
'getresponse',
|
223 |
side_effect=kre('A random error')):
|
224 |
try:
|
225 |
self.resp.text
|
226 |
except kre:
|
227 |
pass
|
228 |
self.HTC.close.assert_called_with()
|
229 |
|
230 |
|
231 |
class KamakiResponse(TestCase): |
232 |
|
233 |
def setUp(self): |
234 |
self.resp = connection.KamakiResponse(
|
235 |
'Abstract class, so test with fake request (str)')
|
236 |
|
237 |
def _mock_get_response(foo): |
238 |
def mocker(self): |
239 |
self.resp._get_response = Mock()
|
240 |
foo(self)
|
241 |
return mocker
|
242 |
|
243 |
def test_release(self): |
244 |
self.assertRaises(NotImplementedError, self.resp.release) |
245 |
|
246 |
def test_prefetched(self): |
247 |
self.assertFalse(self.resp.prefetched) |
248 |
self.resp.prefetched = True |
249 |
self.assertTrue(self.resp.prefetched) |
250 |
|
251 |
@_mock_get_response
|
252 |
def test_content(self): |
253 |
rsp = self.resp
|
254 |
for cont in ('Sample Content', u'\u03c7\u03cd\u03bd\u03c9\x00'): |
255 |
rsp.content = cont |
256 |
self.assertEquals(rsp.content, cont)
|
257 |
|
258 |
( |
259 |
test_text, |
260 |
test_json, |
261 |
test_headers, |
262 |
test_status, |
263 |
test_status_code) = 5 * (test_content,)
|
264 |
|
265 |
def test_request(self): |
266 |
r = self.resp.request
|
267 |
self.assertTrue(isinstance(r, str)) |
268 |
|
269 |
|
270 |
def get_test_classes(module=__import__(__name__), name=''): |
271 |
from inspect import getmembers, isclass |
272 |
for objname, obj in getmembers(module): |
273 |
if (objname == name or not name) and isclass(obj) and ( |
274 |
issubclass(obj, TestCase)):
|
275 |
yield (obj, objname)
|
276 |
|
277 |
|
278 |
def main(argv): |
279 |
for cls, name in get_test_classes(name=argv[1] if len(argv) > 1 else ''): |
280 |
args = argv[2:]
|
281 |
suite = TestSuite() |
282 |
if args:
|
283 |
suite.addTest(cls('_'.join(['test'] + args))) |
284 |
else:
|
285 |
suite.addTest(makeSuite(cls)) |
286 |
print('Test %s' % name)
|
287 |
TextTestRunner(verbosity=2).run(suite)
|
288 |
|
289 |
|
290 |
if __name__ == '__main__': |
291 |
from sys import argv |
292 |
main(argv) |