functest/tests/unit/utils/test_openstack_utils.py

   1 #!/usr/bin/env python
   2
   3 # All rights reserved. This program and the accompanying materials
   4 # are made available under the terms of the Apache License, Version 2.0
   5 # which accompanies this distribution, and is available at
   6 # http://www.apache.org/licenses/LICENSE-2.0
   7
   8 import copy
   9 import logging
  10 import os
  11 import unittest
  12
  13 import mock
  14
  15 from functest.utils import openstack_utils
  16
  17
  18 class OSUtilsTesting(unittest.TestCase):
  19
  20     logging.disable(logging.CRITICAL)
  21
  22     def _get_env_cred_dict(self, os_prefix=''):
  23         return {'OS_USERNAME': os_prefix + 'username',
  24                 'OS_PASSWORD': os_prefix + 'password',
  25                 'OS_AUTH_URL': os_prefix + 'auth_url',
  26                 'OS_TENANT_NAME': os_prefix + 'tenant_name',
  27                 'OS_USER_DOMAIN_NAME': os_prefix + 'user_domain_name',
  28                 'OS_PROJECT_DOMAIN_NAME': os_prefix + 'project_domain_name',
  29                 'OS_PROJECT_NAME': os_prefix + 'project_name',
  30                 'OS_ENDPOINT_TYPE': os_prefix + 'endpoint_type',
  31                 'OS_REGION_NAME': os_prefix + 'region_name'}
  32
  33     def _get_os_env_vars(self):
  34         return {'username': 'test_username', 'password': 'test_password',
  35                 'auth_url': 'test_auth_url', 'tenant_name': 'test_tenant_name',
  36                 'user_domain_name': 'test_user_domain_name',
  37                 'project_domain_name': 'test_project_domain_name',
  38                 'project_name': 'test_project_name',
  39                 'endpoint_type': 'test_endpoint_type',
  40                 'region_name': 'test_region_name'}
  41
  42     def setUp(self):
  43         self.env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD']
  44         self.tenant_name = 'test_tenant_name'
  45         self.env_cred_dict = self._get_env_cred_dict()
  46         self.os_environs = self._get_env_cred_dict(os_prefix='test_')
  47         self.os_env_vars = self._get_os_env_vars()
  48
  49         mock_obj = mock.Mock()
  50         attrs = {'name': 'test_flavor',
  51                  'id': 'flavor_id',
  52                  'ram': 2}
  53         mock_obj.configure_mock(**attrs)
  54         self.flavor = mock_obj
  55
  56         mock_obj = mock.Mock()
  57         attrs = {'name': 'test_aggregate',
  58                  'id': 'aggregate_id',
  59                  'hosts': ['host_name']}
  60         mock_obj.configure_mock(**attrs)
  61         self.aggregate = mock_obj
  62
  63         mock_obj = mock.Mock()
  64         attrs = {'id': 'instance_id',
  65                  'name': 'test_instance',
  66                  'status': 'ok'}
  67         mock_obj.configure_mock(**attrs)
  68         self.instance = mock_obj
  69
  70         mock_obj = mock.Mock()
  71         attrs = {'id': 'azone_id',
  72                  'zoneName': 'test_azone',
  73                  'status': 'ok'}
  74         mock_obj.configure_mock(**attrs)
  75         self.availability_zone = mock_obj
  76
  77         mock_obj = mock.Mock()
  78         attrs = {'id': 'floating_id',
  79                  'zoneName': 'test_floating_ip',
  80                  'status': 'ok'}
  81         mock_obj.configure_mock(**attrs)
  82         self.floating_ip = mock_obj
  83
  84         mock_obj = mock.Mock()
  85         attrs = {'id': 'hypervisor_id',
  86                  'hypervisor_hostname': 'test_hostname',
  87                  'state': 'up'}
  88         mock_obj.configure_mock(**attrs)
  89         self.hypervisor = mock_obj
  90
  91         mock_obj = mock.Mock()
  92         attrs = {'id': 'image_id',
  93                  'name': 'test_image'}
  94         mock_obj.configure_mock(**attrs)
  95         self.image = mock_obj
  96
  97         mock_obj = mock.Mock()
  98         self.mock_return = mock_obj
  99
 100         self.nova_client = mock.Mock()
 101         attrs = {'servers.list.return_value': [self.instance],
 102                  'servers.get.return_value': self.instance,
 103                  'servers.find.return_value': self.instance,
 104                  'servers.create.return_value': self.instance,
 105                  'flavors.list.return_value': [self.flavor],
 106                  'flavors.find.return_value': self.flavor,
 107                  'servers.add_floating_ip.return_value': mock.Mock(),
 108                  'servers.force_delete.return_value': mock.Mock(),
 109                  'aggregates.list.return_value': [self.aggregate],
 110                  'aggregates.add_host.return_value': mock.Mock(),
 111                  'aggregates.remove_host.return_value': mock.Mock(),
 112                  'aggregates.get.return_value': self.aggregate,
 113                  'aggregates.delete.return_value': mock.Mock(),
 114                  'availability_zones.list.return_value':
 115                  [self.availability_zone],
 116                  'floating_ips.list.return_value': [self.floating_ip],
 117                  'floating_ips.delete.return_value': mock.Mock(),
 118                  'hypervisors.list.return_value': [self.hypervisor],
 119                  'create.return_value': mock.Mock(),
 120                  'add_security_group.return_value': mock.Mock(),
 121                  'images.list.return_value': [self.image],
 122                  'images.delete.return_value': mock.Mock(),
 123                  }
 124         self.nova_client.configure_mock(**attrs)
 125
 126         self.glance_client = mock.Mock()
 127         attrs = {'images.list.return_value': [self.image],
 128                  'images.create.return_value': self.image,
 129                  'images.upload.return_value': mock.Mock()}
 130         self.glance_client.configure_mock(**attrs)
 131
 132         mock_obj = mock.Mock()
 133         attrs = {'id': 'volume_id',
 134                  'name': 'test_volume'}
 135         mock_obj.configure_mock(**attrs)
 136         self.volume = mock_obj
 137
 138         mock_obj = mock.Mock()
 139         attrs = {'id': 'volume_type_id',
 140                  'name': 'test_volume_type',
 141                  'is_public': True}
 142         mock_obj.configure_mock(**attrs)
 143         self.volume_types = [mock_obj]
 144
 145         mock_obj = mock.Mock()
 146         attrs = {'id': 'volume_type_id',
 147                  'name': 'test_volume_type',
 148                  'is_public': False}
 149         mock_obj.configure_mock(**attrs)
 150         self.volume_types.append(mock_obj)
 151
 152         self.cinder_client = mock.Mock()
 153         attrs = {'volumes.list.return_value': [self.volume],
 154                  'volume_types.list.return_value': self.volume_types,
 155                  'volume_types.create.return_value': self.volume_types[0],
 156                  'volume_types.delete.return_value': mock.Mock(),
 157                  'quotas.update.return_value': mock.Mock(),
 158                  'volumes.detach.return_value': mock.Mock(),
 159                  'volumes.force_delete.return_value': mock.Mock(),
 160                  'volumes.delete.return_value': mock.Mock()
 161                  }
 162         self.cinder_client.configure_mock(**attrs)
 163
 164         self.resource = mock.Mock()
 165         attrs = {'id': 'resource_test_id',
 166                  'name': 'resource_test_name'
 167                  }
 168
 169         self.heat_client = mock.Mock()
 170         attrs = {'resources.get.return_value': self.resource}
 171         self.heat_client.configure_mock(**attrs)
 172
 173         mock_obj = mock.Mock()
 174         attrs = {'id': 'tenant_id',
 175                  'name': 'test_tenant'}
 176         mock_obj.configure_mock(**attrs)
 177         self.tenant = mock_obj
 178
 179         mock_obj = mock.Mock()
 180         attrs = {'id': 'user_id',
 181                  'name': 'test_user'}
 182         mock_obj.configure_mock(**attrs)
 183         self.user = mock_obj
 184
 185         mock_obj = mock.Mock()
 186         attrs = {'id': 'role_id',
 187                  'name': 'test_role'}
 188         mock_obj.configure_mock(**attrs)
 189         self.role = mock_obj
 190
 191         self.keystone_client = mock.Mock()
 192         attrs = {'projects.list.return_value': [self.tenant],
 193                  'tenants.list.return_value': [self.tenant],
 194                  'users.list.return_value': [self.user],
 195                  'roles.list.return_value': [self.role],
 196                  'projects.create.return_value': self.tenant,
 197                  'tenants.create.return_value': self.tenant,
 198                  'users.create.return_value': self.user,
 199                  'roles.grant.return_value': mock.Mock(),
 200                  'roles.add_user_role.return_value': mock.Mock(),
 201                  'projects.delete.return_value': mock.Mock(),
 202                  'tenants.delete.return_value': mock.Mock(),
 203                  'users.delete.return_value': mock.Mock(),
 204                  }
 205         self.keystone_client.configure_mock(**attrs)
 206
 207         self.router = {'id': 'router_id',
 208                        'name': 'test_router'}
 209
 210         self.subnet = {'id': 'subnet_id',
 211                        'name': 'test_subnet'}
 212
 213         self.networks = [{'id': 'network_id',
 214                           'name': 'test_network',
 215                           'router:external': False,
 216                           'shared': True,
 217                           'subnets': [self.subnet]},
 218                          {'id': 'network_id1',
 219                           'name': 'test_network1',
 220                           'router:external': True,
 221                           'shared': True,
 222                           'subnets': [self.subnet]}]
 223
 224         self.port = {'id': 'port_id',
 225                      'name': 'test_port'}
 226
 227         self.sec_group = {'id': 'sec_group_id',
 228                           'name': 'test_sec_group'}
 229
 230         self.neutron_floatingip = {'id': 'fip_id',
 231                                    'floating_ip_address': 'test_ip'}
 232         self.neutron_client = mock.Mock()
 233         attrs = {'list_networks.return_value': {'networks': self.networks},
 234                  'list_routers.return_value': {'routers': [self.router]},
 235                  'list_ports.return_value': {'ports': [self.port]},
 236                  'list_subnets.return_value': {'subnets': [self.subnet]},
 237                  'create_network.return_value': {'network': self.networks[0]},
 238                  'create_subnet.return_value': {'subnets': [self.subnet]},
 239                  'create_router.return_value': {'router': self.router},
 240                  'create_port.return_value': {'port': self.port},
 241                  'create_floatingip.return_value': {'floatingip':
 242                                                     self.neutron_floatingip},
 243                  'update_network.return_value': mock.Mock(),
 244                  'update_port.return_value': {'port': self.port},
 245                  'add_interface_router.return_value': mock.Mock(),
 246                  'add_gateway_router.return_value': mock.Mock(),
 247                  'delete_network.return_value': mock.Mock(),
 248                  'delete_subnet.return_value': mock.Mock(),
 249                  'delete_router.return_value': mock.Mock(),
 250                  'delete_port.return_value': mock.Mock(),
 251                  'remove_interface_router.return_value': mock.Mock(),
 252                  'remove_gateway_router.return_value': mock.Mock(),
 253                  'create_bgpvpn.return_value': self.mock_return,
 254                  'create_network_association.return_value': self.mock_return,
 255                  'create_router_association.return_value': self.mock_return,
 256                  'update_bgpvpn.return_value': self.mock_return,
 257                  'delete_bgpvpn.return_value': self.mock_return,
 258                  'show_bgpvpn.return_value': self.mock_return,
 259                  'list_security_groups.return_value': {'security_groups':
 260                                                        [self.sec_group]},
 261                  'create_security_group_rule.return_value': mock.Mock(),
 262                  'create_security_group.return_value': {'security_group':
 263                                                         self.sec_group},
 264                  'update_quota.return_value': mock.Mock(),
 265                  'delete_security_group.return_value': mock.Mock()
 266                  }
 267         self.neutron_client.configure_mock(**attrs)
 268
 269         self.empty_client = mock.Mock()
 270         attrs = {'list_networks.return_value': {'networks': []},
 271                  'list_routers.return_value': {'routers': []},
 272                  'list_ports.return_value': {'ports': []},
 273                  'list_subnets.return_value': {'subnets': []}}
 274         self.empty_client.configure_mock(**attrs)
 275
 276     @mock.patch('functest.utils.openstack_utils.os.getenv',
 277                 return_value=None)
 278     def test_is_keystone_v3_missing_identity(self, mock_os_getenv):
 279         self.assertEqual(openstack_utils.is_keystone_v3(), False)
 280
 281     @mock.patch('functest.utils.openstack_utils.os.getenv',
 282                 return_value='3')
 283     def test_is_keystone_v3_default(self, mock_os_getenv):
 284         self.assertEqual(openstack_utils.is_keystone_v3(), True)
 285
 286     @mock.patch('functest.utils.openstack_utils.is_keystone_v3',
 287                 return_value=False)
 288     def test_get_rc_env_vars_missing_identity(self, mock_get_rc_env):
 289         exp_resp = self.env_vars
 290         exp_resp.extend(['OS_TENANT_NAME'])
 291         self.assertEqual(openstack_utils.get_rc_env_vars(), exp_resp)
 292
 293     @mock.patch('functest.utils.openstack_utils.is_keystone_v3',
 294                 return_value=True)
 295     def test_get_rc_env_vars_default(self, mock_get_rc_env):
 296         exp_resp = self.env_vars
 297         exp_resp.extend(['OS_PROJECT_NAME',
 298                          'OS_USER_DOMAIN_NAME',
 299                          'OS_PROJECT_DOMAIN_NAME'])
 300         self.assertEqual(openstack_utils.get_rc_env_vars(), exp_resp)
 301
 302     @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
 303     def test_check_credentials_missing_env(self, mock_get_rc_env):
 304         exp_resp = self.env_vars
 305         exp_resp.extend(['OS_TENANT_NAME'])
 306         mock_get_rc_env.return_value = exp_resp
 307         with mock.patch.dict('functest.utils.openstack_utils.os.environ', {},
 308                              clear=True):
 309             self.assertEqual(openstack_utils.check_credentials(), False)
 310
 311     @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
 312     def test_check_credentials_default(self, mock_get_rc_env):
 313         exp_resp = ['OS_TENANT_NAME']
 314         mock_get_rc_env.return_value = exp_resp
 315         with mock.patch.dict('functest.utils.openstack_utils.os.environ',
 316                              {'OS_TENANT_NAME': self.tenant_name},
 317                              clear=True):
 318             self.assertEqual(openstack_utils.check_credentials(), True)
 319
 320     def test_get_env_cred_dict(self):
 321         self.assertDictEqual(openstack_utils.get_env_cred_dict(),
 322                              self.env_cred_dict)
 323
 324     @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
 325     def test_get_credentials_default(self, mock_get_rc_env):
 326         mock_get_rc_env.return_value = self.env_cred_dict.keys()
 327         with mock.patch.dict('functest.utils.openstack_utils.os.environ',
 328                              self.os_environs,
 329                              clear=True):
 330             self.assertDictEqual(openstack_utils.get_credentials(),
 331                                  self.os_env_vars)
 332
 333     def _get_credentials_missing_env(self, var):
 334         dic = copy.deepcopy(self.os_environs)
 335         dic.pop(var)
 336         with mock.patch('functest.utils.openstack_utils.get_rc_env_vars',
 337                         return_value=self.env_cred_dict.keys()), \
 338                 mock.patch.dict('functest.utils.openstack_utils.os.environ',
 339                                 dic,
 340                                 clear=True):
 341             self.assertRaises(openstack_utils.MissingEnvVar,
 342                               lambda: openstack_utils.get_credentials())
 343
 344     def test_get_credentials_missing_username(self):
 345         self._get_credentials_missing_env('OS_USERNAME')
 346
 347     def test_get_credentials_missing_password(self):
 348         self._get_credentials_missing_env('OS_PASSWORD')
 349
 350     def test_get_credentials_missing_auth_url(self):
 351         self._get_credentials_missing_env('OS_AUTH_URL')
 352
 353     def test_get_credentials_missing_tenantname(self):
 354         self._get_credentials_missing_env('OS_TENANT_NAME')
 355
 356     def test_get_credentials_missing_domainname(self):
 357         self._get_credentials_missing_env('OS_USER_DOMAIN_NAME')
 358
 359     def test_get_credentials_missing_projectname(self):
 360         self._get_credentials_missing_env('OS_PROJECT_NAME')
 361
 362     def test_get_credentials_missing_endpoint_type(self):
 363         self._get_credentials_missing_env('OS_ENDPOINT_TYPE')
 364
 365     def _test_source_credentials(self, msg, key='OS_TENANT_NAME',
 366                                  value='admin'):
 367         try:
 368             del os.environ[key]
 369         except:
 370             pass
 371         f = 'rc_file'
 372         with mock.patch('__builtin__.open', mock.mock_open(read_data=msg),
 373                         create=True) as m:
 374             m.return_value.__iter__ = lambda self: iter(self.readline, '')
 375             openstack_utils.source_credentials(f)
 376             m.assert_called_once_with(f, 'r')
 377             self.assertEqual(os.environ[key], value)
 378
 379     def test_source_credentials(self):
 380         self._test_source_credentials('OS_TENANT_NAME=admin')
 381         self._test_source_credentials('OS_TENANT_NAME= admin')
 382         self._test_source_credentials('OS_TENANT_NAME = admin')
 383         self._test_source_credentials('OS_TENANT_NAME = "admin"')
 384         self._test_source_credentials('export OS_TENANT_NAME=admin')
 385         self._test_source_credentials('export OS_TENANT_NAME =admin')
 386         self._test_source_credentials('export OS_TENANT_NAME = admin')
 387         self._test_source_credentials('export OS_TENANT_NAME = "admin"')
 388         self._test_source_credentials('OS_TENANT_NAME', value='')
 389         self._test_source_credentials('export OS_TENANT_NAME', value='')
 390         # This test will fail as soon as rc_file is fixed
 391         self._test_source_credentials(
 392             'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
 393
 394     @mock.patch('functest.utils.openstack_utils.os.getenv',
 395                 return_value=None)
 396     def test_get_keystone_client_version_missing_env(self, mock_os_getenv):
 397         self.assertEqual(openstack_utils.get_keystone_client_version(),
 398                          openstack_utils.DEFAULT_API_VERSION)
 399
 400     @mock.patch('functest.utils.openstack_utils.logger.info')
 401     @mock.patch('functest.utils.openstack_utils.os.getenv',
 402                 return_value='3')
 403     def test_get_keystone_client_version_default(self, mock_os_getenv,
 404                                                  mock_logger_info):
 405         self.assertEqual(openstack_utils.get_keystone_client_version(),
 406                          '3')
 407         mock_logger_info.assert_called_once_with("OS_IDENTITY_API_VERSION is "
 408                                                  "set in env as '%s'", '3')
 409
 410     def test_get_keystone_client(self):
 411         mock_keystone_obj = mock.Mock()
 412         mock_session_obj = mock.Mock()
 413         with mock.patch('functest.utils.openstack_utils'
 414                         '.get_keystone_client_version', return_value='3'), \
 415             mock.patch('functest.utils.openstack_utils'
 416                        '.keystoneclient.Client',
 417                        return_value=mock_keystone_obj) \
 418             as mock_key_client, \
 419             mock.patch('functest.utils.openstack_utils.get_session',
 420                        return_value=mock_session_obj):
 421             self.assertEqual(openstack_utils.get_keystone_client(),
 422                              mock_keystone_obj)
 423             mock_key_client.assert_called_once_with('3',
 424                                                     session=mock_session_obj)
 425
 426     @mock.patch('functest.utils.openstack_utils.os.getenv',
 427                 return_value=None)
 428     def test_get_nova_client_version_missing_env(self, mock_os_getenv):
 429         self.assertEqual(openstack_utils.get_nova_client_version(),
 430                          openstack_utils.DEFAULT_API_VERSION)
 431
 432     @mock.patch('functest.utils.openstack_utils.logger.info')
 433     @mock.patch('functest.utils.openstack_utils.os.getenv',
 434                 return_value='3')
 435     def test_get_nova_client_version_default(self, mock_os_getenv,
 436                                              mock_logger_info):
 437         self.assertEqual(openstack_utils.get_nova_client_version(),
 438                          '3')
 439         mock_logger_info.assert_called_once_with("OS_COMPUTE_API_VERSION is "
 440                                                  "set in env as '%s'", '3')
 441
 442     def test_get_nova_client(self):
 443         mock_nova_obj = mock.Mock()
 444         mock_session_obj = mock.Mock()
 445         with mock.patch('functest.utils.openstack_utils'
 446                         '.get_nova_client_version', return_value='3'), \
 447             mock.patch('functest.utils.openstack_utils'
 448                        '.novaclient.Client',
 449                        return_value=mock_nova_obj) \
 450             as mock_nova_client, \
 451             mock.patch('functest.utils.openstack_utils.get_session',
 452                        return_value=mock_session_obj):
 453             self.assertEqual(openstack_utils.get_nova_client(),
 454                              mock_nova_obj)
 455             mock_nova_client.assert_called_once_with('3',
 456                                                      session=mock_session_obj)
 457
 458     @mock.patch('functest.utils.openstack_utils.os.getenv',
 459                 return_value=None)
 460     def test_get_cinder_client_version_missing_env(self, mock_os_getenv):
 461         self.assertEqual(openstack_utils.get_cinder_client_version(),
 462                          openstack_utils.DEFAULT_API_VERSION)
 463
 464     @mock.patch('functest.utils.openstack_utils.logger.info')
 465     @mock.patch('functest.utils.openstack_utils.os.getenv',
 466                 return_value='3')
 467     def test_get_cinder_client_version_default(self, mock_os_getenv,
 468                                                mock_logger_info):
 469         self.assertEqual(openstack_utils.get_cinder_client_version(),
 470                          '3')
 471         mock_logger_info.assert_called_once_with("OS_VOLUME_API_VERSION is "
 472                                                  "set in env as '%s'", '3')
 473
 474     def test_get_cinder_client(self):
 475         mock_cinder_obj = mock.Mock()
 476         mock_session_obj = mock.Mock()
 477         with mock.patch('functest.utils.openstack_utils'
 478                         '.get_cinder_client_version', return_value='3'), \
 479             mock.patch('functest.utils.openstack_utils'
 480                        '.cinderclient.Client',
 481                        return_value=mock_cinder_obj) \
 482             as mock_cind_client, \
 483             mock.patch('functest.utils.openstack_utils.get_session',
 484                        return_value=mock_session_obj):
 485             self.assertEqual(openstack_utils.get_cinder_client(),
 486                              mock_cinder_obj)
 487             mock_cind_client.assert_called_once_with('3',
 488                                                      session=mock_session_obj)
 489
 490     @mock.patch('functest.utils.openstack_utils.os.getenv',
 491                 return_value=None)
 492     def test_get_neutron_client_version_missing_env(self, mock_os_getenv):
 493         self.assertEqual(openstack_utils.get_neutron_client_version(),
 494                          openstack_utils.DEFAULT_API_VERSION)
 495
 496     @mock.patch('functest.utils.openstack_utils.logger.info')
 497     @mock.patch('functest.utils.openstack_utils.os.getenv',
 498                 return_value='3')
 499     def test_get_neutron_client_version_default(self, mock_os_getenv,
 500                                                 mock_logger_info):
 501         self.assertEqual(openstack_utils.get_neutron_client_version(),
 502                          '3')
 503         mock_logger_info.assert_called_once_with("OS_NETWORK_API_VERSION is "
 504                                                  "set in env as '%s'", '3')
 505
 506     def test_get_neutron_client(self):
 507         mock_neutron_obj = mock.Mock()
 508         mock_session_obj = mock.Mock()
 509         with mock.patch('functest.utils.openstack_utils'
 510                         '.get_neutron_client_version', return_value='3'), \
 511             mock.patch('functest.utils.openstack_utils'
 512                        '.neutronclient.Client',
 513                        return_value=mock_neutron_obj) \
 514             as mock_neut_client, \
 515             mock.patch('functest.utils.openstack_utils.get_session',
 516                        return_value=mock_session_obj):
 517             self.assertEqual(openstack_utils.get_neutron_client(),
 518                              mock_neutron_obj)
 519             mock_neut_client.assert_called_once_with('3',
 520                                                      session=mock_session_obj)
 521
 522     @mock.patch('functest.utils.openstack_utils.os.getenv',
 523                 return_value=None)
 524     def test_get_glance_client_version_missing_env(self, mock_os_getenv):
 525         self.assertEqual(openstack_utils.get_glance_client_version(),
 526                          openstack_utils.DEFAULT_API_VERSION)
 527
 528     @mock.patch('functest.utils.openstack_utils.logger.info')
 529     @mock.patch('functest.utils.openstack_utils.os.getenv',
 530                 return_value='3')
 531     def test_get_glance_client_version_default(self, mock_os_getenv,
 532                                                mock_logger_info):
 533         self.assertEqual(openstack_utils.get_glance_client_version(),
 534                          '3')
 535         mock_logger_info.assert_called_once_with("OS_IMAGE_API_VERSION is "
 536                                                  "set in env as '%s'", '3')
 537
 538     def test_get_glance_client(self):
 539         mock_glance_obj = mock.Mock()
 540         mock_session_obj = mock.Mock()
 541         with mock.patch('functest.utils.openstack_utils'
 542                         '.get_glance_client_version', return_value='3'), \
 543             mock.patch('functest.utils.openstack_utils'
 544                        '.glanceclient.Client',
 545                        return_value=mock_glance_obj) \
 546             as mock_glan_client, \
 547             mock.patch('functest.utils.openstack_utils.get_session',
 548                        return_value=mock_session_obj):
 549             self.assertEqual(openstack_utils.get_glance_client(),
 550                              mock_glance_obj)
 551             mock_glan_client.assert_called_once_with('3',
 552                                                      session=mock_session_obj)
 553
 554     @mock.patch('functest.utils.openstack_utils.os.getenv',
 555                 return_value=None)
 556     def test_get_heat_client_version_missing_env(self, mock_os_getenv):
 557         self.assertEqual(openstack_utils.get_heat_client_version(),
 558                          openstack_utils.DEFAULT_HEAT_API_VERSION)
 559
 560     @mock.patch('functest.utils.openstack_utils.logger.info')
 561     @mock.patch('functest.utils.openstack_utils.os.getenv', return_value='1')
 562     def test_get_heat_client_version_default(self, mock_os_getenv,
 563                                              mock_logger_info):
 564         self.assertEqual(openstack_utils.get_heat_client_version(), '1')
 565         mock_logger_info.assert_called_once_with(
 566             "OS_ORCHESTRATION_API_VERSION is set in env as '%s'", '1')
 567
 568     def test_get_heat_client(self):
 569         mock_heat_obj = mock.Mock()
 570         mock_session_obj = mock.Mock()
 571         with mock.patch('functest.utils.openstack_utils'
 572                         '.get_heat_client_version', return_value='1'), \
 573             mock.patch('functest.utils.openstack_utils'
 574                        '.heatclient.Client',
 575                        return_value=mock_heat_obj) \
 576             as mock_heat_client, \
 577             mock.patch('functest.utils.openstack_utils.get_session',
 578                        return_value=mock_session_obj):
 579             self.assertEqual(openstack_utils.get_heat_client(),
 580                              mock_heat_obj)
 581             mock_heat_client.assert_called_once_with('1',
 582                                                      session=mock_session_obj)
 583
 584     def test_get_instances_default(self):
 585         self.assertEqual(openstack_utils.get_instances(self.nova_client),
 586                          [self.instance])
 587
 588     @mock.patch('functest.utils.openstack_utils.logger.error')
 589     def test_get_instances_exception(self, mock_logger_error):
 590             self.assertEqual(openstack_utils.
 591                              get_instances(Exception),
 592                              None)
 593             self.assertTrue(mock_logger_error.called)
 594
 595     def test_get_instance_status_default(self):
 596         self.assertEqual(openstack_utils.get_instance_status(self.nova_client,
 597                                                              self.instance),
 598                          'ok')
 599
 600     @mock.patch('functest.utils.openstack_utils.logger.error')
 601     def test_get_instance_status_exception(self, mock_logger_error):
 602             self.assertEqual(openstack_utils.
 603                              get_instance_status(Exception,
 604                                                  self.instance),
 605                              None)
 606             self.assertTrue(mock_logger_error.called)
 607
 608     def test_get_instance_by_name_default(self):
 609         self.assertEqual(openstack_utils.
 610                          get_instance_by_name(self.nova_client,
 611                                               'test_instance'),
 612                          self.instance)
 613
 614     @mock.patch('functest.utils.openstack_utils.logger.error')
 615     def test_get_instance_by_name_exception(self, mock_logger_error):
 616             self.assertEqual(openstack_utils.
 617                              get_instance_by_name(Exception,
 618                                                   'test_instance'),
 619                              None)
 620             self.assertTrue(mock_logger_error.called)
 621
 622     def test_get_flavor_id_default(self):
 623         self.assertEqual(openstack_utils.
 624                          get_flavor_id(self.nova_client,
 625                                        'test_flavor'),
 626                          self.flavor.id)
 627
 628     def test_get_flavor_id_by_ram_range_default(self):
 629         self.assertEqual(openstack_utils.
 630                          get_flavor_id_by_ram_range(self.nova_client,
 631                                                     1, 3),
 632                          self.flavor.id)
 633
 634     def test_get_aggregates_default(self):
 635         self.assertEqual(openstack_utils.
 636                          get_aggregates(self.nova_client),
 637                          [self.aggregate])
 638
 639     @mock.patch('functest.utils.openstack_utils.logger.error')
 640     def test_get_aggregates_exception(self, mock_logger_error):
 641             self.assertEqual(openstack_utils.
 642                              get_aggregates(Exception),
 643                              None)
 644             self.assertTrue(mock_logger_error.called)
 645
 646     def test_get_aggregate_id_default(self):
 647         with mock.patch('functest.utils.openstack_utils.get_aggregates',
 648                         return_value=[self.aggregate]):
 649             self.assertEqual(openstack_utils.
 650                              get_aggregate_id(self.nova_client,
 651                                               'test_aggregate'),
 652                              'aggregate_id')
 653
 654     @mock.patch('functest.utils.openstack_utils.logger.error')
 655     def test_get_aggregate_id_exception(self, mock_logger_error):
 656             self.assertEqual(openstack_utils.
 657                              get_aggregate_id(Exception,
 658                                               'test_aggregate'),
 659                              None)
 660             self.assertTrue(mock_logger_error.called)
 661
 662     def test_get_availability_zone_names_default(self):
 663         with mock.patch('functest.utils.openstack_utils'
 664                         '.get_availability_zones',
 665                         return_value=[self.availability_zone]):
 666             self.assertEqual(openstack_utils.
 667                              get_availability_zone_names(self.nova_client),
 668                              ['test_azone'])
 669
 670     @mock.patch('functest.utils.openstack_utils.logger.error')
 671     def test_get_availability_zone_names_exception(self, mock_logger_error):
 672             self.assertEqual(openstack_utils.
 673                              get_availability_zone_names(Exception),
 674                              None)
 675             self.assertTrue(mock_logger_error.called)
 676
 677     def test_get_availability_zones_default(self):
 678             self.assertEqual(openstack_utils.
 679                              get_availability_zones(self.nova_client),
 680                              [self.availability_zone])
 681
 682     @mock.patch('functest.utils.openstack_utils.logger.error')
 683     def test_get_availability_zones_exception(self, mock_logger_error):
 684             self.assertEqual(openstack_utils.
 685                              get_availability_zones(Exception),
 686                              None)
 687             self.assertTrue(mock_logger_error.called)
 688
 689     def test_get_floating_ips_default(self):
 690             self.assertEqual(openstack_utils.
 691                              get_floating_ips(self.nova_client),
 692                              [self.floating_ip])
 693
 694     @mock.patch('functest.utils.openstack_utils.logger.error')
 695     def test_get_floating_ips_exception(self, mock_logger_error):
 696             self.assertEqual(openstack_utils.
 697                              get_floating_ips(Exception),
 698                              None)
 699             self.assertTrue(mock_logger_error.called)
 700
 701     def test_get_hypervisors_default(self):
 702             self.assertEqual(openstack_utils.
 703                              get_hypervisors(self.nova_client),
 704                              ['test_hostname'])
 705
 706     @mock.patch('functest.utils.openstack_utils.logger.error')
 707     def test_get_hypervisors_exception(self, mock_logger_error):
 708             self.assertEqual(openstack_utils.
 709                              get_hypervisors(Exception),
 710                              None)
 711             self.assertTrue(mock_logger_error.called)
 712
 713     def test_create_aggregate_default(self):
 714             self.assertTrue(openstack_utils.
 715                             create_aggregate(self.nova_client,
 716                                              'test_aggregate',
 717                                              'azone'))
 718
 719     @mock.patch('functest.utils.openstack_utils.logger.error')
 720     def test_create_aggregate_exception(self, mock_logger_error):
 721             self.assertEqual(openstack_utils.
 722                              create_aggregate(Exception,
 723                                               'test_aggregate',
 724                                               'azone'),
 725                              None)
 726             self.assertTrue(mock_logger_error.called)
 727
 728     def test_add_host_to_aggregate_default(self):
 729         with mock.patch('functest.utils.openstack_utils.get_aggregate_id'):
 730                 self.assertTrue(openstack_utils.
 731                                 add_host_to_aggregate(self.nova_client,
 732                                                       'test_aggregate',
 733                                                       'test_hostname'))
 734
 735     @mock.patch('functest.utils.openstack_utils.logger.error')
 736     def test_add_host_to_aggregate_exception(self, mock_logger_error):
 737             self.assertEqual(openstack_utils.
 738                              add_host_to_aggregate(Exception,
 739                                                    'test_aggregate',
 740                                                    'test_hostname'),
 741                              None)
 742             self.assertTrue(mock_logger_error.called)
 743
 744     def test_create_aggregate_with_host_default(self):
 745         with mock.patch('functest.utils.openstack_utils.create_aggregate'), \
 746                 mock.patch('functest.utils.openstack_utils.'
 747                            'add_host_to_aggregate'):
 748             self.assertTrue(openstack_utils.
 749                             create_aggregate_with_host(self.nova_client,
 750                                                        'test_aggregate',
 751                                                        'test_azone',
 752                                                        'test_hostname'))
 753
 754     @mock.patch('functest.utils.openstack_utils.logger.error')
 755     def test_create_aggregate_with_host_exception(self, mock_logger_error):
 756             with mock.patch('functest.utils.openstack_utils.create_aggregate',
 757                             side_effect=Exception):
 758                 self.assertEqual(openstack_utils.
 759                                  create_aggregate_with_host(Exception,
 760                                                             'test_aggregate',
 761                                                             'test_azone',
 762                                                             'test_hostname'),
 763                                  None)
 764                 self.assertTrue(mock_logger_error.called)
 765
 766     def test_create_instance_default(self):
 767         with mock.patch('functest.utils.openstack_utils.'
 768                         'get_nova_client',
 769                         return_value=self.nova_client):
 770             self.assertEqual(openstack_utils.
 771                              create_instance('test_flavor',
 772                                              'image_id',
 773                                              'network_id'),
 774                              self.instance)
 775
 776     @mock.patch('functest.utils.openstack_utils.logger.error')
 777     def test_create_instance_exception(self, mock_logger_error):
 778         with mock.patch('functest.utils.openstack_utils.'
 779                         'get_nova_client',
 780                         return_value=self.nova_client):
 781             self.nova_client.flavors.find.side_effect = Exception
 782             self.assertEqual(openstack_utils.
 783                              create_instance('test_flavor',
 784                                              'image_id',
 785                                              'network_id'),
 786                              None)
 787             self.assertTrue(mock_logger_error)
 788
 789     def test_create_floating_ip_default(self):
 790         with mock.patch('functest.utils.openstack_utils.'
 791                         'get_external_net_id',
 792                         return_value='external_net_id'):
 793             exp_resp = {'fip_addr': 'test_ip', 'fip_id': 'fip_id'}
 794             self.assertEqual(openstack_utils.
 795                              create_floating_ip(self.neutron_client),
 796                              exp_resp)
 797
 798     @mock.patch('functest.utils.openstack_utils.logger.error')
 799     def test_create_floating_ip_exception(self, mock_logger_error):
 800         with mock.patch('functest.utils.openstack_utils.'
 801                         'get_external_net_id',
 802                         return_value='external_net_id'):
 803             self.assertEqual(openstack_utils.
 804                              create_floating_ip(Exception),
 805                              None)
 806             self.assertTrue(mock_logger_error)
 807
 808     def test_add_floating_ip_default(self):
 809         with mock.patch('functest.utils.openstack_utils.get_aggregate_id'):
 810                 self.assertTrue(openstack_utils.
 811                                 add_floating_ip(self.nova_client,
 812                                                 'test_serverid',
 813                                                 'test_floatingip_addr'))
 814
 815     @mock.patch('functest.utils.openstack_utils.logger.error')
 816     def test_add_floating_ip_exception(self, mock_logger_error):
 817             self.assertFalse(openstack_utils.
 818                              add_floating_ip(Exception,
 819                                              'test_serverid',
 820                                              'test_floatingip_addr'))
 821             self.assertTrue(mock_logger_error.called)
 822
 823     def test_delete_instance_default(self):
 824             self.assertTrue(openstack_utils.
 825                             delete_instance(self.nova_client,
 826                                             'instance_id'))
 827
 828     @mock.patch('functest.utils.openstack_utils.logger.error')
 829     def test_delete_instance_exception(self, mock_logger_error):
 830             self.assertFalse(openstack_utils.
 831                              delete_instance(Exception,
 832                                              'instance_id'))
 833             self.assertTrue(mock_logger_error.called)
 834
 835     def test_delete_floating_ip_default(self):
 836             self.assertTrue(openstack_utils.
 837                             delete_floating_ip(self.nova_client,
 838                                                'floating_ip_id'))
 839
 840     @mock.patch('functest.utils.openstack_utils.logger.error')
 841     def test_delete_floating_ip_exception(self, mock_logger_error):
 842             self.assertFalse(openstack_utils.
 843                              delete_floating_ip(Exception,
 844                                                 'floating_ip_id'))
 845             self.assertTrue(mock_logger_error.called)
 846
 847     def test_remove_host_from_aggregate_default(self):
 848         with mock.patch('functest.utils.openstack_utils.'
 849                         'get_aggregate_id'):
 850                 self.assertTrue(openstack_utils.
 851                                 remove_host_from_aggregate(self.nova_client,
 852                                                            'agg_name',
 853                                                            'host_name'))
 854
 855     @mock.patch('functest.utils.openstack_utils.logger.error')
 856     def test_remove_host_from_aggregate_exception(self, mock_logger_error):
 857         with mock.patch('functest.utils.openstack_utils.'
 858                         'get_aggregate_id', side_effect=Exception):
 859             self.assertFalse(openstack_utils.
 860                              remove_host_from_aggregate(self.nova_client,
 861                                                         'agg_name',
 862                                                         'host_name'))
 863             self.assertTrue(mock_logger_error.called)
 864
 865     def test_remove_hosts_from_aggregate_default(self):
 866         with mock.patch('functest.utils.openstack_utils.'
 867                         'get_aggregate_id'), \
 868             mock.patch('functest.utils.openstack_utils.'
 869                        'remove_host_from_aggregate',
 870                        return_value=True) \
 871                 as mock_method:
 872             openstack_utils.remove_hosts_from_aggregate(self.nova_client,
 873                                                         'test_aggregate')
 874             mock_method.assert_any_call(self.nova_client,
 875                                         'test_aggregate',
 876                                         'host_name')
 877
 878     def test_delete_aggregate_default(self):
 879         with mock.patch('functest.utils.openstack_utils.'
 880                         'remove_hosts_from_aggregate'):
 881                 self.assertTrue(openstack_utils.
 882                                 delete_aggregate(self.nova_client,
 883                                                  'agg_name'))
 884
 885     @mock.patch('functest.utils.openstack_utils.logger.error')
 886     def test_delete_aggregate_exception(self, mock_logger_error):
 887         with mock.patch('functest.utils.openstack_utils.'
 888                         'remove_hosts_from_aggregate', side_effect=Exception):
 889             self.assertFalse(openstack_utils.
 890                              delete_aggregate(self.nova_client,
 891                                               'agg_name'))
 892             self.assertTrue(mock_logger_error.called)
 893
 894     def test_get_network_list_default(self):
 895         self.assertEqual(openstack_utils.
 896                          get_network_list(self.neutron_client),
 897                          self.networks)
 898
 899     def test_get_network_list_missing_network(self):
 900         self.assertEqual(openstack_utils.
 901                          get_network_list(self.empty_client),
 902                          None)
 903
 904     def test_get_router_list_default(self):
 905         self.assertEqual(openstack_utils.
 906                          get_router_list(self.neutron_client),
 907                          [self.router])
 908
 909     def test_get_router_list_missing_router(self):
 910         self.assertEqual(openstack_utils.
 911                          get_router_list(self.empty_client),
 912                          None)
 913
 914     def test_get_port_list_default(self):
 915         self.assertEqual(openstack_utils.
 916                          get_port_list(self.neutron_client),
 917                          [self.port])
 918
 919     def test_get_port_list_missing_port(self):
 920         self.assertEqual(openstack_utils.
 921                          get_port_list(self.empty_client),
 922                          None)
 923
 924     def test_get_network_id_default(self):
 925         self.assertEqual(openstack_utils.
 926                          get_network_id(self.neutron_client,
 927                                         'test_network'),
 928                          'network_id')
 929
 930     def test_get_subnet_id_default(self):
 931         self.assertEqual(openstack_utils.
 932                          get_subnet_id(self.neutron_client,
 933                                        'test_subnet'),
 934                          'subnet_id')
 935
 936     def test_get_router_id_default(self):
 937         self.assertEqual(openstack_utils.
 938                          get_router_id(self.neutron_client,
 939                                        'test_router'),
 940                          'router_id')
 941
 942     def test_get_private_net_default(self):
 943         self.assertEqual(openstack_utils.
 944                          get_private_net(self.neutron_client),
 945                          self.networks[0])
 946
 947     def test_get_private_net_missing_net(self):
 948         self.assertEqual(openstack_utils.
 949                          get_private_net(self.empty_client),
 950                          None)
 951
 952     def test_get_external_net_default(self):
 953         self.assertEqual(openstack_utils.
 954                          get_external_net(self.neutron_client),
 955                          'test_network1')
 956
 957     def test_get_external_net_missing_net(self):
 958         self.assertEqual(openstack_utils.
 959                          get_external_net(self.empty_client),
 960                          None)
 961
 962     def test_get_external_net_id_default(self):
 963         self.assertEqual(openstack_utils.
 964                          get_external_net_id(self.neutron_client),
 965                          'network_id1')
 966
 967     def test_get_external_net_id_missing_net(self):
 968         self.assertEqual(openstack_utils.
 969                          get_external_net_id(self.empty_client),
 970                          None)
 971
 972     def test_check_neutron_net_default(self):
 973         self.assertTrue(openstack_utils.
 974                         check_neutron_net(self.neutron_client,
 975                                           'test_network'))
 976
 977     def test_check_neutron_net_missing_net(self):
 978         self.assertFalse(openstack_utils.
 979                          check_neutron_net(self.empty_client,
 980                                            'test_network'))
 981
 982     def test_create_neutron_net_default(self):
 983             self.assertEqual(openstack_utils.
 984                              create_neutron_net(self.neutron_client,
 985                                                 'test_network'),
 986                              'network_id')
 987
 988     @mock.patch('functest.utils.openstack_utils.logger.error')
 989     def test_create_neutron_net_exception(self, mock_logger_error):
 990             self.assertEqual(openstack_utils.
 991                              create_neutron_net(Exception,
 992                                                 'test_network'),
 993                              None)
 994             self.assertTrue(mock_logger_error.called)
 995
 996     def test_create_neutron_subnet_default(self):
 997             self.assertEqual(openstack_utils.
 998                              create_neutron_subnet(self.neutron_client,
 999                                                    'test_subnet',
1000                                                    'test_cidr',
1001                                                    'network_id'),
1002                              'subnet_id')
1003
1004     @mock.patch('functest.utils.openstack_utils.logger.error')
1005     def test_create_neutron_subnet_exception(self, mock_logger_error):
1006             self.assertEqual(openstack_utils.
1007                              create_neutron_subnet(Exception,
1008                                                    'test_subnet',
1009                                                    'test_cidr',
1010                                                    'network_id'),
1011                              None)
1012             self.assertTrue(mock_logger_error.called)
1013
1014     def test_create_neutron_router_default(self):
1015             self.assertEqual(openstack_utils.
1016                              create_neutron_router(self.neutron_client,
1017                                                    'test_router'),
1018                              'router_id')
1019
1020     @mock.patch('functest.utils.openstack_utils.logger.error')
1021     def test_create_neutron_router_exception(self, mock_logger_error):
1022             self.assertEqual(openstack_utils.
1023                              create_neutron_router(Exception,
1024                                                    'test_router'),
1025                              None)
1026             self.assertTrue(mock_logger_error.called)
1027
1028     def test_create_neutron_port_default(self):
1029             self.assertEqual(openstack_utils.
1030                              create_neutron_port(self.neutron_client,
1031                                                  'test_port',
1032                                                  'network_id',
1033                                                  'test_ip'),
1034                              'port_id')
1035
1036     @mock.patch('functest.utils.openstack_utils.logger.error')
1037     def test_create_neutron_port_exception(self, mock_logger_error):
1038             self.assertEqual(openstack_utils.
1039                              create_neutron_port(Exception,
1040                                                  'test_port',
1041                                                  'network_id',
1042                                                  'test_ip'),
1043                              None)
1044             self.assertTrue(mock_logger_error.called)
1045
1046     def test_update_neutron_net_default(self):
1047             self.assertTrue(openstack_utils.
1048                             update_neutron_net(self.neutron_client,
1049                                                'network_id'))
1050
1051     @mock.patch('functest.utils.openstack_utils.logger.error')
1052     def test_update_neutron_net_exception(self, mock_logger_error):
1053             self.assertFalse(openstack_utils.
1054                              update_neutron_net(Exception,
1055                                                 'network_id'))
1056             self.assertTrue(mock_logger_error.called)
1057
1058     def test_update_neutron_port_default(self):
1059             self.assertEqual(openstack_utils.
1060                              update_neutron_port(self.neutron_client,
1061                                                  'port_id',
1062                                                  'test_owner'),
1063                              'port_id')
1064
1065     @mock.patch('functest.utils.openstack_utils.logger.error')
1066     def test_update_neutron_port_exception(self, mock_logger_error):
1067             self.assertEqual(openstack_utils.
1068                              update_neutron_port(Exception,
1069                                                  'port_id',
1070                                                  'test_owner'),
1071                              None)
1072             self.assertTrue(mock_logger_error.called)
1073
1074     def test_add_interface_router_default(self):
1075             self.assertTrue(openstack_utils.
1076                             add_interface_router(self.neutron_client,
1077                                                  'router_id',
1078                                                  'subnet_id'))
1079
1080     @mock.patch('functest.utils.openstack_utils.logger.error')
1081     def test_add_interface_router_exception(self, mock_logger_error):
1082             self.assertFalse(openstack_utils.
1083                              add_interface_router(Exception,
1084                                                   'router_id',
1085                                                   'subnet_id'))
1086             self.assertTrue(mock_logger_error.called)
1087
1088     def test_add_gateway_router_default(self):
1089         with mock.patch('functest.utils.openstack_utils.'
1090                         'get_external_net_id',
1091                         return_value='network_id'):
1092                 self.assertTrue(openstack_utils.
1093                                 add_gateway_router(self.neutron_client,
1094                                                    'router_id'))
1095
1096     @mock.patch('functest.utils.openstack_utils.logger.error')
1097     def test_add_gateway_router_exception(self, mock_logger_error):
1098         with mock.patch('functest.utils.openstack_utils.'
1099                         'get_external_net_id',
1100                         return_value='network_id'):
1101                 self.assertFalse(openstack_utils.
1102                                  add_gateway_router(Exception,
1103                                                     'router_id'))
1104                 self.assertTrue(mock_logger_error.called)
1105
1106     def test_delete_neutron_net_default(self):
1107             self.assertTrue(openstack_utils.
1108                             delete_neutron_net(self.neutron_client,
1109                                                'network_id'))
1110
1111     @mock.patch('functest.utils.openstack_utils.logger.error')
1112     def test_delete_neutron_net_exception(self, mock_logger_error):
1113             self.assertFalse(openstack_utils.
1114                              delete_neutron_net(Exception,
1115                                                 'network_id'))
1116             self.assertTrue(mock_logger_error.called)
1117
1118     def test_delete_neutron_subnet_default(self):
1119             self.assertTrue(openstack_utils.
1120                             delete_neutron_subnet(self.neutron_client,
1121                                                   'subnet_id'))
1122
1123     @mock.patch('functest.utils.openstack_utils.logger.error')
1124     def test_delete_neutron_subnet_exception(self, mock_logger_error):
1125             self.assertFalse(openstack_utils.
1126                              delete_neutron_subnet(Exception,
1127                                                    'subnet_id'))
1128             self.assertTrue(mock_logger_error.called)
1129
1130     def test_delete_neutron_router_default(self):
1131             self.assertTrue(openstack_utils.
1132                             delete_neutron_router(self.neutron_client,
1133                                                   'router_id'))
1134
1135     @mock.patch('functest.utils.openstack_utils.logger.error')
1136     def test_delete_neutron_router_exception(self, mock_logger_error):
1137             self.assertFalse(openstack_utils.
1138                              delete_neutron_router(Exception,
1139                                                    'router_id'))
1140             self.assertTrue(mock_logger_error.called)
1141
1142     def test_delete_neutron_port_default(self):
1143             self.assertTrue(openstack_utils.
1144                             delete_neutron_port(self.neutron_client,
1145                                                 'port_id'))
1146
1147     @mock.patch('functest.utils.openstack_utils.logger.error')
1148     def test_delete_neutron_port_exception(self, mock_logger_error):
1149             self.assertFalse(openstack_utils.
1150                              delete_neutron_port(Exception,
1151                                                  'port_id'))
1152             self.assertTrue(mock_logger_error.called)
1153
1154     def test_remove_interface_router_default(self):
1155             self.assertTrue(openstack_utils.
1156                             remove_interface_router(self.neutron_client,
1157                                                     'router_id',
1158                                                     'subnet_id'))
1159
1160     @mock.patch('functest.utils.openstack_utils.logger.error')
1161     def test_remove_interface_router_exception(self, mock_logger_error):
1162             self.assertFalse(openstack_utils.
1163                              remove_interface_router(Exception,
1164                                                      'router_id',
1165                                                      'subnet_id'))
1166             self.assertTrue(mock_logger_error.called)
1167
1168     def test_remove_gateway_router_default(self):
1169             self.assertTrue(openstack_utils.
1170                             remove_gateway_router(self.neutron_client,
1171                                                   'router_id'))
1172
1173     @mock.patch('functest.utils.openstack_utils.logger.error')
1174     def test_remove_gateway_router_exception(self, mock_logger_error):
1175             self.assertFalse(openstack_utils.
1176                              remove_gateway_router(Exception,
1177                                                    'router_id'))
1178             self.assertTrue(mock_logger_error.called)
1179
1180     def test_create_bgpvpn(self):
1181         self.assertEqual(openstack_utils.
1182                          create_bgpvpn(self.neutron_client),
1183                          self.mock_return)
1184
1185     def test_create_network_association(self):
1186         self.assertEqual(openstack_utils.
1187                          create_network_association(self.neutron_client,
1188                                                     'bgpvpn_id',
1189                                                     'network_id'),
1190                          self.mock_return)
1191
1192     def test_create_router_association(self):
1193         self.assertEqual(openstack_utils.
1194                          create_router_association(self.neutron_client,
1195                                                    'bgpvpn_id',
1196                                                    'router_id'),
1197                          self.mock_return)
1198
1199     def test_update_bgpvpn(self):
1200         self.assertEqual(openstack_utils.
1201                          update_bgpvpn(self.neutron_client,
1202                                        'bgpvpn_id'),
1203                          self.mock_return)
1204
1205     def test_delete_bgpvpn(self):
1206         self.assertEqual(openstack_utils.
1207                          delete_bgpvpn(self.neutron_client,
1208                                        'bgpvpn_id'),
1209                          self.mock_return)
1210
1211     def test_get_bgpvpn(self):
1212         self.assertEqual(openstack_utils.
1213                          get_bgpvpn(self.neutron_client,
1214                                     'bgpvpn_id'),
1215                          self.mock_return)
1216
1217     def test_get_bgpvpn_routers(self):
1218         with mock.patch('functest.utils.openstack_utils.'
1219                         'get_bgpvpn',
1220                         return_value={'bgpvpn':
1221                                       {'routers': [self.router]}}):
1222             self.assertEqual(openstack_utils.
1223                              get_bgpvpn_routers(self.neutron_client,
1224                                                 'bgpvpn_id'),
1225                              [self.router])
1226
1227     def test_get_security_groups_default(self):
1228         self.assertEqual(openstack_utils.
1229                          get_security_groups(self.neutron_client),
1230                          [self.sec_group])
1231
1232     @mock.patch('functest.utils.openstack_utils.logger.error')
1233     def test_get_security_groups_exception(self, mock_logger_error):
1234         self.assertEqual(openstack_utils.
1235                          get_security_groups(Exception),
1236                          None)
1237         self.assertTrue(mock_logger_error.called)
1238
1239     def test_get_security_group_id_default(self):
1240         with mock.patch('functest.utils.openstack_utils.'
1241                         'get_security_groups',
1242                         return_value=[self.sec_group]):
1243             self.assertEqual(openstack_utils.
1244                              get_security_group_id(self.neutron_client,
1245                                                    'test_sec_group'),
1246                              'sec_group_id')
1247
1248     def test_create_security_group_default(self):
1249         self.assertEqual(openstack_utils.
1250                          create_security_group(self.neutron_client,
1251                                                'test_sec_group',
1252                                                'sec_group_desc'),
1253                          self.sec_group)
1254
1255     @mock.patch('functest.utils.openstack_utils.logger.error')
1256     def test_create_security_group_exception(self, mock_logger_error):
1257         self.assertEqual(openstack_utils.
1258                          create_security_group(Exception,
1259                                                'test_sec_group',
1260                                                'sec_group_desc'),
1261                          None)
1262         self.assertTrue(mock_logger_error.called)
1263
1264     def test_create_secgroup_rule_default(self):
1265         self.assertTrue(openstack_utils.
1266                         create_secgroup_rule(self.neutron_client,
1267                                              'sg_id',
1268                                              'direction',
1269                                              'protocol',
1270                                              80,
1271                                              80))
1272         self.assertTrue(openstack_utils.
1273                         create_secgroup_rule(self.neutron_client,
1274                                              'sg_id',
1275                                              'direction',
1276                                              'protocol'))
1277
1278     @mock.patch('functest.utils.openstack_utils.logger.error')
1279     def test_create_secgroup_rule_invalid_port_range(self, mock_logger_error):
1280         self.assertFalse(openstack_utils.
1281                          create_secgroup_rule(self.neutron_client,
1282                                               'sg_id',
1283                                               'direction',
1284                                               'protocol',
1285                                               80))
1286
1287     @mock.patch('functest.utils.openstack_utils.logger.error')
1288     def test_create_secgroup_rule_exception(self, mock_logger_error):
1289         self.assertFalse(openstack_utils.
1290                          create_secgroup_rule(Exception,
1291                                               'sg_id',
1292                                               'direction',
1293                                               'protocol'))
1294
1295     @mock.patch('functest.utils.openstack_utils.logger.info')
1296     def test_create_security_group_full_default(self, mock_logger_info):
1297         with mock.patch('functest.utils.openstack_utils.'
1298                         'get_security_group_id',
1299                         return_value='sg_id'):
1300             self.assertEqual(openstack_utils.
1301                              create_security_group_full(self.neutron_client,
1302                                                         'sg_name',
1303                                                         'sg_desc'),
1304                              'sg_id')
1305             self.assertTrue(mock_logger_info)
1306
1307     @mock.patch('functest.utils.openstack_utils.logger.info')
1308     @mock.patch('functest.utils.openstack_utils.logger.error')
1309     def test_create_security_group_full_sec_group_fail(self,
1310                                                        mock_logger_error,
1311                                                        mock_logger_info):
1312         with mock.patch('functest.utils.openstack_utils.'
1313                         'get_security_group_id',
1314                         return_value=''), \
1315             mock.patch('functest.utils.openstack_utils.'
1316                        'create_security_group',
1317                        return_value=False):
1318             self.assertEqual(openstack_utils.
1319                              create_security_group_full(self.neutron_client,
1320                                                         'sg_name',
1321                                                         'sg_desc'),
1322                              None)
1323             self.assertTrue(mock_logger_error)
1324             self.assertTrue(mock_logger_info)
1325
1326     @mock.patch('functest.utils.openstack_utils.logger.debug')
1327     @mock.patch('functest.utils.openstack_utils.logger.info')
1328     @mock.patch('functest.utils.openstack_utils.logger.error')
1329     def test_create_security_group_full_secgroup_rule_fail(self,
1330                                                            mock_logger_error,
1331                                                            mock_logger_info,
1332                                                            mock_logger_debug):
1333         with mock.patch('functest.utils.openstack_utils.'
1334                         'get_security_group_id',
1335                         return_value=''), \
1336             mock.patch('functest.utils.openstack_utils.'
1337                        'create_security_group',
1338                        return_value={'id': 'sg_id',
1339                                      'name': 'sg_name'}), \
1340             mock.patch('functest.utils.openstack_utils.'
1341                        'create_secgroup_rule',
1342                        return_value=False):
1343             self.assertEqual(openstack_utils.
1344                              create_security_group_full(self.neutron_client,
1345                                                         'sg_name',
1346                                                         'sg_desc'),
1347                              None)
1348             self.assertTrue(mock_logger_error)
1349             self.assertTrue(mock_logger_info)
1350             self.assertTrue(mock_logger_debug)
1351
1352     def test_add_secgroup_to_instance_default(self):
1353         self.assertTrue(openstack_utils.
1354                         add_secgroup_to_instance(self.nova_client,
1355                                                  'instance_id',
1356                                                  'sec_group_id'))
1357
1358     @mock.patch('functest.utils.openstack_utils.logger.error')
1359     def test_add_secgroup_to_instance_exception(self, mock_logger_error):
1360         self.assertFalse(openstack_utils.
1361                          add_secgroup_to_instance(Exception,
1362                                                   'instance_id',
1363                                                   'sec_group_id'))
1364         self.assertTrue(mock_logger_error.called)
1365
1366     def test_update_sg_quota_default(self):
1367         self.assertTrue(openstack_utils.
1368                         update_sg_quota(self.neutron_client,
1369                                         'tenant_id',
1370                                         'sg_quota',
1371                                         'sg_rule_quota'))
1372
1373     @mock.patch('functest.utils.openstack_utils.logger.error')
1374     def test_update_sg_quota_exception(self, mock_logger_error):
1375         self.assertFalse(openstack_utils.
1376                          update_sg_quota(Exception,
1377                                          'tenant_id',
1378                                          'sg_quota',
1379                                          'sg_rule_quota'))
1380         self.assertTrue(mock_logger_error.called)
1381
1382     def test_delete_security_group_default(self):
1383         self.assertTrue(openstack_utils.
1384                         delete_security_group(self.neutron_client,
1385                                               'sec_group_id'))
1386
1387     @mock.patch('functest.utils.openstack_utils.logger.error')
1388     def test_delete_security_group_exception(self, mock_logger_error):
1389         self.assertFalse(openstack_utils.
1390                          delete_security_group(Exception,
1391                                                'sec_group_id'))
1392         self.assertTrue(mock_logger_error.called)
1393
1394     def test_get_images_default(self):
1395         self.assertEqual(openstack_utils.
1396                          get_images(self.nova_client),
1397                          [self.image])
1398
1399     @mock.patch('functest.utils.openstack_utils.logger.error')
1400     def test_get_images_exception(self, mock_logger_error):
1401         self.assertEqual(openstack_utils.
1402                          get_images(Exception),
1403                          None)
1404         self.assertTrue(mock_logger_error.called)
1405
1406     def test_get_image_id_default(self):
1407         self.assertEqual(openstack_utils.
1408                          get_image_id(self.glance_client,
1409                                       'test_image'),
1410                          'image_id')
1411
1412     # create_glance_image, get_or_create_image
1413     @mock.patch('functest.utils.openstack_utils.logger.error')
1414     def test_create_glance_image_file_present(self, mock_logger_error):
1415         with mock.patch('functest.utils.openstack_utils.'
1416                         'os.path.isfile',
1417                         return_value=False):
1418             self.assertEqual(openstack_utils.
1419                              create_glance_image(self.glance_client,
1420                                                  'test_image',
1421                                                  'file_path'),
1422                              None)
1423             self.assertTrue(mock_logger_error.called)
1424
1425     @mock.patch('functest.utils.openstack_utils.logger.info')
1426     def test_create_glance_image_already_exist(self, mock_logger_info):
1427         with mock.patch('functest.utils.openstack_utils.'
1428                         'os.path.isfile',
1429                         return_value=True), \
1430             mock.patch('functest.utils.openstack_utils.get_image_id',
1431                        return_value='image_id'):
1432                 self.assertEqual(openstack_utils.
1433                                  create_glance_image(self.glance_client,
1434                                                      'test_image',
1435                                                      'file_path'),
1436                                  'image_id')
1437                 self.assertTrue(mock_logger_info.called)
1438
1439     @mock.patch('functest.utils.openstack_utils.logger.info')
1440     def test_create_glance_image_default(self, mock_logger_info):
1441         with mock.patch('functest.utils.openstack_utils.'
1442                         'os.path.isfile',
1443                         return_value=True), \
1444             mock.patch('functest.utils.openstack_utils.get_image_id',
1445                        return_value=''), \
1446             mock.patch('__builtin__.open',
1447                        mock.mock_open(read_data='1')) as m:
1448                 self.assertEqual(openstack_utils.
1449                                  create_glance_image(self.glance_client,
1450                                                      'test_image',
1451                                                      'file_path'),
1452                                  'image_id')
1453                 m.assert_called_once_with('file_path')
1454                 self.assertTrue(mock_logger_info.called)
1455
1456     @mock.patch('functest.utils.openstack_utils.logger.error')
1457     def test_create_glance_image_exception(self, mock_logger_error):
1458         with mock.patch('functest.utils.openstack_utils.'
1459                         'os.path.isfile',
1460                         return_value=True), \
1461             mock.patch('functest.utils.openstack_utils.get_image_id',
1462                        side_effect=Exception):
1463                 self.assertEqual(openstack_utils.
1464                                  create_glance_image(self.glance_client,
1465                                                      'test_image',
1466                                                      'file_path'),
1467                                  None)
1468                 self.assertTrue(mock_logger_error.called)
1469
1470     def test_delete_glance_image_default(self):
1471         self.assertTrue(openstack_utils.
1472                         delete_glance_image(self.nova_client,
1473                                             'image_id'))
1474
1475     @mock.patch('functest.utils.openstack_utils.logger.error')
1476     def test_delete_glance_image_exception(self, mock_logger_error):
1477         self.assertFalse(openstack_utils.
1478                          delete_glance_image(Exception,
1479                                              'image_id'))
1480         self.assertTrue(mock_logger_error.called)
1481
1482     def test_get_volumes_default(self):
1483         self.assertEqual(openstack_utils.
1484                          get_volumes(self.cinder_client),
1485                          [self.volume])
1486
1487     @mock.patch('functest.utils.openstack_utils.logger.error')
1488     def test_get_volumes_exception(self, mock_logger_error):
1489         self.assertEqual(openstack_utils.
1490                          get_volumes(Exception),
1491                          None)
1492         self.assertTrue(mock_logger_error.called)
1493
1494     def test_list_volume_types_default_private(self):
1495         self.assertEqual(openstack_utils.
1496                          list_volume_types(self.cinder_client,
1497                                            public=False,
1498                                            private=True),
1499                          [self.volume_types[1]])
1500
1501     def test_list_volume_types_default_public(self):
1502         self.assertEqual(openstack_utils.
1503                          list_volume_types(self.cinder_client,
1504                                            public=True,
1505                                            private=False),
1506                          [self.volume_types[0]])
1507
1508     @mock.patch('functest.utils.openstack_utils.logger.error')
1509     def test_list_volume_types_exception(self, mock_logger_error):
1510         self.assertEqual(openstack_utils.
1511                          list_volume_types(Exception),
1512                          None)
1513         self.assertTrue(mock_logger_error.called)
1514
1515     def test_create_volume_type_default(self):
1516         self.assertEqual(openstack_utils.
1517                          create_volume_type(self.cinder_client,
1518                                             'test_volume_type'),
1519                          self.volume_types[0])
1520
1521     @mock.patch('functest.utils.openstack_utils.logger.error')
1522     def test_create_volume_type_exception(self, mock_logger_error):
1523         self.assertEqual(openstack_utils.
1524                          create_volume_type(Exception,
1525                                             'test_volume_type'),
1526                          None)
1527         self.assertTrue(mock_logger_error.called)
1528
1529     def test_update_cinder_quota_default(self):
1530         self.assertTrue(openstack_utils.
1531                         update_cinder_quota(self.cinder_client,
1532                                             'tenant_id',
1533                                             'vols_quota',
1534                                             'snap_quota',
1535                                             'giga_quota'))
1536
1537     @mock.patch('functest.utils.openstack_utils.logger.error')
1538     def test_update_cinder_quota_exception(self, mock_logger_error):
1539         self.assertFalse(openstack_utils.
1540                          update_cinder_quota(Exception,
1541                                              'tenant_id',
1542                                              'vols_quota',
1543                                              'snap_quota',
1544                                              'giga_quota'))
1545         self.assertTrue(mock_logger_error.called)
1546
1547     def test_delete_volume_default(self):
1548         self.assertTrue(openstack_utils.
1549                         delete_volume(self.cinder_client,
1550                                       'volume_id',
1551                                       forced=False))
1552
1553         self.assertTrue(openstack_utils.
1554                         delete_volume(self.cinder_client,
1555                                       'volume_id',
1556                                       forced=True))
1557
1558     @mock.patch('functest.utils.openstack_utils.logger.error')
1559     def test_delete_volume_exception(self, mock_logger_error):
1560         self.assertFalse(openstack_utils.
1561                          delete_volume(Exception,
1562                                        'volume_id',
1563                                        forced=True))
1564         self.assertTrue(mock_logger_error.called)
1565
1566     def test_delete_volume_type_default(self):
1567         self.assertTrue(openstack_utils.
1568                         delete_volume_type(self.cinder_client,
1569                                            self.volume_types[0]))
1570
1571     @mock.patch('functest.utils.openstack_utils.logger.error')
1572     def test_delete_volume_type_exception(self, mock_logger_error):
1573         self.assertFalse(openstack_utils.
1574                          delete_volume_type(Exception,
1575                                             self.volume_types[0]))
1576         self.assertTrue(mock_logger_error.called)
1577
1578     def test_get_tenants_default(self):
1579         with mock.patch('functest.utils.openstack_utils.'
1580                         'is_keystone_v3', return_value=True):
1581             self.assertEqual(openstack_utils.
1582                              get_tenants(self.keystone_client),
1583                              [self.tenant])
1584         with mock.patch('functest.utils.openstack_utils.'
1585                         'is_keystone_v3', return_value=False):
1586             self.assertEqual(openstack_utils.
1587                              get_tenants(self.keystone_client),
1588                              [self.tenant])
1589
1590     @mock.patch('functest.utils.openstack_utils.logger.error')
1591     def test_get_tenants_exception(self, mock_logger_error):
1592         self.assertEqual(openstack_utils.
1593                          get_tenants(Exception),
1594                          None)
1595         self.assertTrue(mock_logger_error.called)
1596
1597     def test_get_users_default(self):
1598         self.assertEqual(openstack_utils.
1599                          get_users(self.keystone_client),
1600                          [self.user])
1601
1602     @mock.patch('functest.utils.openstack_utils.logger.error')
1603     def test_get_users_exception(self, mock_logger_error):
1604         self.assertEqual(openstack_utils.
1605                          get_users(Exception),
1606                          None)
1607         self.assertTrue(mock_logger_error.called)
1608
1609     def test_get_tenant_id_default(self):
1610         self.assertEqual(openstack_utils.
1611                          get_tenant_id(self.keystone_client,
1612                                        'test_tenant'),
1613                          'tenant_id')
1614
1615     def test_get_user_id_default(self):
1616         self.assertEqual(openstack_utils.
1617                          get_user_id(self.keystone_client,
1618                                      'test_user'),
1619                          'user_id')
1620
1621     def test_get_role_id_default(self):
1622         self.assertEqual(openstack_utils.
1623                          get_role_id(self.keystone_client,
1624                                      'test_role'),
1625                          'role_id')
1626
1627     def test_create_tenant_default(self):
1628         with mock.patch('functest.utils.openstack_utils.'
1629                         'is_keystone_v3', return_value=True):
1630             self.assertEqual(openstack_utils.
1631                              create_tenant(self.keystone_client,
1632                                            'test_tenant',
1633                                            'tenant_desc'),
1634                              'tenant_id')
1635         with mock.patch('functest.utils.openstack_utils.'
1636                         'is_keystone_v3', return_value=False):
1637             self.assertEqual(openstack_utils.
1638                              create_tenant(self.keystone_client,
1639                                            'test_tenant',
1640                                            'tenant_desc'),
1641                              'tenant_id')
1642
1643     @mock.patch('functest.utils.openstack_utils.logger.error')
1644     def test_create_tenant_exception(self, mock_logger_error):
1645         self.assertEqual(openstack_utils.
1646                          create_tenant(Exception,
1647                                        'test_tenant',
1648                                        'tenant_desc'),
1649                          None)
1650         self.assertTrue(mock_logger_error.called)
1651
1652     def test_create_user_default(self):
1653         with mock.patch('functest.utils.openstack_utils.'
1654                         'is_keystone_v3', return_value=True):
1655             self.assertEqual(openstack_utils.
1656                              create_user(self.keystone_client,
1657                                          'test_user',
1658                                          'password',
1659                                          'email',
1660                                          'tenant_id'),
1661                              'user_id')
1662         with mock.patch('functest.utils.openstack_utils.'
1663                         'is_keystone_v3', return_value=False):
1664             self.assertEqual(openstack_utils.
1665                              create_user(self.keystone_client,
1666                                          'test_user',
1667                                          'password',
1668                                          'email',
1669                                          'tenant_id'),
1670                              'user_id')
1671
1672     @mock.patch('functest.utils.openstack_utils.logger.error')
1673     def test_create_user_exception(self, mock_logger_error):
1674         self.assertEqual(openstack_utils.
1675                          create_user(Exception,
1676                                      'test_user',
1677                                      'password',
1678                                      'email',
1679                                      'tenant_id'),
1680                          None)
1681         self.assertTrue(mock_logger_error.called)
1682
1683     def test_add_role_user_default(self):
1684         with mock.patch('functest.utils.openstack_utils.'
1685                         'is_keystone_v3', return_value=True):
1686             self.assertTrue(openstack_utils.
1687                             add_role_user(self.keystone_client,
1688                                           'user_id',
1689                                           'role_id',
1690                                           'tenant_id'))
1691
1692         with mock.patch('functest.utils.openstack_utils.'
1693                         'is_keystone_v3', return_value=False):
1694             self.assertTrue(openstack_utils.
1695                             add_role_user(self.keystone_client,
1696                                           'user_id',
1697                                           'role_id',
1698                                           'tenant_id'))
1699
1700     @mock.patch('functest.utils.openstack_utils.logger.error')
1701     def test_add_role_user_exception(self, mock_logger_error):
1702         self.assertFalse(openstack_utils.
1703                          add_role_user(Exception,
1704                                        'user_id',
1705                                        'role_id',
1706                                        'tenant_id'))
1707         self.assertTrue(mock_logger_error.called)
1708
1709     def test_delete_tenant_default(self):
1710         with mock.patch('functest.utils.openstack_utils.'
1711                         'is_keystone_v3', return_value=True):
1712             self.assertTrue(openstack_utils.
1713                             delete_tenant(self.keystone_client,
1714                                           'tenant_id'))
1715
1716         with mock.patch('functest.utils.openstack_utils.'
1717                         'is_keystone_v3', return_value=False):
1718             self.assertTrue(openstack_utils.
1719                             delete_tenant(self.keystone_client,
1720                                           'tenant_id'))
1721
1722     @mock.patch('functest.utils.openstack_utils.logger.error')
1723     def test_delete_tenant_exception(self, mock_logger_error):
1724         self.assertFalse(openstack_utils.
1725                          delete_tenant(Exception,
1726                                        'tenant_id'))
1727         self.assertTrue(mock_logger_error.called)
1728
1729     def test_delete_user_default(self):
1730         self.assertTrue(openstack_utils.
1731                         delete_user(self.keystone_client,
1732                                     'user_id'))
1733
1734     @mock.patch('functest.utils.openstack_utils.logger.error')
1735     def test_delete_user_exception(self, mock_logger_error):
1736         self.assertFalse(openstack_utils.
1737                          delete_user(Exception,
1738                                      'user_id'))
1739         self.assertTrue(mock_logger_error.called)
1740
1741     def test_get_resource_default(self):
1742         with mock.patch('functest.utils.openstack_utils.'
1743                         'is_keystone_v3', return_value=True):
1744             self.assertEqual(openstack_utils.
1745                              get_resource(self.heat_client,
1746                                           'stack_id',
1747                                           'resource'),
1748                              self.resource)
1749
1750     @mock.patch('functest.utils.openstack_utils.logger.error')
1751     def test_get_resource_exception(self, mock_logger_error):
1752         self.assertEqual(openstack_utils.
1753                          get_resource(Exception,
1754                                       'stack_id',
1755                                       'resource'),
1756                          None)
1757         self.assertTrue(mock_logger_error.called)
1758
1759
1760 if __name__ == "__main__":
1761     unittest.main(verbosity=2)