diff options
Diffstat (limited to 'dovetail/tests/unit/utils/test_dovetail_utils.py')
| -rw-r--r-- | dovetail/tests/unit/utils/test_dovetail_utils.py | 1402 |
1 files changed, 1402 insertions, 0 deletions
diff --git a/dovetail/tests/unit/utils/test_dovetail_utils.py b/dovetail/tests/unit/utils/test_dovetail_utils.py new file mode 100644 index 00000000..7d1fddc1 --- /dev/null +++ b/dovetail/tests/unit/utils/test_dovetail_utils.py @@ -0,0 +1,1402 @@ +#!/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 <mokats@intracom-telecom.com>' + + +class DovetailUtilsTesting(unittest.TestCase): + + def setUp(self): + pass + + def tearDown(self): + pass + + @patch('sys.stdout') + @patch('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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(list(hosts_info.keys())[0], + list(hosts_info.values())[0]) + self.assertEqual(expected, result) + + @patch('dovetail.utils.dovetail_utils.add_hosts_info') + @patch('yaml.safe_load') + @patch('builtins.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 = '0.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() + + def test_get_obj_by_path(self): + obj = {'list': ['a', 'b'], 'name': 'name'} + dst_path = ('name',) + + expected = 'name' + result = dovetail_utils.get_obj_by_path(obj, dst_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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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('builtins.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.encode()] + + expected = (0, "b'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.encode()] + + 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.") + + def test_get_mount_list_error_mount(self): + project_cfg = {'mounts': ['aaa']} + res, msg = dovetail_utils.get_mount_list(project_cfg) + self.assertEqual(None, res) + self.assertEqual('Error mount aaa.', msg) + + def test_get_mount_list_keyerror_exception(self): + project_cfg = {'mounts': ['aaa=a,bbb=b', '']} + res, msg = dovetail_utils.get_mount_list(project_cfg) + self.assertEqual(None, res) + self.assertEqual("'target'", str(msg)) + + def test_get_mount_list(self): + project_cfg = {'mounts': ['target=a,source=b', '']} + res, msg = dovetail_utils.get_mount_list(project_cfg) + expected = [{'Source': 'b', 'Type': 'bind', 'ReadOnly': False, + 'Target': 'a'}] + self.assertEqual(expected, res) + self.assertEqual('Successfully to get mount list.', msg) |