yardstick/tests/unit/benchmark/runner/test_base.py

   1 ##############################################################################
   2 # Copyright (c) 2017 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import time
  11
  12 import mock
  13 import unittest
  14
  15 from yardstick.benchmark.runners import base
  16 from yardstick.benchmark.runners import iteration
  17
  18
  19 class ActionTestCase(unittest.TestCase):
  20
  21     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  22     def test__execute_shell_command(self, mock_subprocess):
  23         mock_subprocess.check_output.side_effect = Exception()
  24
  25         self.assertEqual(base._execute_shell_command("")[0], -1)
  26
  27     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  28     def test__single_action(self, mock_subprocess):
  29         mock_subprocess.check_output.side_effect = Exception()
  30
  31         base._single_action(0, "echo", mock.MagicMock())
  32
  33     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  34     def test__periodic_action(self, mock_subprocess):
  35         mock_subprocess.check_output.side_effect = Exception()
  36
  37         base._periodic_action(0, "echo", mock.MagicMock())
  38
  39
  40 class RunnerTestCase(unittest.TestCase):
  41
  42     def setUp(self):
  43         self.runner = iteration.IterationRunner({})
  44
  45     @mock.patch("yardstick.benchmark.runners.iteration.multiprocessing")
  46     def test_get_output(self, *args):
  47         self.runner.output_queue.put({'case': 'opnfv_yardstick_tc002'})
  48         self.runner.output_queue.put({'criteria': 'PASS'})
  49
  50         idle_result = {
  51             'case': 'opnfv_yardstick_tc002',
  52             'criteria': 'PASS'
  53         }
  54
  55         for _ in range(1000):
  56             time.sleep(0.01)
  57             if not self.runner.output_queue.empty():
  58                 break
  59         actual_result = self.runner.get_output()
  60         self.assertEqual(idle_result, actual_result)
  61
  62     @mock.patch("yardstick.benchmark.runners.iteration.multiprocessing")
  63     def test_get_result(self, *args):
  64         self.runner.result_queue.put({'case': 'opnfv_yardstick_tc002'})
  65         self.runner.result_queue.put({'criteria': 'PASS'})
  66
  67         idle_result = [
  68             {'case': 'opnfv_yardstick_tc002'},
  69             {'criteria': 'PASS'}
  70         ]
  71
  72         for _ in range(1000):
  73             time.sleep(0.01)
  74             if not self.runner.result_queue.empty():
  75                 break
  76         actual_result = self.runner.get_result()
  77         self.assertEqual(idle_result, actual_result)
  78
  79     def test__run_benchmark(self):
  80         runner = base.Runner(mock.Mock())
  81
  82         with self.assertRaises(NotImplementedError):
  83             runner._run_benchmark(mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock())
  84
  85
  86 def main():
  87     unittest.main()
  88
  89
  90 if __name__ == '__main__':
  91     main()