X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=functest%2Ftests%2Funit%2Futils%2Ftest_functest_utils.py;h=07a57a49aa053a7b97cc3f58d97995e1e00f4b43;hb=4fdbdd34fb5fea55571a18b2438ececa953928ff;hp=b4cc5b73ba905af9003ebd961558f91fc8fea4cc;hpb=f49f5f2a94ece7d0e1ae924bc7ab77a38eeeef87;p=functest.git diff --git a/functest/tests/unit/utils/test_functest_utils.py b/functest/tests/unit/utils/test_functest_utils.py index b4cc5b73b..07a57a49a 100644 --- a/functest/tests/unit/utils/test_functest_utils.py +++ b/functest/tests/unit/utils/test_functest_utils.py @@ -7,22 +7,24 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 +# pylint: disable=missing-docstring + import logging -import pkg_resources -import os import time import unittest import mock -import requests -from six.moves import urllib +import pkg_resources +import six -from functest.tests.unit import test_utils from functest.utils import functest_utils -from functest.utils.constants import CONST class FunctestUtilsTesting(unittest.TestCase): + # pylint: disable=too-many-instance-attributes,too-many-public-methods + + readline = 0 + test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15'] def setUp(self): self.url = 'http://www.opnfv.org/' @@ -53,8 +55,6 @@ class FunctestUtilsTesting(unittest.TestCase): self.cmd = 'test_cmd' self.output_file = 'test_output_file' self.testname = 'testname' - self.testcase_dict = {'case_name': 'testname', - 'criteria': self.criteria} self.parameter = 'general.openstack.image_name' self.config_yaml = pkg_resources.resource_filename( 'functest', 'ci/config_functest.yaml') @@ -63,145 +63,6 @@ class FunctestUtilsTesting(unittest.TestCase): self.file_yaml = {'general': {'openstack': {'image_name': 'test_image_name'}}} - @mock.patch('six.moves.urllib.request.urlopen', - side_effect=urllib.error.URLError('no host given')) - def test_check_internet_connectivity_failed(self, mock_method): - self.assertFalse(functest_utils.check_internet_connectivity()) - mock_method.assert_called_once_with(self.url, timeout=self.timeout) - - @mock.patch('six.moves.urllib.request.urlopen') - def test_check_internet_connectivity_default(self, mock_method): - self.assertTrue(functest_utils.check_internet_connectivity()) - mock_method.assert_called_once_with(self.url, timeout=self.timeout) - - @mock.patch('six.moves.urllib.request.urlopen') - def test_check_internet_connectivity_debian(self, mock_method): - self.url = "https://www.debian.org/" - self.assertTrue(functest_utils.check_internet_connectivity(self.url)) - mock_method.assert_called_once_with(self.url, timeout=self.timeout) - - @mock.patch('six.moves.urllib.request.urlopen', - side_effect=urllib.error.URLError('no host given')) - def test_download_url_failed(self, mock_url): - self.assertFalse(functest_utils.download_url(self.url, self.dest_path)) - - @mock.patch('six.moves.urllib.request.urlopen') - def test_download_url_default(self, mock_url): - with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \ - mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\ - as mock_sh: - name = self.url.rsplit('/')[-1] - dest = self.dest_path + "/" + name - self.assertTrue(functest_utils.download_url(self.url, - self.dest_path)) - m.assert_called_once_with(dest, 'wb') - self.assertTrue(mock_sh.called) - - @mock.patch('functest.utils.functest_utils.logger.error') - def test_get_installer_type_failed(self, mock_logger_error): - with mock.patch.dict(os.environ, - {}, - clear=True): - self.assertEqual(functest_utils.get_installer_type(), - "Unknown_installer") - mock_logger_error.assert_called_once_with("Impossible to retrieve" - " the installer type") - - def test_get_installer_type_default(self): - with mock.patch.dict(os.environ, - {'INSTALLER_TYPE': 'test_installer'}, - clear=True): - self.assertEqual(functest_utils.get_installer_type(), - self.installer) - - @mock.patch('functest.utils.functest_utils.logger.info') - def test_get_scenario_failed(self, mock_logger_info): - with mock.patch.dict(os.environ, - {}, - clear=True): - self.assertEqual(functest_utils.get_scenario(), - "os-nosdn-nofeature-noha") - mock_logger_info.assert_called_once_with("Impossible to retrieve " - "the scenario.Use " - "default " - "os-nosdn-nofeature-noha") - - def test_get_scenario_default(self): - with mock.patch.dict(os.environ, - {'DEPLOY_SCENARIO': 'test_scenario'}, - clear=True): - self.assertEqual(functest_utils.get_scenario(), - self.scenario) - - def test_get_version_daily_job(self): - CONST.__setattr__('BUILD_TAG', self.build_tag) - self.assertEqual(functest_utils.get_version(), self.version) - - def test_get_version_weekly_job(self): - CONST.__setattr__('BUILD_TAG', self.build_tag_week) - self.assertEqual(functest_utils.get_version(), self.version) - - def test_get_version_with_dummy_build_tag(self): - CONST.__setattr__('BUILD_TAG', 'whatever') - self.assertEqual(functest_utils.get_version(), 'unknown') - - def test_get_version_unknown(self): - CONST.__setattr__('BUILD_TAG', 'unknown_build_tag') - self.assertEqual(functest_utils.get_version(), "unknown") - - @mock.patch('functest.utils.functest_utils.logger.info') - def test_get_pod_name_failed(self, mock_logger_info): - with mock.patch.dict(os.environ, - {}, - clear=True): - self.assertEqual(functest_utils.get_pod_name(), - "unknown-pod") - mock_logger_info.assert_called_once_with("Unable to retrieve " - "the POD name from " - "environment. Using " - "pod name 'unknown-pod'") - - def test_get_pod_name_default(self): - with mock.patch.dict(os.environ, - {'NODE_NAME': 'test_node_name'}, - clear=True): - self.assertEqual(functest_utils.get_pod_name(), - self.node_name) - - @mock.patch('functest.utils.functest_utils.logger.info') - def test_logger_test_results(self, mock_logger_info): - CONST.__setattr__('results_test_db_url', self.db_url) - CONST.__setattr__('BUILD_TAG', self.build_tag) - with mock.patch('functest.utils.functest_utils.get_pod_name', - return_value=self.node_name), \ - mock.patch('functest.utils.functest_utils.get_scenario', - return_value=self.scenario), \ - mock.patch('functest.utils.functest_utils.get_version', - return_value=self.version): - functest_utils.logger_test_results(self.project, self.case_name, - self.status, self.details) - mock_logger_info.assert_called_once_with( - "\n" - "****************************************\n" - "\t %(p)s/%(n)s results \n\n" - "****************************************\n" - "DB:\t%(db)s\n" - "pod:\t%(pod)s\n" - "version:\t%(v)s\n" - "scenario:\t%(s)s\n" - "status:\t%(c)s\n" - "build tag:\t%(b)s\n" - "details:\t%(d)s\n" - % {'p': self.project, - 'n': self.case_name, - 'db': CONST.__getattribute__('results_test_db_url'), - 'pod': self.node_name, - 'v': self.version, - 's': self.scenario, - 'c': self.status, - 'b': self.build_tag, - 'd': self.details}) - def _get_env_dict(self, var): dic = {'INSTALLER_TYPE': self.installer, 'DEPLOY_SCENARIO': self.scenario, @@ -210,87 +71,6 @@ class FunctestUtilsTesting(unittest.TestCase): dic.pop(var, None) return dic - def _test_push_results_to_db_missing_env(self, env_var): - dic = self._get_env_dict(env_var) - CONST.__setattr__('results_test_db_url', self.db_url) - with mock.patch.dict(os.environ, - dic, - clear=True), \ - mock.patch('functest.utils.functest_utils.logger.error') \ - as mock_logger_error: - functest_utils.push_results_to_db(self.project, self.case_name, - self.start_date, self.stop_date, - self.result, self.details) - mock_logger_error.assert_called_once_with("Please set env var: " + - str("\'" + env_var + - "\'")) - - def test_push_results_to_db_missing_installer(self): - self._test_push_results_to_db_missing_env('INSTALLER_TYPE') - - def test_push_results_to_db_missing_scenario(self): - self._test_push_results_to_db_missing_env('DEPLOY_SCENARIO') - - def test_push_results_to_db_missing_nodename(self): - self._test_push_results_to_db_missing_env('NODE_NAME') - - def test_push_results_to_db_missing_buildtag(self): - self._test_push_results_to_db_missing_env('BUILD_TAG') - - def test_push_results_to_db_request_post_failed(self): - dic = self._get_env_dict(None) - CONST.__setattr__('results_test_db_url', self.db_url) - with mock.patch.dict(os.environ, - dic, - clear=True), \ - mock.patch('functest.utils.functest_utils.logger.error') \ - as mock_logger_error, \ - mock.patch('functest.utils.functest_utils.requests.post', - side_effect=requests.RequestException): - self.assertFalse(functest_utils. - push_results_to_db(self.project, self.case_name, - self.start_date, - self.stop_date, - self.result, self.details)) - mock_logger_error.assert_called_once_with(test_utils. - RegexMatch("Pushing " - "Result to" - " DB" - "(\S+\s*) " - "failed:")) - - def test_push_results_to_db_request_post_exception(self): - dic = self._get_env_dict(None) - CONST.__setattr__('results_test_db_url', self.db_url) - with mock.patch.dict(os.environ, - dic, - clear=True), \ - mock.patch('functest.utils.functest_utils.logger.error') \ - as mock_logger_error, \ - mock.patch('functest.utils.functest_utils.requests.post', - side_effect=Exception): - self.assertFalse(functest_utils. - push_results_to_db(self.project, self.case_name, - self.start_date, - self.stop_date, - self.result, self.details)) - self.assertTrue(mock_logger_error.called) - - def test_push_results_to_db_default(self): - dic = self._get_env_dict(None) - CONST.__setattr__('results_test_db_url', self.db_url) - with mock.patch.dict(os.environ, - dic, - clear=True), \ - mock.patch('functest.utils.functest_utils.requests.post'): - self.assertTrue(functest_utils. - push_results_to_db(self.project, self.case_name, - self.start_date, - self.stop_date, - self.result, self.details)) - readline = 0 - test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15'] - @staticmethod def readline_side(): if FunctestUtilsTesting.readline == \ @@ -299,173 +79,96 @@ class FunctestUtilsTesting(unittest.TestCase): FunctestUtilsTesting.readline += 1 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline] - # TODO: get_resolvconf_ns - @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver') - def test_get_resolvconf_ns_default(self, mock_dns_resolve): - attrs = {'query.return_value': ["test"]} - mock_dns_resolve.configure_mock(**attrs) - - m = mock.Mock() - attrs = {'readline.side_effect': self.readline_side} - m.configure_mock(**attrs) - - with mock.patch("six.moves.builtins.open") as mo: - mo.return_value = m - self.assertEqual(functest_utils.get_resolvconf_ns(), - self.test_ip[1:]) - - def _get_environ(self, var): + def _get_environ(self, var, *args): # pylint: disable=unused-argument if var == 'INSTALLER_TYPE': return self.installer - elif var == 'DEPLOY_SCENARIO': + if var == 'DEPLOY_SCENARIO': return self.scenario return var - def test_get_ci_envvars_default(self): - with mock.patch('os.environ.get', - side_effect=self._get_environ): - dic = {"installer": self.installer, - "scenario": self.scenario} - self.assertDictEqual(functest_utils.get_ci_envvars(), dic) - - def cmd_readline(self): + @staticmethod + def cmd_readline(): return 'test_value\n' - @mock.patch('functest.utils.functest_utils.logger.error') - @mock.patch('functest.utils.functest_utils.logger.info') - def test_execute_command_args_present_with_error(self, mock_logger_info, - mock_logger_error): + @mock.patch('functest.utils.functest_utils.LOGGER.error') + @mock.patch('functest.utils.functest_utils.LOGGER.info') + def test_exec_cmd_args_present_ko(self, mock_logger_info, + mock_logger_error): with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ as mock_subproc_open, \ mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen: - - FunctestUtilsTesting.readline = 0 - - mock_obj = mock.Mock() - attrs = {'readline.side_effect': self.cmd_readline()} - mock_obj.configure_mock(**attrs) - - mock_obj2 = mock.Mock() - attrs = {'stdout': mock_obj, 'wait.return_value': 1} - mock_obj2.configure_mock(**attrs) - - mock_subproc_open.return_value = mock_obj2 - - resp = functest_utils.execute_command(self.cmd, info=True, - error_msg=self.error_msg, - verbose=True, - output_file=self.output_file) + stream = six.BytesIO() + stream.write(self.cmd_readline().encode("utf-8")) + attrs = { + 'return_value.__enter__.return_value.stdout': stream, + 'return_value.__enter__.return_value.wait.return_value': 1} + mock_subproc_open.configure_mock(**attrs) + resp = functest_utils.execute_command( + self.cmd, info=True, error_msg=self.error_msg, verbose=True, + output_file=self.output_file) self.assertEqual(resp, 1) - msg_exec = ("Executing command: '%s'" % self.cmd) + msg_exec = f"Executing command: '{self.cmd}'" mock_logger_info.assert_called_once_with(msg_exec) - mopen.assert_called_once_with(self.output_file, "w") + mopen.assert_called_once_with( + self.output_file, "w", encoding='utf-8') mock_logger_error.assert_called_once_with(self.error_msg) - @mock.patch('functest.utils.functest_utils.logger.info') - def test_execute_command_args_present_with_success(self, mock_logger_info, - ): + @mock.patch('functest.utils.functest_utils.LOGGER.info') + def test_exec_cmd_args_present_ok(self, mock_logger_info): with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ as mock_subproc_open, \ mock.patch('six.moves.builtins.open', mock.mock_open()) as mopen: - - FunctestUtilsTesting.readline = 0 - - mock_obj = mock.Mock() - attrs = {'readline.side_effect': self.cmd_readline()} - mock_obj.configure_mock(**attrs) - - mock_obj2 = mock.Mock() - attrs = {'stdout': mock_obj, 'wait.return_value': 0} - mock_obj2.configure_mock(**attrs) - - mock_subproc_open.return_value = mock_obj2 - - resp = functest_utils.execute_command(self.cmd, info=True, - error_msg=self.error_msg, - verbose=True, - output_file=self.output_file) + stream = six.BytesIO() + stream.write(self.cmd_readline().encode("utf-8")) + attrs = { + 'return_value.__enter__.return_value.stdout': stream, + 'return_value.__enter__.return_value.wait.return_value': 0} + mock_subproc_open.configure_mock(**attrs) + resp = functest_utils.execute_command( + self.cmd, info=True, error_msg=self.error_msg, verbose=True, + output_file=self.output_file) self.assertEqual(resp, 0) - msg_exec = ("Executing command: '%s'" % self.cmd) + msg_exec = (f"Executing command: '{self.cmd}'") mock_logger_info.assert_called_once_with(msg_exec) - mopen.assert_called_once_with(self.output_file, "w") + mopen.assert_called_once_with( + self.output_file, "w", encoding='utf-8') @mock.patch('sys.stdout') - def test_execute_command_args_missing_with_success(self, stdout=None): + def test_exec_cmd_args_missing_ok(self, stdout=None): + # pylint: disable=unused-argument with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ as mock_subproc_open: - - FunctestUtilsTesting.readline = 2 - - mock_obj = mock.Mock() - attrs = {'readline.side_effect': self.cmd_readline()} - mock_obj.configure_mock(**attrs) - - mock_obj2 = mock.Mock() - attrs = {'stdout': mock_obj, 'wait.return_value': 0} - mock_obj2.configure_mock(**attrs) - - mock_subproc_open.return_value = mock_obj2 - - resp = functest_utils.execute_command(self.cmd, info=False, - error_msg="", - verbose=False, - output_file=None) + stream = six.BytesIO() + stream.write(self.cmd_readline().encode("utf-8")) + attrs = { + 'return_value.__enter__.return_value.stdout': stream, + 'return_value.__enter__.return_value.wait.return_value': 0} + mock_subproc_open.configure_mock(**attrs) + resp = functest_utils.execute_command( + self.cmd, info=False, error_msg="", verbose=False, + output_file=None) self.assertEqual(resp, 0) @mock.patch('sys.stdout') - def test_execute_command_args_missing_with_error(self, stdout=None): + def test_exec_cmd_args_missing_ko(self, stdout=None): + # pylint: disable=unused-argument with mock.patch('functest.utils.functest_utils.subprocess.Popen') \ as mock_subproc_open: - - FunctestUtilsTesting.readline = 2 - mock_obj = mock.Mock() - attrs = {'readline.side_effect': self.cmd_readline()} - mock_obj.configure_mock(**attrs) - - mock_obj2 = mock.Mock() - attrs = {'stdout': mock_obj, 'wait.return_value': 1} - mock_obj2.configure_mock(**attrs) - - mock_subproc_open.return_value = mock_obj2 - - resp = functest_utils.execute_command(self.cmd, info=False, - error_msg="", - verbose=False, - output_file=None) + attrs = {} + stream = six.BytesIO() + stream.write(self.cmd_readline().encode("utf-8")) + attrs = { + 'return_value.__enter__.return_value.stdout': stream, + 'return_value.__enter__.return_value.wait.return_value': 1} + mock_subproc_open.configure_mock(**attrs) + resp = functest_utils.execute_command( + self.cmd, info=False, error_msg="", verbose=False, + output_file=None) self.assertEqual(resp, 1) - def _get_functest_config(self, var): - return var - - @mock.patch('functest.utils.functest_utils.logger.error') - def test_get_dict_by_test(self, mock_logger_error): - with mock.patch('six.moves.builtins.open', mock.mock_open()), \ - mock.patch('functest.utils.functest_utils.yaml.safe_load') \ - as mock_yaml: - mock_obj = mock.Mock() - attrs = {'get.return_value': [{'testcases': [self.testcase_dict]}]} - mock_obj.configure_mock(**attrs) - - mock_yaml.return_value = mock_obj - - self.assertDictEqual(functest_utils. - get_dict_by_test(self.testname), - self.testcase_dict) - - @mock.patch('functest.utils.functest_utils.get_dict_by_test') - def test_get_criteria_by_test_default(self, mock_get_dict_by_test): - mock_get_dict_by_test.return_value = self.testcase_dict - self.assertEqual(functest_utils.get_criteria_by_test(self.testname), - self.criteria) - - @mock.patch('functest.utils.functest_utils.get_dict_by_test') - def test_get_criteria_by_test_failed(self, mock_get_dict_by_test): - mock_get_dict_by_test.return_value = None - self.assertIsNone(functest_utils.get_criteria_by_test(self.testname)) - - def test_get_parameter_from_yaml_failed(self): + def test_get_param_from_yaml_failed(self): self.file_yaml['general'] = None with mock.patch('six.moves.builtins.open', mock.mock_open()), \ mock.patch('functest.utils.functest_utils.yaml.safe_load') \ @@ -474,11 +177,11 @@ class FunctestUtilsTesting(unittest.TestCase): mock_yaml.return_value = self.file_yaml functest_utils.get_parameter_from_yaml(self.parameter, self.test_file) - self.assertTrue(("The parameter %s is not" - " defined in config_functest.yaml" % - self.parameter) in excep.exception) + self.assertTrue((f"The parameter {self.parameter} is not" + " defined in config_functest.yaml" + ) in excep.exception) - def test_get_parameter_from_yaml_default(self): + def test_get_param_from_yaml_def(self): with mock.patch('six.moves.builtins.open', mock.mock_open()), \ mock.patch('functest.utils.functest_utils.yaml.safe_load') \ as mock_yaml: @@ -488,28 +191,201 @@ class FunctestUtilsTesting(unittest.TestCase): self.test_file), 'test_image_name') - @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml') - def test_get_functest_config_default(self, mock_get_parameter_from_yaml): - with mock.patch.dict(os.environ, - {'CONFIG_FUNCTEST_YAML': self.config_yaml}): - functest_utils.get_functest_config(self.parameter) - mock_get_parameter_from_yaml. \ - assert_called_once_with(self.parameter, - self.config_yaml) - - def test_get_functest_yaml(self): - with mock.patch('six.moves.builtins.open', mock.mock_open()), \ - mock.patch('functest.utils.functest_utils.yaml.safe_load') \ - as mock_yaml: - mock_yaml.return_value = self.file_yaml - resp = functest_utils.get_functest_yaml() - self.assertEqual(resp, self.file_yaml) - - @mock.patch('functest.utils.functest_utils.logger.info') - def test_print_separator(self, mock_logger_info): - functest_utils.print_separator() - mock_logger_info.assert_called_once_with("=======================" - "=======================") + def test_nova_version_exc1(self): + # pylint: disable=protected-access + cloud = mock.Mock() + cloud._compute_client.request.return_value = None + self.assertEqual(functest_utils.get_nova_version(cloud), None) + cloud._compute_client.request.assert_called_once_with('/', 'GET') + + def test_nova_version_exc2(self): + # pylint: disable=protected-access + cloud = mock.Mock() + cloud._compute_client.request.return_value = {"version": None} + self.assertEqual(functest_utils.get_nova_version(cloud), None) + cloud._compute_client.request.assert_called_once_with('/', 'GET') + + def test_nova_version_exc3(self): + # pylint: disable=protected-access + cloud = mock.Mock() + cloud._compute_client.request.return_value = { + "version": {"version": None}} + self.assertEqual(functest_utils.get_nova_version(cloud), None) + cloud._compute_client.request.assert_called_once_with('/', 'GET') + + def test_nova_version_exc4(self): + # pylint: disable=protected-access + cloud = mock.Mock() + cloud._compute_client.request.return_value = { + "version": {"version": "a.b"}} + self.assertEqual(functest_utils.get_nova_version(cloud), None) + cloud._compute_client.request.assert_called_once_with('/', 'GET') + + def test_nova_version(self): + # pylint: disable=protected-access + cloud = mock.Mock() + cloud._compute_client.request.return_value = { + "version": {"version": "2.1"}} + self.assertEqual(functest_utils.get_nova_version(cloud), (2, 1)) + cloud._compute_client.request.assert_called_once_with('/', 'GET') + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 61)) + def test_openstack_version1(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Rocky") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 60)) + def test_openstack_version2(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Queens") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 43)) + def test_openstack_version3(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Pike") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 39)) + def test_openstack_version4(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Ocata") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 26)) + def test_openstack_version5(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Newton") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 13)) + def test_openstack_version6(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Mitaka") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 4)) + def test_openstack_version7(self, *args): + cloud = mock.Mock() + self.assertEqual( + functest_utils.get_openstack_version(cloud), "Liberty") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 1)) + def test_openstack_version8(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version(cloud), "Kilo") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(1, 9)) + def test_openstack_version9(self, *args): + cloud = mock.Mock() + self.assertEqual( + functest_utils.get_openstack_version(cloud), "Unknown") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(3, 1)) + def test_openstack_version10(self, *args): + cloud = mock.Mock() + self.assertEqual( + functest_utils.get_openstack_version(cloud), "Master") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 66)) + def test_openstack_version11(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Stein") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 78)) + def test_openstack_version12(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Train") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 87)) + def test_openstack_version13(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Ussuri") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 88)) + def test_openstack_version14(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Wallaby") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=None) + def test_openstack_version_exc(self, *args): + cloud = mock.Mock() + self.assertEqual( + functest_utils.get_openstack_version(cloud), "Unknown") + args[0].assert_called_once_with(cloud) + + def test_convert_dict_to_ini(self): + self.assertEqual( + functest_utils.convert_dict_to_ini({}), "") + self.assertEqual( + functest_utils.convert_dict_to_ini({"a": "b"}), "a:b") + value = functest_utils.convert_dict_to_ini({"a": "b", "c": "d"}) + self.assertTrue(value in ('a:b,c:d', 'c:d,a:b')) + with self.assertRaises(AssertionError): + functest_utils.convert_list_to_ini("") + + def test_convert_list_to_ini(self): + self.assertEqual( + functest_utils.convert_list_to_ini([]), "") + self.assertEqual( + functest_utils.convert_list_to_ini(["a"]), "a") + self.assertEqual( + functest_utils.convert_list_to_ini(["a", "b"]), "a,b") + with self.assertRaises(AssertionError): + functest_utils.convert_list_to_ini("") + + def test_convert_ini_to_dict(self): + self.assertEqual( + functest_utils.convert_ini_to_dict(""), {}) + self.assertEqual( + functest_utils.convert_ini_to_dict("a:b"), {"a": "b"}) + self.assertEqual( + functest_utils.convert_ini_to_dict( + "a:b,c:d"), {"a": "b", "c": "d"}) + self.assertEqual( + functest_utils.convert_ini_to_dict( + "a:b:c,d:e:f"), {"a:b": "c", "d:e": "f"}) + with self.assertRaises(AssertionError): + functest_utils.convert_list_to_ini({}) + + def test_convert_ini_to_list(self): + self.assertEqual( + functest_utils.convert_ini_to_list(""), []) + self.assertEqual( + functest_utils.convert_ini_to_list("a"), ["a"]) + self.assertEqual( + functest_utils.convert_ini_to_list("a,b"), ["a", "b"]) + with self.assertRaises(AssertionError): + functest_utils.convert_ini_to_list([]) if __name__ == "__main__":