yardstick/tests/unit/common/test_process.py

   1 # Copyright (c) 2017 Intel Corporation
   2 #
   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
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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.
  14
  15 import mock
  16 import unittest
  17
  18 from oslo_utils import encodeutils
  19
  20 from yardstick.common import exceptions
  21 from yardstick.common import process
  22
  23
  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)
  28
  29     def test_check_if_procces_failed_0(self):
  30         p = mock.MagicMock(**{"exitcode": 0, "name": "debug"})
  31         process.check_if_process_failed(p)
  32
  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)
  37
  38
  39 @mock.patch("yardstick.common.process.multiprocessing")
  40 class TerminateChildrenTestcase(unittest.TestCase):
  41     def test_some_children(self, mock_multiprocessing):
  42         p1 = mock.MagicMock()
  43         p2 = mock.MagicMock()
  44         mock_multiprocessing.active_children.return_value = [p1, p2]
  45         process.terminate_children()
  46
  47     def test_no_children(self, mock_multiprocessing):
  48         mock_multiprocessing.active_children.return_value = []
  49         process.terminate_children()
  50
  51
  52 class ExecuteTestCase(unittest.TestCase):
  53
  54     RET_CODE_OK = 0
  55     RET_CODE_WRONG = 1
  56
  57     def setUp(self):
  58         self._mock_create_process = mock.patch.object(process,
  59                                                       'create_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()
  70
  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)
  82
  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)
  92
  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)
  98
  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)
 109
 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)
 120
 121
 122 class CreateProcessTestCase(unittest.TestCase):
 123
 124     @mock.patch.object(process, 'subprocess_popen')
 125     def test_process_string_command(self, mock_subprocess_popen):
 126         cmd = 'command'
 127         obj = mock.Mock()
 128         mock_subprocess_popen.return_value = obj
 129         out1, out2 = process.create_process(cmd)
 130         self.assertEqual(obj, out1)
 131         self.assertEqual([cmd], out2)
 132
 133     @mock.patch.object(process, 'subprocess_popen')
 134     def test_process_list_command(self, mock_subprocess_popen):
 135         cmd = ['command']
 136         obj = mock.Mock()
 137         mock_subprocess_popen.return_value = obj
 138         out1, out2 = process.create_process(cmd)
 139         self.assertEqual(obj, out1)
 140         self.assertEqual(cmd, out2)
 141
 142     @mock.patch.object(process, 'subprocess_popen')
 143     def test_process_with_env(self, mock_subprocess_popen):
 144         cmd = ['command']
 145         obj = mock.Mock()
 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)