path: root/test/unittests/test_sbd.py

import os
import unittest
import logging

try:
    from unittest import mock
except ImportError:
    import mock

from crmsh import bootstrap
from crmsh import sbd


class TestSBDTimeout(unittest.TestCase):
    """
    Unitary tests for crmsh.sbd.SBDTimeout
    """

    @classmethod
    def setUpClass(cls):
        """
        Global setUp.
        """

    def setUp(self):
        """
        Test setUp.
        """
        _dict = {"sbd.watchdog_timeout": 5, "sbd.msgwait": 10}
        _inst_q = mock.Mock()
        self.sbd_timeout_inst = sbd.SBDTimeout(mock.Mock(profiles_dict=_dict, is_s390=True, qdevice_inst=_inst_q))

    def tearDown(self):
        """
        Test tearDown.
        """

    @classmethod
    def tearDownClass(cls):
        """
        Global tearDown.
        """

    def test_initialize_timeout(self):
        self.sbd_timeout_inst._set_sbd_watchdog_timeout = mock.Mock()
        self.sbd_timeout_inst._set_sbd_msgwait = mock.Mock()
        self.sbd_timeout_inst._adjust_sbd_watchdog_timeout_with_diskless_and_qdevice = mock.Mock()
        self.sbd_timeout_inst.initialize_timeout()
        self.sbd_timeout_inst._set_sbd_watchdog_timeout.assert_called_once()
        self.sbd_timeout_inst._set_sbd_msgwait.assert_not_called()
        self.sbd_timeout_inst._adjust_sbd_watchdog_timeout_with_diskless_and_qdevice.assert_called_once()

    @mock.patch('logging.Logger.warning')
    def test_set_sbd_watchdog_timeout(self, mock_warn):
        self.sbd_timeout_inst._set_sbd_watchdog_timeout()
        mock_warn.assert_called_once_with("sbd_watchdog_timeout is set to %d for s390, it was %d", sbd.SBDTimeout.SBD_WATCHDOG_TIMEOUT_DEFAULT_S390, 5)

    @mock.patch('logging.Logger.warning')
    def test_set_sbd_msgwait(self, mock_warn):
        self.sbd_timeout_inst.sbd_watchdog_timeout = 15
        self.sbd_timeout_inst._set_sbd_msgwait()
        mock_warn.assert_called_once_with("sbd msgwait is set to %d, it was %d", 30, 10)

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.utils.get_qdevice_sync_timeout')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.utils.is_qdevice_configured')
    def test_adjust_sbd_watchdog_timeout_with_diskless_and_qdevice_sbd_stage(self, mock_is_configured, mock_is_active, mock_get_sync, mock_warn):
        mock_is_configured.return_value = True
        mock_is_active.return_value = True
        mock_get_sync.return_value = 15
        self.sbd_timeout_inst.sbd_watchdog_timeout = 5
        self.sbd_timeout_inst._adjust_sbd_watchdog_timeout_with_diskless_and_qdevice()
        mock_warn.assert_called_once_with("sbd_watchdog_timeout is set to 20 for qdevice, it was 5")

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.utils.is_qdevice_configured')
    def test_adjust_sbd_watchdog_timeout_with_diskless_and_qdevice_all(self, mock_is_configured, mock_warn):
        mock_is_configured.return_value = False
        self.sbd_timeout_inst.sbd_watchdog_timeout = 5
        self.sbd_timeout_inst._adjust_sbd_watchdog_timeout_with_diskless_and_qdevice()
        mock_warn.assert_called_once_with("sbd_watchdog_timeout is set to 35 for qdevice, it was 5")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_get_sbd_msgwait_exception(self, mock_run):
        mock_run.return_value = "data"
        with self.assertRaises(ValueError) as err:
            sbd.SBDTimeout.get_sbd_msgwait("/dev/sda1")
        self.assertEqual("Cannot get sbd msgwait for /dev/sda1", str(err.exception))
        mock_run.assert_called_once_with("sbd -d /dev/sda1 dump")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_get_sbd_msgwait(self, mock_run):
        mock_run.return_value = """
        Timeout (loop)     : 1
        Timeout (msgwait)  : 10
        ==Header on disk /dev/sda1 is dumped
        """
        res = sbd.SBDTimeout.get_sbd_msgwait("/dev/sda1")
        assert res == 10
        mock_run.assert_called_once_with("sbd -d /dev/sda1 dump")

    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_get_sbd_watchdog_timeout_exception(self, mock_get):
        mock_get.return_value = None
        with self.assertRaises(ValueError) as err:
            sbd.SBDTimeout.get_sbd_watchdog_timeout()
        self.assertEqual("Cannot get the value of SBD_WATCHDOG_TIMEOUT", str(err.exception))
        mock_get.assert_called_once_with("SBD_WATCHDOG_TIMEOUT")

    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_get_sbd_watchdog_timeout(self, mock_get):
        mock_get.return_value = 5
        res = sbd.SBDTimeout.get_sbd_watchdog_timeout()
        assert res == 5
        mock_get.assert_called_once_with("SBD_WATCHDOG_TIMEOUT")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    def test_get_stonith_watchdog_timeout_return(self, mock_active):
        mock_active.return_value = False
        res = sbd.SBDTimeout.get_stonith_watchdog_timeout()
        assert res == sbd.SBDTimeout.STONITH_WATCHDOG_TIMEOUT_DEFAULT
        mock_active.assert_called_once_with("pacemaker.service")

    @mock.patch('crmsh.utils.get_property')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    def test_get_stonith_watchdog_timeout(self, mock_active, mock_get_property):
        mock_active.return_value = True
        mock_get_property.return_value = "60s"
        res = sbd.SBDTimeout.get_stonith_watchdog_timeout()
        assert res == 60
        mock_active.assert_called_once_with("pacemaker.service")

    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    @mock.patch('crmsh.utils.detect_virt')
    @mock.patch('crmsh.sbd.SBDTimeout.get_sbd_delay_start_expected')
    @mock.patch('crmsh.utils.get_pcmk_delay_max')
    @mock.patch('crmsh.sbd.SBDTimeout.get_sbd_msgwait')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_device_from_config')
    @mock.patch('crmsh.utils.is_2node_cluster_without_qdevice')
    def test_load_configurations(self, mock_2node, mock_get_sbd_dev, mock_get_msgwait, mock_pcmk_delay, mock_delay_expected, mock_detect, mock_get_sbd_value, mock_debug):
        mock_2node.return_value = True
        mock_debug.return_value = False
        mock_get_sbd_value.return_value = "no"
        mock_get_sbd_dev.return_value = ["/dev/sda1"]
        mock_get_msgwait.return_value = 30
        mock_pcmk_delay.return_value = 30

        self.sbd_timeout_inst._load_configurations()

        mock_2node.assert_called_once_with()
        mock_get_sbd_dev.assert_called_once_with()
        mock_get_msgwait.assert_called_once_with("/dev/sda1")
        mock_pcmk_delay.assert_called_once_with(True)

    @mock.patch('logging.Logger.debug')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    @mock.patch('crmsh.utils.detect_virt')
    @mock.patch('crmsh.sbd.SBDTimeout.get_sbd_delay_start_expected')
    @mock.patch('crmsh.sbd.SBDTimeout.get_stonith_watchdog_timeout')
    @mock.patch('crmsh.sbd.SBDTimeout.get_sbd_watchdog_timeout')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_device_from_config')
    @mock.patch('crmsh.utils.is_2node_cluster_without_qdevice')
    def test_load_configurations_diskless(self, mock_2node, mock_get_sbd_dev, mock_get_watchdog_timeout, mock_get_stonith_watchdog_timeout, mock_delay_expected, mock_detect, mock_get_sbd_value, mock_debug):
        mock_2node.return_value = True
        mock_debug.return_value = False
        mock_get_sbd_value.return_value = "no"
        mock_get_sbd_dev.return_value = []
        mock_get_watchdog_timeout.return_value = 30
        mock_get_stonith_watchdog_timeout.return_value = 30

        self.sbd_timeout_inst._load_configurations()

        mock_2node.assert_called_once_with()
        mock_get_sbd_dev.assert_called_once_with()
        mock_get_watchdog_timeout.assert_called_once_with()
        mock_get_stonith_watchdog_timeout.assert_called_once_with()

    @mock.patch('crmsh.corosync.token_and_consensus_timeout')
    @mock.patch('logging.Logger.debug')
    def test_get_stonith_timeout_expected(self, mock_debug, mock_general):
        self.sbd_timeout_inst.disk_based = True
        self.sbd_timeout_inst.pcmk_delay_max = 30
        self.sbd_timeout_inst.msgwait = 30
        mock_general.return_value = 11
        res = self.sbd_timeout_inst.get_stonith_timeout_expected()
        assert res == 83

    @mock.patch('crmsh.corosync.token_and_consensus_timeout')
    @mock.patch('logging.Logger.debug')
    def test_get_stonith_timeout_expected_diskless(self, mock_debug, mock_general):
        self.sbd_timeout_inst.disk_based = False
        self.sbd_timeout_inst.stonith_watchdog_timeout = -1
        self.sbd_timeout_inst.sbd_watchdog_timeout = 15
        mock_general.return_value = 11
        res = self.sbd_timeout_inst.get_stonith_timeout_expected()
        assert res == 71

    @mock.patch('crmsh.corosync.token_and_consensus_timeout')
    def test_get_sbd_delay_start_expected(self, mock_corosync):
        mock_corosync.return_value = 30
        self.sbd_timeout_inst.disk_based = True
        self.sbd_timeout_inst.pcmk_delay_max = 30
        self.sbd_timeout_inst.msgwait = 30
        res = self.sbd_timeout_inst.get_sbd_delay_start_expected()
        assert res == 90

    @mock.patch('crmsh.corosync.token_and_consensus_timeout')
    def test_get_sbd_delay_start_expected_diskless(self, mock_corosync):
        mock_corosync.return_value = 30
        self.sbd_timeout_inst.disk_based = False
        self.sbd_timeout_inst.sbd_watchdog_timeout = 30
        res = self.sbd_timeout_inst.get_sbd_delay_start_expected()
        assert res == 90

    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_is_sbd_delay_start(self, mock_get_sbd_value):
        mock_get_sbd_value.return_value = "100"
        assert sbd.SBDTimeout.is_sbd_delay_start() is True
        mock_get_sbd_value.assert_called_once_with("SBD_DELAY_START")

    @mock.patch('crmsh.sbd.SBDManager.update_configuration')
    def test_adjust_sbd_delay_start_return(self, mock_update):
        self.sbd_timeout_inst.sbd_delay_start_value_expected = 100
        self.sbd_timeout_inst.sbd_delay_start_value_from_config = "100"
        self.sbd_timeout_inst.adjust_sbd_delay_start()
        mock_update.assert_not_called()

    @mock.patch('crmsh.sbd.SBDManager.update_configuration')
    def test_adjust_sbd_delay_start(self, mock_update):
        self.sbd_timeout_inst.sbd_delay_start_value_expected = 100
        self.sbd_timeout_inst.sbd_delay_start_value_from_config = "no"
        self.sbd_timeout_inst.adjust_sbd_delay_start()
        mock_update.assert_called_once_with({"SBD_DELAY_START": "100"})

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_adjust_systemd_start_timeout_no_delay_start_no(self, mock_get_sbd_value, mock_run):
        mock_get_sbd_value.return_value = "no"
        self.sbd_timeout_inst.adjust_systemd_start_timeout()
        mock_run.assert_not_called()

    @mock.patch('crmsh.utils.mkdirp')
    @mock.patch('crmsh.utils.get_systemd_timeout_start_in_sec')
    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_adjust_systemd_start_timeout_no_delay_start_return(self, mock_get_sbd_value, mock_run, mock_get_systemd_sec, mock_mkdirp):
        mock_get_sbd_value.return_value = "10"
        mock_run.return_value = "1min 30s"
        mock_get_systemd_sec.return_value = 90
        self.sbd_timeout_inst.adjust_systemd_start_timeout()
        mock_run.assert_called_once_with("systemctl show -p TimeoutStartUSec sbd --value")
        mock_get_systemd_sec.assert_called_once_with("1min 30s")
        mock_mkdirp.assert_not_called()

    @mock.patch('crmsh.utils.cluster_run_cmd')
    @mock.patch('crmsh.bootstrap.sync_file')
    @mock.patch('crmsh.utils.str2file')
    @mock.patch('crmsh.utils.mkdirp')
    @mock.patch('crmsh.utils.get_systemd_timeout_start_in_sec')
    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_adjust_systemd_start_timeout_no_delay_start(self, mock_get_sbd_value, mock_run, mock_get_systemd_sec, mock_mkdirp, mock_str2file, mock_csync2, mock_cluster_run):
        mock_get_sbd_value.return_value = "100"
        mock_run.return_value = "1min 30s"
        mock_get_systemd_sec.return_value = 90
        self.sbd_timeout_inst.adjust_systemd_start_timeout()
        mock_run.assert_called_once_with("systemctl show -p TimeoutStartUSec sbd --value")
        mock_get_systemd_sec.assert_called_once_with("1min 30s")
        mock_mkdirp.assert_called_once_with(bootstrap.SBD_SYSTEMD_DELAY_START_DIR)
        mock_str2file.assert_called_once_with('[Service]\nTimeoutSec=120', '/etc/systemd/system/sbd.service.d/sbd_delay_start.conf')
        mock_csync2.assert_called_once_with(bootstrap.SBD_SYSTEMD_DELAY_START_DIR)
        mock_cluster_run.assert_called_once_with("systemctl daemon-reload")

    @mock.patch('crmsh.sbd.SBDTimeout.get_sbd_watchdog_timeout')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_get_sbd_delay_start_sec_from_sysconfig_yes(self, mock_get_sbd_value, mock_get_sbd_timeout):
        mock_get_sbd_value.return_value = "yes"
        mock_get_sbd_timeout.return_value = 30
        assert sbd.SBDTimeout.get_sbd_delay_start_sec_from_sysconfig() == 60
        mock_get_sbd_value.assert_called_once_with("SBD_DELAY_START")

    @mock.patch('crmsh.sbd.SBDManager.get_sbd_value_from_config')
    def test_get_sbd_delay_start_sec_from_sysconfig(self, mock_get_sbd_value):
        mock_get_sbd_value.return_value = "30"
        assert sbd.SBDTimeout.get_sbd_delay_start_sec_from_sysconfig() == 30
        mock_get_sbd_value.assert_called_once_with("SBD_DELAY_START")


