diff options
Diffstat (limited to 'functest/tests/unit/utils/test_functest_utils.py')
| -rw-r--r-- | functest/tests/unit/utils/test_functest_utils.py | 337 |
1 files changed, 261 insertions, 76 deletions
diff --git a/functest/tests/unit/utils/test_functest_utils.py b/functest/tests/unit/utils/test_functest_utils.py index d35ed8ced..4b642ff9d 100644 --- a/functest/tests/unit/utils/test_functest_utils.py +++ b/functest/tests/unit/utils/test_functest_utils.py @@ -15,12 +15,13 @@ import unittest import mock import pkg_resources +import six from functest.utils import functest_utils class FunctestUtilsTesting(unittest.TestCase): - # pylint: disable=too-many-instance-attributes + # 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'] @@ -81,7 +82,7 @@ class FunctestUtilsTesting(unittest.TestCase): 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 @@ -97,27 +98,20 @@ class FunctestUtilsTesting(unittest.TestCase): 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') @@ -126,50 +120,35 @@ class FunctestUtilsTesting(unittest.TestCase): 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_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') @@ -177,22 +156,16 @@ class FunctestUtilsTesting(unittest.TestCase): # 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 test_get_param_from_yaml_failed(self): @@ -204,9 +177,9 @@ 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_param_from_yaml_def(self): with mock.patch('six.moves.builtins.open', mock.mock_open()), \ @@ -218,6 +191,218 @@ class FunctestUtilsTesting(unittest.TestCase): self.test_file), 'test_image_name') + 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=(2, 89)) + def test_openstack_version15(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Xena") + args[0].assert_called_once_with(cloud) + + @mock.patch('functest.utils.functest_utils.get_nova_version', + return_value=(2, 92)) + def test_openstack_version16(self, *args): + cloud = mock.Mock() + self.assertEqual(functest_utils.get_openstack_version( + cloud), "Zed") + 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__": logging.disable(logging.CRITICAL) |