81bcd8c3330ed5b72059f763586a092f1f04328a
[yardstick.git] / yardstick / tests / unit / common / test_openstack_utils.py
1 ##############################################################################
2 # Copyright (c) 2016 Huawei Technologies Co.,Ltd and others.
3 #
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
9
10 from oslo_utils import uuidutils
11 import unittest
12 import mock
13
14 from shade import exc
15 from yardstick.common import openstack_utils
16
17
18 class GetCredentialsTestCase(unittest.TestCase):
19
20     @mock.patch('yardstick.common.openstack_utils.os')
21     def test_get_credentials(self, _):
22         with mock.patch.dict('os.environ', {'OS_IDENTITY_API_VERSION': '2'},
23                              clear=True):
24             openstack_utils.get_credentials()
25
26
27 class GetHeatApiVersionTestCase(unittest.TestCase):
28
29     def test_get_heat_api_version_check_result(self):
30         API = 'HEAT_API_VERSION'
31         expected_result = '2'
32
33         with mock.patch.dict('os.environ', {API: '2'}, clear=True):
34             api_version = openstack_utils.get_heat_api_version()
35             self.assertEqual(api_version, expected_result)
36
37
38 class DeleteNeutronNetTestCase(unittest.TestCase):
39
40     def setUp(self):
41         self.mock_shade_client = mock.Mock()
42
43     def test_delete_neutron_net(self):
44         self.mock_shade_client.delete_network.return_value = True
45         output = openstack_utils.delete_neutron_net(self.mock_shade_client,
46                                                     'network_name_or_id')
47         self.assertTrue(output)
48
49     def test_delete_neutron_net_fail(self):
50         self.mock_shade_client.delete_network.return_value = False
51         output = openstack_utils.delete_neutron_net(self.mock_shade_client,
52                                                     'network_name_or_id')
53         self.assertFalse(output)
54
55     @mock.patch.object(openstack_utils, 'log')
56     def test_delete_neutron_net_exception(self, mock_logger):
57         self.mock_shade_client.delete_network.side_effect = (
58             exc.OpenStackCloudException('error message'))
59         output = openstack_utils.delete_neutron_net(self.mock_shade_client,
60                                                     'network_name_or_id')
61         self.assertFalse(output)
62         mock_logger.error.assert_called_once()
63
64
65 class CreateNeutronNetTestCase(unittest.TestCase):
66
67     def setUp(self):
68         self.mock_shade_client = mock.Mock()
69         self.network_name = 'name'
70         self.mock_shade_client.create_network = mock.Mock()
71
72     def test_create_neutron_net(self):
73         _uuid = uuidutils.generate_uuid()
74         self.mock_shade_client.create_network.return_value = {'id': _uuid}
75         output = openstack_utils.create_neutron_net(self.mock_shade_client,
76                                                     self.network_name)
77         self.assertEqual(_uuid, output)
78
79     @mock.patch.object(openstack_utils, 'log')
80     def test_create_neutron_net_exception(self, mock_logger):
81         self.mock_shade_client.create_network.side_effect = (
82             exc.OpenStackCloudException('error message'))
83
84         output = openstack_utils.create_neutron_net(self.mock_shade_client,
85                                                     self.network_name)
86         mock_logger.error.assert_called_once()
87         self.assertIsNone(output)
88
89
90 class CreateNeutronSubnetTestCase(unittest.TestCase):
91
92     def setUp(self):
93         self.mock_shade_client = mock.Mock()
94         self.network_name_or_id = 'name_or_id'
95         self.mock_shade_client.create_subnet = mock.Mock()
96
97     def test_create_neutron_subnet(self):
98         _uuid = uuidutils.generate_uuid()
99         self.mock_shade_client.create_subnet.return_value = {'id': _uuid}
100         output = openstack_utils.create_neutron_subnet(
101             self.mock_shade_client, self.network_name_or_id)
102         self.assertEqual(_uuid, output)
103
104     @mock.patch.object(openstack_utils, 'log')
105     def test_create_neutron_subnet_exception(self, mock_logger):
106         self.mock_shade_client.create_subnet.side_effect = (
107             exc.OpenStackCloudException('error message'))
108
109         output = openstack_utils.create_neutron_subnet(
110             self.mock_shade_client, self.network_name_or_id)
111         mock_logger.error.assert_called_once()
112         self.assertIsNone(output)
113
114
115 class DeleteNeutronRouterTestCase(unittest.TestCase):
116
117     def setUp(self):
118         self.mock_shade_client = mock.Mock()
119         self.mock_shade_client.delete_router = mock.Mock()
120
121     def test_delete_neutron_router(self):
122         self.mock_shade_client.delete_router.return_value = True
123         output = openstack_utils.delete_neutron_router(self.mock_shade_client,
124                                                        'router_id')
125         self.assertTrue(output)
126
127     def test_delete_neutron_router_fail(self):
128         self.mock_shade_client.delete_router.return_value = False
129         output = openstack_utils.delete_neutron_router(self.mock_shade_client,
130                                                        'router_id')
131         self.assertFalse(output)
132
133     @mock.patch.object(openstack_utils, 'log')
134     def test_delete_neutron_router_exception(self, mock_logger):
135         self.mock_shade_client.delete_router.side_effect = (
136             exc.OpenStackCloudException('error message'))
137         output = openstack_utils.delete_neutron_router(self.mock_shade_client,
138                                                        'router_id')
139         mock_logger.error.assert_called_once()
140         self.assertFalse(output)
141
142
143 class CreateNeutronRouterTestCase(unittest.TestCase):
144
145     def setUp(self):
146         self.mock_shade_client = mock.Mock()
147         self.mock_shade_client.create_subnet = mock.Mock()
148
149     def test_create_neutron_router(self):
150         _uuid = uuidutils.generate_uuid()
151         self.mock_shade_client.create_router.return_value = {'id': _uuid}
152         output = openstack_utils.create_neutron_router(
153             self.mock_shade_client)
154         self.assertEqual(_uuid, output)
155
156     @mock.patch.object(openstack_utils, 'log')
157     def test_create_neutron_subnet_exception(self, mock_logger):
158         self.mock_shade_client.create_router.side_effect = (
159             exc.OpenStackCloudException('error message'))
160
161         output = openstack_utils.create_neutron_router(
162             self.mock_shade_client)
163         mock_logger.error.assert_called_once()
164         self.assertIsNone(output)
165
166
167 class RemoveRouterInterfaceTestCase(unittest.TestCase):
168
169     def setUp(self):
170         self.mock_shade_client = mock.Mock()
171         self.router = 'router'
172         self.mock_shade_client.remove_router_interface = mock.Mock()
173
174     def test_remove_router_interface(self):
175         self.mock_shade_client.remove_router_interface.return_value = True
176         output = openstack_utils.remove_router_interface(
177             self.mock_shade_client, self.router)
178         self.assertTrue(output)
179
180     @mock.patch.object(openstack_utils, 'log')
181     def test_remove_router_interface_exception(self, mock_logger):
182         self.mock_shade_client.remove_router_interface.side_effect = (
183             exc.OpenStackCloudException('error message'))
184         output = openstack_utils.remove_router_interface(
185             self.mock_shade_client, self.router)
186         mock_logger.error.assert_called_once()
187         self.assertFalse(output)
188
189
190 class CreateFloatingIpTestCase(unittest.TestCase):
191
192     def setUp(self):
193         self.mock_shade_client = mock.Mock()
194         self.network_name_or_id = 'name'
195         self.mock_shade_client.create_floating_ip = mock.Mock()
196
197     def test_create_floating_ip(self):
198         self.mock_shade_client.create_floating_ip.return_value = \
199             {'floating_ip_address': 'value1', 'id': 'value2'}
200         output = openstack_utils.create_floating_ip(self.mock_shade_client,
201                                                     self.network_name_or_id)
202         self.assertEqual({'fip_addr': 'value1', 'fip_id': 'value2'}, output)
203
204     @mock.patch.object(openstack_utils, 'log')
205     def test_create_floating_ip_exception(self, mock_logger):
206         self.mock_shade_client.create_floating_ip.side_effect = (
207             exc.OpenStackCloudException('error message'))
208         output = openstack_utils.create_floating_ip(
209             self.mock_shade_client, self.network_name_or_id)
210         mock_logger.error.assert_called_once()
211         self.assertIsNone(output)
212
213
214 class DeleteFloatingIpTestCase(unittest.TestCase):
215
216     def setUp(self):
217         self.mock_shade_client = mock.Mock()
218         self.floating_ip_id = 'floating_ip_id'
219         self.mock_shade_client.delete_floating_ip = mock.Mock()
220
221     def test_delete_floating_ip(self):
222         self.mock_shade_client.delete_floating_ip.return_value = True
223         output = openstack_utils.delete_floating_ip(self.mock_shade_client,
224                                                     'floating_ip_id')
225         self.assertTrue(output)
226
227     def test_delete_floating_ip_fail(self):
228         self.mock_shade_client.delete_floating_ip.return_value = False
229         output = openstack_utils.delete_floating_ip(self.mock_shade_client,
230                                                     'floating_ip_id')
231         self.assertFalse(output)
232
233     @mock.patch.object(openstack_utils, 'log')
234     def test_delete_floating_ip_exception(self, mock_logger):
235         self.mock_shade_client.delete_floating_ip.side_effect = (
236             exc.OpenStackCloudException('error message'))
237         output = openstack_utils.delete_floating_ip(self.mock_shade_client,
238                                                     'floating_ip_id')
239         mock_logger.error.assert_called_once()
240         self.assertFalse(output)
241
242
243 class CreateSecurityGroupRuleTestCase(unittest.TestCase):
244
245     def setUp(self):
246         self.mock_shade_client = mock.Mock()
247         self.secgroup_name_or_id = 'sg_name_id'
248         self.mock_shade_client.create_security_group_rule = mock.Mock()
249
250     def test_create_security_group_rule(self):
251         self.mock_shade_client.create_security_group_rule.return_value = (
252             {'security_group_rule'})
253         output = openstack_utils.create_security_group_rule(
254             self.mock_shade_client, self.secgroup_name_or_id)
255         self.assertTrue(output)
256
257     @mock.patch.object(openstack_utils, 'log')
258     def test_create_security_group_rule_exception(self, mock_logger):
259         self.mock_shade_client.create_security_group_rule.side_effect = (
260             exc.OpenStackCloudException('error message'))
261
262         output = openstack_utils.create_security_group_rule(
263             self.mock_shade_client, self.secgroup_name_or_id)
264         mock_logger.error.assert_called_once()
265         self.assertFalse(output)
266
267
268 class ListImageTestCase(unittest.TestCase):
269
270     def test_list_images(self):
271         mock_shade_client = mock.MagicMock()
272         mock_shade_client.list_images.return_value = []
273         openstack_utils.list_images(mock_shade_client)
274
275     @mock.patch.object(openstack_utils, 'log')
276     def test_list_images_exception(self, mock_logger):
277         mock_shade_client = mock.MagicMock()
278         mock_shade_client.list_images = mock.MagicMock()
279         mock_shade_client.list_images.side_effect = (
280             exc.OpenStackCloudException('error message'))
281         images = openstack_utils.list_images(mock_shade_client)
282         mock_logger.error.assert_called_once()
283         self.assertFalse(images)
284
285
286 class SecurityGroupTestCase(unittest.TestCase):
287
288     def setUp(self):
289         self.mock_shade_client = mock.Mock()
290         self.sg_name = 'sg_name'
291         self.sg_description = 'sg_description'
292         self._uuid = uuidutils.generate_uuid()
293
294     def test_create_security_group_full_existing_security_group(self):
295         self.mock_shade_client.get_security_group.return_value = (
296             {'name': 'name', 'id': self._uuid})
297         output = openstack_utils.create_security_group_full(
298             self.mock_shade_client, self.sg_name, self.sg_description)
299         self.mock_shade_client.get_security_group.assert_called_once()
300         self.assertEqual(self._uuid, output)
301
302     @mock.patch.object(openstack_utils, 'log')
303     def test_create_security_group_full_non_existing_security_group(
304             self, mock_logger):
305         self.mock_shade_client.get_security_group.return_value = None
306         self.mock_shade_client.create_security_group.side_effect = (
307             exc.OpenStackCloudException('error message'))
308         output = openstack_utils.create_security_group_full(
309             self.mock_shade_client, self.sg_name, self.sg_description)
310         mock_logger.error.assert_called_once()
311         self.assertIsNone(output)
312
313     @mock.patch.object(openstack_utils, 'create_security_group_rule')
314     @mock.patch.object(openstack_utils, 'log')
315     def test_create_security_group_full_create_rule_fail(
316             self, mock_logger, mock_create_security_group_rule):
317         self.mock_shade_client.get_security_group.return_value = None
318         self.mock_shade_client.create_security_group.return_value = (
319             {'name': 'name', 'id': self._uuid})
320         mock_create_security_group_rule.return_value = False
321         output = openstack_utils.create_security_group_full(
322             self.mock_shade_client, self.sg_name, self.sg_description)
323         mock_create_security_group_rule.assert_called()
324         self.mock_shade_client.delete_security_group(self.sg_name)
325         mock_logger.error.assert_called_once()
326         self.assertIsNone(output)
327
328     @mock.patch.object(openstack_utils, 'create_security_group_rule')
329     def test_create_security_group_full(
330             self, mock_create_security_group_rule):
331         self.mock_shade_client.get_security_group.return_value = None
332         self.mock_shade_client.create_security_group.return_value = (
333             {'name': 'name', 'id': self._uuid})
334         mock_create_security_group_rule.return_value = True
335         output = openstack_utils.create_security_group_full(
336             self.mock_shade_client, self.sg_name, self.sg_description)
337         mock_create_security_group_rule.assert_called()
338         self.mock_shade_client.delete_security_group(self.sg_name)
339         self.assertEqual(self._uuid, output)
340
341 # *********************************************
342 #   NOVA
343 # *********************************************
344
345
346 class CreateInstanceTestCase(unittest.TestCase):
347
348     def test_create_instance_and_wait_for_active(self):
349         self.mock_shade_client = mock.Mock()
350         name = 'server_name'
351         image = 'image_name'
352         flavor = 'flavor_name'
353         self.mock_shade_client.create_server.return_value = (
354             {'name': name, 'image': image, 'flavor': flavor})
355         output = openstack_utils.create_instance_and_wait_for_active(
356             self.mock_shade_client, name, image, flavor)
357         self.assertEqual(
358             {'name': name, 'image': image, 'flavor': flavor}, output)
359
360     @mock.patch.object(openstack_utils, 'log')
361     def test_create_instance_and_wait_for_active_fail(self, mock_logger):
362         self.mock_shade_client = mock.Mock()
363         self.mock_shade_client.create_server.side_effect = (
364             exc.OpenStackCloudException('error message'))
365         output = openstack_utils.create_instance_and_wait_for_active(
366             self.mock_shade_client, 'server_name', 'image_name', 'flavor_name')
367         mock_logger.error.assert_called_once()
368         self.assertIsNone(output)
369
370
371 class DeleteInstanceTestCase(unittest.TestCase):
372
373     def setUp(self):
374         self.mock_shade_client = mock.Mock()
375
376     def test_delete_instance(self):
377         self.mock_shade_client.delete_server.return_value = True
378         output = openstack_utils.delete_instance(self.mock_shade_client,
379                                                  'instance_name_id')
380         self.assertTrue(output)
381
382     def test_delete_instance_fail(self):
383         self.mock_shade_client.delete_server.return_value = False
384         output = openstack_utils.delete_instance(self.mock_shade_client,
385                                                  'instance_name_id')
386         self.assertFalse(output)
387
388     @mock.patch.object(openstack_utils, 'log')
389     def test_delete_instance_exception(self, mock_logger):
390         self.mock_shade_client.delete_server.side_effect = (
391             exc.OpenStackCloudException('error message'))
392         output = openstack_utils.delete_instance(self.mock_shade_client,
393                                                  'instance_name_id')
394         mock_logger.error.assert_called_once()
395         self.assertFalse(output)
396
397
398 class CreateKeypairTestCase(unittest.TestCase):
399
400     def setUp(self):
401         self.mock_shade_client = mock.Mock()
402         self.name = 'key_name'
403
404     def test_create_keypair(self):
405         self.mock_shade_client.create_keypair.return_value = (
406             {'name': 'key-name', 'type': 'ssh'})
407         output = openstack_utils.create_keypair(
408             self.mock_shade_client, self.name)
409         self.assertEqual(
410             {'name': 'key-name', 'type': 'ssh'},
411             output)
412
413     @mock.patch.object(openstack_utils, 'log')
414     def test_create_keypair_fail(self, mock_logger):
415         self.mock_shade_client.create_keypair.side_effect = (
416             exc.OpenStackCloudException('error message'))
417         output = openstack_utils.create_keypair(
418             self.mock_shade_client, self.name)
419         mock_logger.error.assert_called_once()
420         self.assertIsNone(output)
421
422
423 class DeleteKeypairTestCase(unittest.TestCase):
424
425     def setUp(self):
426         self.mock_shade_client = mock.Mock()
427
428     def test_delete_keypair(self):
429         self.mock_shade_client.delete_keypair.return_value = True
430         output = openstack_utils.delete_keypair(self.mock_shade_client,
431                                                 'key_name')
432         self.assertTrue(output)
433
434     def test_delete_keypair_fail(self):
435         self.mock_shade_client.delete_keypair.return_value = False
436         output = openstack_utils.delete_keypair(self.mock_shade_client,
437                                                 'key_name')
438         self.assertFalse(output)
439
440     @mock.patch.object(openstack_utils, 'log')
441     def test_delete_keypair_exception(self, mock_logger):
442         self.mock_shade_client.delete_keypair.side_effect = (
443             exc.OpenStackCloudException('error message'))
444         output = openstack_utils.delete_keypair(self.mock_shade_client,
445                                                 'key_name')
446         mock_logger.error.assert_called_once()
447         self.assertFalse(output)
448
449
450 class AttachVolumeToServerTestCase(unittest.TestCase):
451
452     def test_attach_volume_to_server(self):
453         self.mock_shade_client = mock.Mock()
454         self.mock_shade_client.get_server.return_value = {'server_dict'}
455         self.mock_shade_client.get_volume.return_value = {'volume_dict'}
456         self.mock_shade_client.attach_volume.return_value = True
457         output = openstack_utils.attach_volume_to_server(
458             self.mock_shade_client, 'server_name_or_id', 'volume_name_or_id')
459         self.assertTrue(output)
460
461     @mock.patch.object(openstack_utils, 'log')
462     def test_attach_volume_to_server_fail(self, mock_logger):
463         self.mock_shade_client = mock.Mock()
464         self.mock_shade_client.attach_volume.side_effect = (
465             exc.OpenStackCloudException('error message'))
466         output = openstack_utils.attach_volume_to_server(
467             self.mock_shade_client, 'server_name_or_id', 'volume_name_or_id')
468         mock_logger.error.assert_called_once()
469         self.assertFalse(output)
470
471
472 class GetServerTestCase(unittest.TestCase):
473
474     def test_get_server(self):
475         self.mock_shade_client = mock.Mock()
476         _uuid = uuidutils.generate_uuid()
477         self.mock_shade_client.get_server.return_value = {
478             'name': 'server_name', 'id': _uuid}
479         output = openstack_utils.get_server(self.mock_shade_client,
480                                             'server_name_or_id')
481         self.assertEqual({'name': 'server_name', 'id': _uuid}, output)
482
483     @mock.patch.object(openstack_utils, 'log')
484     def test_get_server_exception(self, mock_logger):
485         self.mock_shade_client = mock.Mock()
486         self.mock_shade_client.get_server.side_effect = (
487             exc.OpenStackCloudException('error message'))
488         output = openstack_utils.get_server(self.mock_shade_client,
489                                             'server_name_or_id')
490         mock_logger.error.assert_called_once()
491         self.assertIsNone(output)
492
493
494 class GetFlavorTestCase(unittest.TestCase):
495
496     def test_get_flavor(self):
497         self.mock_shade_client = mock.Mock()
498         _uuid = uuidutils.generate_uuid()
499         self.mock_shade_client.get_flavor.return_value = {
500             'name': 'flavor_name', 'id': _uuid}
501         output = openstack_utils.get_flavor(self.mock_shade_client,
502                                             'flavor_name_or_id')
503         self.assertEqual({'name': 'flavor_name', 'id': _uuid}, output)
504
505     @mock.patch.object(openstack_utils, 'log')
506     def test_get_flavor_exception(self, mock_logger):
507         self.mock_shade_client = mock.Mock()
508         self.mock_shade_client.get_flavor.side_effect = (
509             exc.OpenStackCloudException('error message'))
510         output = openstack_utils.get_flavor(self.mock_shade_client,
511                                             'flavor_name_or_id')
512         mock_logger.error.assert_called_once()
513         self.assertIsNone(output)