path: root/test/unittests/test_crashtest_check.py

import os
import sys

try:
    from unittest import mock, TestCase
except ImportError:
    import mock

from crmsh import utils as crmshutils
from crmsh.crash_test import check, config


class TestCheck(TestCase):

    @mock.patch('crmsh.crash_test.check.check_cluster')
    def test_check(self, mock_cluster_check):
        ctx = mock.Mock(cluster_check=True)
        check.check(ctx)
        mock_cluster_check.assert_called_once_with()

    @mock.patch('crmsh.crash_test.check.check_firewall')
    @mock.patch('crmsh.crash_test.check.check_time_service')
    @mock.patch('crmsh.crash_test.check.check_my_hostname_resolves')
    def test_check_environment(self, mock_hostname, mock_time, mock_firewall):
        check.check_environment()
        mock_hostname.assert_called_once_with()
        mock_time.assert_called_once_with()
        mock_firewall.assert_called_once_with()

    @mock.patch('crmsh.crash_test.utils.this_node')
    @mock.patch('crmsh.crash_test.check.crmshboot.my_hostname_resolves')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_my_hostname_resolves(self, mock_task_check, mock_hostname, mock_this_node):
        mock_task_inst = mock.Mock()
        mock_task_check.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_hostname.return_value = False
        mock_this_node.return_value = "node1"

        check.check_my_hostname_resolves()

        mock_task_check.assert_called_once_with("Checking hostname resolvable")
        mock_hostname.assert_called_once_with()
        mock_task_inst.error.assert_called_once_with('Hostname "node1" is unresolvable.\n  Please add an entry to /etc/hosts or configure DNS.')

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_available')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_time_service_none(self, mock_task, mock_service_available, mock_service_enabled, mock_service_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_service_available.side_effect = [False, False, False]

        check.check_time_service()

        mock_task.assert_called_once_with("Checking time service")
        mock_service_available.assert_has_calls([
            mock.call('chronyd.service'),
            mock.call('ntp.service'),
            mock.call('ntpd.service')
            ])
        mock_task_inst.warn.assert_called_once_with("No NTP service found.")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_available')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_time_service_warn(self, mock_task, mock_service_available, mock_service_enabled, mock_service_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_service_available.return_value = True
        mock_service_enabled.return_value = False
        mock_service_active.return_value = False

        check.check_time_service()

        mock_task.assert_called_once_with("Checking time service")
        mock_service_available.assert_called_once_with("chronyd.service")
        mock_task_inst.info.assert_called_once_with("chronyd.service is available")
        mock_task_inst.warn.assert_has_calls([
            mock.call("chronyd.service is disabled"),
            mock.call("chronyd.service is not active"),
            ])

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_available')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_time_service(self, mock_task, mock_service_available, mock_service_enabled, mock_service_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_service_available.return_value = True
        mock_service_enabled.return_value = True
        mock_service_active.return_value = True

        check.check_time_service()

        mock_task.assert_called_once_with("Checking time service")
        mock_service_available.assert_called_once_with("chronyd.service")
        mock_task_inst.info.assert_has_calls([
            mock.call("chronyd.service is available"),
            mock.call("chronyd.service is enabled"),
            mock.call("chronyd.service is active")
            ])

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.utils.corosync_port_list')
    def test_check_port_open_return(self, mock_corosync_port, mock_run):
        mock_corosync_port.return_value = ["1234", "5678"]
        mock_run.return_value = (1, None, "error")
        task_inst = mock.Mock()

        check.check_port_open(task_inst, "firewalld")

        mock_corosync_port.assert_called_once_with()
        task_inst.error.assert_called_once_with("error")
        mock_run.assert_called_once_with("firewall-cmd --list-port")

    @mock.patch('crmsh.crash_test.utils.corosync_port_list')
    def test_check_port_open_fail_to_get_port(self, mock_corosync_port):
        mock_corosync_port.return_value = []
        task_inst = mock.Mock()

        check.check_port_open(task_inst, "firewalld")

        mock_corosync_port.assert_called_once_with()
        task_inst.error.assert_called_once_with("Can not get corosync's port")

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.utils.corosync_port_list')
    def test_check_port_open(self, mock_corosync_port, mock_run):
        mock_corosync_port.return_value = ["1234", "5678"]
        output_cmd = """
 1234/udp
 4444/tcp
        """
        mock_run.return_value = (0, output_cmd, None)
        task_inst = mock.Mock()

        check.check_port_open(task_inst, "firewalld")

        mock_corosync_port.assert_called_once_with()
        task_inst.error.assert_called_once_with("UDP port 5678 should open in firewalld")
        mock_run.assert_called_once_with("firewall-cmd --list-port")
        task_inst.info.assert_called_once_with("UDP port 1234 is opened in firewalld")

    @mock.patch('crmsh.crash_test.check.crmshutils.package_is_installed')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_firewall_not_intalled(self, mock_task, mock_installed):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_installed.side_effect = [False, False]

        check.check_firewall()

        mock_task.assert_called_once_with("Checking firewall")
        mock_installed.assert_has_calls([
            mock.call("firewalld"),
            mock.call("SuSEfirewall2")
            ])
        mock_task_inst.warn.assert_called_once_with("Failed to detect firewall")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.crash_test.check.crmshutils.package_is_installed')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_firewall_warn(self, mock_task, mock_installed, mock_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_installed.return_value = True
        mock_active.return_value = False

        check.check_firewall()

        mock_task.assert_called_once_with("Checking firewall")
        mock_installed.assert_called_once_with("firewalld")
        mock_task_inst.info.assert_called_once_with("firewalld.service is available")
        mock_task_inst.warn.assert_called_once_with("firewalld.service is not active")

    @mock.patch('crmsh.crash_test.check.check_port_open')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.crash_test.check.crmshutils.package_is_installed')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_firewall(self, mock_task, mock_installed, mock_active, mock_check_port):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_installed.return_value = True
        mock_active.return_value = True

        check.check_firewall()

        mock_task.assert_called_once_with("Checking firewall")
        mock_installed.assert_called_once_with("firewalld")
        mock_task_inst.info.assert_has_calls([
            mock.call("firewalld.service is available"),
            mock.call("firewalld.service is active")
            ])
        mock_active.assert_called_once_with("firewalld")
        mock_check_port.assert_called_once_with(mock_task_inst, "firewalld")

    @mock.patch('crmsh.crash_test.check.check_cluster_service')
    def test_check_cluster_return(self, mock_check_cluster):
        mock_check_cluster.return_value = False
        check.check_cluster()
        mock_check_cluster.assert_called_once_with()

    @mock.patch('crmsh.crash_test.check.check_resources')
    @mock.patch('crmsh.crash_test.check.check_nodes')
    @mock.patch('crmsh.crash_test.check.check_fencing')
    @mock.patch('crmsh.crash_test.check.check_cluster_service')
    def test_check_cluster(self, mock_check_cluster, mock_check_fencing, mock_check_nodes, mock_check_resources):
        mock_check_cluster.return_value = True
        check.check_cluster()
        mock_check_cluster.assert_called_once_with()
        mock_check_fencing.assert_called_once_with()
        mock_check_nodes.assert_called_once_with()
        mock_check_resources.assert_called_once_with()

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_cluster_service_pacemaker_disable(self, mock_task, mock_enabled, mock_active):
        mock_task_inst = mock.Mock(passed=False)
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_enabled.side_effect = [False, True]
        mock_active.side_effect = [True, False]

        res = check.check_cluster_service()
        self.assertEqual(res, False)

        mock_task.assert_called_once_with("Checking cluster service", quiet=False)
        mock_enabled.assert_has_calls([
            mock.call("pacemaker"),
            mock.call("corosync")
            ])
        mock_task_inst.warn.assert_has_calls([
            mock.call("pacemaker.service is disabled"),
            mock.call("corosync.service is enabled")
            ])
        mock_active.assert_has_calls([
            mock.call("corosync"),
            mock.call("pacemaker")
            ])
        mock_task_inst.info.assert_called_once_with("corosync.service is running")
        mock_task_inst.error.assert_called_once_with("pacemaker.service is not running!")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_cluster_service(self, mock_task, mock_enabled, mock_active):
        mock_task_inst = mock.Mock(passed=True)
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_enabled.side_effect = [True, True]
        mock_active.side_effect = [True, True]

        res = check.check_cluster_service()
        self.assertEqual(res, True)

        mock_task.assert_called_once_with("Checking cluster service", quiet=False)
        mock_enabled.assert_has_calls([
            mock.call("pacemaker"),
            mock.call("corosync")
            ])
        mock_active.assert_has_calls([
            mock.call("corosync"),
            mock.call("pacemaker")
            ])
        mock_task_inst.info.assert_has_calls([
            mock.call("pacemaker.service is enabled"),
            mock.call("corosync.service is running"),
            mock.call("pacemaker.service is running")
            ])
        mock_task_inst.warn.assert_called_once_with("corosync.service is enabled")

    @mock.patch('crmsh.crash_test.utils.FenceInfo')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_fencing_no_stonith(self, mock_task, mock_fence_info):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_fence_info_inst = mock.Mock(fence_enabled=False)
        mock_fence_info.return_value = mock_fence_info_inst

        check.check_fencing()

        mock_task.assert_called_once_with("Checking STONITH/Fence")
        mock_fence_info.assert_called_once_with()
        mock_task_inst.warn.assert_called_once_with("stonith is disabled")

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.utils.FenceInfo')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_fencing_no_resources(self, mock_task, mock_fence_info, mock_run):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_fence_info_inst = mock.Mock(fence_enabled=True)
        mock_fence_info.return_value = mock_fence_info_inst
        mock_run.return_value = (1, None, None)

        check.check_fencing()

        mock_task.assert_called_once_with("Checking STONITH/Fence")
        mock_fence_info.assert_called_once_with()
        mock_run.assert_called_once_with("crm_mon -r1 | grep '(stonith:.*):'")
        mock_task_inst.info.assert_called_once_with("stonith is enabled")
        mock_task_inst.warn.assert_called_once_with("No stonith resource configured!")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.utils.FenceInfo')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_fencing_has_warn(self, mock_task, mock_fence_info, mock_run, mock_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_fence_info_inst = mock.Mock(fence_enabled=True)
        mock_fence_info.return_value = mock_fence_info_inst
        mock_run.return_value = (0, "* stonith-sbd  (stonith:external/sbd):  Stopped (disabled)", None)
        mock_active.return_value = False

        check.check_fencing()

        mock_task.assert_called_once_with("Checking STONITH/Fence")
        mock_fence_info.assert_called_once_with()
        mock_run.assert_called_once_with("crm_mon -r1 | grep '(stonith:.*):'")
        mock_task_inst.info.assert_has_calls([
            mock.call("stonith is enabled"),
            mock.call("stonith resource stonith-sbd(external/sbd) is configured")
            ])
        mock_task_inst.warn.assert_has_calls([
            mock.call("stonith resource stonith-sbd(external/sbd) is Stopped"),
            mock.call("sbd service is not running!")
            ])

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.utils.FenceInfo')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_fencing(self, mock_task, mock_fence_info, mock_run, mock_active):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_fence_info_inst = mock.Mock(fence_enabled=True)
        mock_fence_info.return_value = mock_fence_info_inst
        mock_run.return_value = (0, "* stonith-sbd  (stonith:external/sbd):  Started node2", None)
        mock_active.return_value = True

        check.check_fencing()

        mock_task.assert_called_once_with("Checking STONITH/Fence")
        mock_fence_info.assert_called_once_with()
        mock_run.assert_called_once_with("crm_mon -r1 | grep '(stonith:.*):'")
        mock_task_inst.info.assert_has_calls([
            mock.call("stonith is enabled"),
            mock.call("stonith resource stonith-sbd(external/sbd) is configured"),
            mock.call("stonith resource stonith-sbd(external/sbd) is Started"),
            mock.call("sbd service is running")
            ])
        mock_active.assert_called_once_with("sbd")

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_nodes_error(self, mock_task, mock_run):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_run.return_value = (1, None, "error data")

        check.check_nodes()

        mock_task.assert_called_once_with("Checking nodes")
        mock_run.assert_called_once_with("crm_mon -1")
        mock_task_inst.error.assert_called_once_with("run \"crm_mon -1\" error: error data")

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_nodes(self, mock_task, mock_run):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        output = """
Cluster Summary:
  * Stack: corosync
  * Current DC: 15sp2-1 (version 2.0.3+20200511.2b248d828-1.10-2.0.3+20200511.2b248d828) - partition with quorum
  * Last updated: Tue Nov  3 14:09:29 2020
  * Last change:  Tue Nov  3 13:47:29 2020 by root via cibadmin on 15sp2-1
  * 2 nodes configured
  * 1 resource instance configured (1 DISABLED)

Node List:
  * Online: [ 15sp2-1 ]
  * OFFLINE: [ 15sp2-2 ]
        """
        mock_run.return_value = (0, output, None)

        check.check_nodes()

        mock_task.assert_called_once_with("Checking nodes")
        mock_run.assert_called_once_with("crm_mon -1")
        mock_task_inst.info.assert_has_calls([
            mock.call("DC node: 15sp2-1"),
            mock.call("Cluster have quorum"),
            mock.call("Online nodes: [ 15sp2-1 ]")
            ])
        mock_task_inst.warn.assert_called_once_with("OFFLINE nodes: [ 15sp2-2 ]")

    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_nodes_warn(self, mock_task, mock_run):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        output = """
Cluster Summary:
  * Stack: corosync
  * Current DC: NONE
  * Last updated: Tue Nov  3 14:16:49 2020
  * Last change:  Tue Nov  3 14:09:29 2020 by root via cibadmin on 15sp2-1
  * 2 nodes configured
  * 1 resource instance configured (1 DISABLED)

Node List:
  * Node 15sp2-1: UNCLEAN (offline)
  * Node 15sp2-2: UNCLEAN (offline)

Active Resources:
  * No active resources
        """
        mock_run.return_value = (0, output, None)

        check.check_nodes()

        mock_task.assert_called_once_with("Checking nodes")
        mock_run.assert_called_once_with("crm_mon -1")
        mock_task_inst.warn.assert_has_calls([
            mock.call("Cluster lost quorum!"),
            mock.call("Node 15sp2-1 is UNCLEAN!"),
            mock.call("Node 15sp2-2 is UNCLEAN!")
            ])

    @mock.patch('crmsh.crash_test.check.completers.resources_stopped')
    @mock.patch('crmsh.crash_test.check.completers.resources_started')
    @mock.patch('crmsh.crash_test.task.TaskCheck')
    def test_check_resources(self, mock_task, mock_started, mock_stopped):
        mock_task_inst = mock.Mock()
        mock_task.return_value = mock_task_inst
        mock_task_inst.run.return_value.__enter__ = mock.Mock()
        mock_task_inst.run.return_value.__exit__ = mock.Mock()
        mock_started.return_value = ["r1", "r2"]
        mock_stopped.return_value = ["r3", "r4"]

        check.check_resources()

        mock_task.assert_called_once_with("Checking resources")
        mock_task_inst.info.assert_has_calls([
            mock.call("Started resources: r1,r2"),
            mock.call("Stopped resources: r3,r4")
            ])

    # Test fix()
    @classmethod
    @mock.patch('crmsh.crash_test.check.correct_sbd')
    @mock.patch('crmsh.crash_test.check.check_sbd')
    def test_fix_no_candidate(cls, mock_check_sbd, mock_correct_sbd):
        """
        Test fix() has no valid candidate
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        ctx = mock.Mock(fix_conf=True)
        mock_check_sbd.return_value = dev
        check.fix(ctx)
        mock_correct_sbd.assert_called_once_with(ctx, dev)

    @classmethod
    @mock.patch('crmsh.crash_test.check.correct_sbd')
    @mock.patch('crmsh.crash_test.check.check_sbd')
    def test_fix_has_candidate(cls, mock_check_sbd, mock_correct_sbd):
        """
        Test fix() has valid candidate
        """
        ctx = mock.Mock(fix_conf=True)
        mock_check_sbd.return_value = ""
        mock_correct_sbd.return_value = ""
        check.fix(ctx)
        mock_correct_sbd.assert_not_called()

    # Test check_sbd()
    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('os.path.exists')
    def test_check_sbd_no_conf(cls, mock_os_path_exists,
                               mock_utils_msg_info, mock_run):
        """
        Test no configuration file
        """
        mock_os_path_exists.return_value = False
        check.check_sbd()
        mock_utils_msg_info.assert_called_with("SBD configuration file {} not found.".
                                               format(config.SBD_CONF), to_stdout=False)
        mock_run.assert_called_once_with()

    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    def test_check_sbd_not_configured(cls, mock_os_path_exists, mock_utils_parse_sysconf,
                                      mock_utils_msg_info, mock_run):
        """
        Test SBD device not configured
        """
        mock_os_path_exists.return_value = True
        mock_utils_parse_sysconf.return_value = {}
        check.check_sbd()
        mock_utils_msg_info.assert_called_with("SBD DEVICE not used.", to_stdout=False)
        mock_run.assert_called_once_with()

    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    def test_check_sbd_exist_and_valid(cls, mock_os_path_exists,
                                       mock_utils_parse_sysconf, mock_find_hexdump,
                                       mock_msg_info, mock_is_valid_sbd, mock_run):
        """
        Test configured SBD device exist and valid
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        mock_os_path_exists.side_effect = [True, True, True]
        mock_utils_parse_sysconf.return_value = {"SBD_DEVICE": dev}
        mock_find_hexdump.return_value = (0, "/usr/bin/hexdump", None)
        mock_is_valid_sbd.return_value = True

        check.check_sbd()
        mock_msg_info.assert_called_with("'{}' is a valid SBD device.".format(dev),
                                         to_stdout=False)
        mock_run.assert_called_once_with()

    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.find_candidate_sbd')
    @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
    @mock.patch('crmsh.crash_test.utils.msg_warn')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    def test_check_sbd_exist_and_not_valid_but_no_can(cls, mock_os_path_exists,
                                                      mock_utils_parse_sysconf, mock_find_hexdump,
                                                      mock_msg_warn, mock_is_valid_sbd,
                                                      mock_find_can_sbd, mock_run):
        """
        Test configured SBD device not valid and no candidate
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        mock_os_path_exists.side_effect = [True, True, True]
        mock_utils_parse_sysconf.return_value = {"SBD_DEVICE": dev}
        mock_find_hexdump.return_value = (0, "/usr/bin/hexdump", None)
        mock_is_valid_sbd.return_value = False
        mock_find_can_sbd.return_value = ""

        check.check_sbd()
        mock_msg_warn.assert_has_calls(
            [mock.call("Device '{}' is not valid for SBD, may need initialize.".
                       format(dev), to_stdout=False),
             mock.call("Fail to find a valid candidate SBD device.",
                       to_stdout=False)])
        mock_run.assert_called_once_with()

    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.find_candidate_sbd')
    @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.crash_test.utils.msg_warn')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    def test_check_sbd_exist_and_not_exist_has_can(cls, mock_os_path_exists,
                                                   mock_utils_parse_sysconf, mock_find_hexdump,
                                                   mock_msg_warn, mock_msg_info, mock_is_valid_sbd,
                                                   mock_find_can_sbd, mock_run):
        """
        Test configured SBD device not valid but has candidate
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        candev = "/dev/disk/by-id/scsi-SATA_ST2037LM010-2R82_WDZ5J36B"
        mock_os_path_exists.side_effect = [True, False]
        mock_utils_parse_sysconf.return_value = {"SBD_DEVICE": dev}
        mock_find_hexdump.return_value = (0, "/usr/bin/hexdump", None)
        mock_is_valid_sbd.return_value = False
        mock_find_can_sbd.return_value = candev

        check.check_sbd()
        mock_msg_warn.assert_called_once_with(
            "SBD device '{}' is not exist.".format(dev),
            to_stdout=False)
        mock_msg_info.assert_called_with("Found '{}' with SBD header exist.".format(candev),
                                         to_stdout=False)
        mock_run.assert_called_once_with()

    @classmethod
    @mock.patch('crmsh.crash_test.task.TaskCheck.print_result')
    @mock.patch('crmsh.crash_test.utils.find_candidate_sbd')
    @mock.patch('crmsh.crash_test.utils.is_valid_sbd')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.crash_test.utils.msg_warn')
    @mock.patch('crmsh.sh.ShellUtils.get_stdout_stderr')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    def test_check_sbd_exist_and_not_valid_has_can(cls, mock_os_path_exists,
                                                   mock_utils_parse_sysconf, mock_find_hexdump,
                                                   mock_msg_warn, mock_msg_info, mock_is_valid_sbd,
                                                   mock_find_can_sbd, mock_run):
        """
        Test configured SBD device not valid but has candidate
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        candev = "/dev/disk/by-id/scsi-SATA_ST2037LM010-2R82_WDZ5J36B"
        mock_os_path_exists.side_effect = [True, True, True]
        mock_utils_parse_sysconf.return_value = {"SBD_DEVICE": dev}
        mock_find_hexdump.return_value = (0, "/usr/bin/hexdump", None)
        mock_is_valid_sbd.return_value = False
        mock_find_can_sbd.return_value = candev

        check.check_sbd()
        mock_msg_warn.assert_called_once_with(
            "Device '{}' is not valid for SBD, may need initialize.".format(dev),
            to_stdout=False)
        mock_msg_info.assert_called_with("Found '{}' with SBD header exist.".format(candev),
                                         to_stdout=False)
        mock_run.assert_called_once_with()

    # Test correct_sbd()
    @mock.patch('crmsh.crash_test.task.Task.error')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.crash_test.main.Context')
    def test_correct_sbd_exception_no_conf(self, mock_context, mock_os_path_exists,
                                           mock_utils_parse_sysconf, mock_msg_info,
                                           mock_error):
        """
        Test correct_sbd with exception
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        mock_context = mock.Mock(yes=True)
        mock_os_path_exists.side_effect = [False, True]
        mock_utils_parse_sysconf.retrun_value = {"SBD_DEVICE": dev}

        with self.assertRaises(crmshutils.TerminateSubCommand):
            check.correct_sbd(mock_context, dev)

        mock_msg_info.assert_called_once_with('Replace SBD_DEVICE with candidate {}'.
                                              format(dev), to_stdout=False)
        mock_error.assert_called_once_with('Configure file {} not exist!'.
                                           format(config.SBD_CONF))

    @mock.patch('crmsh.crash_test.task.Task.error')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.crash_test.main.Context')
    def test_correct_sbd_exception_no_dev(self, mock_context, mock_os_path_exists,
                                          mock_utils_parse_sysconf, mock_msg_info,
                                          mock_error):
        """
        Test correct_sbd with exception
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        mock_context = mock.Mock(yes=True)
        mock_os_path_exists.side_effect = [True, False]
        mock_utils_parse_sysconf.retrun_value = {"SBD_DEVICE": dev}

        with self.assertRaises(crmshutils.TerminateSubCommand):
            check.correct_sbd(mock_context, dev)

        mock_msg_info.assert_called_once_with('Replace SBD_DEVICE with candidate {}'.
                                              format(dev), to_stdout=False)
        mock_error.assert_called_once_with('Device {} not exist!'.format(dev))

    @classmethod
    @mock.patch('builtins.open')
    @mock.patch('crmsh.crash_test.task.TaskFixSBD.verify')
    @mock.patch('tempfile.mkstemp')
    @mock.patch('os.remove')
    @mock.patch('shutil.move')
    @mock.patch('shutil.copymode')
    @mock.patch('shutil.copyfile')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.crash_test.main.Context')
    def test_correct_sbd(cls, mock_context, mock_os_path_exists,
                         mock_utils_parse_sysconf, mock_msg_info, mock_copyfile,
                         mock_copymode, mock_move, mock_remove,
                         mock_mkstemp, mock_sbd_verify, mock_open):
        """
        Test correct_sbd
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        bak = "/tmp/tmpmby3ty9g"
        edit = "/tmp/tmpnic4t30s"
        mock_context.return_value = mock.Mock(yes=True)
        mock_os_path_exists.side_effect = [True, True]
        mock_utils_parse_sysconf.retrun_value = {"SBD_DEVICE": dev}
        mock_open.side_effect = [
            mock.mock_open(read_data="data1").return_value,
            mock.mock_open(read_data="SBD_DEVICE={}".format(dev)).return_value
        ]
        mock_mkstemp.side_effect = [(1, bak), (2, edit)]

        check.correct_sbd(mock_context, dev)

        mock_msg_info.assert_called_once_with('Replace SBD_DEVICE with candidate {}'.
                                              format(dev), to_stdout=False)
        mock_copyfile.assert_called_once_with(config.SBD_CONF, bak)
        mock_copymode.assert_called_once_with(config.SBD_CONF, edit)
        mock_move.assert_called_once_with(edit, config.SBD_CONF)
        mock_remove.assert_called()
        mock_sbd_verify.assert_called_once_with()

    @classmethod
    @mock.patch('builtins.open')
    @mock.patch('crmsh.crash_test.task.Task.error')
    @mock.patch('tempfile.mkstemp')
    @mock.patch('shutil.copymode')
    @mock.patch('shutil.copyfile')
    @mock.patch('crmsh.crash_test.utils.msg_info')
    @mock.patch('crmsh.utils.parse_sysconfig')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.crash_test.main.Context')
    def test_correct_sbd_run_exception(cls, mock_context, mock_os_path_exists,
                                       mock_utils_parse_sysconf, mock_msg_info, mock_copyfile,
                                       mock_copymode, mock_mkstemp, mock_msg_error,
                                       mock_open):
        """
        Test correct_sbd
        """
        dev = "/dev/disk/by-id/scsi-SATA_ST2000LM007-1R81_WDZ5J42A"
        bak = "/tmp/tmpmby3ty9g"
        edit = "/tmp/tmpnic4t30s"
        mock_context.return_value = mock.Mock(yes=True)
        mock_os_path_exists.side_effect = [True, True]
        mock_utils_parse_sysconf.retrun_value = {"SBD_DEVICE": dev}
        mock_open.side_effect = [
            mock.mock_open(read_data="data1").return_value,
            mock.mock_open(read_data="data2").return_value
        ]
        mock_mkstemp.side_effect = [(1, bak), (2, edit)]
        mock_copymode.side_effect = Exception('Copy file error!')

        with cls.assertRaises(cls, crmshutils.TerminateSubCommand):
            check.correct_sbd(mock_context, dev)

        mock_msg_info.assert_called_once_with('Replace SBD_DEVICE with candidate {}'.
                                              format(dev), to_stdout=False)
        mock_copyfile.assert_has_calls([mock.call(config.SBD_CONF, bak),
                                        mock.call(bak, config.SBD_CONF)])
        mock_copymode.assert_called_once_with(config.SBD_CONF, edit)
        mock_msg_error.assert_called_once_with('Fail to modify file {}'.
                                               format(config.SBD_CONF))