Code Review / functest.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | review | tree
raw | inline | side by side
Update Flake8
[functest.git] / functest / tests / unit / utils / test_functest_utils.py
diff --git a/functest/tests/unit/utils/test_functest_utils.py b/functest/tests/unit/utils/test_functest_utils.py
index ab90de5..4f3f16f 100644 (file)
--- 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,23 +98,15 @@ 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)
-
+            stream = six.BytesIO()
+            stream.write(self.cmd_readline().encode("utf-8"))
             mock_obj2 = mock.Mock()
-            attrs = {'stdout': mock_obj, 'wait.return_value': 1}
+            attrs = {'stdout': stream, '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)
+            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)
             mock_logger_info.assert_called_once_with(msg_exec)
@@ -126,23 +119,15 @@ 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)
-
+            stream = six.BytesIO()
+            stream.write(self.cmd_readline().encode("utf-8"))
             mock_obj2 = mock.Mock()
-            attrs = {'stdout': mock_obj, 'wait.return_value': 0}
+            attrs = {'stdout': stream, '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)
+            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)
             mock_logger_info.assert_called_once_with(msg_exec)
@@ -153,23 +138,15 @@ 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)
-
+            stream = six.BytesIO()
+            stream.write(self.cmd_readline().encode("utf-8"))
             mock_obj2 = mock.Mock()
-            attrs = {'stdout': mock_obj, 'wait.return_value': 0}
+            attrs = {'stdout': stream, '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)
+            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 +154,15 @@ 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)
-
+            stream = six.BytesIO()
+            stream.write(self.cmd_readline().encode("utf-8"))
             mock_obj2 = mock.Mock()
-            attrs = {'stdout': mock_obj, 'wait.return_value': 1}
+            attrs = {'stdout': stream, '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)
+            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):
@@ -330,6 +300,22 @@ class FunctestUtilsTesting(unittest.TestCase):
             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=None)
     def test_openstack_version_exc(self, *args):
@@ -338,6 +324,50 @@ class FunctestUtilsTesting(unittest.TestCase):
             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)
Test suites and cases to verify OPNFV Platform functionality