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)
 514
 515 # *********************************************
 516 #   CINDER
 517 # *********************************************
 518
 519
 520 class GetVolumeIDTestCase(unittest.TestCase):
 521
 522     def test_get_volume_id(self):
 523         self.mock_shade_client = mock.Mock()
 524         _uuid = uuidutils.generate_uuid()
 525         self.mock_shade_client.get_volume_id.return_value = _uuid
 526         output = openstack_utils.get_volume_id(self.mock_shade_client,
 527                                                'volume_name')
 528         self.assertEqual(_uuid, output)
 529
 530     def test_get_volume_id_None(self):
 531         self.mock_shade_client = mock.Mock()
 532         self.mock_shade_client.get_volume_id.return_value = None
 533         output = openstack_utils.get_volume_id(self.mock_shade_client,
 534                                                'volume_name')
 535         self.assertIsNone(output)
 536
 537
 538 class GetVolumeTestCase(unittest.TestCase):
 539
 540     def setUp(self):
 541         self.mock_shade_client = mock.Mock()
 542         self.mock_shade_client.get_volume = mock.Mock()
 543
 544     def test_get_volume(self):
 545         self.mock_shade_client.get_volume.return_value = {'volume'}
 546         output = openstack_utils.get_volume(self.mock_shade_client,
 547                                             'volume_name_or_id')
 548         self.assertEqual({'volume'}, output)
 549
 550     def test_get_volume_None(self):
 551         self.mock_shade_client.get_volume.return_value = None
 552         output = openstack_utils.get_volume(self.mock_shade_client,
 553                                             'volume_name_or_id')
 554         self.assertIsNone(output)
 555
 556
 557 class CreateVolumeTestCase(unittest.TestCase):
 558
 559     def setUp(self):
 560         self.mock_shade_client = mock.Mock()
 561         self.size = 1
 562
 563     def test_create_volume(self):
 564         self.mock_shade_client.create_volume.return_value = (
 565             {'name': 'volume-name', 'size': self.size})
 566         output = openstack_utils.create_volume(
 567             self.mock_shade_client, self.size)
 568         self.assertEqual(
 569             {'name': 'volume-name', 'size': self.size},
 570             output)
 571
 572     @mock.patch.object(openstack_utils, 'log')
 573     def test_create_volume_fail(self, mock_logger):
 574         self.mock_shade_client.create_volume.side_effect = (
 575             exc.OpenStackCloudException('error message'))
 576         output = openstack_utils.create_volume(self.mock_shade_client,
 577                                                self.size)
 578         mock_logger.error.assert_called_once()
 579         self.assertIsNone(output)