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