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