summaryrefslogtreecommitdiffstats
path: root/test/unittests/test_crashtest_utils.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/unittests/test_crashtest_utils.py')
-rw-r--r--test/unittests/test_crashtest_utils.py540
1 files changed, 540 insertions, 0 deletions
diff --git a/test/unittests/test_crashtest_utils.py b/test/unittests/test_crashtest_utils.py
new file mode 100644
index 0000000..f8a579b
--- /dev/null
+++ b/test/unittests/test_crashtest_utils.py
@@ -0,0 +1,540 @@
+import os
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
+
+try:
+ from unittest import mock, TestCase
+except ImportError:
+ import mock
+import logging
+
+from crmsh.crash_test import utils, main, config
+
+
+class TestMyLoggingFormatter(TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ """
+ Global setUp.
+ """
+
+ def setUp(self):
+ """
+ Test setUp.
+ """
+ self.fence_info_inst = utils.FenceInfo()
+
+ def tearDown(self):
+ """
+ Test tearDown.
+ """
+
+ @classmethod
+ def tearDownClass(cls):
+ """
+ Global tearDown.
+ """
+
+
+class TestFenceInfo(TestCase):
+
+ @classmethod
+ def setUpClass(cls):
+ """
+ Global setUp.
+ """
+
+ def setUp(self):
+ """
+ Test setUp.
+ """
+ self.fence_info_inst = utils.FenceInfo()
+
+ def tearDown(self):
+ """
+ Test tearDown.
+ """
+
+ @classmethod
+ def tearDownClass(cls):
+ """
+ Global tearDown.
+ """
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_enabled_false(self, mock_get_property):
+ mock_get_property.return_value = None
+ res = self.fence_info_inst.fence_enabled
+ self.assertEqual(res, False)
+ mock_get_property.assert_called_once_with("stonith-enabled")
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_enabled_true(self, mock_get_property):
+ mock_get_property.return_value = "True"
+ res = self.fence_info_inst.fence_enabled
+ self.assertEqual(res, True)
+ mock_get_property.assert_called_once_with("stonith-enabled")
+
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_action_none(self, mock_get_property, mock_error):
+ mock_get_property.return_value = None
+ res = self.fence_info_inst.fence_action
+ self.assertEqual(res, None)
+ mock_get_property.assert_called_once_with("stonith-action")
+ mock_error.assert_called_once_with('Cluster property "stonith-action" should be reboot|off|poweroff')
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_action(self, mock_get_property):
+ mock_get_property.return_value = "reboot"
+ res = self.fence_info_inst.fence_action
+ self.assertEqual(res, "reboot")
+ mock_get_property.assert_called_once_with("stonith-action")
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_timeout(self, mock_get_property):
+ mock_get_property.return_value = "60s"
+ res = self.fence_info_inst.fence_timeout
+ self.assertEqual(res, "60")
+ mock_get_property.assert_called_once_with("stonith-timeout")
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.get_property')
+ def test_fence_timeout_default(self, mock_get_property):
+ mock_get_property.return_value = None
+ res = self.fence_info_inst.fence_timeout
+ self.assertEqual(res, config.FENCE_TIMEOUT)
+ mock_get_property.assert_called_once_with("stonith-timeout")
+
+
+class TestUtils(TestCase):
+ '''
+ Unitary tests for crash_test/utils.py
+ '''
+
+ @classmethod
+ def setUpClass(cls):
+ """
+ Global setUp.
+ """
+
+ def setUp(self):
+ """
+ Test setUp.
+ """
+
+ def tearDown(self):
+ """
+ Test tearDown.
+ """
+
+ @classmethod
+ def tearDownClass(cls):
+ """
+ Global tearDown.
+ """
+
+ @mock.patch('crmsh.crash_test.utils.datetime')
+ def test_now(self, mock_datetime):
+ mock_now = mock.Mock()
+ mock_datetime.now.return_value = mock_now
+ mock_now.strftime.return_value = "2019/07/05 14:44:55"
+
+ result = utils.now()
+
+ self.assertEqual(result, "2019/07/05 14:44:55")
+ mock_datetime.now.assert_called_once_with()
+ mock_now.strftime.assert_called_once_with("%Y/%m/%d %H:%M:%S")
+
+ @mock.patch('crmsh.crash_test.utils.get_handler')
+ def test_manage_handler(self, mock_get_handler):
+ mock_get_handler.return_value = "handler"
+ utils.logger = mock.Mock()
+ utils.logger.removeHandler = mock.Mock()
+ utils.logger.addHandler = mock.Mock()
+
+ with utils.manage_handler("type1", keep=False):
+ pass
+
+ mock_get_handler.assert_called_once_with(utils.logger, "type1")
+ utils.logger.removeHandler.assert_called_once_with("handler")
+ utils.logger.addHandler.assert_called_once_with("handler")
+
+ @mock.patch('crmsh.crash_test.utils.manage_handler')
+ def test_msg_raw(self, mock_handler):
+ utils.logger = mock.Mock()
+ utils.logger.log = mock.Mock()
+ utils.msg_raw("level1", "msg1")
+ mock_handler.assert_called_once_with("console", True)
+ utils.logger.log.assert_called_once_with("level1", "msg1")
+
+ @mock.patch('crmsh.crash_test.utils.msg_raw')
+ def test_msg_info(self, mock_raw):
+ utils.msg_info("msg1")
+ mock_raw.assert_called_once_with(logging.INFO, "msg1", True)
+
+ @mock.patch('crmsh.crash_test.utils.msg_raw')
+ def test_msg_warn(self, mock_raw):
+ utils.msg_warn("msg1")
+ mock_raw.assert_called_once_with(logging.WARNING, "msg1", True)
+
+ @mock.patch('crmsh.crash_test.utils.msg_raw')
+ def test_msg_error(self, mock_raw):
+ utils.msg_error("msg1")
+ mock_raw.assert_called_once_with(logging.ERROR, "msg1", True)
+
+ @mock.patch('os.fsync')
+ @mock.patch('json.dumps')
+ @mock.patch('builtins.open', create=True)
+ def test_json_dumps(self, mock_open_file, mock_dumps, mock_fsync):
+ main.ctx = mock.Mock(jsonfile="file1", task_list={"process_name": "xin", "age": 38})
+ mock_open_write = mock.mock_open()
+ file_handle = mock_open_write.return_value.__enter__.return_value
+ mock_open_file.return_value = mock_open_write.return_value
+ mock_dumps.return_value = "data"
+
+ utils.json_dumps()
+
+ mock_open_file.assert_called_once_with("file1", "w")
+ mock_dumps.assert_called_once_with(main.ctx.task_list, indent=2)
+ file_handle.write.assert_called_once_with("data")
+ file_handle.flush.assert_called_once_with()
+ mock_fsync.assert_called_once_with(file_handle)
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.this_node')
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_this_node_false(self, mock_run, mock_error, mock_this_node):
+ mock_run.return_value = (1, None, "error data")
+ mock_this_node.return_value = "node1"
+
+ res = utils.this_node()
+ self.assertEqual(res, "node1")
+
+ mock_run.assert_called_once_with("crm_node --name")
+ mock_error.assert_called_once_with("error data")
+ mock_this_node.assert_called_once_with()
+
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_this_node(self, mock_run):
+ mock_run.return_value = (0, "data", None)
+ res = utils.this_node()
+ self.assertEqual(res, "data")
+ mock_run.assert_called_once_with("crm_node --name")
+
+ @mock.patch('crmsh.crash_test.utils.datetime')
+ def test_str_to_datetime(self, mock_datetime):
+ utils.str_to_datetime("Mon Nov 2 15:37:11 2020", "%a %b %d %H:%M:%S %Y")
+ mock_datetime.strptime.assert_called_once_with("Mon Nov 2 15:37:11 2020", "%a %b %d %H:%M:%S %Y")
+
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_corosync_port_list(self, mock_run):
+ output = """
+totem.interface.0.bindnetaddr (str) = 10.10.10.121
+totem.interface.0.mcastaddr (str) = 239.101.40.63
+totem.interface.0.mcastport (u16) = 5405
+totem.interface.0.ttl (u8) = 1
+totem.interface.1.bindnetaddr (str) = 20.20.20.121
+totem.interface.1.mcastaddr (str) = 239.6.213.31
+totem.interface.1.mcastport (u16) = 5407
+totem.interface.1.ttl (u8) = 1
+ """
+ mock_run.return_value = (0, output, None)
+ result = utils.corosync_port_list()
+ expected = ['5405', '5407']
+ self.assertListEqual(result, expected)
+ mock_run.assert_called_once_with("corosync-cmapctl totem.interface")
+
+ def test_get_handler(self):
+ mock_handler1 = mock.Mock(_name="test1_handler")
+ mock_handler2 = mock.Mock(_name="test2_handler")
+ mock_logger = mock.Mock(handlers=[mock_handler1, mock_handler2])
+ res = utils.get_handler(mock_logger, "test1_handler")
+ self.assertEqual(res, mock_handler1)
+
+ @mock.patch('os.getuid')
+ def test_is_root(self, mock_getuid):
+ mock_getuid.return_value = 0
+ self.assertEqual(utils.is_root(), True)
+ mock_getuid.assert_called_once_with()
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.to_ascii')
+ @mock.patch('os.path.basename')
+ @mock.patch('builtins.open')
+ @mock.patch('os.path.join')
+ @mock.patch('os.listdir')
+ def test_get_process_status_false(self, mock_listdir, mock_join, mock_open_file, mock_basename, mock_to_ascii):
+ mock_listdir.return_value = ['1', '2', 'none']
+ mock_join.side_effect = ['/proc/1/cmdline', '/proc/2/cmdline']
+ mock_open_read_1 = mock.mock_open(read_data=b'/usr/sbin/cmd1\x00--user\x00')
+ mock_open_read_2 = mock.mock_open(read_data=b'/usr/sbin/cmd2\x00')
+ mock_open_file.side_effect = [
+ mock_open_read_1.return_value,
+ mock_open_read_2.return_value
+ ]
+ mock_to_ascii.side_effect = [
+ "/usr/sbin/cmd1\x00--user\x00",
+ "/usr/sbin/cmd2\x00"
+ ]
+ mock_basename.side_effect = ["cmd1", "cmd2"]
+
+ rc, pid = utils.get_process_status("sbd")
+ self.assertEqual(rc, False)
+ self.assertEqual(pid, -1)
+
+ mock_listdir.assert_called_once_with('/proc')
+ mock_join.assert_has_calls([
+ mock.call('/proc', '1', 'cmdline'),
+ mock.call('/proc', '2', 'cmdline')
+ ])
+ mock_open_file.assert_has_calls([
+ mock.call('/proc/1/cmdline', 'rb'),
+ mock.call('/proc/2/cmdline', 'rb')
+ ])
+ mock_to_ascii.assert_has_calls([
+ mock.call(b'/usr/sbin/cmd1\x00--user\x00'),
+ mock.call(b'/usr/sbin/cmd2\x00')
+ ])
+
+ @mock.patch('crmsh.crash_test.utils.crmshutils.to_ascii')
+ @mock.patch('os.path.basename')
+ @mock.patch('builtins.open')
+ @mock.patch('os.path.join')
+ @mock.patch('os.listdir')
+ def test_get_process_status(self, mock_listdir, mock_join, mock_open_file, mock_basename, mock_to_ascii):
+ mock_listdir.return_value = ['1', '2', 'none']
+ mock_join.side_effect = ['/proc/1/cmdline', '/proc/2/cmdline']
+ mock_open_read_1 = mock.mock_open(read_data=b'/usr/sbin/cmd1\x00--user\x00')
+ mock_open_read_2 = mock.mock_open(read_data=b'/usr/sbin/sbd\x00')
+ mock_open_file.side_effect = [
+ mock_open_read_1.return_value,
+ mock_open_read_2.return_value
+ ]
+ mock_to_ascii.side_effect = [
+ "/usr/sbin/cmd1\x00--user\x00",
+ "/usr/sbin/sbd\x00"
+ ]
+ mock_basename.side_effect = ["cmd1", "sbd"]
+
+ rc, pid = utils.get_process_status("sbd")
+ self.assertEqual(rc, True)
+ self.assertEqual(pid, 2)
+
+ mock_listdir.assert_called_once_with('/proc')
+ mock_join.assert_has_calls([
+ mock.call('/proc', '1', 'cmdline'),
+ mock.call('/proc', '2', 'cmdline')
+ ])
+ mock_open_file.assert_has_calls([
+ mock.call('/proc/1/cmdline', 'rb'),
+ mock.call('/proc/2/cmdline', 'rb')
+ ])
+ mock_to_ascii.assert_has_calls([
+ mock.call(b'/usr/sbin/cmd1\x00--user\x00'),
+ mock.call(b'/usr/sbin/sbd\x00')
+ ])
+
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_check_node_status_error_cmd(self, mock_run, mock_error):
+ mock_run.return_value = (1, None, "error")
+ res = utils.check_node_status("node1", "member")
+ self.assertEqual(res, False)
+ mock_run.assert_called_once_with("crm_node -l")
+ mock_error.assert_called_once_with("error")
+
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_check_node_status(self, mock_run, mock_error):
+ output = """
+1084783297 15sp2-1 member
+1084783193 15sp2-2 lost
+ """
+ mock_run.return_value = (0, output, None)
+
+ res = utils.check_node_status("15sp2-2", "member")
+ self.assertEqual(res, False)
+ res = utils.check_node_status("15sp2-1", "member")
+ self.assertEqual(res, True)
+
+ mock_run.assert_has_calls([
+ mock.call("crm_node -l"),
+ mock.call("crm_node -l")
+ ])
+ mock_error.assert_not_called()
+
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_online_nodes_empty(self, mock_run):
+ mock_run.return_value = (0, "data", None)
+ res = utils.online_nodes()
+ self.assertEqual(res, [])
+ mock_run.assert_called_once_with("crm_mon -1")
+
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ def test_online_nodes(self, mock_run):
+ output = """
+Node List:
+ * Online: [ 15sp2-1 15sp2-2 ]
+ """
+ mock_run.return_value = (0, output, None)
+ res = utils.online_nodes()
+ self.assertEqual(res, ["15sp2-1", "15sp2-2"])
+ mock_run.assert_called_once_with("crm_mon -1")
+
+ @mock.patch('crmsh.crash_test.utils.online_nodes')
+ def test_peer_node_list_empty(self, mock_online):
+ mock_online.return_value = None
+ res = utils.peer_node_list()
+ self.assertEqual(res, [])
+ mock_online.assert_called_once_with()
+
+ @mock.patch('crmsh.crash_test.utils.this_node')
+ @mock.patch('crmsh.crash_test.utils.online_nodes')
+ def test_peer_node_list(self, mock_online, mock_this_node):
+ mock_online.return_value = ["node1", "node2"]
+ mock_this_node.return_value = "node1"
+ res = utils.peer_node_list()
+ self.assertEqual(res, ["node2"])
+ mock_online.assert_called_once_with()
+
+ # Test is_valid_sbd():
+ @classmethod
+ @mock.patch('os.path.exists')
+ def test_is_valid_sbd_not_exist(cls, mock_os_path_exists):
+ """
+ Test device not exist
+ """
+ dev = "/dev/disk/by-id/scsi-device1"
+ mock_os_path_exists.return_value = False
+
+ res = utils.is_valid_sbd(dev)
+ assert res is False
+
+ @classmethod
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ @mock.patch('os.path.exists')
+ def test_is_valid_sbd_cmd_error(cls, mock_os_path_exists,
+ mock_sbd_check_header, mock_msg_err):
+ """
+ Test device is not valid sbd
+ """
+ dev = "/dev/disk/by-id/scsi-device1"
+ mock_os_path_exists.return_value = True
+ mock_sbd_check_header.return_value = (-1, None, "Unknown error!")
+ mock_msg_err.return_value = ""
+
+ res = utils.is_valid_sbd(dev)
+ mock_msg_err.assert_called_once_with("Unknown error!")
+ assert res is False
+
+ @classmethod
+ @mock.patch('crmsh.crash_test.utils.msg_error')
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ @mock.patch('os.path.exists')
+ def test_is_valid_sbd_not_sbd(cls, mock_os_path_exists,
+ mock_sbd_check_header, mock_msg_err):
+ """
+ Test device is not SBD device
+ """
+ dev = "/dev/disk/by-id/scsi-device1"
+ err_output = """
+==Dumping header on disk {}
+==Header on disk {} NOT dumped
+sbd failed; please check the logs.
+""".format(dev, dev)
+ mock_os_path_exists.return_value = True
+ mock_sbd_check_header.return_value = (1, "==Dumping header on disk {}".format(dev),
+ err_output)
+
+ res = utils.is_valid_sbd(dev)
+ assert res is False
+ mock_msg_err.assert_called_once_with(err_output)
+
+ @classmethod
+ @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
+ @mock.patch('os.path.exists')
+ def test_is_valid_sbd_is_sbd(cls, mock_os_path_exists,
+ mock_sbd_check_header):
+ """
+ Test device is not SBD device
+ """
+ dev = "/dev/disk/by-id/scsi-device1"
+ std_output = """
+==Dumping header on disk {}
+Header version : 2.1
+UUID : f4c99362-6522-46fc-8ce4-7db60aff19bb
+Number of slots : 255
+Sector size : 512
+Timeout (watchdog) : 5
+Timeout (allocate) : 2
+Timeout (loop) : 1
+Timeout (msgwait) : 10
+==Header on disk {} is dumped
+""".format(dev, dev)
+ mock_os_path_exists.return_value = True
+ mock_sbd_check_header.return_value = (0, std_output, None)
+
+ res = utils.is_valid_sbd(dev)
+ assert res is True
+
+ # Test find_candidate_sbd() and _find_match_count()
+ @classmethod
+ @mock.patch('glob.glob')
+ @mock.patch('os.path.basename')
+ @mock.patch('os.path.dirname')
+ def test_find_candidate_no_dev(cls, mock_os_path_dname, mock_os_path_bname,
+ mock_glob):
+ """
+ Test no suitable device
+ """
+ mock_os_path_dname.return_value = "/dev/disk/by-id"
+ mock_os_path_bname.return_value = "scsi-label_CN_devA"
+ mock_glob.return_value = []
+
+ res = utils.find_candidate_sbd("/not-exist-folder/not-exist-dev")
+ assert res == ""
+
+ @classmethod
+ @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
+ @mock.patch('glob.glob')
+ @mock.patch('os.path.basename')
+ @mock.patch('os.path.dirname')
+ def test_find_candidate_no_can(cls, mock_os_path_dname, mock_os_path_bname,
+ mock_glob, mock_is_valid_sbd):
+ """
+ Test no valid candidate device
+ """
+ mock_os_path_dname.return_value = "/dev/disk/by-id"
+ mock_os_path_bname.return_value = "scsi-label_CN_devA"
+ mock_glob.return_value = ["/dev/disk/by-id/scsi-label_DE_devA",
+ "/dev/disk/by-id/scsi-label_DE_devB",
+ "/dev/disk/by-id/scsi-label_DE_devC",
+ "/dev/disk/by-id/scsi-label_DE_devD"]
+ mock_is_valid_sbd.side_effect = [False, False, False, False]
+
+ res = utils.find_candidate_sbd("/dev/disk/by-id/scsi-label_CN_devA")
+ assert res == ""
+
+ @classmethod
+ @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
+ @mock.patch('glob.glob')
+ @mock.patch('os.path.basename')
+ @mock.patch('os.path.dirname')
+ def test_find_candidate_has_multi(cls, mock_os_path_dname, mock_os_path_bname,
+ mock_glob, mock_is_valid_sbd):
+ """
+ Test has multiple valid candidate devices
+ """
+ mock_os_path_dname.return_value = "/dev/disk/by-id"
+ mock_os_path_bname.return_value = "scsi-label_CN_devA"
+ mock_glob.return_value = ["/dev/disk/by-id/scsi-label_DE_devA",
+ "/dev/disk/by-id/scsi-label_DE_devB",
+ "/dev/disk/by-id/scsi-label_CN_devC",
+ "/dev/disk/by-id/scsi-label_CN_devD",
+ "/dev/disk/by-id/scsi-mp_China_devE",
+ "/dev/disk/by-id/scsi-mp_China_devF"]
+ mock_is_valid_sbd.side_effect = [True, False, False, True, True, False]
+
+ res = utils.find_candidate_sbd("/dev/disk/by-id/scsi-label_CN_devA")
+ assert res == "/dev/disk/by-id/scsi-label_CN_devD"
generated by cgit v1.2.3 (git 2.39.1) at 2024-10-03 22:45:40 +0000