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     @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)
  99
 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)
 110
 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)
 122
 123
 124 class CreateProcessTestCase(unittest.TestCase):
 125
 126     @mock.patch.object(process, 'subprocess_popen')
 127     def test_process_string_command(self, mock_subprocess_popen):
 128         cmd = 'command'
 129         obj = mock.Mock()
 130         mock_subprocess_popen.return_value = obj
 131         out1, out2 = process.create_process(cmd)
 132         self.assertEqual(obj, out1)
 133         self.assertEqual([cmd], out2)
 134
 135     @mock.patch.object(process, 'subprocess_popen')
 136     def test_process_list_command(self, mock_subprocess_popen):
 137         cmd = ['command']
 138         obj = mock.Mock()
 139         mock_subprocess_popen.return_value = obj
 140         out1, out2 = process.create_process(cmd)
 141         self.assertEqual(obj, out1)
 142         self.assertEqual(cmd, out2)
 143
 144     @mock.patch.object(process, 'subprocess_popen')
 145     def test_process_with_env(self, mock_subprocess_popen):
 146         cmd = ['command']
 147         obj = mock.Mock()
 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)