1 ##############################################################################
2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
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 ##############################################################################
15 from yardstick.benchmark.scenarios.availability.monitor import basemonitor
18 class MonitorMgrTestCase(unittest.TestCase):
21 self.monitor_configs = [
23 "monitor_type": "openstack-cmd",
24 "command_name": "openstack router list",
32 "monitor_type": "process",
33 "process_name": "neutron-server",
38 "max_recover_time": 20
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}
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()
55 # TODO(elfoley): Check the return value
56 ret = instance.verify_SLA() # pylint: disable=unused-variable
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)
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}
68 self.MonitorMgr.store_result(result)
69 self.assertDictEqual(result, expect)
72 class BaseMonitorTestCase(unittest.TestCase):
74 class MonitorSimple(basemonitor.BaseMonitor):
75 __monitor_type__ = "MonitorForTest"
78 self.monitor_result = False
80 def monitor_func(self):
81 return self.monitor_result
85 'monitor_type': 'MonitorForTest',
86 'command_name': 'nova image-list',
88 'sla': {'max_outage_time': 5}
91 def _close_queue(self, instace):
93 instace._queue.close()
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)
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)
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)
113 mock_multiprocess.Event().is_set.return_value = False
117 def test__basemonitor_getmonitorcls_successfule(self):
118 with self.assertRaises(RuntimeError):
119 basemonitor.BaseMonitor.get_monitor_cls(self.monitor_cfg)