Adding upstream version 4.6.0.upstream/4.6.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

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"