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 def test_execute_exception(self):
94 self.obj.returncode = self.RET_CODE_WRONG
95 self.assertRaises(exceptions.ProcessExecutionError, process.execute,
96 self.input_cmd, additional_env=self.additional_env)
97 self.obj.communicate.assert_called_once_with(None)
99 def test_execute_with_extra_code(self):
100 self.obj.returncode = self.RET_CODE_WRONG
101 out = process.execute(self.input_cmd,
102 additional_env=self.additional_env,
103 extra_ok_codes=[self.RET_CODE_WRONG])
104 self.obj.communicate.assert_called_once_with(None)
105 self.mock_create_process.assert_called_once_with(
106 self.input_cmd, run_as_root=False,
107 additional_env=self.additional_env)
108 self.assertEqual(self.stdout, out)
110 def test_execute_exception_no_check(self):
111 self.obj.returncode = self.RET_CODE_WRONG
112 out = process.execute(self.input_cmd,
113 additional_env=self.additional_env,
114 check_exit_code=False)
115 self.obj.communicate.assert_called_once_with(None)
116 self.mock_create_process.assert_called_once_with(
117 self.input_cmd, run_as_root=False,
118 additional_env=self.additional_env)
119 self.assertEqual(self.stdout, out)
122 class CreateProcessTestCase(unittest.TestCase):
124 @mock.patch.object(process, 'subprocess_popen')
125 def test_process_string_command(self, mock_subprocess_popen):
128 mock_subprocess_popen.return_value = obj
129 out1, out2 = process.create_process(cmd)
130 self.assertEqual(obj, out1)
131 self.assertEqual([cmd], out2)
133 @mock.patch.object(process, 'subprocess_popen')
134 def test_process_list_command(self, mock_subprocess_popen):
137 mock_subprocess_popen.return_value = obj
138 out1, out2 = process.create_process(cmd)
139 self.assertEqual(obj, out1)
140 self.assertEqual(cmd, out2)
142 @mock.patch.object(process, 'subprocess_popen')
143 def test_process_with_env(self, mock_subprocess_popen):
146 additional_env = {'var1': 'value1'}
147 mock_subprocess_popen.return_value = obj
148 out1, out2 = process.create_process(cmd, additional_env=additional_env)
149 self.assertEqual(obj, out1)
150 self.assertEqual(['env', 'var1=value1'] + cmd, out2)