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 from subprocess import CalledProcessError
  15
  16
  17 from yardstick.benchmark.runners import base
  18 from yardstick.benchmark.runners import iteration
  19
  20
  21 class ActionTestCase(unittest.TestCase):
  22
  23     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  24     def test__execute_shell_command(self, mock_subprocess):
  25         mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
  26
  27         self.assertEqual(base._execute_shell_command("")[0], -1)
  28
  29     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  30     def test__single_action(self, mock_subprocess):
  31         mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
  32
  33         base._single_action(0, "echo", mock.MagicMock())
  34
  35     @mock.patch("yardstick.benchmark.runners.base.subprocess")
  36     def test__periodic_action(self, mock_subprocess):
  37         mock_subprocess.check_output.side_effect = CalledProcessError(-1, '')
  38
  39         base._periodic_action(0, "echo", mock.MagicMock())
  40
  41
  42 class RunnerTestCase(unittest.TestCase):
  43
  44     def setUp(self):
  45         config = {
  46             'output_config': {
  47                 'DEFAULT': {
  48                     'dispatcher': 'file'
  49                 }
  50             }
  51         }
  52         self.runner = iteration.IterationRunner(config)
  53
  54     @mock.patch("yardstick.benchmark.runners.iteration.multiprocessing")
  55     def test_get_output(self, *args):
  56         self.runner.output_queue.put({'case': 'opnfv_yardstick_tc002'})
  57         self.runner.output_queue.put({'criteria': 'PASS'})
  58
  59         idle_result = {
  60             'case': 'opnfv_yardstick_tc002',
  61             'criteria': 'PASS'
  62         }
  63
  64         for _ in range(1000):
  65             time.sleep(0.01)
  66             if not self.runner.output_queue.empty():
  67                 break
  68         actual_result = self.runner.get_output()
  69         self.assertEqual(idle_result, actual_result)
  70
  71     @mock.patch("yardstick.benchmark.runners.iteration.multiprocessing")
  72     def test_get_result(self, *args):
  73         self.runner.result_queue.put({'case': 'opnfv_yardstick_tc002'})
  74         self.runner.result_queue.put({'criteria': 'PASS'})
  75
  76         idle_result = [
  77             {'case': 'opnfv_yardstick_tc002'},
  78             {'criteria': 'PASS'}
  79         ]
  80
  81         for _ in range(1000):
  82             time.sleep(0.01)
  83             if not self.runner.result_queue.empty():
  84                 break
  85         actual_result = self.runner.get_result()
  86         self.assertEqual(idle_result, actual_result)
  87
  88     def test__run_benchmark(self):
  89         runner = base.Runner(mock.Mock())
  90
  91         with self.assertRaises(NotImplementedError):
  92             runner._run_benchmark(mock.Mock(), mock.Mock(), mock.Mock(), mock.Mock())
  93
  94
  95 def main():
  96     unittest.main()
  97
  98
  99 if __name__ == '__main__':
 100     main()