tests/unit/benchmark/runner/test_base.py

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