5 from tasks.mgr.mgr_test_case import MgrTestCase
8 class TestModuleSelftest(MgrTestCase):
10 That modules with a self-test command can be loaded and execute it
13 This is not a substitute for really testing the modules, but it
14 is quick and is designed to catch regressions that could occur
15 if data structures change in a way that breaks how the modules
20 def _selftest_plugin(self, module_name):
21 self._load_module(module_name)
23 # Execute the module's self-test routine
24 self.mgr_cluster.mon_manager.raw_cluster_cmd(module_name, "self-test")
26 def test_zabbix(self):
27 self._selftest_plugin("zabbix")
29 def test_prometheus(self):
30 self._selftest_plugin("prometheus")
32 def test_influx(self):
33 self._selftest_plugin("influx")
35 def test_selftest_run(self):
36 self._load_module("selftest")
37 self.mgr_cluster.mon_manager.raw_cluster_cmd("mgr", "self-test", "run")
39 def test_selftest_command_spam(self):
40 # Use the selftest module to stress the mgr daemon
41 self._load_module("selftest")
43 # Use the dashboard to test that the mgr is still able to do its job
44 self._assign_ports("dashboard", "server_port")
45 self._load_module("dashboard")
47 original_active = self.mgr_cluster.get_active_id()
48 original_standbys = self.mgr_cluster.get_standby_ids()
50 self.mgr_cluster.mon_manager.raw_cluster_cmd("mgr", "self-test",
51 "background", "start",
54 dashboard_uri = self._get_uri("dashboard")
58 for i in range(0, periods):
60 # Check that an HTTP module remains responsive
61 r = requests.get(dashboard_uri)
62 self.assertEqual(r.status_code, 200)
64 # Check that a native non-module command remains responsive
65 self.mgr_cluster.mon_manager.raw_cluster_cmd("osd", "df")
67 time.sleep(delay - (time.time() - t1))
69 self.mgr_cluster.mon_manager.raw_cluster_cmd("mgr", "self-test",
72 # Check that all mgr daemons are still running
73 self.assertEqual(original_active, self.mgr_cluster.get_active_id())
74 self.assertEqual(original_standbys, self.mgr_cluster.get_standby_ids())