From d835b2cae8abc71958b69362162e6a70c3d7ef63 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Wed, 17 Apr 2024 08:48:59 +0200 Subject: Adding upstream version 4.6.0. Signed-off-by: Daniel Baumann --- test/unittests/test_crashtest_utils.py | 540 +++++++++++++++++++++++++++++++++ 1 file changed, 540 insertions(+) create mode 100644 test/unittests/test_crashtest_utils.py (limited to 'test/unittests/test_crashtest_utils.py') 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" -- cgit v1.2.3