class TestSBDManager(unittest.TestCase):
    """
    Unitary tests for crmsh.sbd.SBDManager
    """

    @classmethod
    def setUpClass(cls):
        """
        Global setUp.
        """

    def setUp(self):
        """
        Test setUp.
        """
        self.sbd_inst = sbd.SBDManager(mock.Mock(sbd_devices=["/dev/sdb1", "/dev/sdc1"], diskless_sbd=False))
        self.sbd_inst_devices_gt_3 = sbd.SBDManager(mock.Mock(sbd_devices=["/dev/sdb1", "/dev/sdc1", "/dev/sdd1", "/dev/sde1"]))
        self.sbd_inst_interactive = sbd.SBDManager(mock.Mock(sbd_devices=[], diskless_sbd=False))
        self.sbd_inst_diskless = sbd.SBDManager(mock.Mock(sbd_devices=[], diskless_sbd=True))

    def tearDown(self):
        """
        Test tearDown.
        """

    @classmethod
    def tearDownClass(cls):
        """
        Global tearDown.
        """

    @mock.patch('logging.Logger.warning')
    def test_get_sbd_device_interactive_yes_to_all(self, mock_warn):
        self.sbd_inst._context = mock.Mock(yes_to_all=True)
        self.sbd_inst._get_sbd_device_interactive()
        mock_warn.assert_called_once_with(sbd.SBDManager.SBD_WARNING)

    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('logging.Logger.info')
    @mock.patch('logging.Logger.warning')
    def test_get_sbd_device_interactive_not_confirm(self, mock_warn, mock_status, mock_confirm):
        self.sbd_inst._context.yes_to_all = False
        mock_confirm.return_value = False
        self.sbd_inst._get_sbd_device_interactive()
        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_warn.assert_called_once_with("Not configuring SBD - STONITH will be disabled.")

    @mock.patch('crmsh.sbd.SBDManager._no_overwrite_check')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('logging.Logger.info')
    def test_get_sbd_device_interactive_already_configured(self, mock_status, mock_confirm, mock_from_config, mock_no_overwrite):
        self.sbd_inst._context = mock.Mock(yes_to_all=False)
        mock_confirm.return_value = True
        mock_from_config.return_value = ["/dev/sda1"]
        mock_no_overwrite.return_value = True

        res = self.sbd_inst._get_sbd_device_interactive()
        self.assertEqual(res, ["/dev/sda1"])

        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_confirm.assert_has_calls([
            mock.call("Do you wish to use SBD?"),
            ])
        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_from_config.assert_called_once_with()

    @mock.patch('crmsh.bootstrap.prompt_for_string')
    @mock.patch('crmsh.sbd.SBDManager._no_overwrite_check')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('logging.Logger.info')
    def test_get_sbd_device_interactive_diskless(self, mock_status, mock_confirm, mock_from_config, mock_no_overwrite, mock_prompt):
        self.sbd_inst._context = mock.Mock(yes_to_all=False)
        mock_confirm.return_value = True
        mock_no_overwrite.return_value = False
        mock_from_config.return_value = []
        mock_prompt.return_value = "none"

        self.sbd_inst._get_sbd_device_interactive()

        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_from_config.assert_called_once_with()
        mock_prompt.assert_called_once_with('Path to storage device (e.g. /dev/disk/by-id/...), or "none" for diskless sbd, use ";" as separator for multi path', 'none|\\/.*')

    @mock.patch('crmsh.bootstrap.prompt_for_string')
    @mock.patch('crmsh.sbd.SBDManager._no_overwrite_check')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('logging.Logger.info')
    def test_get_sbd_device_interactive_null_and_diskless(self, mock_status, mock_confirm, mock_from_config, mock_no_overwrite, mock_prompt):
        self.sbd_inst._context = mock.Mock(yes_to_all=False)
        mock_confirm.return_value = True
        mock_no_overwrite.return_value = False
        mock_from_config.return_value = []
        mock_prompt.return_value = "none"

        self.sbd_inst._get_sbd_device_interactive()

        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_confirm.assert_called_once_with("Do you wish to use SBD?")
        mock_from_config.assert_called_once_with()
        mock_prompt.assert_has_calls([
            mock.call('Path to storage device (e.g. /dev/disk/by-id/...), or "none" for diskless sbd, use ";" as separator for multi path', 'none|\\/.*')
            ])

    @mock.patch('crmsh.utils.re_split_string')
    @mock.patch('logging.Logger.warning')
    @mock.patch('logging.Logger.error')
    @mock.patch('crmsh.sbd.SBDManager._verify_sbd_device')
    @mock.patch('crmsh.bootstrap.prompt_for_string')
    @mock.patch('crmsh.sbd.SBDManager._no_overwrite_check')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('logging.Logger.info')
    def test_get_sbd_device_interactive(self, mock_status, mock_confirm, mock_from_config, mock_no_overwrite, mock_prompt, mock_verify, mock_error_msg, mock_warn, mock_split):
        self.sbd_inst._context = mock.Mock(yes_to_all=False)
        mock_confirm.side_effect = [True, False, True]
        mock_from_config.return_value = []
        mock_no_overwrite.return_value = False
        mock_prompt.side_effect = ["/dev/test1", "/dev/sda1", "/dev/sdb1"]
        mock_split.side_effect = [["/dev/test1"], ["/dev/sda1"], ["/dev/sdb1"]]
        mock_verify.side_effect = [ValueError("/dev/test1 error"), None, None]

        res = self.sbd_inst._get_sbd_device_interactive()
        self.assertEqual(res, ["/dev/sdb1"])

        mock_status.assert_called_once_with(sbd.SBDManager.SBD_STATUS_DESCRIPTION)
        mock_confirm.assert_has_calls([
            mock.call("Do you wish to use SBD?"),
            mock.call("Are you sure you wish to use this device?")
            ])
        mock_from_config.assert_called_once_with()
        mock_error_msg.assert_called_once_with("/dev/test1 error")
        mock_warn.assert_has_calls([
            mock.call("All data on /dev/sda1 will be destroyed!"),
            mock.call("All data on /dev/sdb1 will be destroyed!")
            ])
        mock_prompt.assert_has_calls([
            mock.call('Path to storage device (e.g. /dev/disk/by-id/...), or "none" for diskless sbd, use ";" as separator for multi path', 'none|\\/.*') for x in range(3)
            ])
        mock_split.assert_has_calls([
            mock.call(sbd.SBDManager.PARSE_RE, "/dev/test1"),
            mock.call(sbd.SBDManager.PARSE_RE, "/dev/sda1"),
            mock.call(sbd.SBDManager.PARSE_RE, "/dev/sdb1"),
            ])

    def test_verify_sbd_device_gt_3(self):
        assert self.sbd_inst_devices_gt_3.sbd_devices_input == ["/dev/sdb1", "/dev/sdc1", "/dev/sdd1", "/dev/sde1"]
        dev_list = self.sbd_inst_devices_gt_3.sbd_devices_input
        with self.assertRaises(ValueError) as err:
            self.sbd_inst_devices_gt_3._verify_sbd_device(dev_list)
        self.assertEqual("Maximum number of SBD device is 3", str(err.exception))

    @mock.patch('crmsh.sbd.SBDManager._compare_device_uuid')
    @mock.patch('crmsh.utils.is_block_device')
    def test_verify_sbd_device_not_block(self, mock_block_device, mock_compare):
        assert self.sbd_inst.sbd_devices_input == ["/dev/sdb1", "/dev/sdc1"]
        dev_list = self.sbd_inst.sbd_devices_input
        mock_block_device.side_effect = [True, False]

        with self.assertRaises(ValueError) as err:
            self.sbd_inst._verify_sbd_device(dev_list)
        self.assertEqual("/dev/sdc1 doesn't look like a block device", str(err.exception))

        mock_block_device.assert_has_calls([mock.call("/dev/sdb1"), mock.call("/dev/sdc1")])
        mock_compare.assert_called_once_with("/dev/sdb1", [])

    @mock.patch('crmsh.sbd.SBDManager._verify_sbd_device')
    def test_get_sbd_device_from_option(self, mock_verify):
        self.sbd_inst._get_sbd_device()
        mock_verify.assert_called_once_with(['/dev/sdb1', '/dev/sdc1'])

    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_interactive')
    def test_get_sbd_device_from_interactive(self, mock_interactive):
        mock_interactive.return_value = ["/dev/sdb1", "/dev/sdc1"]
        self.sbd_inst_interactive._get_sbd_device()
        mock_interactive.assert_called_once_with()

    def test_get_sbd_device_diskless(self):
        self.sbd_inst_diskless._get_sbd_device()

    @mock.patch('crmsh.sbd.SBDTimeout')
    @mock.patch('logging.Logger.info')
    def test_initialize_sbd_return(self, mock_info, mock_sbd_timeout):
        mock_inst = mock.Mock()
        mock_sbd_timeout.return_value = mock_inst
        self.sbd_inst_diskless._context = mock.Mock(profiles_dict={})
        self.sbd_inst_diskless._initialize_sbd()
        mock_info.assert_called_once_with("Configuring diskless SBD")
        mock_inst.initialize_timeout.assert_called_once_with()

    @mock.patch('crmsh.utils.fatal')
    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('crmsh.sbd.SBDTimeout')
    @mock.patch('logging.Logger.info')
    def test_initialize_sbd(self, mock_info, mock_sbd_timeout, mock_invoke, mock_error):
        mock_inst = mock.Mock(sbd_msgwait=10, sbd_watchdog_timeout=5)
        mock_sbd_timeout.return_value = mock_inst
        mock_inst.set_sbd_watchdog_timeout = mock.Mock()
        mock_inst.set_sbd_msgwait = mock.Mock()
        self.sbd_inst._sbd_devices = ["/dev/sdb1", "/dev/sdc1"]
        mock_invoke.side_effect = [(True, None, None), (False, None, "error")]
        mock_error.side_effect = ValueError

        with self.assertRaises(ValueError):
            self.sbd_inst._initialize_sbd()

        mock_invoke.assert_has_calls([
            mock.call("sbd -4 10 -1 5 -d /dev/sdb1 create"),
            mock.call("sbd -4 10 -1 5 -d /dev/sdc1 create")
            ])
        mock_error.assert_called_once_with("Failed to initialize SBD device /dev/sdc1: error")

    @mock.patch('crmsh.bootstrap.sync_file')
    @mock.patch('crmsh.utils.sysconfig_set')
    @mock.patch('shutil.copyfile')
    def test_update_configuration(self, mock_copy, mock_sysconfig, mock_update):
        self.sbd_inst._sbd_devices = ["/dev/sdb1", "/dev/sdc1"]
        self.sbd_inst._watchdog_inst = mock.Mock(watchdog_device_name="/dev/watchdog")
        self.sbd_inst.timeout_inst = mock.Mock(sbd_watchdog_timeout=15)

        self.sbd_inst._update_sbd_configuration()

        mock_copy.assert_called_once_with("/usr/share/fillup-templates/sysconfig.sbd", "/etc/sysconfig/sbd")
        mock_sysconfig.assert_called_once_with("/etc/sysconfig/sbd", SBD_WATCHDOG_DEV='/dev/watchdog', SBD_DEVICE='/dev/sdb1;/dev/sdc1', SBD_WATCHDOG_TIMEOUT="15")
        mock_update.assert_called_once_with("/etc/sysconfig/sbd")

    @mock.patch('crmsh.bootstrap.utils.parse_sysconfig')
    def test_get_sbd_device_from_config_none(self, mock_parse):
        mock_parse_inst = mock.Mock()
        mock_parse.return_value = mock_parse_inst
        mock_parse_inst.get.return_value = None

        res = self.sbd_inst._get_sbd_device_from_config()
        assert res == []

        mock_parse.assert_called_once_with("/etc/sysconfig/sbd")
        mock_parse_inst.get.assert_called_once_with("SBD_DEVICE")

    @mock.patch('crmsh.utils.re_split_string')
    @mock.patch('crmsh.bootstrap.utils.parse_sysconfig')
    def test_get_sbd_device_from_config(self, mock_parse, mock_split):
        mock_parse_inst = mock.Mock()
        mock_parse.return_value = mock_parse_inst
        mock_parse_inst.get.return_value = "/dev/sdb1;/dev/sdc1"
        mock_split.return_value = ["/dev/sdb1", "/dev/sdc1"]

        res = self.sbd_inst._get_sbd_device_from_config()
        assert res == ["/dev/sdb1", "/dev/sdc1"]

        mock_parse.assert_called_once_with("/etc/sysconfig/sbd")
        mock_parse_inst.get.assert_called_once_with("SBD_DEVICE")
        mock_split.assert_called_once_with(sbd.SBDManager.PARSE_RE, "/dev/sdb1;/dev/sdc1")

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.utils.get_quorum_votes_dict')
    def test_warn_diskless_sbd_diskless(self, mock_vote, mock_warn):
        self.sbd_inst_diskless._context = mock.Mock(cluster_is_running=False)
        self.sbd_inst_diskless._warn_diskless_sbd()
        mock_vote.assert_not_called()
        mock_warn.assert_called_once_with(sbd.SBDManager.DISKLESS_SBD_WARNING)

    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.utils.get_quorum_votes_dict')
    def test_warn_diskless_sbd_peer(self, mock_vote, mock_warn):
        mock_vote.return_value = {'Expected': '1'}
        self.sbd_inst_diskless._warn_diskless_sbd("node2")
        mock_vote.assert_called_once_with("node2")
        mock_warn.assert_called_once_with(sbd.SBDManager.DISKLESS_SBD_WARNING)

    @mock.patch('crmsh.utils.package_is_installed')
    def test_sbd_init_not_installed(self, mock_package):
        mock_package.return_value = False
        self.sbd_inst.sbd_init()
        mock_package.assert_called_once_with("sbd")

    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('crmsh.sbd.SBDManager._update_sbd_configuration')
    @mock.patch('crmsh.sbd.SBDManager._initialize_sbd')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device')
    @mock.patch('crmsh.watchdog.Watchdog')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_sbd_init_return(self, mock_package, mock_watchdog, mock_get_device, mock_initialize, mock_update, mock_invoke):
        mock_package.return_value = True
        self.sbd_inst._sbd_devices = None
        self.sbd_inst.diskless_sbd = False
        self.sbd_inst._context = mock.Mock(watchdog=None)
        mock_watchdog_inst = mock.Mock()
        mock_watchdog.return_value = mock_watchdog_inst
        mock_watchdog_inst.init_watchdog = mock.Mock()

        self.sbd_inst.sbd_init()

        mock_package.assert_called_once_with("sbd")
        mock_get_device.assert_called_once_with()
        mock_initialize.assert_not_called()
        mock_update.assert_not_called()
        mock_watchdog.assert_called_once_with(_input=None)
        mock_watchdog_inst.init_watchdog.assert_called_once_with()
        mock_invoke.assert_called_once_with("systemctl disable sbd.service")
 
    @mock.patch('crmsh.sbd.SBDManager._enable_sbd_service')
    @mock.patch('crmsh.sbd.SBDManager._warn_diskless_sbd')
    @mock.patch('crmsh.sbd.SBDManager._update_sbd_configuration')
    @mock.patch('crmsh.sbd.SBDManager._initialize_sbd')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device')
    @mock.patch('crmsh.watchdog.Watchdog')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_sbd_init(self, mock_package, mock_watchdog, mock_get_device, mock_initialize, mock_update, mock_warn, mock_enable_sbd):
        mock_package.return_value = True
        self.sbd_inst_diskless._context = mock.Mock(watchdog=None)
        mock_watchdog_inst = mock.Mock()
        mock_watchdog.return_value = mock_watchdog_inst
        mock_watchdog_inst.init_watchdog = mock.Mock()
        self.sbd_inst_diskless.sbd_init()

        mock_package.assert_called_once_with("sbd")
        mock_get_device.assert_called_once_with()
        mock_initialize.assert_called_once_with()
        mock_update.assert_called_once_with()
        mock_watchdog.assert_called_once_with(_input=None)
        mock_watchdog_inst.init_watchdog.assert_called_once_with()
        mock_warn.assert_called_once_with()
        mock_enable_sbd.assert_called_once_with()

    @mock.patch('crmsh.sbd.SBDManager.configure_sbd_resource_and_properties')
    @mock.patch('crmsh.bootstrap.wait_for_cluster')
    @mock.patch('crmsh.utils.cluster_run_cmd')
    @mock.patch('logging.Logger.info')
    @mock.patch('crmsh.xmlutil.CrmMonXmlParser')
    def test_restart_cluster_on_needed_no_ra_running(self, mock_parser, mock_status, mock_cluster_run, mock_wait, mock_config_sbd_ra):
        mock_parser().is_any_resource_running.return_value = False
        self.sbd_inst._restart_cluster_and_configure_sbd_ra()
        mock_status.assert_called_once_with("Restarting cluster service")
        mock_cluster_run.assert_called_once_with("crm cluster restart")
        mock_wait.assert_called_once_with()
        mock_config_sbd_ra.assert_called_once_with()

    @mock.patch('crmsh.sbd.SBDTimeout.get_stonith_timeout')
    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.xmlutil.CrmMonXmlParser')
    def test_restart_cluster_on_needed_diskless(self, mock_parser, mock_warn, mock_get_timeout):
        mock_parser().is_any_resource_running.return_value = True
        mock_get_timeout.return_value = 60
        self.sbd_inst_diskless.timeout_inst = mock.Mock(stonith_watchdog_timeout=-1)
        self.sbd_inst_diskless._restart_cluster_and_configure_sbd_ra()
        mock_warn.assert_has_calls([
            mock.call("To start sbd.service, need to restart cluster service manually on each node"),
            mock.call("Then run \"crm configure property stonith-enabled=true stonith-watchdog-timeout=-1 stonith-timeout=60\" on any node")
            ])

    @mock.patch('crmsh.sbd.SBDManager.configure_sbd_resource_and_properties')
    @mock.patch('logging.Logger.warning')
    @mock.patch('crmsh.xmlutil.CrmMonXmlParser')
    def test_restart_cluster_on_needed(self, mock_parser, mock_warn, mock_config_sbd_ra):
        mock_parser().is_any_resource_running.return_value = True
        self.sbd_inst._restart_cluster_and_configure_sbd_ra()
        mock_warn.assert_has_calls([
            mock.call("To start sbd.service, need to restart cluster service manually on each node"),
            ])

    @mock.patch('crmsh.bootstrap.invoke')
    def test_enable_sbd_service_init(self, mock_invoke):
        self.sbd_inst._context = mock.Mock(cluster_is_running=False)
        self.sbd_inst._enable_sbd_service()
        mock_invoke.assert_called_once_with("systemctl enable sbd.service")

    @mock.patch('crmsh.sbd.SBDManager._restart_cluster_and_configure_sbd_ra')
    @mock.patch('crmsh.utils.cluster_run_cmd')
    def test_enable_sbd_service_restart(self, mock_cluster_run, mock_restart):
        self.sbd_inst._context = mock.Mock(cluster_is_running=True)
        self.sbd_inst._enable_sbd_service()
        mock_cluster_run.assert_has_calls([
            mock.call("systemctl enable sbd.service"),
            ])
        mock_restart.assert_called_once_with()

    @mock.patch('crmsh.utils.package_is_installed')
    def test_configure_sbd_resource_and_properties_not_installed(self, mock_package):
        mock_package.return_value = False
        self.sbd_inst.configure_sbd_resource_and_properties()
        mock_package.assert_called_once_with("sbd")

    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.sbd.SBDTimeout.adjust_sbd_timeout_related_cluster_configuration')
    @mock.patch('crmsh.utils.set_property')
    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.xmlutil.CrmMonXmlParser')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_configure_sbd_resource_and_properties(
            self,
            mock_package, mock_enabled, mock_parser, mock_run, mock_set_property, sbd_adjust, mock_is_active,
    ):
        mock_package.return_value = True
        mock_enabled.return_value = True
        mock_parser().is_resource_configured.return_value = False
        mock_is_active.return_value = False
        self.sbd_inst._context = mock.Mock(cluster_is_running=True)
        self.sbd_inst._get_sbd_device_from_config = mock.Mock()
        self.sbd_inst._get_sbd_device_from_config.return_value = ["/dev/sda1"]

        self.sbd_inst.configure_sbd_resource_and_properties()

        mock_package.assert_called_once_with("sbd")
        mock_enabled.assert_called_once_with("sbd.service")
        mock_run.assert_called_once_with("crm configure primitive {} {}".format(sbd.SBDManager.SBD_RA_ID, sbd.SBDManager.SBD_RA))
        mock_set_property.assert_called_once_with("stonith-enabled", "true")

    @mock.patch('crmsh.utils.package_is_installed')
    def test_join_sbd_config_not_installed(self, mock_package):
        mock_package.return_value = False
        self.sbd_inst.join_sbd("alice", "node1")
        mock_package.assert_called_once_with("sbd")

    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_join_sbd_config_not_exist(self, mock_package, mock_exists, mock_invoke):
        mock_package.return_value = True
        mock_exists.return_value = False
        self.sbd_inst.join_sbd("alice", "node1")
        mock_package.assert_called_once_with("sbd")
        mock_exists.assert_called_once_with("/etc/sysconfig/sbd")
        mock_invoke.assert_called_once_with("systemctl disable sbd.service")

    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_join_sbd_config_disabled(self, mock_package, mock_exists, mock_enabled, mock_invoke):
        mock_package.return_value = True
        mock_exists.return_value = True
        mock_enabled.return_value = False

        self.sbd_inst.join_sbd("alice", "node1")

        mock_package.assert_called_once_with("sbd")
        mock_exists.assert_called_once_with("/etc/sysconfig/sbd")
        mock_invoke.assert_called_once_with("systemctl disable sbd.service")
        mock_enabled.assert_called_once_with("sbd.service", "node1")

    @mock.patch('logging.Logger.info')
    @mock.patch('crmsh.sbd.SBDManager._verify_sbd_device')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.watchdog.Watchdog')
    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_join_sbd(self, mock_package, mock_exists, mock_enabled, mock_invoke, mock_watchdog, mock_get_device, mock_verify, mock_status):
        mock_package.return_value = True
        mock_exists.return_value = True
        mock_enabled.return_value = True
        mock_get_device.return_value = ["/dev/sdb1"]
        mock_watchdog_inst = mock.Mock()
        mock_watchdog.return_value = mock_watchdog_inst
        mock_watchdog_inst.join_watchdog = mock.Mock()

        self.sbd_inst.join_sbd("alice", "node1")

        mock_package.assert_called_once_with("sbd")
        mock_exists.assert_called_once_with("/etc/sysconfig/sbd")
        mock_invoke.assert_called_once_with("systemctl enable sbd.service")
        mock_get_device.assert_called_once_with()
        mock_verify.assert_called_once_with(["/dev/sdb1"], ["node1"])
        mock_enabled.assert_called_once_with("sbd.service", "node1")
        mock_status.assert_called_once_with("Got SBD configuration")
        mock_watchdog.assert_called_once_with(remote_user="alice", peer_host="node1")
        mock_watchdog_inst.join_watchdog.assert_called_once_with()

    @mock.patch('crmsh.utils.sysconfig_set')
    @mock.patch('logging.Logger.info')
    @mock.patch('crmsh.sbd.SBDManager._warn_diskless_sbd')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.watchdog.Watchdog')
    @mock.patch('crmsh.bootstrap.invoke')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    @mock.patch('os.path.exists')
    @mock.patch('crmsh.utils.package_is_installed')
    def test_join_sbd_diskless(self, mock_package, mock_exists, mock_enabled, mock_invoke, mock_watchdog, mock_get_device, mock_warn, mock_status, mock_set):
        mock_package.return_value = True
        mock_exists.return_value = True
        mock_enabled.return_value = True
        mock_get_device.return_value = []
        mock_watchdog_inst = mock.Mock()
        mock_watchdog.return_value = mock_watchdog_inst
        mock_watchdog_inst.join_watchdog = mock.Mock()

        self.sbd_inst.join_sbd("alice", "node1")

        mock_package.assert_called_once_with("sbd")
        mock_exists.assert_called_once_with("/etc/sysconfig/sbd")
        mock_invoke.assert_called_once_with("systemctl enable sbd.service")
        mock_get_device.assert_called_once_with()
        mock_warn.assert_called_once_with("node1")
        mock_enabled.assert_called_once_with("sbd.service", "node1")
        mock_status.assert_called_once_with("Got diskless SBD configuration")
        mock_watchdog.assert_called_once_with(remote_user="alice", peer_host="node1")
        mock_watchdog_inst.join_watchdog.assert_called_once_with()

    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    def test_verify_sbd_device_classmethod_exception(self, mock_get_config):
        mock_get_config.return_value = []
        with self.assertRaises(ValueError) as err:
            sbd.SBDManager.verify_sbd_device()
        self.assertEqual("No sbd device configured", str(err.exception))
        mock_get_config.assert_called_once_with()

    @mock.patch('crmsh.sbd.SBDManager._verify_sbd_device')
    @mock.patch('crmsh.utils.list_cluster_nodes_except_me')
    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    def test_verify_sbd_device_classmethod(self, mock_get_config, mock_list_nodes, mock_verify):
        mock_get_config.return_value = ["/dev/sda1"]
        mock_list_nodes.return_value = ["node1"]
        sbd.SBDManager.verify_sbd_device()
        mock_get_config.assert_called_once_with()
        mock_verify.assert_called_once_with(["/dev/sda1"], ["node1"])

    @mock.patch('crmsh.sbd.SBDManager._get_device_uuid')
    def test_compare_device_uuid_return(self, mock_get_uuid):
        self.sbd_inst._compare_device_uuid("/dev/sdb1", None)
        mock_get_uuid.assert_not_called()

    @mock.patch('crmsh.sbd.SBDManager._get_device_uuid')
    def test_compare_device_uuid(self, mock_get_uuid):
        mock_get_uuid.side_effect = ["1234", "5678"]
        with self.assertRaises(ValueError) as err:
            self.sbd_inst._compare_device_uuid("/dev/sdb1", ["node1"])
        self.assertEqual("Device /dev/sdb1 doesn't have the same UUID with node1", str(err.exception))
        mock_get_uuid.assert_has_calls([mock.call("/dev/sdb1"), mock.call("/dev/sdb1", "node1")])

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_get_device_uuid_not_match(self, mock_run):
        mock_run.return_value = "data"
        with self.assertRaises(ValueError) as err:
            self.sbd_inst._get_device_uuid("/dev/sdb1")
        self.assertEqual("Cannot find sbd device UUID for /dev/sdb1", str(err.exception))
        mock_run.assert_called_once_with("sbd -d /dev/sdb1 dump", None)

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_get_device_uuid(self, mock_run):
        output = """
        ==Dumping header on disk /dev/sda1
        Header version     : 2.1
        UUID               : a2e9a92c-cc72-4ef9-ac55-ccc342f3546b
        Number of slots    : 255
        Sector size        : 512
        Timeout (watchdog) : 5
        Timeout (allocate) : 2
        Timeout (loop)     : 1
        Timeout (msgwait)  : 10
        ==Header on disk /dev/sda1 is dumped
        """
        mock_run.return_value = output
        res = self.sbd_inst._get_device_uuid("/dev/sda1", node="node1")
        self.assertEqual(res, "a2e9a92c-cc72-4ef9-ac55-ccc342f3546b")
        mock_run.assert_called_once_with("sbd -d /dev/sda1 dump", "node1")

    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_active')
    @mock.patch('crmsh.bootstrap.Context')
    def test_is_using_diskless_sbd_true(self, mock_context, mock_is_active, mock_get_sbd):
        context_inst = mock.Mock()
        mock_context.return_value = context_inst
        mock_get_sbd.return_value = []
        mock_is_active.return_value = True
        assert sbd.SBDManager.is_using_diskless_sbd() is True
        mock_context.assert_called_once_with()
        mock_get_sbd.assert_called_once_with()
        mock_is_active.assert_called_once_with("sbd.service")

    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.Context')
    def test_is_using_diskless_sbd_false(self, mock_context, mock_get_sbd):
        context_inst = mock.Mock()
        mock_context.return_value = context_inst
        mock_get_sbd.return_value = ["/dev/sda1"]
        assert sbd.SBDManager.is_using_diskless_sbd() is False
        mock_context.assert_called_once_with()
        mock_get_sbd.assert_called_once_with()

    @mock.patch('crmsh.sbd.SBDManager._get_sbd_device_from_config')
    @mock.patch('crmsh.bootstrap.Context')
    def test_get_sbd_device_from_config_classmethod(self, mock_context, mock_get_sbd):
        context_inst = mock.Mock()
        mock_context.return_value = context_inst
        mock_get_sbd.return_value = ["/dev/sda1"]
        assert sbd.SBDManager.get_sbd_device_from_config() == ["/dev/sda1"]
        mock_context.assert_called_once_with()
        mock_get_sbd.assert_called_once_with()

    @mock.patch('crmsh.bootstrap.sync_file')
    @mock.patch('crmsh.utils.sysconfig_set')
    def test_update_configuration_static(self, mock_config_set, mock_csync2):
        sbd_config_dict = {
                "SBD_PACEMAKER": "yes",
                "SBD_STARTMODE": "always",
                "SBD_DELAY_START": "no",
                }
        self.sbd_inst.update_configuration(sbd_config_dict)
        mock_config_set.assert_called_once_with(bootstrap.SYSCONFIG_SBD, **sbd_config_dict)
        mock_csync2.assert_called_once_with(bootstrap.SYSCONFIG_SBD)