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']
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
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)
mock_logger_info.assert_called_once_with(msg_exec)
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)
mock_logger_info.assert_called_once_with(msg_exec)
# 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')
# 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):
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=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)
Code Review / functest.git / blobdiff