#!/usr/bin/env python # # Copyright (c) 2018 mokats@intracom-telecom.com and others. # # All rights reserved. This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 ## import io import unittest from mock import patch, call, Mock from dovetail import constants from dovetail.utils import dovetail_utils __author__ = 'Stamatis Katsaounis ' class DovetailUtilsTesting(unittest.TestCase): def setUp(self): pass def tearDown(self): pass @patch('sys.stdout') @patch('__builtin__.print') def test_exec_log_no_verbose(self, mock_print, mock_stdout): dovetail_utils.exec_log(verbose=False, logger=None, msg='', level='info', flush=True) mock_print.assert_not_called() mock_stdout.flush.assert_not_called() @patch('sys.stdout') @patch('__builtin__.print') def test_exec_log_no_logger_flush(self, mock_print, mock_stdout): message = 'message' dovetail_utils.exec_log(verbose=True, logger=False, msg=message, level='info', flush=True) mock_print.assert_has_calls([call(message)]) mock_stdout.flush.assert_called_once() @patch('sys.stdout') @patch('__builtin__.print') def test_exec_log_no_logger_no_flush(self, mock_print, mock_stdout): message = 'message' dovetail_utils.exec_log(verbose=True, logger=False, msg=message, level='info', flush=False) mock_print.assert_has_calls([call(message)]) mock_stdout.flush.assert_not_called() def test_exec_log_logger_info(self): message = 'message' logger = Mock() dovetail_utils.exec_log(verbose=True, logger=logger, msg=message, level='info', flush=True) logger.info.assert_called_once_with(message) def test_exec_log_logger_error(self): message = 'message' logger = Mock() dovetail_utils.exec_log(verbose=True, logger=logger, msg=message, level='error', flush=True) logger.error.assert_called_once_with(message) def test_exec_log_logger_debug(self): message = 'message' logger = Mock() dovetail_utils.exec_log(verbose=True, logger=logger, msg=message, level='debug', flush=True) logger.debug.assert_called_once_with(message) def test_get_value_from_dict(self): key_path = 'a.b' input_dict = {'a': {'b': 'c'}} expected = 'c' result = dovetail_utils.get_value_from_dict(key_path, input_dict) self.assertEqual(expected, result) def test_get_value_from_dict_key_path_not_str(self): key_path = 1 input_dict = {'a': {'b': 'c'}} expected = None result = dovetail_utils.get_value_from_dict(key_path, input_dict) self.assertEqual(expected, result) def test_get_value_from_dict_input_dict_not_dict(self): key_path = 'a.b' input_dict = 'dictionary' expected = None result = dovetail_utils.get_value_from_dict(key_path, input_dict) self.assertEqual(expected, result) def test_get_value_from_dict_no_value_for_key_in_key_path(self): key_path = 'a.b' input_dict = {'a': {'c': 'b'}} expected = None result = dovetail_utils.get_value_from_dict(key_path, input_dict) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_read_plain_file_not_exist(self, mock_path): file_path = 'unknown_file' logger = Mock() mock_path.isfile.return_value = False expected = None result = dovetail_utils.read_plain_file(file_path, logger) mock_path.isfile.assert_called_once_with(file_path) logger.error.assert_called_once_with("File {} doesn't exist." .format(file_path)) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path', autospec=True) def test_read_plain_file(self, mock_path, mock_open): file_path = 'known_file' file_data = u'file data' mock_path.isfile.return_value = True mock_open.return_value.__enter__.return_value = io.StringIO(file_data) expected = file_data result = dovetail_utils.read_plain_file(file_path) mock_path.isfile.assert_called_once_with(file_path) mock_open.assert_called_once_with(file_path, 'r') self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path', autospec=True) def test_read_plain_file_raised_exception(self, mock_path, mock_open): logger = Mock() file_path = 'known_file' errorMSG = 'Exception was raised' exception = Exception(errorMSG) mock_open.side_effect = exception expected = None result = dovetail_utils.read_plain_file(file_path, logger) mock_open.assert_called_once_with(file_path, 'r') logger.exception.assert_called_once_with( 'Failed to read file {}, exception: {}' .format(file_path, exception)) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_read_yaml_file_not_exist(self, mock_path): file_path = 'unknown_file' logger = Mock() mock_path.isfile.return_value = False expected = None result = dovetail_utils.read_yaml_file(file_path, logger) mock_path.isfile.assert_called_once_with(file_path) logger.error.assert_called_once_with("File {} doesn't exist." .format(file_path)) self.assertEqual(expected, result) @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_read_yaml_file(self, mock_path, mock_open, mock_load): file_obj = Mock() file_path = 'known_file' file_data = 'file data' mock_path.isfile.return_value = True mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = file_data expected = file_data result = dovetail_utils.read_yaml_file(file_path) mock_path.isfile.assert_called_once_with(file_path) mock_open.assert_called_once_with(file_path, 'r') mock_load.assert_called_once_with(file_obj) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path', autospec=True) def test_read_yaml_file_raised_exception(self, mock_path, mock_open): logger = Mock() file_path = 'known_file' errorMSG = 'Exception was raised' exception = Exception(errorMSG) mock_open.side_effect = exception expected = None result = dovetail_utils.read_yaml_file(file_path, logger) mock_open.assert_called_once_with(file_path, 'r') logger.exception.assert_called_once_with( 'Failed to read file {}, exception: {}' .format(file_path, exception)) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_get_hosts_info_not_exist(self, mock_path): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = False logger = Mock() expected = {} result = dovetail_utils.get_hosts_info(logger) mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) self.assertEqual(expected, result) @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_get_hosts_info_not_yaml(self, mock_path, mock_open, mock_load): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = True file_obj = Mock() mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = None logger = Mock() expected = {} result = dovetail_utils.get_hosts_info(logger) mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) mock_open.assert_called_once_with(file_complete_name) mock_load.assert_called_once_with(file_obj) logger.debug.assert_called_once_with( 'File {} is empty.'.format(file_complete_name)) self.assertEqual(expected, result) @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_get_hosts_info_no_hosts_info(self, mock_path, mock_open, mock_load): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = True file_obj = Mock() mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = {'a': 'b'} logger = Mock() expected = {} result = dovetail_utils.get_hosts_info(logger) mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) mock_open.assert_called_once_with(file_complete_name) mock_load.assert_called_once_with(file_obj) logger.error.assert_called_once_with( 'There is no key hosts_info in file {}' .format(file_complete_name)) self.assertEqual(expected, result) @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_get_hosts_info_no_hostname(self, mock_path, mock_open, mock_load): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = True file_obj = Mock() mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = {'hosts_info': {'127.0.0.1': []}} expected = {} result = dovetail_utils.get_hosts_info() mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) mock_open.assert_called_once_with(file_complete_name) mock_load.assert_called_once_with(file_obj) self.assertEqual(expected, result) @patch('dovetail.utils.dovetail_utils.add_hosts_info') @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_get_hosts_info_no_valid_hostname(self, mock_path, mock_open, mock_load, mock_fn): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = True file_obj = Mock() mock_open.return_value.__enter__.return_value = file_obj hosts_info = {'127.0.0.1': [None]} mock_load.return_value = {'hosts_info': hosts_info} expected = {} result = dovetail_utils.get_hosts_info() mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) mock_open.assert_called_once_with(file_complete_name) mock_load.assert_called_once_with(file_obj) mock_fn.assert_called_once_with(hosts_info.keys()[0], hosts_info.values()[0]) self.assertEqual(expected, result) @patch('dovetail.utils.dovetail_utils.add_hosts_info') @patch('yaml.safe_load') @patch('__builtin__.open') @patch('os.path', autospec=True) def test_get_hosts_info(self, mock_path, mock_open, mock_load, mock_fn): file_path = 'file_path' file_complete_name = '/'.join((file_path, 'hosts.yaml')) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} mock_path.join.return_value = file_complete_name mock_path.isfile.return_value = True file_obj = Mock() mock_open.return_value.__enter__.return_value = file_obj hosts_ip = '127.0.0.1' hostnames = ['host_one', 'host_two'] mock_load.return_value = {'hosts_info': {hosts_ip: hostnames}} logger = Mock() names_str = ' '.join(hostnames) expected = {names_str: hosts_ip} result = dovetail_utils.get_hosts_info(logger) mock_path.join.assert_called_once_with(file_path, 'hosts.yaml') mock_path.isfile.assert_called_once_with(file_complete_name) mock_open.assert_called_once_with(file_complete_name) mock_load.assert_called_once_with(file_obj) mock_fn.assert_called_once_with(hosts_ip, hostnames) logger.debug.assert_called_once_with('Get hosts info {}:{}.' .format(hosts_ip, names_str)) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_check_cacert_false(self, mock_path): file_one_path = 'path_one' file_two_path = 'path_two' logger = Mock() mock_path.isfile.return_value = True mock_path.dirname.return_value = file_one_path dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_two_path} expected = False result = dovetail_utils.check_cacert_file(file_one_path, logger) mock_path.isfile.assert_called_once_with(file_one_path) mock_path.dirname.assert_called_once_with(file_one_path) logger.error.assert_called_once_with( 'Credential file must be put under {}, ' 'which can be mounted into other container.' .format(file_two_path)) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_check_cacert_true(self, mock_path): file_path = 'path_one' mock_path.isfile.return_value = True mock_path.dirname.return_value = file_path dovetail_utils.dt_cfg.dovetail_config = {'config_dir': file_path} expected = True result = dovetail_utils.check_cacert_file(file_path) mock_path.isfile.assert_called_once_with(file_path) mock_path.dirname.assert_called_once_with(file_path) self.assertEqual(expected, result) @patch('os.path', autospec=True) def test_check_cacert_not_exist(self, mock_path): file_path = 'path_one' mock_path.isfile.return_value = False logger = Mock() expected = False result = dovetail_utils.check_cacert_file(file_path, logger) mock_path.isfile.assert_called_once_with(file_path) logger.error.assert_called_once_with( 'OS_CACERT is {}, but the file does not exist.' .format(file_path)) self.assertEqual(expected, result) @patch('sys.stdout') def test_show_progress_bar(self, mock_stdout): length = 99 max_length = 50 expect_length = length % max_length calls = [call('Running ' + ' ' * max_length + '\r'), call('Running ' + '.' * expect_length + '\r')] dovetail_utils.show_progress_bar(length) mock_stdout.write.assert_has_calls(calls) mock_stdout.flush.assert_has_calls([call(), call()]) def test_get_duration(self): date = '2018-08-10 05:12:27' logger = Mock() expected = '0m0s' result = dovetail_utils.get_duration(date, date, logger) self.assertEqual(expected, result) def test_get_duration_invalid_time(self): date = 'invalid' logger = Mock() expected = None result = dovetail_utils.get_duration(date, date, logger) logger.exception.assert_called_once() self.assertEqual(expected, result) @patch('os.getenv') def test_check_https_enabled(self, mock_getenv): auth_url = 'https://valid-url.com' mock_getenv.return_value = auth_url logger = Mock() expected = True result = dovetail_utils.check_https_enabled(logger) logger.debug.assert_has_calls( [call('Checking if https enabled or not...'), call('https is enabled')]) self.assertEqual(expected, result) @patch('os.getenv') def test_check_https_disabled(self, mock_getenv): auth_url = 'invalid' mock_getenv.return_value = auth_url logger = Mock() expected = False result = dovetail_utils.check_https_enabled(logger) logger.debug.assert_has_calls( [call('Checking if https enabled or not...'), call('https is not enabled')]) self.assertEqual(expected, result) @patch('python_hosts.Hosts', autospec=True) @patch('python_hosts.HostsEntry', autospec=True) def test_add_hosts_info_no_ip(self, mock_python_h_entry, mock_python_hosts): dovetail_utils.add_hosts_info(None, ['host_one']) mock_python_hosts.assert_called_once_with(path='/etc/hosts') mock_python_h_entry.assert_not_called() @patch('python_hosts.Hosts', autospec=True) @patch('python_hosts.HostsEntry', autospec=True) def test_add_hosts_info_no_hosts(self, mock_python_h_entry, mock_python_hosts): dovetail_utils.add_hosts_info('127.0.0.1', []) mock_python_hosts.assert_called_once_with(path='/etc/hosts') mock_python_h_entry.assert_not_called() @patch('python_hosts.Hosts', autospec=True) @patch('python_hosts.HostsEntry', autospec=True) def test_add_hosts_info(self, mock_python_h_entry, mock_python_hosts): ip = '127.0.0.1' hostnames = ['host_one'] hosts_obj = Mock() entry_obj = Mock() mock_python_hosts.return_value = hosts_obj mock_python_h_entry.return_value = entry_obj dovetail_utils.add_hosts_info(ip, hostnames) mock_python_hosts.assert_called_once_with(path='/etc/hosts') mock_python_h_entry.assert_called_once_with(entry_type='ipv4', address=ip, names=hostnames) hosts_obj.add.assert_called_once_with([entry_obj]) hosts_obj.write.assert_called_once() @patch('dovetail.utils.dovetail_utils.objwalk') def test_get_obj_by_path(self, mock_walk): path = dist_path = 'path' obj = 'obj' mock_walk.return_value = [(path, obj)] expected = obj result = dovetail_utils.get_obj_by_path(obj, dist_path) self.assertEqual(expected, result) @patch('dovetail.utils.dovetail_utils.objwalk') def test_get_obj_by_path_none(self, mock_walk): path = 'path' dist_path = 'dst_path' obj = 'obj' mock_walk.return_value = [(path, obj)] expected = None result = dovetail_utils.get_obj_by_path(obj, dist_path) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.environ') def test_source_env(self, mock_env, mock_open): file_path = 'file_path' env_name = 'name' env_value = 'value' file_data = u'export %s=%s' % (env_name, env_value) mock_open.return_value.__enter__.return_value = io.StringIO(file_data) dovetail_utils.source_env(file_path) mock_open.assert_called_once_with(file_path, 'r') mock_env.update.assert_called_once_with({env_name: env_value}) @patch('dovetail.utils.dovetail_utils.docker') def test_check_docker_version(self, mock_docker): server_version = '1.12.3' client_obj = Mock() mock_docker.from_env.return_value = client_obj client_obj.version.return_value = {'Version': server_version} logger = Mock() dovetail_utils.check_docker_version(logger) logger.debug.assert_has_calls( [call('Docker server version: {}'.format(server_version))]) @patch('dovetail.utils.dovetail_utils.docker') def test_check_docker_version_error(self, mock_docker): client_obj = Mock() mock_docker.from_env.return_value = client_obj client_obj.version.return_value = {} logger = Mock() dovetail_utils.check_docker_version(logger) logger.error.assert_has_calls( [call("Don't support this Docker server version. " "Docker server should be updated to at least 1.12.3.")]) @patch('dovetail.utils.dovetail_utils.docker') def test_check_docker_version_less_than(self, mock_docker): server_version = '1.12.1' client_obj = Mock() mock_docker.from_env.return_value = client_obj client_obj.version.return_value = {'Version': server_version} logger = Mock() dovetail_utils.check_docker_version(logger) logger.error.assert_has_calls( [call("Don't support this Docker server version. " "Docker server should be updated to at least 1.12.3.")]) @patch('__builtin__.open') @patch('os.path') @patch('os.listdir') @patch('json.load') @patch('json.dumps') def test_combine_files(self, mock_dumps, mock_load, mock_listdir, mock_path, mock_open): file_path = 'file_path' file_name = 'file_name' file_complete_name = '/'.join([file_path, file_name]) file_content_dict = {'key': 'value'} file_content_str = '{"key": "value"}' file_obj = Mock() mock_listdir.return_value = [file_name] mock_path.join.return_value = file_complete_name mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = file_content_dict mock_dumps.return_value = file_content_str logger = Mock() expected = 'result_file' result = dovetail_utils.combine_files(file_path, expected, logger) mock_listdir.assert_called_once_with(file_path) mock_path.join.assert_called_once_with(file_path, file_name) mock_open.assert_any_call(file_complete_name, 'r') mock_load.assert_called_once_with(file_obj) mock_open.assert_any_call(expected, 'w') mock_dumps.assert_called_once_with({file_name: file_content_dict}) file_obj.write.assert_called_once_with(file_content_str) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path') @patch('os.listdir') def test_combine_files_read_exception(self, mock_listdir, mock_path, mock_open): file_path = 'file_path' file_name = 'file_name' file_complete_name = '/'.join([file_path, file_name]) mock_listdir.return_value = [file_name] mock_path.join.return_value = file_complete_name mock_open.side_effect = Exception() logger = Mock() expected = None result = dovetail_utils.combine_files(file_path, expected, logger) mock_listdir.assert_called_once_with(file_path) mock_path.join.assert_called_once_with(file_path, file_name) mock_open.assert_any_call(file_complete_name, 'r') logger.error.assert_called_once_with( 'Failed to read file {}.'.format(file_complete_name)) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path') @patch('os.listdir') @patch('json.load') @patch('json.dumps') def test_combine_files_write_exception(self, mock_dumps, mock_load, mock_listdir, mock_path, mock_open): file_path = 'file_path' file_name = 'file_name' file_complete_name = '/'.join([file_path, file_name]) file_content_dict = {'key': 'value'} file_content_str = '{"key": "value"}' file_obj = Mock() file_obj.write.side_effect = Exception() mock_listdir.return_value = [file_name] mock_path.join.return_value = file_complete_name mock_open.return_value.__enter__.return_value = file_obj mock_load.return_value = file_content_dict mock_dumps.return_value = file_content_str logger = Mock() expected = None result = dovetail_utils.combine_files(file_path, expected, logger) mock_listdir.assert_called_once_with(file_path) mock_path.join.assert_called_once_with(file_path, file_name) mock_open.assert_any_call(file_complete_name, 'r') mock_load.assert_called_once_with(file_obj) mock_open.assert_any_call(expected, 'w') mock_dumps.assert_called_once_with({file_name: file_content_dict}) file_obj.write.assert_called_once_with(file_content_str) logger.exception.assert_called_once_with( 'Failed to write file {}.'.format(expected)) self.assertEqual(expected, result) @patch('json.dump') @patch('__builtin__.open') @patch('os.path') @patch('dovetail.utils.dovetail_utils.check_https_enabled') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.OS_Utils', autospec=True) def test_get_openstack_endpoint(self, mock_utils, mock_getenv, mock_https_check, mock_path, mock_open, mock_dump): mock_https_check.return_value = True mock_getenv.return_value = 'True' endpoints_ret = True endpoint_url = 'http://www.abc.com' endpoint_enabled = True service_id = '123456789' service_type = 'type' service_name = 'name' endpoints = [{'url': endpoint_url, 'enabled': endpoint_enabled, 'service_id': service_id}] services_ret = True services = [{'service_type': service_type, 'name': service_name}] file_path = 'file_path' file_name = 'endpoint_info.json' file_complete_name = '/'.join([file_path, file_name]) utils_obj = Mock() file_obj = Mock() logger = Mock() mock_utils.return_value = utils_obj utils_obj.search_endpoints.return_value = (endpoints_ret, endpoints) utils_obj.search_services.return_value = (services_ret, services) dovetail_utils.dt_cfg.dovetail_config = {'result_dir': file_path} mock_path.join.return_value = file_complete_name mock_open.return_value.__enter__.return_value = file_obj expected = [{'Service Type': service_type, 'URL': endpoint_url, 'Enabled': endpoint_enabled, 'Service Name': service_name}] result = dovetail_utils.get_openstack_endpoint(logger=logger) mock_https_check.assert_called_once_with(logger) mock_getenv.assert_called_once_with('OS_INSECURE') mock_utils.assert_called_once_with(verify=False) utils_obj.search_endpoints.assert_called_once() utils_obj.search_services.assert_called_once_with( service_id=service_id) mock_path.join.assert_called_once_with(file_path, file_name) mock_open.assert_any_call(file_complete_name, 'w') mock_dump.assert_called_once_with(expected, file_obj) logger.debug.assert_called_once_with( 'Record all endpoint info into file {}.' .format(file_complete_name)) self.assertEqual(expected, result) @patch('dovetail.utils.dovetail_utils.check_https_enabled') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.OS_Utils', autospec=True) def test_get_openstack_endpoint_no_endpoints(self, mock_utils, mock_getenv, mock_https_check): mock_https_check.return_value = True mock_getenv.return_value = 'True' endpoints_ret = False endpoints_exception_msg = 'An exception occured' utils_obj = Mock() logger = Mock() mock_utils.return_value = utils_obj utils_obj.search_endpoints.return_value = (endpoints_ret, endpoints_exception_msg) expected = None result = dovetail_utils.get_openstack_endpoint(logger=logger) mock_https_check.assert_called_once_with(logger) mock_getenv.assert_called_once_with('OS_INSECURE') mock_utils.assert_called_once_with(verify=False) utils_obj.search_endpoints.assert_called_once() logger.error.assert_called_once_with( 'Failed to list endpoints. Exception message, {}' .format(endpoints_exception_msg)) self.assertEqual(expected, result) @patch('dovetail.utils.dovetail_utils.check_https_enabled') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.OS_Utils', autospec=True) def test_get_openstack_endpoint_no_services(self, mock_utils, mock_getenv, mock_https_check): mock_https_check.return_value = True mock_getenv.return_value = 'True' endpoints_ret = True endpoint_url = 'http://www.abc.com' endpoint_enabled = True service_id = '123456789' endpoints = [{'url': endpoint_url, 'enabled': endpoint_enabled, 'service_id': service_id}] services_ret = False services_exception_msg = 'An exception occured' utils_obj = Mock() logger = Mock() mock_utils.return_value = utils_obj utils_obj.search_endpoints.return_value = (endpoints_ret, endpoints) utils_obj.search_services.return_value = (services_ret, services_exception_msg) expected = None result = dovetail_utils.get_openstack_endpoint(logger=logger) mock_https_check.assert_called_once_with(logger) mock_getenv.assert_called_once_with('OS_INSECURE') mock_utils.assert_called_once_with(verify=False) utils_obj.search_endpoints.assert_called_once() utils_obj.search_services.assert_called_once_with( service_id=service_id) logger.error.assert_called_once_with( 'Failed to list services. Exception message, {}' .format(services_exception_msg)) self.assertEqual(expected, result) @patch('__builtin__.open') @patch('os.path') @patch('dovetail.utils.dovetail_utils.check_https_enabled') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.OS_Utils', autospec=True) def test_get_openstack_endpoint_exception(self, mock_utils, mock_getenv, mock_https_check, mock_path, mock_open): mock_https_check.return_value = False mock_getenv.return_value = 'True' endpoints_ret = True endpoint_url = 'http://www.abc.com' endpoint_enabled = True service_id = '123456789' service_type = 'type' service_name = 'name' endpoints = [{'url': endpoint_url, 'enabled': endpoint_enabled, 'service_id': service_id}] services_ret = True services = [{'service_type': service_type, 'name': service_name}] file_path = 'file_path' file_name = 'endpoint_info.json' file_complete_name = '/'.join([file_path, file_name]) utils_obj = Mock() logger = Mock() mock_utils.return_value = utils_obj utils_obj.search_endpoints.return_value = (endpoints_ret, endpoints) utils_obj.search_services.return_value = (services_ret, services) dovetail_utils.dt_cfg.dovetail_config = {'result_dir': file_path} mock_path.join.return_value = file_complete_name errorMSG = 'Exception was raised' exception = Exception(errorMSG) mock_open.side_effect = exception expected = None result = dovetail_utils.get_openstack_endpoint(logger=logger) mock_https_check.assert_called_once_with(logger) mock_getenv.assert_called_once_with('OS_INSECURE') mock_utils.assert_called_once_with() utils_obj.search_endpoints.assert_called_once() utils_obj.search_services.assert_called_once_with( service_id=service_id) mock_path.join.assert_called_once_with(file_path, file_name) mock_open.assert_any_call(file_complete_name, 'w') logger.exception.assert_called_once_with( 'Failed to write endpoint info into file.') self.assertEqual(expected, result) @patch('os.path') @patch('dovetail.utils.dovetail_utils.get_inventory_file') @patch('dovetail.utils.dovetail_utils.exec_cmd') @patch('dovetail.utils.dovetail_utils.combine_files') def test_get_hardware_info(self, mock_combine, mock_cmd, mock_inventory, mock_path): logger = Mock() config_dir = 'config_dir' result_dir = 'result_dir' pod_file = 'pod_file' dovetail_utils.dt_cfg.dovetail_config = {'config_dir': config_dir, 'result_dir': result_dir, 'pod_file': pod_file} mock_inventory.return_value = Mock() ret = 0 msg = '' mock_cmd.return_value = (ret, msg) inventory_file = '/'.join([result_dir, 'inventory.ini']) info_file_path = '/'.join([result_dir, 'sut_hardware_info']) all_info_file = '/'.join([result_dir, 'all_hosts_info.json']) mock_path.join.side_effect = [ '/'.join([config_dir, pod_file]), info_file_path, all_info_file, inventory_file] mock_path.exists.return_value = True mock_combine.return_value = True expected = all_info_file result = dovetail_utils.get_hardware_info(logger=logger) join_calls = [call(config_dir, pod_file), call(result_dir, 'sut_hardware_info'), call(result_dir, 'all_hosts_info.json'), call(result_dir, 'inventory.ini')] mock_path.join.assert_has_calls(join_calls) log_calls = [ call('Get hardware info of all nodes list in file {} ...' .format('/'.join([config_dir, pod_file]))), call('Hardware info of all nodes are stored in file {}.' .format(expected))] logger.info.assert_has_calls(log_calls) mock_cmd.assert_called_once_with( 'cd {} && ansible all -m setup -i {} --tree {}' .format(constants.USERCONF_PATH, inventory_file, info_file_path), verbose=False) mock_path.exists.assert_called_once_with(info_file_path) mock_combine.assert_called_once_with(info_file_path, all_info_file, logger) self.assertEqual(expected, result) @patch('os.path') @patch('dovetail.utils.dovetail_utils.get_inventory_file') def test_get_hardware_info_no_inventory(self, mock_inventory, mock_path): logger = Mock() config_dir = 'config_dir' result_dir = 'result_dir' pod_file = 'pod_file' dovetail_utils.dt_cfg.dovetail_config = {'config_dir': config_dir, 'result_dir': result_dir, 'pod_file': pod_file} mock_inventory.return_value = None mock_path.join.side_effect = [ '/'.join([config_dir, pod_file]), '/'.join([result_dir, 'sut_hardware_info']), '/'.join([result_dir, 'all_hosts_info.json']), '/'.join([result_dir, 'inventory.ini'])] expected = None result = dovetail_utils.get_hardware_info(logger=logger) join_calls = [call(config_dir, pod_file), call(result_dir, 'sut_hardware_info'), call(result_dir, 'all_hosts_info.json'), call(result_dir, 'inventory.ini')] mock_path.join.assert_has_calls(join_calls) logger.info.assert_called_once_with( 'Get hardware info of all nodes list in file {} ...' .format('/'.join([config_dir, pod_file]))) logger.error.assert_called_once_with( 'Failed to get SUT hardware info.') self.assertEqual(expected, result) @patch('os.path') @patch('dovetail.utils.dovetail_utils.get_inventory_file') @patch('dovetail.utils.dovetail_utils.exec_cmd') def test_get_hardware_info_no_info(self, mock_cmd, mock_inventory, mock_path): logger = Mock() config_dir = 'config_dir' result_dir = 'result_dir' pod_file = 'pod_file' dovetail_utils.dt_cfg.dovetail_config = {'config_dir': config_dir, 'result_dir': result_dir, 'pod_file': pod_file} mock_inventory.return_value = Mock() ret = 0 msg = '' mock_cmd.return_value = (ret, msg) inventory_file = '/'.join([result_dir, 'inventory.ini']) info_file_path = '/'.join([result_dir, 'sut_hardware_info']) all_info_file = '/'.join([result_dir, 'all_hosts_info.json']) mock_path.join.side_effect = [ '/'.join([config_dir, pod_file]), info_file_path, all_info_file, inventory_file] mock_path.exists.return_value = False expected = None result = dovetail_utils.get_hardware_info(logger=logger) join_calls = [call(config_dir, pod_file), call(result_dir, 'sut_hardware_info'), call(result_dir, 'all_hosts_info.json'), call(result_dir, 'inventory.ini')] mock_path.join.assert_has_calls(join_calls) logger.info.assert_called_once_with( 'Get hardware info of all nodes list in file {} ...' .format('/'.join([config_dir, pod_file]))) logger.error.assert_called_once_with( 'Failed to get SUT hardware info.') mock_cmd.assert_called_once_with( 'cd {} && ansible all -m setup -i {} --tree {}' .format(constants.USERCONF_PATH, inventory_file, info_file_path), verbose=False) mock_path.exists.assert_called_once_with(info_file_path) self.assertEqual(expected, result) @patch('os.path') @patch('dovetail.utils.dovetail_utils.get_inventory_file') @patch('dovetail.utils.dovetail_utils.exec_cmd') @patch('dovetail.utils.dovetail_utils.combine_files') def test_get_hardware_info_no_combine(self, mock_combine, mock_cmd, mock_inventory, mock_path): logger = Mock() config_dir = 'config_dir' result_dir = 'result_dir' pod_file = 'pod_file' dovetail_utils.dt_cfg.dovetail_config = {'config_dir': config_dir, 'result_dir': result_dir, 'pod_file': pod_file} mock_inventory.return_value = Mock() ret = 0 msg = '' mock_cmd.return_value = (ret, msg) inventory_file = '/'.join([result_dir, 'inventory.ini']) info_file_path = '/'.join([result_dir, 'sut_hardware_info']) all_info_file = '/'.join([result_dir, 'all_hosts_info.json']) mock_path.join.side_effect = [ '/'.join([config_dir, pod_file]), info_file_path, all_info_file, inventory_file] mock_path.exists.return_value = True mock_combine.return_value = False expected = None result = dovetail_utils.get_hardware_info(logger=logger) join_calls = [call(config_dir, pod_file), call(result_dir, 'sut_hardware_info'), call(result_dir, 'all_hosts_info.json'), call(result_dir, 'inventory.ini')] mock_path.join.assert_has_calls(join_calls) logger.info.assert_called_once_with( 'Get hardware info of all nodes list in file {} ...' .format('/'.join([config_dir, pod_file]))) logger.error.assert_called_once_with( 'Failed to get all hardware info.') mock_cmd.assert_called_once_with( 'cd {} && ansible all -m setup -i {} --tree {}' .format(constants.USERCONF_PATH, inventory_file, info_file_path), verbose=False) mock_path.exists.assert_called_once_with(info_file_path) mock_combine.assert_called_once_with(info_file_path, all_info_file, logger) self.assertEqual(expected, result) @patch('os.path') @patch('__builtin__.open') @patch('yaml.safe_load') def test_get_inventory_password(self, mock_load, mock_open, mock_path): name = 'name' ip = 'ip' user = 'user' password = 'password' pod_file_data = {'nodes': [{'name': name, 'ip': ip, 'user': user, 'password': password}]} inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() pod_file_obj = Mock() inventory_file_obj = Mock() mock_path.isfile.return_value = True mock_open.return_value.__enter__.side_effect = [pod_file_obj, inventory_file_obj] mock_load.return_value = pod_file_data expected = True result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) mock_open.assert_any_call(pod_file_name, 'r') mock_open.assert_any_call(inventory_file_name, 'w') mock_load.assert_called_once_with(pod_file_obj) inventory_file_obj.write.assert_called_once_with( '{name} ansible_host={ip} ansible_user={user} ' 'ansible_ssh_pass={password}\n' .format(name=name, ip=ip, user=user, password=password)) logger.debug.assert_called_once_with( 'Ansible inventory file is {}.'.format(inventory_file_name)) self.assertEqual(expected, result) @patch('os.path') @patch('__builtin__.open') @patch('yaml.safe_load') def test_get_inventory_key_filename(self, mock_load, mock_open, mock_path): name = 'name' ip = 'ip' user = 'user' password = 'password' pod_file_data = {'nodes': [{'name': name, 'ip': ip, 'user': user, 'key_filename': password}]} inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() pod_file_obj = Mock() inventory_file_obj = Mock() mock_path.isfile.return_value = True mock_open.return_value.__enter__.side_effect = [pod_file_obj, inventory_file_obj] mock_load.return_value = pod_file_data config_dir = 'config_dir' key_file = '/'.join([config_dir, 'id_rsa']) dovetail_utils.dt_cfg.dovetail_config = {'config_dir': config_dir} mock_path.join.return_value = key_file expected = True result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) mock_open.assert_any_call(pod_file_name, 'r') mock_open.assert_any_call(inventory_file_name, 'w') mock_load.assert_called_once_with(pod_file_obj) mock_path.join.assert_called_once_with(config_dir, 'id_rsa') inventory_file_obj.write.assert_called_once_with( '{name} ansible_host={ip} ansible_user={user} ' 'ansible_ssh_private_key_file={key_file}\n' .format(name=name, ip=ip, user=user, key_file=key_file)) logger.debug.assert_called_once_with( 'Ansible inventory file is {}.'.format(inventory_file_name)) self.assertEqual(expected, result) @patch('os.path') @patch('__builtin__.open') @patch('yaml.safe_load') def test_get_inventory_other(self, mock_load, mock_open, mock_path): name = 'name' ip = 'ip' user = 'user' pod_file_data = {'nodes': [{'name': name, 'ip': ip, 'user': user}]} inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() pod_file_obj = Mock() inventory_file_obj = Mock() mock_path.isfile.return_value = True mock_open.return_value.__enter__.side_effect = [pod_file_obj, inventory_file_obj] mock_load.return_value = pod_file_data expected = False result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) mock_open.assert_any_call(pod_file_name, 'r') mock_open.assert_any_call(inventory_file_name, 'w') mock_load.assert_called_once_with(pod_file_obj) logger.error.assert_called_once_with( 'No password or key_filename in file {}.'.format(pod_file_name)) self.assertEqual(expected, result) @patch('os.path') @patch('__builtin__.open') @patch('yaml.safe_load') def test_get_inventory_keyerror(self, mock_load, mock_open, mock_path): name = 'name' ip = 'ip' pod_file_data = {'nodes': [{'name': name, 'ip': ip}]} inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() pod_file_obj = Mock() inventory_file_obj = Mock() mock_path.isfile.return_value = True mock_open.return_value.__enter__.side_effect = [pod_file_obj, inventory_file_obj] mock_load.return_value = pod_file_data expected = False result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) mock_open.assert_any_call(pod_file_name, 'r') mock_open.assert_any_call(inventory_file_name, 'w') mock_load.assert_called_once_with(pod_file_obj) logger.exception.assert_called_once_with( "KeyError 'user'.") self.assertEqual(expected, result) @patch('os.path') @patch('__builtin__.open') def test_get_inventory_exception(self, mock_open, mock_path): inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() mock_path.isfile.return_value = True mock_open.return_value.__enter__.side_effect = Exception() expected = False result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) mock_open.assert_called_once_with(pod_file_name, 'r') logger.exception.assert_called_once_with( 'Failed to read file {}.'.format(pod_file_name)) self.assertEqual(expected, result) @patch('os.path') def test_get_inventory_invalid_pod_file(self, mock_path): inventory_file_name = 'inventory' pod_file_name = 'pod' logger = Mock() mock_path.isfile.return_value = False expected = False result = dovetail_utils.get_inventory_file(pod_file_name, inventory_file_name, logger=logger) mock_path.isfile.assert_called_once_with(pod_file_name) logger.error.assert_called_once_with( "File {} doesn't exist.".format(pod_file_name)) self.assertEqual(expected, result) @patch('subprocess.Popen') @patch('subprocess.PIPE') @patch('subprocess.STDOUT') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.exec_log') @patch('dovetail.utils.dovetail_utils.show_progress_bar') def test_exec_cmd(self, mock_bar, mock_log, mock_getenv, mock_stdout, mock_pipe, mock_open): logger = Mock() cmd = 'cmd' verbose = True subprocess_obj = Mock() cmd_output = 'line' mock_open.return_value = subprocess_obj pip_obj = Mock() stdout_obj = Mock() mock_pipe.return_value = pip_obj mock_stdout.return_value = stdout_obj subp_stdout = Mock() subprocess_obj.stdout = subp_stdout subprocess_obj.wait.return_value = 0 subp_stdout.readline.side_effect = [cmd_output, ''] expected = (0, 'line') result = dovetail_utils.exec_cmd( cmd, logger=logger, exit_on_error=True, info=False, exec_msg_on=True, err_msg='', verbose=verbose, progress_bar=True) log_calls = [ call(verbose, logger, "Executing command: '%s'" % cmd, 'debug'), call(verbose, logger, cmd_output, 'debug', True)] mock_log.assert_has_calls(log_calls) mock_open.assert_called_once_with(cmd, shell=True, stdout=mock_pipe, stderr=mock_stdout) subp_stdout.readline.assert_has_calls([call(), call()]) subp_stdout.close.assert_called_once_with() subprocess_obj.wait.assert_called_once_with() mock_getenv.assert_called_once_with('DEBUG') mock_bar.assert_called_once_with(1) self.assertEqual(expected, result) @patch('sys.exit') @patch('subprocess.Popen') @patch('subprocess.PIPE') @patch('subprocess.STDOUT') @patch('os.getenv') @patch('dovetail.utils.dovetail_utils.exec_log') @patch('dovetail.utils.dovetail_utils.show_progress_bar') def test_exec_cmd_error(self, mock_bar, mock_log, mock_getenv, mock_stdout, mock_pipe, mock_open, mock_exit): logger = Mock() cmd = 'cmd' verbose = True subprocess_obj = Mock() cmd_output = 'line' mock_open.return_value = subprocess_obj pip_obj = Mock() stdout_obj = Mock() mock_pipe.return_value = pip_obj mock_stdout.return_value = stdout_obj subp_stdout = Mock() subprocess_obj.stdout = subp_stdout subprocess_obj.wait.return_value = 1 subp_stdout.readline.side_effect = [cmd_output, ''] dovetail_utils.exec_cmd( cmd, logger=logger, exit_on_error=True, info=False, exec_msg_on=True, err_msg='', verbose=verbose, progress_bar=True) log_calls = [ call(verbose, logger, "Executing command: '%s'" % cmd, 'debug'), call(verbose, logger, cmd_output, 'debug', True), call(verbose, logger, "The command '%s' failed." % cmd, 'error')] mock_log.assert_has_calls(log_calls) mock_open.assert_called_once_with(cmd, shell=True, stdout=mock_pipe, stderr=mock_stdout) subp_stdout.readline.assert_has_calls([call(), call()]) subp_stdout.close.assert_called_once_with() subprocess_obj.wait.assert_called_once_with() mock_getenv.assert_called_once_with('DEBUG') mock_bar.assert_called_once_with(1) mock_exit.assert_called_once_with(1) @patch('os.path', autospec=True) def test_get_openstack_info_no_openrc(self, mock_path): logger = Mock() config_dir = 'config_dir' env_file = 'env_file' dovetail_utils.dt_cfg.dovetail_config = { 'config_dir': config_dir, 'env_file': env_file} mock_path.join.side_effect = ['openrc'] mock_path.isfile.return_value = False dovetail_utils.get_openstack_info(logger) mock_path.join.assert_called_once_with(config_dir, env_file) mock_path.isfile.assert_called_once_with('openrc') logger.error.assert_called_once_with('File openrc does not exist.') @patch('dovetail.utils.dovetail_utils.source_env') @patch('dovetail.utils.dovetail_utils.get_hosts_info') @patch('dovetail.utils.dovetail_utils.get_openstack_endpoint') @patch('dovetail.utils.dovetail_utils.get_hardware_info') @patch('os.path', autospec=True) def test_get_openstack_info(self, mock_path, mock_hardware, mock_endpoint, mock_host, mock_env): logger = Mock() config_dir = 'config_dir' env_file = 'env_file' dovetail_utils.dt_cfg.dovetail_config = { 'config_dir': config_dir, 'env_file': env_file} mock_path.join.side_effect = ['openrc'] mock_path.isfile.return_value = True dovetail_utils.get_openstack_info(logger) mock_path.join.assert_called_once_with(config_dir, env_file) mock_path.isfile.assert_called_once_with('openrc') mock_env.assert_called_once() mock_host.assert_called_once() mock_endpoint.assert_called_once() mock_hardware.assert_called_once() @patch('json.dumps') @patch('dovetail.utils.dovetail_utils.requests') @patch('os.getenv') def test_push_results_to_db(self, mock_getenv, mock_requests, mock_dumps): logger = Mock() case_name = 'case_name' details = {'criteria': 'PASS'} start_date = 'start_date' stop_date = 'stop_date' mock_getenv.side_effect = [ 'url', 'installer', 'scenario', 'pod_name', 'build_tag', 'version'] post_req = Mock() post_req.raise_for_status.return_value = None mock_requests.post.return_value = post_req mock_dumps.return_value = {"project_name": "dovetail"} dovetail_utils.push_results_to_db( case_name, details, start_date, stop_date, logger) mock_requests.post.assert_called_once_with( 'url', data={"project_name": "dovetail"}, headers={"Content-Type": "application/json"}) logger.debug.assert_called_once_with( "The results were successfully pushed to DB.") @patch('json.dumps') @patch('dovetail.utils.dovetail_utils.requests') @patch('os.getenv') def test_push_results_to_db_exception(self, mock_getenv, mock_requests, mock_dumps): logger = Mock() case_name = 'case_name' details = {'criteria': 'PASS'} start_date = 'start_date' stop_date = 'stop_date' mock_getenv.side_effect = [ 'url', 'installer', 'scenario', 'pod_name', 'build_tag', 'version'] post_req = Mock() post_req.raise_for_status.side_effect = Exception() mock_requests.post.return_value = post_req mock_dumps.return_value = {"project_name": "dovetail"} dovetail_utils.push_results_to_db( case_name, details, start_date, stop_date, logger) logger.debug.assert_not_called() logger.exception.assert_called_once_with( "The results cannot be pushed to DB.")