import time
import mock
-import unittest
-from subprocess import CalledProcessError
+import subprocess
-
-from yardstick.benchmark.runners import base
+from yardstick.benchmark.runners import base as runner_base
from yardstick.benchmark.runners import iteration
+from yardstick.tests.unit import base as ut_base
-class ActionTestCase(unittest.TestCase):
+class ActionTestCase(ut_base.BaseUnitTestCase):
- @mock.patch("yardstick.benchmark.runners.base.subprocess")
- def test__execute_shell_command(self, mock_subprocess):
- mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
+ def setUp(self):
+ self._mock_log = mock.patch.object(runner_base.log, 'error')
+ self.mock_log = self._mock_log.start()
+ self.addCleanup(self._stop_mocks)
- self.assertEqual(base._execute_shell_command("")[0], -1)
+ def _stop_mocks(self):
+ self._mock_log.stop()
- @mock.patch("yardstick.benchmark.runners.base.subprocess")
- def test__single_action(self, mock_subprocess):
- mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
+ @mock.patch.object(subprocess, 'check_output')
+ def test__execute_shell_command(self, mock_subprocess):
+ mock_subprocess.side_effect = subprocess.CalledProcessError(-1, '')
+ self.assertEqual(runner_base._execute_shell_command("")[0], -1)
- base._single_action(0, "echo", mock.MagicMock())
+ @mock.patch.object(subprocess, 'check_output')
+ def test__single_action(self, mock_subprocess):
+ mock_subprocess.side_effect = subprocess.CalledProcessError(-1, '')
+ runner_base._single_action(0, 'echo', mock.Mock())
- @mock.patch("yardstick.benchmark.runners.base.subprocess")
+ @mock.patch.object(subprocess, 'check_output')
def test__periodic_action(self, mock_subprocess):
- mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
+ mock_subprocess.side_effect = subprocess.CalledProcessError(-1, '')
+ runner_base._periodic_action(0, 'echo', mock.Mock())
+
+
+class ScenarioOutputTestCase(ut_base.BaseUnitTestCase):
+
+ def setUp(self):
+ self.output_queue = mock.Mock()
+ self.scenario_output = runner_base.ScenarioOutput(self.output_queue,
+ sequence=1)
+
+ @mock.patch.object(time, 'time')
+ def test_push(self, mock_time):
+ mock_time.return_value = 2
+ data = {"value1": 1}
+ self.scenario_output.push(data)
+ self.output_queue.put.assert_called_once_with({'timestamp': 2,
+ 'sequence': 1,
+ 'data': data}, True, 10)
- base._periodic_action(0, "echo", mock.MagicMock())
+ def test_push_no_timestamp(self):
+ self.scenario_output["value1"] = 1
+ self.scenario_output.push(None, False)
+ self.output_queue.put.assert_called_once_with({'sequence': 1,
+ 'value1': 1}, True, 10)
-class RunnerTestCase(unittest.TestCase):
+class RunnerTestCase(ut_base.BaseUnitTestCase):
def setUp(self):
config = {
self.assertEqual(idle_result, actual_result)
def test__run_benchmark(self):
- runner = base.Runner(mock.Mock())
+ runner = runner_base.Runner(mock.Mock())
with self.assertRaises(NotImplementedError):
runner._run_benchmark(mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock())