« back to all changes in this revision
Viewing changes to tests/test_service.py
- browse files at revision 21
- compare with another revision
- download diff
- download tarball
- view history from revision 21
- Committer: Package Import Robot
- Author(s): James Page
- Date: 2015-08-11 09:25:22 UTC
- mfrom: (1.1.12) (2.1.3 experimental)
- Revision ID: package-import@ubuntu.com-20150811092522-6epbeuzn6a0jt750
Tags: 1.18.0-2ubuntu1
* Resync with Debian experimental.
* d/pydist-overrides: Map suds-jurko -> suds to ease backporting.
* d/pydist-overrides: Map suds-jurko -> suds to ease backporting.
- files removed:
- oslo
- oslo/vmware
- oslo/vmware/objects
- tests
- tests/objects
- files modified:
- bandit.yaml
- setup.py *
added
removed
tests/test_service.py
1
# Copyright (c) 2014 VMware, Inc.
2
# All Rights Reserved.
3
#
4
# Licensed under the Apache License, Version 2.0 (the "License"); you may
5
# not use this file except in compliance with the License. You may obtain
6
# a copy of the License at
7
#
8
# http://www.apache.org/licenses/LICENSE-2.0
9
#
10
# Unless required by applicable law or agreed to in writing, software
11
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13
# License for the specific language governing permissions and limitations
14
# under the License.
15
16
import mock
17
import requests
18
import six
19
import six.moves.http_client as httplib
20
import suds
21
22
from oslo.vmware import exceptions
23
from oslo.vmware import service
24
from tests import base
25
26
27
class ServiceMessagePluginTest(base.TestCase):
28
"""Test class for ServiceMessagePlugin."""
29
30
def test_add_attribute_for_value(self):
31
node = mock.Mock()
32
node.name = 'value'
33
plugin = service.ServiceMessagePlugin()
34
plugin.add_attribute_for_value(node)
35
node.set.assert_called_once_with('xsi:type', 'xsd:string')
36
37
def test_marshalled(self):
38
plugin = service.ServiceMessagePlugin()
39
context = mock.Mock()
40
plugin.marshalled(context)
41
context.envelope.prune.assert_called_once_with()
42
context.envelope.walk.assert_called_once_with(
43
plugin.add_attribute_for_value)
44
45
46
class ServiceTest(base.TestCase):
47
48
def setUp(self):
49
super(ServiceTest, self).setUp()
50
patcher = mock.patch('suds.client.Client')
51
self.addCleanup(patcher.stop)
52
self.SudsClientMock = patcher.start()
53
54
def test_retrieve_properties_ex_fault_checker_with_empty_response(self):
55
try:
56
service.Service._retrieve_properties_ex_fault_checker(None)
57
assert False
58
except exceptions.VimFaultException as ex:
59
self.assertEqual([exceptions.NOT_AUTHENTICATED],
60
ex.fault_list)
61
62
def test_retrieve_properties_ex_fault_checker(self):
63
fault_list = ['FileFault', 'VimFault']
64
missing_set = []
65
for fault in fault_list:
66
missing_elem = mock.Mock()
67
missing_elem.fault.fault.__class__.__name__ = fault
68
missing_set.append(missing_elem)
69
obj_cont = mock.Mock()
70
obj_cont.missingSet = missing_set
71
response = mock.Mock()
72
response.objects = [obj_cont]
73
74
try:
75
service.Service._retrieve_properties_ex_fault_checker(response)
76
assert False
77
except exceptions.VimFaultException as ex:
78
self.assertEqual(fault_list, ex.fault_list)
79
80
def test_request_handler(self):
81
managed_object = 'VirtualMachine'
82
resp = mock.Mock()
83
84
def side_effect(mo, **kwargs):
85
self.assertEqual(managed_object, mo._type)
86
self.assertEqual(managed_object, mo.value)
87
return resp
88
89
svc_obj = service.Service()
90
attr_name = 'powerOn'
91
service_mock = svc_obj.client.service
92
setattr(service_mock, attr_name, side_effect)
93
ret = svc_obj.powerOn(managed_object)
94
self.assertEqual(resp, ret)
95
96
def test_request_handler_with_retrieve_properties_ex_fault(self):
97
managed_object = 'Datacenter'
98
99
def side_effect(mo, **kwargs):
100
self.assertEqual(managed_object, mo._type)
101
self.assertEqual(managed_object, mo.value)
102
return None
103
104
svc_obj = service.Service()
105
attr_name = 'retrievePropertiesEx'
106
service_mock = svc_obj.client.service
107
setattr(service_mock, attr_name, side_effect)
108
self.assertRaises(exceptions.VimFaultException,
109
svc_obj.retrievePropertiesEx,
110
managed_object)
111
112
def test_request_handler_with_web_fault(self):
113
managed_object = 'VirtualMachine'
114
fault_list = ['Fault']
115
116
doc = mock.Mock()
117
118
def side_effect(mo, **kwargs):
119
self.assertEqual(managed_object, mo._type)
120
self.assertEqual(managed_object, mo.value)
121
fault = mock.Mock(faultstring="MyFault")
122
123
fault_children = mock.Mock()
124
fault_children.name = "name"
125
fault_children.getText.return_value = "value"
126
child = mock.Mock()
127
child.get.return_value = fault_list[0]
128
child.getChildren.return_value = [fault_children]
129
detail = mock.Mock()
130
detail.getChildren.return_value = [child]
131
doc.childAtPath.return_value = detail
132
raise suds.WebFault(fault, doc)
133
134
svc_obj = service.Service()
135
service_mock = svc_obj.client.service
136
setattr(service_mock, 'powerOn', side_effect)
137
138
ex = self.assertRaises(exceptions.VimFaultException, svc_obj.powerOn,
139
managed_object)
140
141
self.assertEqual(fault_list, ex.fault_list)
142
self.assertEqual({'name': 'value'}, ex.details)
143
self.assertEqual("MyFault", ex.msg)
144
doc.childAtPath.assertCalledOnceWith('/detail')
145
146
def test_request_handler_with_empty_web_fault_doc(self):
147
148
def side_effect(mo, **kwargs):
149
fault = mock.Mock(faultstring="MyFault")
150
raise suds.WebFault(fault, None)
151
152
svc_obj = service.Service()
153
service_mock = svc_obj.client.service
154
setattr(service_mock, 'powerOn', side_effect)
155
156
ex = self.assertRaises(exceptions.VimFaultException,
157
svc_obj.powerOn,
158
'VirtualMachine')
159
self.assertEqual([], ex.fault_list)
160
self.assertEqual({}, ex.details)
161
self.assertEqual("MyFault", ex.msg)
162
163
def test_request_handler_with_vc51_web_fault(self):
164
managed_object = 'VirtualMachine'
165
fault_list = ['Fault']
166
167
doc = mock.Mock()
168
169
def side_effect(mo, **kwargs):
170
self.assertEqual(managed_object, mo._type)
171
self.assertEqual(managed_object, mo.value)
172
fault = mock.Mock(faultstring="MyFault")
173
174
fault_children = mock.Mock()
175
fault_children.name = "name"
176
fault_children.getText.return_value = "value"
177
child = mock.Mock()
178
child.get.return_value = fault_list[0]
179
child.getChildren.return_value = [fault_children]
180
detail = mock.Mock()
181
detail.getChildren.return_value = [child]
182
doc.childAtPath.side_effect = [None, detail]
183
raise suds.WebFault(fault, doc)
184
185
svc_obj = service.Service()
186
service_mock = svc_obj.client.service
187
setattr(service_mock, 'powerOn', side_effect)
188
189
ex = self.assertRaises(exceptions.VimFaultException, svc_obj.powerOn,
190
managed_object)
191
192
self.assertEqual(fault_list, ex.fault_list)
193
self.assertEqual({'name': 'value'}, ex.details)
194
self.assertEqual("MyFault", ex.msg)
195
exp_calls = [mock.call('/detail'),
196
mock.call('/Envelope/Body/Fault/detail')]
197
self.assertEqual(exp_calls, doc.childAtPath.call_args_list)
198
199
def test_request_handler_with_attribute_error(self):
200
managed_object = 'VirtualMachine'
201
svc_obj = service.Service()
202
# no powerOn method in Service
203
service_mock = mock.Mock(spec=service.Service)
204
svc_obj.client.service = service_mock
205
self.assertRaises(exceptions.VimAttributeException,
206
svc_obj.powerOn,
207
managed_object)
208
209
def test_request_handler_with_http_cannot_send_error(self):
210
managed_object = 'VirtualMachine'
211
212
def side_effect(mo, **kwargs):
213
self.assertEqual(managed_object, mo._type)
214
self.assertEqual(managed_object, mo.value)
215
raise httplib.CannotSendRequest()
216
217
svc_obj = service.Service()
218
attr_name = 'powerOn'
219
service_mock = svc_obj.client.service
220
setattr(service_mock, attr_name, side_effect)
221
self.assertRaises(exceptions.VimSessionOverLoadException,
222
svc_obj.powerOn,
223
managed_object)
224
225
def test_request_handler_with_http_response_not_ready_error(self):
226
managed_object = 'VirtualMachine'
227
228
def side_effect(mo, **kwargs):
229
self.assertEqual(managed_object, mo._type)
230
self.assertEqual(managed_object, mo.value)
231
raise httplib.ResponseNotReady()
232
233
svc_obj = service.Service()
234
attr_name = 'powerOn'
235
service_mock = svc_obj.client.service
236
setattr(service_mock, attr_name, side_effect)
237
self.assertRaises(exceptions.VimSessionOverLoadException,
238
svc_obj.powerOn,
239
managed_object)
240
241
def test_request_handler_with_http_cannot_send_header_error(self):
242
managed_object = 'VirtualMachine'
243
244
def side_effect(mo, **kwargs):
245
self.assertEqual(managed_object, mo._type)
246
self.assertEqual(managed_object, mo.value)
247
raise httplib.CannotSendHeader()
248
249
svc_obj = service.Service()
250
attr_name = 'powerOn'
251
service_mock = svc_obj.client.service
252
setattr(service_mock, attr_name, side_effect)
253
self.assertRaises(exceptions.VimSessionOverLoadException,
254
svc_obj.powerOn,
255
managed_object)
256
257
def test_request_handler_with_connection_error(self):
258
managed_object = 'VirtualMachine'
259
260
def side_effect(mo, **kwargs):
261
self.assertEqual(managed_object, mo._type)
262
self.assertEqual(managed_object, mo.value)
263
raise requests.ConnectionError()
264
265
svc_obj = service.Service()
266
attr_name = 'powerOn'
267
service_mock = svc_obj.client.service
268
setattr(service_mock, attr_name, side_effect)
269
self.assertRaises(exceptions.VimConnectionException,
270
svc_obj.powerOn,
271
managed_object)
272
273
def test_request_handler_with_http_error(self):
274
managed_object = 'VirtualMachine'
275
276
def side_effect(mo, **kwargs):
277
self.assertEqual(managed_object, mo._type)
278
self.assertEqual(managed_object, mo.value)
279
raise requests.HTTPError()
280
281
svc_obj = service.Service()
282
attr_name = 'powerOn'
283
service_mock = svc_obj.client.service
284
setattr(service_mock, attr_name, side_effect)
285
self.assertRaises(exceptions.VimConnectionException,
286
svc_obj.powerOn,
287
managed_object)
288
289
@mock.patch('oslo_vmware.vim_util.get_moref', return_value=None)
290
def test_request_handler_no_value(self, mock_moref):
291
managed_object = 'VirtualMachine'
292
svc_obj = service.Service()
293
ret = svc_obj.UnregisterVM(managed_object)
294
self.assertIsNone(ret)
295
296
def _test_request_handler_with_exception(self, message, exception):
297
managed_object = 'VirtualMachine'
298
299
def side_effect(mo, **kwargs):
300
self.assertEqual(managed_object, mo._type)
301
self.assertEqual(managed_object, mo.value)
302
raise Exception(message)
303
304
svc_obj = service.Service()
305
attr_name = 'powerOn'
306
service_mock = svc_obj.client.service
307
setattr(service_mock, attr_name, side_effect)
308
self.assertRaises(exception, svc_obj.powerOn, managed_object)
309
310
def test_request_handler_with_address_in_use_error(self):
311
self._test_request_handler_with_exception(
312
service.ADDRESS_IN_USE_ERROR,
313
exceptions.VimSessionOverLoadException)
314
315
def test_request_handler_with_conn_abort_error(self):
316
self._test_request_handler_with_exception(
317
service.CONN_ABORT_ERROR, exceptions.VimSessionOverLoadException)
318
319
def test_request_handler_with_resp_not_xml_error(self):
320
self._test_request_handler_with_exception(
321
service.RESP_NOT_XML_ERROR, exceptions.VimSessionOverLoadException)
322
323
def test_request_handler_with_generic_error(self):
324
self._test_request_handler_with_exception(
325
'GENERIC_ERROR', exceptions.VimException)
326
327
def test_get_session_cookie(self):
328
svc_obj = service.Service()
329
cookie_value = 'xyz'
330
cookie = mock.Mock()
331
cookie.name = 'vmware_soap_session'
332
cookie.value = cookie_value
333
svc_obj.client.options.transport.cookiejar = [cookie]
334
self.assertEqual(cookie_value, svc_obj.get_http_cookie())
335
336
def test_get_session_cookie_with_no_cookie(self):
337
svc_obj = service.Service()
338
cookie = mock.Mock()
339
cookie.name = 'cookie'
340
cookie.value = 'xyz'
341
svc_obj.client.options.transport.cookiejar = [cookie]
342
self.assertIsNone(svc_obj.get_http_cookie())
343
344
345
class MemoryCacheTest(base.TestCase):
346
"""Test class for MemoryCache."""
347
348
def test_get_set(self):
349
cache = service.MemoryCache()
350
cache.put('key1', 'value1')
351
cache.put('key2', 'value2')
352
self.assertEqual('value1', cache.get('key1'))
353
self.assertEqual('value2', cache.get('key2'))
354
self.assertEqual(None, cache.get('key3'))
355
356
@mock.patch('suds.reader.DocumentReader.download')
357
def test_shared_cache(self, mock_reader):
358
cache1 = service.Service().client.options.cache
359
cache2 = service.Service().client.options.cache
360
self.assertIs(cache1, cache2)
361
362
@mock.patch('oslo_utils.timeutils.utcnow_ts')
363
def test_cache_timeout(self, mock_utcnow_ts):
364
mock_utcnow_ts.side_effect = [100, 125, 150, 175, 195, 200, 225]
365
366
cache = service.MemoryCache()
367
cache.put('key1', 'value1', 10)
368
cache.put('key2', 'value2', 75)
369
cache.put('key3', 'value3', 100)
370
371
self.assertIsNone(cache.get('key1'))
372
self.assertEqual('value2', cache.get('key2'))
373
self.assertIsNone(cache.get('key2'))
374
self.assertEqual('value3', cache.get('key3'))
375
376
377
class RequestsTransportTest(base.TestCase):
378
"""Tests for RequestsTransport."""
379
380
def test_open(self):
381
transport = service.RequestsTransport()
382
383
data = "Hello World"
384
resp = mock.Mock(content=data)
385
transport.session.get = mock.Mock(return_value=resp)
386
387
request = mock.Mock(url=mock.sentinel.url)
388
self.assertEqual(data,
389
transport.open(request).getvalue())
390
transport.session.get.assert_called_once_with(mock.sentinel.url,
391
verify=transport.verify)
392
393
def test_send(self):
394
transport = service.RequestsTransport()
395
396
resp = mock.Mock(status_code=mock.sentinel.status_code,
397
headers=mock.sentinel.headers,
398
content=mock.sentinel.content)
399
transport.session.post = mock.Mock(return_value=resp)
400
401
request = mock.Mock(url=mock.sentinel.url,
402
message=mock.sentinel.message,
403
headers=mock.sentinel.req_headers)
404
reply = transport.send(request)
405
406
self.assertEqual(mock.sentinel.status_code, reply.code)
407
self.assertEqual(mock.sentinel.headers, reply.headers)
408
self.assertEqual(mock.sentinel.content, reply.message)
409
410
@mock.patch('os.path.getsize')
411
def test_send_with_local_file_url(self, get_size_mock):
412
transport = service.RequestsTransport()
413
414
url = 'file:///foo'
415
request = requests.PreparedRequest()
416
request.url = url
417
418
data = b"Hello World"
419
get_size_mock.return_value = len(data)
420
421
def readinto_mock(buf):
422
buf[0:] = data
423
424
if six.PY3:
425
builtin_open = 'builtins.open'
426
open_mock = mock.MagicMock(name='file_handle',
427
spec=open)
428
import _io
429
file_spec = list(set(dir(_io.TextIOWrapper)).union(
430
set(dir(_io.BytesIO))))
431
else:
432
builtin_open = '__builtin__.open'
433
open_mock = mock.MagicMock(name='file_handle',
434
spec=file)
435
file_spec = file
436
437
file_handle = mock.MagicMock(spec=file_spec)
438
file_handle.write.return_value = None
439
file_handle.__enter__.return_value = file_handle
440
file_handle.readinto.side_effect = readinto_mock
441
open_mock.return_value = file_handle
442
443
with mock.patch(builtin_open, open_mock, create=True):
444
resp = transport.session.send(request)
445
self.assertEqual(data, resp.content)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1