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