aboutsummaryrefslogtreecommitdiffstats
path: root/functest/tests/unit/utils/test_functest_utils.py
diff options
context:
space:
mode:
Diffstat (limited to 'functest/tests/unit/utils/test_functest_utils.py')
-rw-r--r--functest/tests/unit/utils/test_functest_utils.py173
1 files changed, 97 insertions, 76 deletions
diff --git a/functest/tests/unit/utils/test_functest_utils.py b/functest/tests/unit/utils/test_functest_utils.py
index dcbbd79e5..4b642ff9d 100644
--- a/functest/tests/unit/utils/test_functest_utils.py
+++ b/functest/tests/unit/utils/test_functest_utils.py
@@ -15,6 +15,7 @@ import unittest
import mock
import pkg_resources
+import six
from functest.utils import functest_utils
@@ -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()), \
@@ -331,6 +304,54 @@ class FunctestUtilsTesting(unittest.TestCase):
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()
@@ -344,7 +365,7 @@ class FunctestUtilsTesting(unittest.TestCase):
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 == "a:b,c:d" or value == "c:d,a:b")
+ self.assertTrue(value in ('a:b,c:d', 'c:d,a:b'))
with self.assertRaises(AssertionError):
functest_utils.convert_list_to_ini("")