1 # Copyright (c) 2017 Intel Corporation
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
18 from oslo_utils import encodeutils
20 from yardstick.common import exceptions
21 from yardstick.common import process
24 class ProcessTestcase(unittest.TestCase):
25 def test_check_if_procces_failed_None(self):
26 p = mock.MagicMock(**{"exitcode": None, "name": "debug"})
27 process.check_if_process_failed(p)
29 def test_check_if_procces_failed_0(self):
30 p = mock.MagicMock(**{"exitcode": 0, "name": "debug"})
31 process.check_if_process_failed(p)
33 def test_check_if_procces_failed_1(self):
34 p = mock.MagicMock(**{"exitcode": 1, "name": "debug"})
35 with self.assertRaises(RuntimeError):
36 process.check_if_process_failed(p)
39 @mock.patch("yardstick.common.process.multiprocessing")
40 class TerminateChildrenTestcase(unittest.TestCase):
41 def test_some_children(self, mock_multiprocessing):
44 mock_multiprocessing.active_children.return_value = [p1, p2]
45 process.terminate_children()
47 def test_no_children(self, mock_multiprocessing):
48 mock_multiprocessing.active_children.return_value = []
49 process.terminate_children()
52 class ExecuteTestCase(unittest.TestCase):
58 self._mock_create_process = mock.patch.object(process,
60 self.mock_create_process = self._mock_create_process.start()
61 self.obj = mock.Mock()
62 self.cmd = mock.Mock()
63 self.obj.communicate = mock.Mock()
64 self.stdout = 'std out'
65 self.stderr = 'std err'
66 self.obj.communicate.return_value = (self.stdout, self.stderr)
67 self.mock_create_process.return_value = (self.obj, self.cmd)
68 self.input_cmd = 'input cmd'
69 self.additional_env = mock.Mock()
71 def test_execute_with_input(self):
72 process_input = 'process input'
73 self.obj.returncode = self.RET_CODE_OK
74 out = process.execute(self.input_cmd, process_input=process_input,
75 additional_env=self.additional_env)
76 self.obj.communicate.assert_called_once_with(
77 encodeutils.to_utf8(process_input))
78 self.mock_create_process.assert_called_once_with(
79 self.input_cmd, run_as_root=False,
80 additional_env=self.additional_env)
81 self.assertEqual(self.stdout, out)
83 def test_execute_no_input(self):
84 self.obj.returncode = self.RET_CODE_OK
85 out = process.execute(self.input_cmd,
86 additional_env=self.additional_env)
87 self.obj.communicate.assert_called_once_with(None)
88 self.mock_create_process.assert_called_once_with(
89 self.input_cmd, run_as_root=False,
90 additional_env=self.additional_env)
91 self.assertEqual(self.stdout, out)
93 @mock.patch.object(process, 'LOG')
94 def test_execute_exception(self, *args):
95 self.obj.returncode = self.RET_CODE_WRONG
96 with self.assertRaises(exceptions.ProcessExecutionError):
97 process.execute(self.input_cmd, additional_env=self.additional_env)
98 self.obj.communicate.assert_called_once_with(None)
100 def test_execute_with_extra_code(self):
101 self.obj.returncode = self.RET_CODE_WRONG
102 out = process.execute(self.input_cmd,
103 additional_env=self.additional_env,
104 extra_ok_codes=[self.RET_CODE_WRONG])
105 self.obj.communicate.assert_called_once_with(None)
106 self.mock_create_process.assert_called_once_with(
107 self.input_cmd, run_as_root=False,
108 additional_env=self.additional_env)
109 self.assertEqual(self.stdout, out)
111 @mock.patch.object(process, 'LOG')
112 def test_execute_exception_no_check(self, *args):
113 self.obj.returncode = self.RET_CODE_WRONG
114 out = process.execute(self.input_cmd,
115 additional_env=self.additional_env,
116 check_exit_code=False)
117 self.obj.communicate.assert_called_once_with(None)
118 self.mock_create_process.assert_called_once_with(
119 self.input_cmd, run_as_root=False,
120 additional_env=self.additional_env)
121 self.assertEqual(self.stdout, out)
124 class CreateProcessTestCase(unittest.TestCase):
126 @mock.patch.object(process, 'subprocess_popen')
127 def test_process_string_command(self, mock_subprocess_popen):
130 mock_subprocess_popen.return_value = obj
131 out1, out2 = process.create_process(cmd)
132 self.assertEqual(obj, out1)
133 self.assertEqual([cmd], out2)
135 @mock.patch.object(process, 'subprocess_popen')
136 def test_process_list_command(self, mock_subprocess_popen):
139 mock_subprocess_popen.return_value = obj
140 out1, out2 = process.create_process(cmd)
141 self.assertEqual(obj, out1)
142 self.assertEqual(cmd, out2)
144 @mock.patch.object(process, 'subprocess_popen')
145 def test_process_with_env(self, mock_subprocess_popen):
148 additional_env = {'var1': 'value1'}
149 mock_subprocess_popen.return_value = obj
150 out1, out2 = process.create_process(cmd, additional_env=additional_env)
151 self.assertEqual(obj, out1)
152 self.assertEqual(['env', 'var1=value1'] + cmd, out2)