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