yardstick/tests/unit/benchmark/scenarios/availability/test_basemonitor.py

   1 ##############################################################################
   2 # Copyright (c) 2015 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.scenarios.availability.monitor import basemonitor
  16
  17
  18 class MonitorMgrTestCase(unittest.TestCase):
  19
  20     def setUp(self):
  21         self.monitor_configs = [
  22             {
  23                 "monitor_type": "openstack-cmd",
  24                 "command_name": "openstack router list",
  25                 "monitor_time": 10,
  26                 "monitor_number": 3,
  27                 "sla": {
  28                     "max_outage_time": 5
  29                 }
  30             },
  31             {
  32                 "monitor_type": "process",
  33                 "process_name": "neutron-server",
  34                 "host": "node1",
  35                 "monitor_time": 20,
  36                 "monitor_number": 3,
  37                 "sla": {
  38                     "max_recover_time": 20
  39                 }
  40             }
  41         ]
  42         self.MonitorMgr = basemonitor.MonitorMgr([])
  43         self.MonitorMgr.init_monitors(self.monitor_configs, None)
  44         self.monitor_list = self.MonitorMgr._monitor_list
  45         for mo in self.monitor_list:
  46             mo._result = {"outage_time": 10}
  47
  48     @mock.patch.object(basemonitor, 'BaseMonitor')
  49     def test__MonitorMgr_setup_successful(self, *args):
  50         instance = basemonitor.MonitorMgr({"nova-api": 10})
  51         instance.init_monitors(self.monitor_configs, None)
  52         instance.start_monitors()
  53         instance.wait_monitors()
  54
  55         # TODO(elfoley): Check the return value
  56         ret = instance.verify_SLA()  # pylint: disable=unused-variable
  57
  58     @mock.patch.object(basemonitor, 'BaseMonitor')
  59     def test_MonitorMgr_getitem(self, *args):
  60         monitorMgr = basemonitor.MonitorMgr({"nova-api": 10})
  61         monitorMgr.init_monitors(self.monitor_configs, None)
  62
  63     @mock.patch.object(basemonitor, 'BaseMonitor')
  64     def test_store_result(self, *args):
  65         expect = {'process_neutron-server_outage_time': 10,
  66                   'openstack-router-list_outage_time': 10}
  67         result = {}
  68         self.MonitorMgr.store_result(result)
  69         self.assertDictEqual(result, expect)
  70
  71
  72 class BaseMonitorTestCase(unittest.TestCase):
  73
  74     class MonitorSimple(basemonitor.BaseMonitor):
  75         __monitor_type__ = "MonitorForTest"
  76
  77         def setup(self):
  78             self.monitor_result = False
  79
  80         def monitor_func(self):
  81             return self.monitor_result
  82
  83     def setUp(self):
  84         self.monitor_cfg = {
  85             'monitor_type': 'MonitorForTest',
  86             'command_name': 'nova image-list',
  87             'monitor_time': 0.01,
  88             'sla': {'max_outage_time': 5}
  89         }
  90
  91     def _close_queue(self, instace):
  92         time.sleep(0.1)
  93         instace._queue.close()
  94
  95     def test__basemonitor_start_wait_successful(self):
  96         ins = basemonitor.BaseMonitor(self.monitor_cfg, None, {"nova-api": 10})
  97         self.addCleanup(self._close_queue, ins)
  98         ins.start_monitor()
  99         ins.wait_monitor()
 100
 101     def test__basemonitor_all_successful(self):
 102         ins = self.MonitorSimple(self.monitor_cfg, None, {"nova-api": 10})
 103         self.addCleanup(self._close_queue, ins)
 104         ins.setup()
 105         ins.run()
 106         ins.verify_SLA()
 107
 108     @mock.patch.object(basemonitor, 'multiprocessing')
 109     def test__basemonitor_func_false(self, mock_multiprocess):
 110         ins = self.MonitorSimple(self.monitor_cfg, None, {"nova-api": 10})
 111         self.addCleanup(self._close_queue, ins)
 112         ins.setup()
 113         mock_multiprocess.Event().is_set.return_value = False
 114         ins.run()
 115         ins.verify_SLA()
 116
 117     def test__basemonitor_getmonitorcls_successfule(self):
 118         with self.assertRaises(RuntimeError):
 119             basemonitor.BaseMonitor.get_monitor_cls(self.monitor_cfg)