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
15 from functest.utils import openstack_utils
18 class OSUtilsTesting(unittest.TestCase):
20 def _get_env_cred_dict(self, os_prefix=''):
21 return {'OS_USERNAME': os_prefix + 'username',
22 'OS_PASSWORD': os_prefix + 'password',
23 'OS_AUTH_URL': os_prefix + 'auth_url',
24 'OS_TENANT_NAME': os_prefix + 'tenant_name',
25 'OS_USER_DOMAIN_NAME': os_prefix + 'user_domain_name',
26 'OS_PROJECT_DOMAIN_NAME': os_prefix + 'project_domain_name',
27 'OS_PROJECT_NAME': os_prefix + 'project_name',
28 'OS_ENDPOINT_TYPE': os_prefix + 'endpoint_type',
29 'OS_REGION_NAME': os_prefix + 'region_name',
30 'OS_CACERT': os_prefix + 'https_cacert',
31 'OS_INSECURE': os_prefix + 'https_insecure'}
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 'https_cacert': 'test_https_cacert',
42 'https_insecure': 'test_https_insecure'}
45 self.env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD']
46 self.tenant_name = 'test_tenant_name'
47 self.env_cred_dict = self._get_env_cred_dict()
48 self.os_environs = self._get_env_cred_dict(os_prefix='test_')
49 self.os_env_vars = self._get_os_env_vars()
51 mock_obj = mock.Mock()
52 attrs = {'name': 'test_flavor',
55 mock_obj.configure_mock(**attrs)
56 self.flavor = mock_obj
58 mock_obj = mock.Mock()
59 attrs = {'name': 'test_aggregate',
61 'hosts': ['host_name']}
62 mock_obj.configure_mock(**attrs)
63 self.aggregate = mock_obj
65 mock_obj = mock.Mock()
66 attrs = {'id': 'instance_id',
67 'name': 'test_instance',
69 mock_obj.configure_mock(**attrs)
70 self.instance = mock_obj
72 mock_obj = mock.Mock()
73 attrs = {'id': 'azone_id',
74 'zoneName': 'test_azone',
76 mock_obj.configure_mock(**attrs)
77 self.availability_zone = mock_obj
79 mock_obj = mock.Mock()
80 attrs = {'floating_network_id': 'floating_id',
81 'floating_ip_address': 'test_floating_ip'}
82 mock_obj.configure_mock(**attrs)
83 self.floating_ip = mock_obj
85 mock_obj = mock.Mock()
86 attrs = {'id': 'hypervisor_id',
87 'hypervisor_hostname': 'test_hostname',
89 mock_obj.configure_mock(**attrs)
90 self.hypervisor = mock_obj
92 mock_obj = mock.Mock()
93 attrs = {'id': 'image_id',
95 mock_obj.configure_mock(**attrs)
98 mock_obj = mock.Mock()
99 self.mock_return = mock_obj
101 self.nova_client = mock.Mock()
102 attrs = {'servers.list.return_value': [self.instance],
103 'servers.get.return_value': self.instance,
104 'servers.find.return_value': self.instance,
105 'servers.create.return_value': self.instance,
106 'flavors.list.return_value': [self.flavor],
107 'flavors.find.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 'hypervisors.list.return_value': [self.hypervisor],
118 'create.return_value': mock.Mock(),
119 'add_security_group.return_value': mock.Mock(),
120 'images.list.return_value': [self.image],
121 'images.delete.return_value': mock.Mock(),
123 self.nova_client.configure_mock(**attrs)
125 self.glance_client = mock.Mock()
126 attrs = {'images.list.return_value': [self.image],
127 'images.create.return_value': self.image,
128 'images.upload.return_value': mock.Mock()}
129 self.glance_client.configure_mock(**attrs)
131 mock_obj = mock.Mock()
132 attrs = {'id': 'volume_id',
133 'name': 'test_volume'}
134 mock_obj.configure_mock(**attrs)
135 self.volume = mock_obj
137 mock_obj = mock.Mock()
138 attrs = {'id': 'volume_type_id',
139 'name': 'test_volume_type',
141 mock_obj.configure_mock(**attrs)
142 self.volume_types = [mock_obj]
144 mock_obj = mock.Mock()
145 attrs = {'id': 'volume_type_id',
146 'name': 'test_volume_type',
148 mock_obj.configure_mock(**attrs)
149 self.volume_types.append(mock_obj)
151 self.cinder_client = mock.Mock()
152 attrs = {'volumes.list.return_value': [self.volume],
153 'volume_types.list.return_value': self.volume_types,
154 'volume_types.create.return_value': self.volume_types[0],
155 'volume_types.delete.return_value': mock.Mock(),
156 'quotas.update.return_value': mock.Mock(),
157 'volumes.detach.return_value': mock.Mock(),
158 'volumes.force_delete.return_value': mock.Mock(),
159 'volumes.delete.return_value': mock.Mock()
161 self.cinder_client.configure_mock(**attrs)
163 self.resource = mock.Mock()
164 attrs = {'id': 'resource_test_id',
165 'name': 'resource_test_name'
168 self.heat_client = mock.Mock()
169 attrs = {'resources.get.return_value': self.resource}
170 self.heat_client.configure_mock(**attrs)
172 mock_obj = mock.Mock()
173 attrs = {'id': 'tenant_id',
174 'name': 'test_tenant'}
175 mock_obj.configure_mock(**attrs)
176 self.tenant = mock_obj
178 mock_obj = mock.Mock()
179 attrs = {'id': 'user_id',
181 mock_obj.configure_mock(**attrs)
184 mock_obj = mock.Mock()
185 attrs = {'id': 'role_id',
187 mock_obj.configure_mock(**attrs)
190 self.keystone_client = mock.Mock()
191 attrs = {'projects.list.return_value': [self.tenant],
192 'tenants.list.return_value': [self.tenant],
193 'users.list.return_value': [self.user],
194 'roles.list.return_value': [self.role],
195 'projects.create.return_value': self.tenant,
196 'tenants.create.return_value': self.tenant,
197 'users.create.return_value': self.user,
198 'roles.grant.return_value': mock.Mock(),
199 'roles.add_user_role.return_value': mock.Mock(),
200 'projects.delete.return_value': mock.Mock(),
201 'tenants.delete.return_value': mock.Mock(),
202 'users.delete.return_value': mock.Mock(),
204 self.keystone_client.configure_mock(**attrs)
206 self.router = {'id': 'router_id',
207 'name': 'test_router'}
209 self.subnet = {'id': 'subnet_id',
210 'name': 'test_subnet'}
212 self.networks = [{'id': 'network_id',
213 'name': 'test_network',
214 'router:external': False,
216 'subnets': [self.subnet]},
217 {'id': 'network_id1',
218 'name': 'test_network1',
219 'router:external': True,
221 'subnets': [self.subnet]}]
223 self.port = {'id': 'port_id',
226 self.sec_group = {'id': 'sec_group_id',
227 'name': 'test_sec_group'}
229 self.sec_group_rule = {'id': 'sec_group_rule_id',
230 'direction': 'direction',
231 'protocol': 'protocol',
232 'port_range_max': 'port_max',
233 'security_group_id': self.sec_group['id'],
234 'port_range_min': 'port_min'}
235 self.neutron_floatingip = {'id': 'fip_id',
236 'floating_ip_address': 'test_ip'}
237 self.neutron_client = mock.Mock()
238 attrs = {'list_networks.return_value': {'networks': self.networks},
239 'list_routers.return_value': {'routers': [self.router]},
240 'list_ports.return_value': {'ports': [self.port]},
241 'list_subnets.return_value': {'subnets': [self.subnet]},
242 'create_network.return_value': {'network': self.networks[0]},
243 'create_subnet.return_value': {'subnets': [self.subnet]},
244 'create_router.return_value': {'router': self.router},
245 'create_port.return_value': {'port': self.port},
246 'create_floatingip.return_value': {'floatingip':
247 self.neutron_floatingip},
248 'update_network.return_value': mock.Mock(),
249 'update_port.return_value': {'port': self.port},
250 'add_interface_router.return_value': mock.Mock(),
251 'add_gateway_router.return_value': mock.Mock(),
252 'delete_network.return_value': mock.Mock(),
253 'delete_subnet.return_value': mock.Mock(),
254 'delete_router.return_value': mock.Mock(),
255 'delete_port.return_value': mock.Mock(),
256 'remove_interface_router.return_value': mock.Mock(),
257 'remove_gateway_router.return_value': mock.Mock(),
258 'list_security_groups.return_value': {'security_groups':
260 'list_security_group_rules.'
261 'return_value': {'security_group_rules':
262 [self.sec_group_rule]},
263 'create_security_group_rule.return_value': mock.Mock(),
264 'create_security_group.return_value': {'security_group':
266 'update_quota.return_value': mock.Mock(),
267 'delete_security_group.return_value': mock.Mock(),
268 'list_floatingips.return_value': {'floatingips':
270 'delete_floatingip.return_value': mock.Mock(),
272 self.neutron_client.configure_mock(**attrs)
274 self.empty_client = mock.Mock()
275 attrs = {'list_networks.return_value': {'networks': []},
276 'list_routers.return_value': {'routers': []},
277 'list_ports.return_value': {'ports': []},
278 'list_subnets.return_value': {'subnets': []}}
279 self.empty_client.configure_mock(**attrs)
281 @mock.patch('functest.utils.openstack_utils.os.getenv',
283 def test_is_keystone_v3_missing_identity(self, mock_os_getenv):
284 self.assertEqual(openstack_utils.is_keystone_v3(), False)
286 @mock.patch('functest.utils.openstack_utils.os.getenv',
288 def test_is_keystone_v3_default(self, mock_os_getenv):
289 self.assertEqual(openstack_utils.is_keystone_v3(), True)
291 @mock.patch('functest.utils.openstack_utils.is_keystone_v3',
293 def test_get_rc_env_vars_missing_identity(self, mock_get_rc_env):
294 exp_resp = self.env_vars
295 exp_resp.extend(['OS_TENANT_NAME'])
296 self.assertEqual(openstack_utils.get_rc_env_vars(), exp_resp)
298 @mock.patch('functest.utils.openstack_utils.is_keystone_v3',
300 def test_get_rc_env_vars_default(self, mock_get_rc_env):
301 exp_resp = self.env_vars
302 exp_resp.extend(['OS_PROJECT_NAME',
303 'OS_USER_DOMAIN_NAME',
304 'OS_PROJECT_DOMAIN_NAME'])
305 self.assertEqual(openstack_utils.get_rc_env_vars(), exp_resp)
307 @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
308 def test_check_credentials_missing_env(self, mock_get_rc_env):
309 exp_resp = self.env_vars
310 exp_resp.extend(['OS_TENANT_NAME'])
311 mock_get_rc_env.return_value = exp_resp
312 with mock.patch.dict('functest.utils.openstack_utils.os.environ', {},
314 self.assertEqual(openstack_utils.check_credentials(), False)
316 @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
317 def test_check_credentials_default(self, mock_get_rc_env):
318 exp_resp = ['OS_TENANT_NAME']
319 mock_get_rc_env.return_value = exp_resp
320 with mock.patch.dict('functest.utils.openstack_utils.os.environ',
321 {'OS_TENANT_NAME': self.tenant_name},
323 self.assertEqual(openstack_utils.check_credentials(), True)
325 def test_get_env_cred_dict(self):
326 self.assertDictEqual(openstack_utils.get_env_cred_dict(),
329 @mock.patch('functest.utils.openstack_utils.get_rc_env_vars')
330 def test_get_credentials_default(self, mock_get_rc_env):
331 mock_get_rc_env.return_value = self.env_cred_dict.keys()
332 with mock.patch.dict('functest.utils.openstack_utils.os.environ',
335 self.assertDictEqual(openstack_utils.get_credentials(),
338 def _get_credentials_missing_env(self, var):
339 dic = copy.deepcopy(self.os_environs)
341 with mock.patch('functest.utils.openstack_utils.get_rc_env_vars',
342 return_value=self.env_cred_dict.keys()), \
343 mock.patch.dict('functest.utils.openstack_utils.os.environ',
346 self.assertRaises(openstack_utils.MissingEnvVar,
347 lambda: openstack_utils.get_credentials())
349 def test_get_credentials_missing_username(self):
350 self._get_credentials_missing_env('OS_USERNAME')
352 def test_get_credentials_missing_password(self):
353 self._get_credentials_missing_env('OS_PASSWORD')
355 def test_get_credentials_missing_auth_url(self):
356 self._get_credentials_missing_env('OS_AUTH_URL')
358 def test_get_credentials_missing_tenantname(self):
359 self._get_credentials_missing_env('OS_TENANT_NAME')
361 def test_get_credentials_missing_domainname(self):
362 self._get_credentials_missing_env('OS_USER_DOMAIN_NAME')
364 def test_get_credentials_missing_projectname(self):
365 self._get_credentials_missing_env('OS_PROJECT_NAME')
367 def test_get_credentials_missing_endpoint_type(self):
368 self._get_credentials_missing_env('OS_ENDPOINT_TYPE')
370 def _test_source_credentials(self, msg, key='OS_TENANT_NAME',
377 with mock.patch('__builtin__.open', mock.mock_open(read_data=msg),
379 m.return_value.__iter__ = lambda self: iter(self.readline, '')
380 openstack_utils.source_credentials(f)
381 m.assert_called_once_with(f, 'r')
382 self.assertEqual(os.environ[key], value)
384 def test_source_credentials(self):
385 self._test_source_credentials('OS_TENANT_NAME=admin')
386 self._test_source_credentials('OS_TENANT_NAME= admin')
387 self._test_source_credentials('OS_TENANT_NAME = admin')
388 self._test_source_credentials('OS_TENANT_NAME = "admin"')
389 self._test_source_credentials('export OS_TENANT_NAME=admin')
390 self._test_source_credentials('export OS_TENANT_NAME =admin')
391 self._test_source_credentials('export OS_TENANT_NAME = admin')
392 self._test_source_credentials('export OS_TENANT_NAME = "admin"')
393 # This test will fail as soon as rc_file is fixed
394 self._test_source_credentials(
395 'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
397 @mock.patch('functest.utils.openstack_utils.os.getenv',
399 def test_get_keystone_client_version_missing_env(self, mock_os_getenv):
400 self.assertEqual(openstack_utils.get_keystone_client_version(),
401 openstack_utils.DEFAULT_API_VERSION)
403 @mock.patch('functest.utils.openstack_utils.logger.info')
404 @mock.patch('functest.utils.openstack_utils.os.getenv',
406 def test_get_keystone_client_version_default(self, mock_os_getenv,
408 self.assertEqual(openstack_utils.get_keystone_client_version(),
410 mock_logger_info.assert_called_once_with("OS_IDENTITY_API_VERSION is "
411 "set in env as '%s'", '3')
413 @mock.patch('functest.utils.openstack_utils.get_session')
414 @mock.patch('functest.utils.openstack_utils.keystoneclient.Client')
415 @mock.patch('functest.utils.openstack_utils.get_keystone_client_version',
417 @mock.patch('functest.utils.openstack_utils.os.getenv',
418 return_value='public')
419 def test_get_keystone_client_with_interface(self, mock_os_getenv,
420 mock_keystoneclient_version,
423 mock_keystone_obj = mock.Mock()
424 mock_session_obj = mock.Mock()
425 mock_key_client.return_value = mock_keystone_obj
426 mock_get_session.return_value = mock_session_obj
427 self.assertEqual(openstack_utils.get_keystone_client(),
429 mock_key_client.assert_called_once_with('3',
430 session=mock_session_obj,
433 @mock.patch('functest.utils.openstack_utils.get_session')
434 @mock.patch('functest.utils.openstack_utils.keystoneclient.Client')
435 @mock.patch('functest.utils.openstack_utils.get_keystone_client_version',
437 @mock.patch('functest.utils.openstack_utils.os.getenv',
438 return_value='admin')
439 def test_get_keystone_client_no_interface(self, mock_os_getenv,
440 mock_keystoneclient_version,
443 mock_keystone_obj = mock.Mock()
444 mock_session_obj = mock.Mock()
445 mock_key_client.return_value = mock_keystone_obj
446 mock_get_session.return_value = mock_session_obj
447 self.assertEqual(openstack_utils.get_keystone_client(),
449 mock_key_client.assert_called_once_with('3',
450 session=mock_session_obj,
453 @mock.patch('functest.utils.openstack_utils.os.getenv',
455 def test_get_nova_client_version_missing_env(self, mock_os_getenv):
456 self.assertEqual(openstack_utils.get_nova_client_version(),
457 openstack_utils.DEFAULT_API_VERSION)
459 @mock.patch('functest.utils.openstack_utils.logger.info')
460 @mock.patch('functest.utils.openstack_utils.os.getenv',
462 def test_get_nova_client_version_default(self, mock_os_getenv,
464 self.assertEqual(openstack_utils.get_nova_client_version(),
466 mock_logger_info.assert_called_once_with("OS_COMPUTE_API_VERSION is "
467 "set in env as '%s'", '3')
469 def test_get_nova_client(self):
470 mock_nova_obj = mock.Mock()
471 mock_session_obj = mock.Mock()
472 with mock.patch('functest.utils.openstack_utils'
473 '.get_nova_client_version', return_value='3'), \
474 mock.patch('functest.utils.openstack_utils'
475 '.novaclient.Client',
476 return_value=mock_nova_obj) \
477 as mock_nova_client, \
478 mock.patch('functest.utils.openstack_utils.get_session',
479 return_value=mock_session_obj):
480 self.assertEqual(openstack_utils.get_nova_client(),
482 mock_nova_client.assert_called_once_with('3',
483 session=mock_session_obj)
485 @mock.patch('functest.utils.openstack_utils.os.getenv',
487 def test_get_cinder_client_version_missing_env(self, mock_os_getenv):
488 self.assertEqual(openstack_utils.get_cinder_client_version(),
489 openstack_utils.DEFAULT_API_VERSION)
491 @mock.patch('functest.utils.openstack_utils.logger.info')
492 @mock.patch('functest.utils.openstack_utils.os.getenv',
494 def test_get_cinder_client_version_default(self, mock_os_getenv,
496 self.assertEqual(openstack_utils.get_cinder_client_version(),
498 mock_logger_info.assert_called_once_with("OS_VOLUME_API_VERSION is "
499 "set in env as '%s'", '3')
501 def test_get_cinder_client(self):
502 mock_cinder_obj = mock.Mock()
503 mock_session_obj = mock.Mock()
504 with mock.patch('functest.utils.openstack_utils'
505 '.get_cinder_client_version', return_value='3'), \
506 mock.patch('functest.utils.openstack_utils'
507 '.cinderclient.Client',
508 return_value=mock_cinder_obj) \
509 as mock_cind_client, \
510 mock.patch('functest.utils.openstack_utils.get_session',
511 return_value=mock_session_obj):
512 self.assertEqual(openstack_utils.get_cinder_client(),
514 mock_cind_client.assert_called_once_with('3',
515 session=mock_session_obj)
517 @mock.patch('functest.utils.openstack_utils.os.getenv',
519 def test_get_neutron_client_version_missing_env(self, mock_os_getenv):
520 self.assertEqual(openstack_utils.get_neutron_client_version(),
521 openstack_utils.DEFAULT_API_VERSION)
523 @mock.patch('functest.utils.openstack_utils.logger.info')
524 @mock.patch('functest.utils.openstack_utils.os.getenv',
526 def test_get_neutron_client_version_default(self, mock_os_getenv,
528 self.assertEqual(openstack_utils.get_neutron_client_version(),
530 mock_logger_info.assert_called_once_with("OS_NETWORK_API_VERSION is "
531 "set in env as '%s'", '3')
533 def test_get_neutron_client(self):
534 mock_neutron_obj = mock.Mock()
535 mock_session_obj = mock.Mock()
536 with mock.patch('functest.utils.openstack_utils'
537 '.get_neutron_client_version', return_value='3'), \
538 mock.patch('functest.utils.openstack_utils'
539 '.neutronclient.Client',
540 return_value=mock_neutron_obj) \
541 as mock_neut_client, \
542 mock.patch('functest.utils.openstack_utils.get_session',
543 return_value=mock_session_obj):
544 self.assertEqual(openstack_utils.get_neutron_client(),
546 mock_neut_client.assert_called_once_with('3',
547 session=mock_session_obj)
549 @mock.patch('functest.utils.openstack_utils.os.getenv',
551 def test_get_glance_client_version_missing_env(self, mock_os_getenv):
552 self.assertEqual(openstack_utils.get_glance_client_version(),
553 openstack_utils.DEFAULT_API_VERSION)
555 @mock.patch('functest.utils.openstack_utils.logger.info')
556 @mock.patch('functest.utils.openstack_utils.os.getenv',
558 def test_get_glance_client_version_default(self, mock_os_getenv,
560 self.assertEqual(openstack_utils.get_glance_client_version(),
562 mock_logger_info.assert_called_once_with("OS_IMAGE_API_VERSION is "
563 "set in env as '%s'", '3')
565 def test_get_glance_client(self):
566 mock_glance_obj = mock.Mock()
567 mock_session_obj = mock.Mock()
568 with mock.patch('functest.utils.openstack_utils'
569 '.get_glance_client_version', return_value='3'), \
570 mock.patch('functest.utils.openstack_utils'
571 '.glanceclient.Client',
572 return_value=mock_glance_obj) \
573 as mock_glan_client, \
574 mock.patch('functest.utils.openstack_utils.get_session',
575 return_value=mock_session_obj):
576 self.assertEqual(openstack_utils.get_glance_client(),
578 mock_glan_client.assert_called_once_with('3',
579 session=mock_session_obj)
581 @mock.patch('functest.utils.openstack_utils.os.getenv',
583 def test_get_heat_client_version_missing_env(self, mock_os_getenv):
584 self.assertEqual(openstack_utils.get_heat_client_version(),
585 openstack_utils.DEFAULT_HEAT_API_VERSION)
587 @mock.patch('functest.utils.openstack_utils.logger.info')
588 @mock.patch('functest.utils.openstack_utils.os.getenv', return_value='1')
589 def test_get_heat_client_version_default(self, mock_os_getenv,
591 self.assertEqual(openstack_utils.get_heat_client_version(), '1')
592 mock_logger_info.assert_called_once_with(
593 "OS_ORCHESTRATION_API_VERSION is set in env as '%s'", '1')
595 def test_get_heat_client(self):
596 mock_heat_obj = mock.Mock()
597 mock_session_obj = mock.Mock()
598 with mock.patch('functest.utils.openstack_utils'
599 '.get_heat_client_version', return_value='1'), \
600 mock.patch('functest.utils.openstack_utils'
601 '.heatclient.Client',
602 return_value=mock_heat_obj) \
603 as mock_heat_client, \
604 mock.patch('functest.utils.openstack_utils.get_session',
605 return_value=mock_session_obj):
606 self.assertEqual(openstack_utils.get_heat_client(),
608 mock_heat_client.assert_called_once_with('1',
609 session=mock_session_obj)
611 def test_get_instances_default(self):
612 self.assertEqual(openstack_utils.get_instances(self.nova_client),
615 @mock.patch('functest.utils.openstack_utils.logger.error')
616 def test_get_instances_exception(self, mock_logger_error):
617 self.assertEqual(openstack_utils.
618 get_instances(Exception),
620 self.assertTrue(mock_logger_error.called)
622 def test_get_instance_status_default(self):
623 self.assertEqual(openstack_utils.get_instance_status(self.nova_client,
627 @mock.patch('functest.utils.openstack_utils.logger.error')
628 def test_get_instance_status_exception(self, mock_logger_error):
629 self.assertEqual(openstack_utils.
630 get_instance_status(Exception,
633 self.assertTrue(mock_logger_error.called)
635 def test_get_instance_by_name_default(self):
636 self.assertEqual(openstack_utils.
637 get_instance_by_name(self.nova_client,
641 @mock.patch('functest.utils.openstack_utils.logger.error')
642 def test_get_instance_by_name_exception(self, mock_logger_error):
643 self.assertEqual(openstack_utils.
644 get_instance_by_name(Exception,
647 self.assertTrue(mock_logger_error.called)
649 def test_get_flavor_id_default(self):
650 self.assertEqual(openstack_utils.
651 get_flavor_id(self.nova_client,
655 def test_get_flavor_id_by_ram_range_default(self):
656 self.assertEqual(openstack_utils.
657 get_flavor_id_by_ram_range(self.nova_client,
661 def test_get_aggregates_default(self):
662 self.assertEqual(openstack_utils.
663 get_aggregates(self.nova_client),
666 @mock.patch('functest.utils.openstack_utils.logger.error')
667 def test_get_aggregates_exception(self, mock_logger_error):
668 self.assertEqual(openstack_utils.
669 get_aggregates(Exception),
671 self.assertTrue(mock_logger_error.called)
673 def test_get_aggregate_id_default(self):
674 with mock.patch('functest.utils.openstack_utils.get_aggregates',
675 return_value=[self.aggregate]):
676 self.assertEqual(openstack_utils.
677 get_aggregate_id(self.nova_client,
681 @mock.patch('functest.utils.openstack_utils.logger.error')
682 def test_get_aggregate_id_exception(self, mock_logger_error):
683 self.assertEqual(openstack_utils.
684 get_aggregate_id(Exception,
687 self.assertTrue(mock_logger_error.called)
689 def test_get_availability_zone_names_default(self):
690 with mock.patch('functest.utils.openstack_utils'
691 '.get_availability_zones',
692 return_value=[self.availability_zone]):
693 self.assertEqual(openstack_utils.
694 get_availability_zone_names(self.nova_client),
697 @mock.patch('functest.utils.openstack_utils.logger.error')
698 def test_get_availability_zone_names_exception(self, mock_logger_error):
699 self.assertEqual(openstack_utils.
700 get_availability_zone_names(Exception),
702 self.assertTrue(mock_logger_error.called)
704 def test_get_availability_zones_default(self):
705 self.assertEqual(openstack_utils.
706 get_availability_zones(self.nova_client),
707 [self.availability_zone])
709 @mock.patch('functest.utils.openstack_utils.logger.error')
710 def test_get_availability_zones_exception(self, mock_logger_error):
711 self.assertEqual(openstack_utils.
712 get_availability_zones(Exception),
714 self.assertTrue(mock_logger_error.called)
716 def test_get_floating_ips_default(self):
717 self.assertEqual(openstack_utils.
718 get_floating_ips(self.neutron_client),
721 @mock.patch('functest.utils.openstack_utils.logger.error')
722 def test_get_floating_ips_exception(self, mock_logger_error):
723 self.assertEqual(openstack_utils.
724 get_floating_ips(Exception),
726 self.assertTrue(mock_logger_error.called)
728 def test_get_hypervisors_default(self):
729 self.assertEqual(openstack_utils.
730 get_hypervisors(self.nova_client),
733 @mock.patch('functest.utils.openstack_utils.logger.error')
734 def test_get_hypervisors_exception(self, mock_logger_error):
735 self.assertEqual(openstack_utils.
736 get_hypervisors(Exception),
738 self.assertTrue(mock_logger_error.called)
740 def test_create_aggregate_default(self):
741 self.assertTrue(openstack_utils.
742 create_aggregate(self.nova_client,
746 @mock.patch('functest.utils.openstack_utils.logger.error')
747 def test_create_aggregate_exception(self, mock_logger_error):
748 self.assertEqual(openstack_utils.
749 create_aggregate(Exception,
753 self.assertTrue(mock_logger_error.called)
755 def test_add_host_to_aggregate_default(self):
756 with mock.patch('functest.utils.openstack_utils.get_aggregate_id'):
757 self.assertTrue(openstack_utils.
758 add_host_to_aggregate(self.nova_client,
762 @mock.patch('functest.utils.openstack_utils.logger.error')
763 def test_add_host_to_aggregate_exception(self, mock_logger_error):
764 self.assertEqual(openstack_utils.
765 add_host_to_aggregate(Exception,
769 self.assertTrue(mock_logger_error.called)
771 def test_create_aggregate_with_host_default(self):
772 with mock.patch('functest.utils.openstack_utils.create_aggregate'), \
773 mock.patch('functest.utils.openstack_utils.'
774 'add_host_to_aggregate'):
775 self.assertTrue(openstack_utils.
776 create_aggregate_with_host(self.nova_client,
781 @mock.patch('functest.utils.openstack_utils.logger.error')
782 def test_create_aggregate_with_host_exception(self, mock_logger_error):
783 with mock.patch('functest.utils.openstack_utils.create_aggregate',
784 side_effect=Exception):
785 self.assertEqual(openstack_utils.
786 create_aggregate_with_host(Exception,
791 self.assertTrue(mock_logger_error.called)
793 def test_create_instance_default(self):
794 with mock.patch('functest.utils.openstack_utils.'
796 return_value=self.nova_client):
797 self.assertEqual(openstack_utils.
798 create_instance('test_flavor',
803 @mock.patch('functest.utils.openstack_utils.logger.error')
804 def test_create_instance_exception(self, mock_logger_error):
805 with mock.patch('functest.utils.openstack_utils.'
807 return_value=self.nova_client):
808 self.nova_client.flavors.find.side_effect = Exception
809 self.assertEqual(openstack_utils.
810 create_instance('test_flavor',
814 self.assertTrue(mock_logger_error)
816 def test_create_floating_ip_default(self):
817 with mock.patch('functest.utils.openstack_utils.'
818 'get_external_net_id',
819 return_value='external_net_id'):
820 exp_resp = {'fip_addr': 'test_ip', 'fip_id': 'fip_id'}
821 self.assertEqual(openstack_utils.
822 create_floating_ip(self.neutron_client),
825 @mock.patch('functest.utils.openstack_utils.logger.error')
826 def test_create_floating_ip_exception(self, mock_logger_error):
827 with mock.patch('functest.utils.openstack_utils.'
828 'get_external_net_id',
829 return_value='external_net_id'):
830 self.assertEqual(openstack_utils.
831 create_floating_ip(Exception),
833 self.assertTrue(mock_logger_error)
835 def test_add_floating_ip_default(self):
836 with mock.patch('functest.utils.openstack_utils.get_aggregate_id'):
837 self.assertTrue(openstack_utils.
838 add_floating_ip(self.nova_client,
840 'test_floatingip_addr'))
842 @mock.patch('functest.utils.openstack_utils.logger.error')
843 def test_add_floating_ip_exception(self, mock_logger_error):
844 self.assertFalse(openstack_utils.
845 add_floating_ip(Exception,
847 'test_floatingip_addr'))
848 self.assertTrue(mock_logger_error.called)
850 def test_delete_instance_default(self):
851 self.assertTrue(openstack_utils.
852 delete_instance(self.nova_client,
855 @mock.patch('functest.utils.openstack_utils.logger.error')
856 def test_delete_instance_exception(self, mock_logger_error):
857 self.assertFalse(openstack_utils.
858 delete_instance(Exception,
860 self.assertTrue(mock_logger_error.called)
862 def test_delete_floating_ip_default(self):
863 self.assertTrue(openstack_utils.
864 delete_floating_ip(self.neutron_client,
867 @mock.patch('functest.utils.openstack_utils.logger.error')
868 def test_delete_floating_ip_exception(self, mock_logger_error):
869 self.assertFalse(openstack_utils.
870 delete_floating_ip(Exception,
872 self.assertTrue(mock_logger_error.called)
874 def test_remove_host_from_aggregate_default(self):
875 with mock.patch('functest.utils.openstack_utils.'
877 self.assertTrue(openstack_utils.
878 remove_host_from_aggregate(self.nova_client,
882 @mock.patch('functest.utils.openstack_utils.logger.error')
883 def test_remove_host_from_aggregate_exception(self, mock_logger_error):
884 with mock.patch('functest.utils.openstack_utils.'
885 'get_aggregate_id', side_effect=Exception):
886 self.assertFalse(openstack_utils.
887 remove_host_from_aggregate(self.nova_client,
890 self.assertTrue(mock_logger_error.called)
892 def test_remove_hosts_from_aggregate_default(self):
893 with mock.patch('functest.utils.openstack_utils.'
894 'get_aggregate_id'), \
895 mock.patch('functest.utils.openstack_utils.'
896 'remove_host_from_aggregate',
899 openstack_utils.remove_hosts_from_aggregate(self.nova_client,
901 mock_method.assert_any_call(self.nova_client,
905 def test_delete_aggregate_default(self):
906 with mock.patch('functest.utils.openstack_utils.'
907 'remove_hosts_from_aggregate'):
908 self.assertTrue(openstack_utils.
909 delete_aggregate(self.nova_client,
912 @mock.patch('functest.utils.openstack_utils.logger.error')
913 def test_delete_aggregate_exception(self, mock_logger_error):
914 with mock.patch('functest.utils.openstack_utils.'
915 'remove_hosts_from_aggregate', side_effect=Exception):
916 self.assertFalse(openstack_utils.
917 delete_aggregate(self.nova_client,
919 self.assertTrue(mock_logger_error.called)
921 def test_get_network_list_default(self):
922 self.assertEqual(openstack_utils.
923 get_network_list(self.neutron_client),
926 def test_get_network_list_missing_network(self):
927 self.assertEqual(openstack_utils.
928 get_network_list(self.empty_client),
931 def test_get_router_list_default(self):
932 self.assertEqual(openstack_utils.
933 get_router_list(self.neutron_client),
936 def test_get_router_list_missing_router(self):
937 self.assertEqual(openstack_utils.
938 get_router_list(self.empty_client),
941 def test_get_port_list_default(self):
942 self.assertEqual(openstack_utils.
943 get_port_list(self.neutron_client),
946 def test_get_port_list_missing_port(self):
947 self.assertEqual(openstack_utils.
948 get_port_list(self.empty_client),
951 def test_get_network_id_default(self):
952 self.assertEqual(openstack_utils.
953 get_network_id(self.neutron_client,
957 def test_get_subnet_id_default(self):
958 self.assertEqual(openstack_utils.
959 get_subnet_id(self.neutron_client,
963 def test_get_router_id_default(self):
964 self.assertEqual(openstack_utils.
965 get_router_id(self.neutron_client,
969 def test_get_private_net_default(self):
970 self.assertEqual(openstack_utils.
971 get_private_net(self.neutron_client),
974 def test_get_private_net_missing_net(self):
975 self.assertEqual(openstack_utils.
976 get_private_net(self.empty_client),
979 def test_get_external_net_default(self):
980 self.assertEqual(openstack_utils.
981 get_external_net(self.neutron_client),
984 def test_get_external_net_missing_net(self):
985 self.assertEqual(openstack_utils.
986 get_external_net(self.empty_client),
989 def test_get_external_net_id_default(self):
990 self.assertEqual(openstack_utils.
991 get_external_net_id(self.neutron_client),
994 def test_get_external_net_id_missing_net(self):
995 self.assertEqual(openstack_utils.
996 get_external_net_id(self.empty_client),
999 def test_check_neutron_net_default(self):
1000 self.assertTrue(openstack_utils.
1001 check_neutron_net(self.neutron_client,
1004 def test_check_neutron_net_missing_net(self):
1005 self.assertFalse(openstack_utils.
1006 check_neutron_net(self.empty_client,
1009 def test_create_neutron_net_default(self):
1010 self.assertEqual(openstack_utils.
1011 create_neutron_net(self.neutron_client,
1015 @mock.patch('functest.utils.openstack_utils.logger.error')
1016 def test_create_neutron_net_exception(self, mock_logger_error):
1017 self.assertEqual(openstack_utils.
1018 create_neutron_net(Exception,
1021 self.assertTrue(mock_logger_error.called)
1023 def test_create_neutron_subnet_default(self):
1024 self.assertEqual(openstack_utils.
1025 create_neutron_subnet(self.neutron_client,
1031 @mock.patch('functest.utils.openstack_utils.logger.error')
1032 def test_create_neutron_subnet_exception(self, mock_logger_error):
1033 self.assertEqual(openstack_utils.
1034 create_neutron_subnet(Exception,
1039 self.assertTrue(mock_logger_error.called)
1041 def test_create_neutron_router_default(self):
1042 self.assertEqual(openstack_utils.
1043 create_neutron_router(self.neutron_client,
1047 @mock.patch('functest.utils.openstack_utils.logger.error')
1048 def test_create_neutron_router_exception(self, mock_logger_error):
1049 self.assertEqual(openstack_utils.
1050 create_neutron_router(Exception,
1053 self.assertTrue(mock_logger_error.called)
1055 def test_create_neutron_port_default(self):
1056 self.assertEqual(openstack_utils.
1057 create_neutron_port(self.neutron_client,
1063 @mock.patch('functest.utils.openstack_utils.logger.error')
1064 def test_create_neutron_port_exception(self, mock_logger_error):
1065 self.assertEqual(openstack_utils.
1066 create_neutron_port(Exception,
1071 self.assertTrue(mock_logger_error.called)
1073 def test_update_neutron_net_default(self):
1074 self.assertTrue(openstack_utils.
1075 update_neutron_net(self.neutron_client,
1078 @mock.patch('functest.utils.openstack_utils.logger.error')
1079 def test_update_neutron_net_exception(self, mock_logger_error):
1080 self.assertFalse(openstack_utils.
1081 update_neutron_net(Exception,
1083 self.assertTrue(mock_logger_error.called)
1085 def test_update_neutron_port_default(self):
1086 self.assertEqual(openstack_utils.
1087 update_neutron_port(self.neutron_client,
1092 @mock.patch('functest.utils.openstack_utils.logger.error')
1093 def test_update_neutron_port_exception(self, mock_logger_error):
1094 self.assertEqual(openstack_utils.
1095 update_neutron_port(Exception,
1099 self.assertTrue(mock_logger_error.called)
1101 def test_add_interface_router_default(self):
1102 self.assertTrue(openstack_utils.
1103 add_interface_router(self.neutron_client,
1107 @mock.patch('functest.utils.openstack_utils.logger.error')
1108 def test_add_interface_router_exception(self, mock_logger_error):
1109 self.assertFalse(openstack_utils.
1110 add_interface_router(Exception,
1113 self.assertTrue(mock_logger_error.called)
1115 def test_add_gateway_router_default(self):
1116 with mock.patch('functest.utils.openstack_utils.'
1117 'get_external_net_id',
1118 return_value='network_id'):
1119 self.assertTrue(openstack_utils.
1120 add_gateway_router(self.neutron_client,
1123 @mock.patch('functest.utils.openstack_utils.logger.error')
1124 def test_add_gateway_router_exception(self, mock_logger_error):
1125 with mock.patch('functest.utils.openstack_utils.'
1126 'get_external_net_id',
1127 return_value='network_id'):
1128 self.assertFalse(openstack_utils.
1129 add_gateway_router(Exception,
1131 self.assertTrue(mock_logger_error.called)
1133 def test_delete_neutron_net_default(self):
1134 self.assertTrue(openstack_utils.
1135 delete_neutron_net(self.neutron_client,
1138 @mock.patch('functest.utils.openstack_utils.logger.error')
1139 def test_delete_neutron_net_exception(self, mock_logger_error):
1140 self.assertFalse(openstack_utils.
1141 delete_neutron_net(Exception,
1143 self.assertTrue(mock_logger_error.called)
1145 def test_delete_neutron_subnet_default(self):
1146 self.assertTrue(openstack_utils.
1147 delete_neutron_subnet(self.neutron_client,
1150 @mock.patch('functest.utils.openstack_utils.logger.error')
1151 def test_delete_neutron_subnet_exception(self, mock_logger_error):
1152 self.assertFalse(openstack_utils.
1153 delete_neutron_subnet(Exception,
1155 self.assertTrue(mock_logger_error.called)
1157 def test_delete_neutron_router_default(self):
1158 self.assertTrue(openstack_utils.
1159 delete_neutron_router(self.neutron_client,
1162 @mock.patch('functest.utils.openstack_utils.logger.error')
1163 def test_delete_neutron_router_exception(self, mock_logger_error):
1164 self.assertFalse(openstack_utils.
1165 delete_neutron_router(Exception,
1167 self.assertTrue(mock_logger_error.called)
1169 def test_delete_neutron_port_default(self):
1170 self.assertTrue(openstack_utils.
1171 delete_neutron_port(self.neutron_client,
1174 @mock.patch('functest.utils.openstack_utils.logger.error')
1175 def test_delete_neutron_port_exception(self, mock_logger_error):
1176 self.assertFalse(openstack_utils.
1177 delete_neutron_port(Exception,
1179 self.assertTrue(mock_logger_error.called)
1181 def test_remove_interface_router_default(self):
1182 self.assertTrue(openstack_utils.
1183 remove_interface_router(self.neutron_client,
1187 @mock.patch('functest.utils.openstack_utils.logger.error')
1188 def test_remove_interface_router_exception(self, mock_logger_error):
1189 self.assertFalse(openstack_utils.
1190 remove_interface_router(Exception,
1193 self.assertTrue(mock_logger_error.called)
1195 def test_remove_gateway_router_default(self):
1196 self.assertTrue(openstack_utils.
1197 remove_gateway_router(self.neutron_client,
1200 @mock.patch('functest.utils.openstack_utils.logger.error')
1201 def test_remove_gateway_router_exception(self, mock_logger_error):
1202 self.assertFalse(openstack_utils.
1203 remove_gateway_router(Exception,
1205 self.assertTrue(mock_logger_error.called)
1207 def test_get_security_groups_default(self):
1208 self.assertEqual(openstack_utils.
1209 get_security_groups(self.neutron_client),
1212 @mock.patch('functest.utils.openstack_utils.logger.error')
1213 def test_get_security_groups_exception(self, mock_logger_error):
1214 self.assertEqual(openstack_utils.
1215 get_security_groups(Exception),
1217 self.assertTrue(mock_logger_error.called)
1219 def test_get_security_group_id_default(self):
1220 with mock.patch('functest.utils.openstack_utils.'
1221 'get_security_groups',
1222 return_value=[self.sec_group]):
1223 self.assertEqual(openstack_utils.
1224 get_security_group_id(self.neutron_client,
1228 def test_get_security_group_rules_default(self):
1229 self.assertEqual(openstack_utils.
1230 get_security_group_rules(self.neutron_client,
1231 self.sec_group['id']),
1232 [self.sec_group_rule])
1234 @mock.patch('functest.utils.openstack_utils.logger.error')
1235 def test_get_security_group_rules_exception(self, mock_logger_error):
1236 self.assertEqual(openstack_utils.
1237 get_security_group_rules(Exception,
1240 self.assertTrue(mock_logger_error.called)
1242 def test_check_security_group_rules_not_exists(self):
1243 self.assertEqual(openstack_utils.
1244 check_security_group_rules(self.neutron_client,
1252 def test_check_security_group_rules_exists(self):
1253 self.assertEqual(openstack_utils.
1254 check_security_group_rules(self.neutron_client,
1255 self.sec_group['id'],
1262 @mock.patch('functest.utils.openstack_utils.logger.error')
1263 def test_check_security_group_rules_exception(self, mock_logger_error):
1264 self.assertEqual(openstack_utils.
1265 check_security_group_rules(Exception,
1272 self.assertTrue(mock_logger_error.called)
1274 def test_create_security_group_default(self):
1275 self.assertEqual(openstack_utils.
1276 create_security_group(self.neutron_client,
1281 @mock.patch('functest.utils.openstack_utils.logger.error')
1282 def test_create_security_group_exception(self, mock_logger_error):
1283 self.assertEqual(openstack_utils.
1284 create_security_group(Exception,
1288 self.assertTrue(mock_logger_error.called)
1290 def test_create_secgroup_rule_default(self):
1291 self.assertTrue(openstack_utils.
1292 create_secgroup_rule(self.neutron_client,
1298 self.assertTrue(openstack_utils.
1299 create_secgroup_rule(self.neutron_client,
1304 @mock.patch('functest.utils.openstack_utils.logger.error')
1305 def test_create_secgroup_rule_invalid_port_range(self, mock_logger_error):
1306 self.assertFalse(openstack_utils.
1307 create_secgroup_rule(self.neutron_client,
1313 @mock.patch('functest.utils.openstack_utils.logger.error')
1314 def test_create_secgroup_rule_exception(self, mock_logger_error):
1315 self.assertFalse(openstack_utils.
1316 create_secgroup_rule(Exception,
1321 @mock.patch('functest.utils.openstack_utils.logger.info')
1322 def test_create_security_group_full_default(self, mock_logger_info):
1323 with mock.patch('functest.utils.openstack_utils.'
1324 'get_security_group_id',
1325 return_value='sg_id'):
1326 self.assertEqual(openstack_utils.
1327 create_security_group_full(self.neutron_client,
1331 self.assertTrue(mock_logger_info)
1333 @mock.patch('functest.utils.openstack_utils.logger.info')
1334 @mock.patch('functest.utils.openstack_utils.logger.error')
1335 def test_create_security_group_full_sec_group_fail(self,
1338 with mock.patch('functest.utils.openstack_utils.'
1339 'get_security_group_id',
1341 mock.patch('functest.utils.openstack_utils.'
1342 'create_security_group',
1343 return_value=False):
1344 self.assertEqual(openstack_utils.
1345 create_security_group_full(self.neutron_client,
1349 self.assertTrue(mock_logger_error)
1350 self.assertTrue(mock_logger_info)
1352 @mock.patch('functest.utils.openstack_utils.logger.debug')
1353 @mock.patch('functest.utils.openstack_utils.logger.info')
1354 @mock.patch('functest.utils.openstack_utils.logger.error')
1355 def test_create_security_group_full_secgroup_rule_fail(self,
1359 with mock.patch('functest.utils.openstack_utils.'
1360 'get_security_group_id',
1362 mock.patch('functest.utils.openstack_utils.'
1363 'create_security_group',
1364 return_value={'id': 'sg_id',
1365 'name': 'sg_name'}), \
1366 mock.patch('functest.utils.openstack_utils.'
1367 'create_secgroup_rule',
1368 return_value=False):
1369 self.assertEqual(openstack_utils.
1370 create_security_group_full(self.neutron_client,
1374 self.assertTrue(mock_logger_error)
1375 self.assertTrue(mock_logger_info)
1376 self.assertTrue(mock_logger_debug)
1378 def test_add_secgroup_to_instance_default(self):
1379 self.assertTrue(openstack_utils.
1380 add_secgroup_to_instance(self.nova_client,
1384 @mock.patch('functest.utils.openstack_utils.logger.error')
1385 def test_add_secgroup_to_instance_exception(self, mock_logger_error):
1386 self.assertFalse(openstack_utils.
1387 add_secgroup_to_instance(Exception,
1390 self.assertTrue(mock_logger_error.called)
1392 def test_update_sg_quota_default(self):
1393 self.assertTrue(openstack_utils.
1394 update_sg_quota(self.neutron_client,
1399 @mock.patch('functest.utils.openstack_utils.logger.error')
1400 def test_update_sg_quota_exception(self, mock_logger_error):
1401 self.assertFalse(openstack_utils.
1402 update_sg_quota(Exception,
1406 self.assertTrue(mock_logger_error.called)
1408 def test_delete_security_group_default(self):
1409 self.assertTrue(openstack_utils.
1410 delete_security_group(self.neutron_client,
1413 @mock.patch('functest.utils.openstack_utils.logger.error')
1414 def test_delete_security_group_exception(self, mock_logger_error):
1415 self.assertFalse(openstack_utils.
1416 delete_security_group(Exception,
1418 self.assertTrue(mock_logger_error.called)
1420 def test_get_images_default(self):
1421 self.assertEqual(openstack_utils.
1422 get_images(self.glance_client),
1425 @mock.patch('functest.utils.openstack_utils.logger.error')
1426 def test_get_images_exception(self, mock_logger_error):
1427 self.assertEqual(openstack_utils.
1428 get_images(Exception),
1430 self.assertTrue(mock_logger_error.called)
1432 def test_get_image_id_default(self):
1433 self.assertEqual(openstack_utils.
1434 get_image_id(self.glance_client,
1438 # create_glance_image, get_or_create_image
1439 @mock.patch('functest.utils.openstack_utils.logger.error')
1440 def test_create_glance_image_file_present(self, mock_logger_error):
1441 with mock.patch('functest.utils.openstack_utils.'
1443 return_value=False):
1444 self.assertEqual(openstack_utils.
1445 create_glance_image(self.glance_client,
1449 self.assertTrue(mock_logger_error.called)
1451 @mock.patch('functest.utils.openstack_utils.logger.info')
1452 def test_create_glance_image_already_exist(self, mock_logger_info):
1453 with mock.patch('functest.utils.openstack_utils.'
1455 return_value=True), \
1456 mock.patch('functest.utils.openstack_utils.get_image_id',
1457 return_value='image_id'):
1458 self.assertEqual(openstack_utils.
1459 create_glance_image(self.glance_client,
1463 self.assertTrue(mock_logger_info.called)
1465 @mock.patch('functest.utils.openstack_utils.logger.info')
1466 def test_create_glance_image_default(self, mock_logger_info):
1467 with mock.patch('functest.utils.openstack_utils.'
1469 return_value=True), \
1470 mock.patch('functest.utils.openstack_utils.get_image_id',
1472 mock.patch('__builtin__.open',
1473 mock.mock_open(read_data='1')) as m:
1474 self.assertEqual(openstack_utils.
1475 create_glance_image(self.glance_client,
1479 m.assert_called_once_with('file_path')
1480 self.assertTrue(mock_logger_info.called)
1482 @mock.patch('functest.utils.openstack_utils.logger.error')
1483 def test_create_glance_image_exception(self, mock_logger_error):
1484 with mock.patch('functest.utils.openstack_utils.'
1486 return_value=True), \
1487 mock.patch('functest.utils.openstack_utils.get_image_id',
1488 side_effect=Exception):
1489 self.assertEqual(openstack_utils.
1490 create_glance_image(self.glance_client,
1494 self.assertTrue(mock_logger_error.called)
1496 def test_delete_glance_image_default(self):
1497 self.assertTrue(openstack_utils.
1498 delete_glance_image(self.nova_client,
1501 @mock.patch('functest.utils.openstack_utils.logger.error')
1502 def test_delete_glance_image_exception(self, mock_logger_error):
1503 self.assertFalse(openstack_utils.
1504 delete_glance_image(Exception,
1506 self.assertTrue(mock_logger_error.called)
1508 def test_get_volumes_default(self):
1509 self.assertEqual(openstack_utils.
1510 get_volumes(self.cinder_client),
1513 @mock.patch('functest.utils.openstack_utils.logger.error')
1514 def test_get_volumes_exception(self, mock_logger_error):
1515 self.assertEqual(openstack_utils.
1516 get_volumes(Exception),
1518 self.assertTrue(mock_logger_error.called)
1520 def test_list_volume_types_default_private(self):
1521 self.assertEqual(openstack_utils.
1522 list_volume_types(self.cinder_client,
1525 [self.volume_types[1]])
1527 def test_list_volume_types_default_public(self):
1528 self.assertEqual(openstack_utils.
1529 list_volume_types(self.cinder_client,
1532 [self.volume_types[0]])
1534 @mock.patch('functest.utils.openstack_utils.logger.error')
1535 def test_list_volume_types_exception(self, mock_logger_error):
1536 self.assertEqual(openstack_utils.
1537 list_volume_types(Exception),
1539 self.assertTrue(mock_logger_error.called)
1541 def test_create_volume_type_default(self):
1542 self.assertEqual(openstack_utils.
1543 create_volume_type(self.cinder_client,
1544 'test_volume_type'),
1545 self.volume_types[0])
1547 @mock.patch('functest.utils.openstack_utils.logger.error')
1548 def test_create_volume_type_exception(self, mock_logger_error):
1549 self.assertEqual(openstack_utils.
1550 create_volume_type(Exception,
1551 'test_volume_type'),
1553 self.assertTrue(mock_logger_error.called)
1555 def test_update_cinder_quota_default(self):
1556 self.assertTrue(openstack_utils.
1557 update_cinder_quota(self.cinder_client,
1563 @mock.patch('functest.utils.openstack_utils.logger.error')
1564 def test_update_cinder_quota_exception(self, mock_logger_error):
1565 self.assertFalse(openstack_utils.
1566 update_cinder_quota(Exception,
1571 self.assertTrue(mock_logger_error.called)
1573 def test_delete_volume_default(self):
1574 self.assertTrue(openstack_utils.
1575 delete_volume(self.cinder_client,
1579 self.assertTrue(openstack_utils.
1580 delete_volume(self.cinder_client,
1584 @mock.patch('functest.utils.openstack_utils.logger.error')
1585 def test_delete_volume_exception(self, mock_logger_error):
1586 self.assertFalse(openstack_utils.
1587 delete_volume(Exception,
1590 self.assertTrue(mock_logger_error.called)
1592 def test_delete_volume_type_default(self):
1593 self.assertTrue(openstack_utils.
1594 delete_volume_type(self.cinder_client,
1595 self.volume_types[0]))
1597 @mock.patch('functest.utils.openstack_utils.logger.error')
1598 def test_delete_volume_type_exception(self, mock_logger_error):
1599 self.assertFalse(openstack_utils.
1600 delete_volume_type(Exception,
1601 self.volume_types[0]))
1602 self.assertTrue(mock_logger_error.called)
1604 def test_get_tenants_default(self):
1605 with mock.patch('functest.utils.openstack_utils.'
1606 'is_keystone_v3', return_value=True):
1607 self.assertEqual(openstack_utils.
1608 get_tenants(self.keystone_client),
1610 with mock.patch('functest.utils.openstack_utils.'
1611 'is_keystone_v3', return_value=False):
1612 self.assertEqual(openstack_utils.
1613 get_tenants(self.keystone_client),
1616 @mock.patch('functest.utils.openstack_utils.logger.error')
1617 def test_get_tenants_exception(self, mock_logger_error):
1618 self.assertEqual(openstack_utils.
1619 get_tenants(Exception),
1621 self.assertTrue(mock_logger_error.called)
1623 def test_get_users_default(self):
1624 self.assertEqual(openstack_utils.
1625 get_users(self.keystone_client),
1628 @mock.patch('functest.utils.openstack_utils.logger.error')
1629 def test_get_users_exception(self, mock_logger_error):
1630 self.assertEqual(openstack_utils.
1631 get_users(Exception),
1633 self.assertTrue(mock_logger_error.called)
1635 def test_get_tenant_id_default(self):
1636 self.assertEqual(openstack_utils.
1637 get_tenant_id(self.keystone_client,
1641 def test_get_user_id_default(self):
1642 self.assertEqual(openstack_utils.
1643 get_user_id(self.keystone_client,
1647 def test_get_role_id_default(self):
1648 self.assertEqual(openstack_utils.
1649 get_role_id(self.keystone_client,
1653 def test_create_tenant_default(self):
1654 with mock.patch('functest.utils.openstack_utils.'
1655 'is_keystone_v3', return_value=True):
1656 self.assertEqual(openstack_utils.
1657 create_tenant(self.keystone_client,
1661 with mock.patch('functest.utils.openstack_utils.'
1662 'is_keystone_v3', return_value=False):
1663 self.assertEqual(openstack_utils.
1664 create_tenant(self.keystone_client,
1669 @mock.patch('functest.utils.openstack_utils.logger.error')
1670 def test_create_tenant_exception(self, mock_logger_error):
1671 self.assertEqual(openstack_utils.
1672 create_tenant(Exception,
1676 self.assertTrue(mock_logger_error.called)
1678 def test_create_user_default(self):
1679 with mock.patch('functest.utils.openstack_utils.'
1680 'is_keystone_v3', return_value=True):
1681 self.assertEqual(openstack_utils.
1682 create_user(self.keystone_client,
1688 with mock.patch('functest.utils.openstack_utils.'
1689 'is_keystone_v3', return_value=False):
1690 self.assertEqual(openstack_utils.
1691 create_user(self.keystone_client,
1698 @mock.patch('functest.utils.openstack_utils.logger.error')
1699 def test_create_user_exception(self, mock_logger_error):
1700 self.assertEqual(openstack_utils.
1701 create_user(Exception,
1707 self.assertTrue(mock_logger_error.called)
1709 def test_add_role_user_default(self):
1710 with mock.patch('functest.utils.openstack_utils.'
1711 'is_keystone_v3', return_value=True):
1712 self.assertTrue(openstack_utils.
1713 add_role_user(self.keystone_client,
1718 with mock.patch('functest.utils.openstack_utils.'
1719 'is_keystone_v3', return_value=False):
1720 self.assertTrue(openstack_utils.
1721 add_role_user(self.keystone_client,
1726 @mock.patch('functest.utils.openstack_utils.logger.error')
1727 def test_add_role_user_exception(self, mock_logger_error):
1728 self.assertFalse(openstack_utils.
1729 add_role_user(Exception,
1733 self.assertTrue(mock_logger_error.called)
1735 def test_delete_tenant_default(self):
1736 with mock.patch('functest.utils.openstack_utils.'
1737 'is_keystone_v3', return_value=True):
1738 self.assertTrue(openstack_utils.
1739 delete_tenant(self.keystone_client,
1742 with mock.patch('functest.utils.openstack_utils.'
1743 'is_keystone_v3', return_value=False):
1744 self.assertTrue(openstack_utils.
1745 delete_tenant(self.keystone_client,
1748 @mock.patch('functest.utils.openstack_utils.logger.error')
1749 def test_delete_tenant_exception(self, mock_logger_error):
1750 self.assertFalse(openstack_utils.
1751 delete_tenant(Exception,
1753 self.assertTrue(mock_logger_error.called)
1755 def test_delete_user_default(self):
1756 self.assertTrue(openstack_utils.
1757 delete_user(self.keystone_client,
1760 @mock.patch('functest.utils.openstack_utils.logger.error')
1761 def test_delete_user_exception(self, mock_logger_error):
1762 self.assertFalse(openstack_utils.
1763 delete_user(Exception,
1765 self.assertTrue(mock_logger_error.called)
1767 def test_get_resource_default(self):
1768 with mock.patch('functest.utils.openstack_utils.'
1769 'is_keystone_v3', return_value=True):
1770 self.assertEqual(openstack_utils.
1771 get_resource(self.heat_client,
1776 @mock.patch('functest.utils.openstack_utils.logger.error')
1777 def test_get_resource_exception(self, mock_logger_error):
1778 self.assertEqual(openstack_utils.
1779 get_resource(Exception,
1783 self.assertTrue(mock_logger_error.called)
1785 def test_get_or_create_user_for_vnf_get(self):
1786 with mock.patch('functest.utils.openstack_utils.'
1788 return_value='user_id'), \
1789 mock.patch('functest.utils.openstack_utils.get_tenant_id',
1790 return_value='tenant_id'):
1791 self.assertFalse(openstack_utils.
1792 get_or_create_user_for_vnf(self.keystone_client,
1795 def test_get_or_create_user_for_vnf_create(self):
1796 with mock.patch('functest.utils.openstack_utils.'
1798 return_value=None), \
1799 mock.patch('functest.utils.openstack_utils.get_tenant_id',
1800 return_value='tenant_id'):
1801 self.assertTrue(openstack_utils.
1802 get_or_create_user_for_vnf(self.keystone_client,
1805 def test_get_or_create_user_for_vnf_error_get_user_id(self):
1806 with mock.patch('functest.utils.openstack_utils.'
1808 side_effect=Exception):
1809 self.assertRaises(Exception)
1811 def test_get_or_create_user_for_vnf_error_get_tenant_id(self):
1812 with mock.patch('functest.utils.openstack_utils.'
1814 return_value='user_id'), \
1815 mock.patch('functest.utils.openstack_utils.get_tenant_id',
1816 side_effect='Exception'):
1817 self.assertRaises(Exception)
1819 def test_get_or_create_tenant_for_vnf_get(self):
1820 with mock.patch('functest.utils.openstack_utils.'
1822 return_value='tenant_id'):
1824 openstack_utils.get_or_create_tenant_for_vnf(
1825 self.keystone_client, 'tenant_name', 'tenant_description'))
1827 def test_get_or_create_tenant_for_vnf_create(self):
1828 with mock.patch('functest.utils.openstack_utils.get_tenant_id',
1831 openstack_utils.get_or_create_tenant_for_vnf(
1832 self.keystone_client, 'tenant_name', 'tenant_description'))
1834 def test_get_or_create_tenant_for_vnf_error_get_tenant_id(self):
1835 with mock.patch('functest.utils.openstack_utils.'
1837 side_effect=Exception):
1838 self.assertRaises(Exception)
1840 def test_download_and_add_image_on_glance_image_creation_failure(self):
1841 with mock.patch('functest.utils.openstack_utils.'
1843 mock.patch('functest.utils.openstack_utils.'
1844 'ft_utils.download_url',
1845 return_value=True), \
1846 mock.patch('functest.utils.openstack_utils.'
1847 'create_glance_image',
1849 resp = openstack_utils.download_and_add_image_on_glance(
1854 self.assertEqual(resp, False)
1857 if __name__ == "__main__":
1858 logging.disable(logging.CRITICAL)
1859 unittest.main(verbosity=2)