import copy
import logging
+import os
import unittest
import mock
class OSUtilsTesting(unittest.TestCase):
- logging.disable(logging.CRITICAL)
-
def _get_env_cred_dict(self, os_prefix=''):
return {'OS_USERNAME': os_prefix + 'username',
'OS_PASSWORD': os_prefix + 'password',
'OS_PROJECT_DOMAIN_NAME': os_prefix + 'project_domain_name',
'OS_PROJECT_NAME': os_prefix + 'project_name',
'OS_ENDPOINT_TYPE': os_prefix + 'endpoint_type',
- 'OS_REGION_NAME': os_prefix + 'region_name'}
+ 'OS_REGION_NAME': os_prefix + 'region_name',
+ 'OS_CACERT': os_prefix + 'https_cacert'}
def _get_os_env_vars(self):
return {'username': 'test_username', 'password': 'test_password',
'project_domain_name': 'test_project_domain_name',
'project_name': 'test_project_name',
'endpoint_type': 'test_endpoint_type',
- 'region_name': 'test_region_name'}
+ 'region_name': 'test_region_name',
+ 'https_cacert': 'test_https_cacert'}
def setUp(self):
self.env_vars = ['OS_AUTH_URL', 'OS_USERNAME', 'OS_PASSWORD']
'servers.create.return_value': self.instance,
'flavors.list.return_value': [self.flavor],
'flavors.find.return_value': self.flavor,
- 'flavors.list.return_value': [self.flavor],
'servers.add_floating_ip.return_value': mock.Mock(),
'servers.force_delete.return_value': mock.Mock(),
'aggregates.list.return_value': [self.aggregate],
}
self.cinder_client.configure_mock(**attrs)
+ self.resource = mock.Mock()
+ attrs = {'id': 'resource_test_id',
+ 'name': 'resource_test_name'
+ }
+
+ self.heat_client = mock.Mock()
+ attrs = {'resources.get.return_value': self.resource}
+ self.heat_client.configure_mock(**attrs)
+
mock_obj = mock.Mock()
attrs = {'id': 'tenant_id',
'name': 'test_tenant'}
self.sec_group = {'id': 'sec_group_id',
'name': 'test_sec_group'}
+ self.sec_group_rule = {'id': 'sec_group_rule_id',
+ 'direction': 'direction',
+ 'protocol': 'protocol',
+ 'port_range_max': 'port_max',
+ 'security_group_id': self.sec_group['id'],
+ 'port_range_min': 'port_min'}
self.neutron_floatingip = {'id': 'fip_id',
'floating_ip_address': 'test_ip'}
self.neutron_client = mock.Mock()
'show_bgpvpn.return_value': self.mock_return,
'list_security_groups.return_value': {'security_groups':
[self.sec_group]},
+ 'list_security_group_rules.'
+ 'return_value': {'security_group_rules':
+ [self.sec_group_rule]},
'create_security_group_rule.return_value': mock.Mock(),
'create_security_group.return_value': {'security_group':
self.sec_group},
def test_get_credentials_missing_endpoint_type(self):
self._get_credentials_missing_env('OS_ENDPOINT_TYPE')
+ def _test_source_credentials(self, msg, key='OS_TENANT_NAME',
+ value='admin'):
+ try:
+ del os.environ[key]
+ except:
+ pass
+ f = 'rc_file'
+ with mock.patch('__builtin__.open', mock.mock_open(read_data=msg),
+ create=True) as m:
+ m.return_value.__iter__ = lambda self: iter(self.readline, '')
+ openstack_utils.source_credentials(f)
+ m.assert_called_once_with(f, 'r')
+ self.assertEqual(os.environ[key], value)
+
def test_source_credentials(self):
- with mock.patch('functest.utils.openstack_utils.subprocess.Popen') \
- as mock_subproc_popen, \
- mock.patch('functest.utils.openstack_utils.os.environ'):
- process_mock = mock.Mock()
- attrs = {'communicate.return_value': ('OS_USER_NAME=test_name',
- 'success')}
- process_mock.configure_mock(**attrs)
- mock_subproc_popen.return_value = process_mock
-
- self.assertDictEqual(openstack_utils.source_credentials('rc_file'),
- {'OS_USER_NAME': 'test_name'})
+ self._test_source_credentials('OS_TENANT_NAME=admin')
+ self._test_source_credentials('OS_TENANT_NAME= admin')
+ self._test_source_credentials('OS_TENANT_NAME = admin')
+ self._test_source_credentials('OS_TENANT_NAME = "admin"')
+ self._test_source_credentials('export OS_TENANT_NAME=admin')
+ self._test_source_credentials('export OS_TENANT_NAME =admin')
+ self._test_source_credentials('export OS_TENANT_NAME = admin')
+ self._test_source_credentials('export OS_TENANT_NAME = "admin"')
+ self._test_source_credentials('OS_TENANT_NAME', value='')
+ self._test_source_credentials('export OS_TENANT_NAME', value='')
+ # This test will fail as soon as rc_file is fixed
+ self._test_source_credentials(
+ 'export "\'OS_TENANT_NAME\'" = "\'admin\'"')
@mock.patch('functest.utils.openstack_utils.os.getenv',
return_value=None)
mock_logger_info.assert_called_once_with("OS_IDENTITY_API_VERSION is "
"set in env as '%s'", '3')
- def test_get_keystone_client(self):
+ @mock.patch('functest.utils.openstack_utils.get_session')
+ @mock.patch('functest.utils.openstack_utils.keystoneclient.Client')
+ @mock.patch('functest.utils.openstack_utils.get_keystone_client_version',
+ return_value='3')
+ @mock.patch('functest.utils.openstack_utils.os.getenv',
+ return_value='public')
+ def test_get_keystone_client_with_interface(self, mock_os_getenv,
+ mock_keystoneclient_version,
+ mock_key_client,
+ mock_get_session):
mock_keystone_obj = mock.Mock()
mock_session_obj = mock.Mock()
- with mock.patch('functest.utils.openstack_utils'
- '.get_keystone_client_version', return_value='3'), \
- mock.patch('functest.utils.openstack_utils'
- '.keystoneclient.Client',
- return_value=mock_keystone_obj) \
- as mock_key_client, \
- mock.patch('functest.utils.openstack_utils.get_session',
- return_value=mock_session_obj):
- self.assertEqual(openstack_utils.get_keystone_client(),
- mock_keystone_obj)
- mock_key_client.assert_called_once_with('3',
- session=mock_session_obj)
+ mock_key_client.return_value = mock_keystone_obj
+ mock_get_session.return_value = mock_session_obj
+ self.assertEqual(openstack_utils.get_keystone_client(),
+ mock_keystone_obj)
+ mock_key_client.assert_called_once_with('3',
+ session=mock_session_obj,
+ interface='public')
+
+ @mock.patch('functest.utils.openstack_utils.get_session')
+ @mock.patch('functest.utils.openstack_utils.keystoneclient.Client')
+ @mock.patch('functest.utils.openstack_utils.get_keystone_client_version',
+ return_value='3')
+ @mock.patch('functest.utils.openstack_utils.os.getenv',
+ return_value='admin')
+ def test_get_keystone_client_no_interface(self, mock_os_getenv,
+ mock_keystoneclient_version,
+ mock_key_client,
+ mock_get_session):
+ mock_keystone_obj = mock.Mock()
+ mock_session_obj = mock.Mock()
+ mock_key_client.return_value = mock_keystone_obj
+ mock_get_session.return_value = mock_session_obj
+ self.assertEqual(openstack_utils.get_keystone_client(),
+ mock_keystone_obj)
+ mock_key_client.assert_called_once_with('3',
+ session=mock_session_obj,
+ interface='admin')
@mock.patch('functest.utils.openstack_utils.os.getenv',
return_value=None)
mock_glan_client.assert_called_once_with('3',
session=mock_session_obj)
+ @mock.patch('functest.utils.openstack_utils.os.getenv',
+ return_value=None)
+ def test_get_heat_client_version_missing_env(self, mock_os_getenv):
+ self.assertEqual(openstack_utils.get_heat_client_version(),
+ openstack_utils.DEFAULT_HEAT_API_VERSION)
+
+ @mock.patch('functest.utils.openstack_utils.logger.info')
+ @mock.patch('functest.utils.openstack_utils.os.getenv', return_value='1')
+ def test_get_heat_client_version_default(self, mock_os_getenv,
+ mock_logger_info):
+ self.assertEqual(openstack_utils.get_heat_client_version(), '1')
+ mock_logger_info.assert_called_once_with(
+ "OS_ORCHESTRATION_API_VERSION is set in env as '%s'", '1')
+
+ def test_get_heat_client(self):
+ mock_heat_obj = mock.Mock()
+ mock_session_obj = mock.Mock()
+ with mock.patch('functest.utils.openstack_utils'
+ '.get_heat_client_version', return_value='1'), \
+ mock.patch('functest.utils.openstack_utils'
+ '.heatclient.Client',
+ return_value=mock_heat_obj) \
+ as mock_heat_client, \
+ mock.patch('functest.utils.openstack_utils.get_session',
+ return_value=mock_session_obj):
+ self.assertEqual(openstack_utils.get_heat_client(),
+ mock_heat_obj)
+ mock_heat_client.assert_called_once_with('1',
+ session=mock_session_obj)
+
def test_get_instances_default(self):
self.assertEqual(openstack_utils.get_instances(self.nova_client),
[self.instance])
'test_sec_group'),
'sec_group_id')
+ def test_get_security_group_rules_default(self):
+ self.assertEqual(openstack_utils.
+ get_security_group_rules(self.neutron_client,
+ self.sec_group['id']),
+ [self.sec_group_rule])
+
+ @mock.patch('functest.utils.openstack_utils.logger.error')
+ def test_get_security_group_rules_exception(self, mock_logger_error):
+ self.assertEqual(openstack_utils.
+ get_security_group_rules(Exception,
+ 'sec_group_id'),
+ None)
+ self.assertTrue(mock_logger_error.called)
+
+ def test_check_security_group_rules_not_exists(self):
+ self.assertEqual(openstack_utils.
+ check_security_group_rules(self.neutron_client,
+ 'sec_group_id_2',
+ 'direction',
+ 'protocol',
+ 'port_min',
+ 'port_max'),
+ True)
+
+ def test_check_security_group_rules_exists(self):
+ self.assertEqual(openstack_utils.
+ check_security_group_rules(self.neutron_client,
+ self.sec_group['id'],
+ 'direction',
+ 'protocol',
+ 'port_min',
+ 'port_max'),
+ False)
+
+ @mock.patch('functest.utils.openstack_utils.logger.error')
+ def test_check_security_group_rules_exception(self, mock_logger_error):
+ self.assertEqual(openstack_utils.
+ check_security_group_rules(Exception,
+ 'sec_group_id',
+ 'direction',
+ 'protocol',
+ 'port_max',
+ 'port_min'),
+ None)
+ self.assertTrue(mock_logger_error.called)
+
def test_create_security_group_default(self):
self.assertEqual(openstack_utils.
create_security_group(self.neutron_client,
'user_id'))
self.assertTrue(mock_logger_error.called)
+ def test_get_resource_default(self):
+ with mock.patch('functest.utils.openstack_utils.'
+ 'is_keystone_v3', return_value=True):
+ self.assertEqual(openstack_utils.
+ get_resource(self.heat_client,
+ 'stack_id',
+ 'resource'),
+ self.resource)
+
+ @mock.patch('functest.utils.openstack_utils.logger.error')
+ def test_get_resource_exception(self, mock_logger_error):
+ self.assertEqual(openstack_utils.
+ get_resource(Exception,
+ 'stack_id',
+ 'resource'),
+ None)
+ self.assertTrue(mock_logger_error.called)
+
if __name__ == "__main__":
+ logging.disable(logging.CRITICAL)
unittest.main(verbosity=2)
Code Review / functest.git / blobdiff