path: root/test/unittests/test_ocfs2.py

import logging
import unittest
try:
    from unittest import mock
except ImportError:
    import mock
from crmsh import ocfs2, utils, ra, constants

logging.basicConfig(level=logging.INFO)

class TestOCFS2Manager(unittest.TestCase):
    """
    Unitary tests for crmsh.bootstrap.SBDManager
    """

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

    def setUp(self):
        """
        Test setUp.
        """
        context1 = mock.Mock(ocfs2_devices=[])
        self.ocfs2_inst1 = ocfs2.OCFS2Manager(context1)

        context2 = mock.Mock(ocfs2_devices=[],
                stage="ocfs2",
                yes_to_all=True)
        self.ocfs2_inst2 = ocfs2.OCFS2Manager(context2)

        context3 = mock.Mock(ocfs2_devices=["/dev/sdb2", "/dev/sdc2"],
                use_cluster_lvm2=False)
        self.ocfs2_inst3 = ocfs2.OCFS2Manager(context3)

        context4 = mock.Mock(ocfs2_devices=[],
                use_cluster_lvm2=True)
        self.ocfs2_inst4 = ocfs2.OCFS2Manager(context4)

        context5 = mock.Mock(ocfs2_devices=["/dev/sda2", "/dev/sda2"])
        self.ocfs2_inst5 = ocfs2.OCFS2Manager(context5)

        context6 = mock.Mock(ocfs2_devices=["/dev/sda2"],
                mount_point="/data")
        self.ocfs2_inst6 = ocfs2.OCFS2Manager(context6)

        context7 = mock.Mock(ocfs2_devices=["/dev/sdb2"],
                use_cluster_lvm2=True)
        self.ocfs2_inst7 = ocfs2.OCFS2Manager(context7)

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

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

    @mock.patch('crmsh.utils.package_is_installed')
    def test_verify_packages(self, mock_installed):
        mock_installed.side_effect = [True, False]
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst1._verify_packages(use_cluster_lvm2=True)
        self.assertEqual("Missing required package for configuring OCFS2: lvm2-lockd", str(err.exception))
        mock_installed.assert_has_calls([
            mock.call("ocfs2-tools"),
            mock.call("lvm2-lockd")
            ])

    def test_verify_options_stage_miss_option(self):
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst2._verify_options()
        self.assertEqual("ocfs2 stage require -o option", str(err.exception))

    def test_verify_options_two_devices(self):
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst3._verify_options()
        self.assertEqual("Without Cluster LVM2 (-C option), -o option only support one device", str(err.exception))

    def test_verify_options_only_C(self):
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst4._verify_options()
        self.assertEqual("-C option only valid together with -o option", str(err.exception))

    @mock.patch('crmsh.utils.has_mount_point_used')
    def test_verify_options_mount(self, mock_mount):
        mock_mount.return_value = True
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst6._verify_options()
        self.assertEqual("Mount point /data already mounted", str(err.exception))
        mock_mount.assert_called_once_with("/data")

    @mock.patch('crmsh.utils.is_block_device')
    def test_verify_devices_not_block(self, mock_is_block):
        mock_is_block.return_value = False
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst3._verify_devices()
        self.assertEqual("/dev/sdb2 doesn't look like a block device", str(err.exception))
        mock_is_block.assert_called_once_with("/dev/sdb2")

    @mock.patch('crmsh.utils.is_dev_used_for_lvm')
    @mock.patch('crmsh.utils.is_block_device')
    def test_verify_devices_lvm(self, mock_is_block, mock_lvm):
        mock_lvm.return_value = True
        mock_is_block.return_value = True
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst7._verify_devices()
        self.assertEqual("/dev/sdb2 is a Logical Volume, cannot be used with the -C option", str(err.exception))
        mock_is_block.assert_called_once_with("/dev/sdb2")
        mock_lvm.assert_called_once_with("/dev/sdb2")

    @mock.patch('crmsh.utils.has_disk_mounted')
    @mock.patch('crmsh.utils.is_dev_used_for_lvm')
    @mock.patch('crmsh.utils.is_block_device')
    def test_verify_devices_mounted(self, mock_is_block, mock_lvm, mock_mounted):
        mock_lvm.return_value = False
        mock_is_block.return_value = True
        mock_mounted.return_value = True
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst7._verify_devices()
        self.assertEqual("/dev/sdb2 already mounted", str(err.exception))
        mock_is_block.assert_called_once_with("/dev/sdb2")
        mock_lvm.assert_called_once_with("/dev/sdb2")
        mock_mounted.assert_called_once_with("/dev/sdb2")

    def test_check_if_already_configured_return(self):
        self.ocfs2_inst3._check_if_already_configured()

    @mock.patch('logging.Logger.info')
    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_check_if_already_configured(self, mock_run, mock_info):
        mock_run.return_value = "data xxx fstype=ocfs2  sss"
        with self.assertRaises(utils.TerminateSubCommand):
            self.ocfs2_inst2._check_if_already_configured()
        mock_run.assert_called_once_with("crm configure show")
        mock_info.assert_called_once_with("Already configured OCFS2 related resources")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._verify_devices')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._check_if_already_configured')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._verify_options')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._verify_packages')
    def test_static_verify(self, mock_verify_packages, mock_verify_options, mock_configured, mock_verify_devices):
        self.ocfs2_inst3._static_verify()
        mock_verify_packages.assert_called_once_with(False)
        mock_verify_options.assert_called_once_with()
        mock_configured.assert_called_once_with()
        mock_verify_devices.assert_called_once_with()

    def test_dynamic_raise_error(self):
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst2._dynamic_raise_error("error messages")
        self.assertEqual("error messages", str(err.exception))

    @mock.patch('crmsh.ocfs2.OCFS2Manager._dynamic_raise_error')
    @mock.patch('crmsh.sbd.SBDManager.get_sbd_device_from_config')
    @mock.patch('crmsh.service_manager.ServiceManager.service_is_enabled')
    def test_check_sbd_and_ocfs2_dev(self, mock_enabled, mock_get_device, mock_error):
        mock_enabled.return_value = True
        mock_get_device.return_value = ["/dev/sdb2"]
        self.ocfs2_inst3._check_sbd_and_ocfs2_dev()
        mock_enabled.assert_called_once_with("sbd.service")
        mock_get_device.assert_called_once_with()
        mock_error.assert_called_once_with("/dev/sdb2 cannot be the same with SBD device")

    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('crmsh.utils.get_dev_fs_type')
    @mock.patch('crmsh.utils.has_dev_partitioned')
    def test_confirm_to_overwrite_ocfs2_dev(self, mock_has_parted, mock_fstype, mock_confirm):
        mock_has_parted.side_effect = [True, False]
        mock_fstype.return_value = "ext4"
        mock_confirm.side_effect = [True, False]
        with self.assertRaises(utils.TerminateSubCommand) as err:
            self.ocfs2_inst3._confirm_to_overwrite_ocfs2_dev()
        mock_has_parted.assert_has_calls([
            mock.call("/dev/sdb2"),
            mock.call("/dev/sdc2")
            ])
        mock_fstype.assert_called_once_with("/dev/sdc2")
        mock_confirm.assert_has_calls([
            mock.call("Found a partition table in /dev/sdb2 - Proceed anyway?"),
            mock.call("/dev/sdc2 contains a ext4 file system - Proceed anyway?")
            ])

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.bootstrap.confirm')
    @mock.patch('crmsh.utils.get_dev_fs_type')
    @mock.patch('crmsh.utils.has_dev_partitioned')
    def test_confirm_to_overwrite_ocfs2_dev_confirmed(self, mock_has_parted, mock_fstype, mock_confirm, mock_run):
        mock_has_parted.side_effect = [True, False]
        mock_fstype.return_value = "ext4"
        mock_confirm.side_effect = [True, True]
        self.ocfs2_inst3._confirm_to_overwrite_ocfs2_dev()
        mock_has_parted.assert_has_calls([
            mock.call("/dev/sdb2"),
            mock.call("/dev/sdc2")
            ])
        mock_fstype.assert_called_once_with("/dev/sdc2")
        mock_confirm.assert_has_calls([
            mock.call("Found a partition table in /dev/sdb2 - Proceed anyway?"),
            mock.call("/dev/sdc2 contains a ext4 file system - Proceed anyway?")
            ])
        mock_run.assert_has_calls([
            mock.call("wipefs -a /dev/sdb2"),
            mock.call("wipefs -a /dev/sdc2")
            ])

    @mock.patch('crmsh.ocfs2.OCFS2Manager._dynamic_raise_error')
    @mock.patch('crmsh.utils.has_stonith_running')
    def test_dynamic_verify_error(self, mock_has_stonith, mock_error):
        mock_has_stonith.return_value = False
        mock_error.side_effect = SystemExit
        with self.assertRaises(SystemExit):
            self.ocfs2_inst3._dynamic_verify()
        mock_has_stonith.assert_called_once_with()
        mock_error.assert_called_once_with("OCFS2 requires stonith device configured and running")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._confirm_to_overwrite_ocfs2_dev')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._check_sbd_and_ocfs2_dev')
    @mock.patch('crmsh.utils.has_stonith_running')
    def test_dynamic_verify(self, mock_has_stonith, mock_check_dev, mock_confirm):
        mock_has_stonith.return_value = True
        self.ocfs2_inst3._dynamic_verify()
        mock_has_stonith.assert_called_once_with()
        mock_check_dev.assert_called_once_with()
        mock_confirm.assert_called_once_with()

    @mock.patch('crmsh.utils.gen_unused_id')
    def test_gen_ra_scripts(self, mock_gen_unused):
        self.ocfs2_inst3.exist_ra_id_list = []
        mock_gen_unused.return_value = "g1"
        res = self.ocfs2_inst3._gen_ra_scripts("GROUP", {"id": "g1", "ra_string": "d vip"})
        assert res == ("g1", "\ngroup g1 d vip")
        mock_gen_unused.assert_called_once_with([], "g1")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.corosync.get_value')
    @mock.patch('crmsh.log.LoggerUtils.status_long')
    def test_mkfs(self, mock_long, mock_get_value, mock_run):
        mock_get_value.return_value = "hacluster"
        self.ocfs2_inst3._mkfs("/dev/sdb2")
        mock_long.assert_called_once_with("  Creating OCFS2 filesystem for /dev/sdb2")
        mock_get_value.assert_called_once_with("totem.cluster_name")
        mock_run.assert_called_once_with("mkfs.ocfs2 --cluster-stack pcmk --cluster-name hacluster -N 8 -x /dev/sdb2")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_vg_change(self, mock_run):
        self.ocfs2_inst3.vg_id = "vg1"
        with self.ocfs2_inst3._vg_change():
            pass
        mock_run.assert_has_calls([
            mock.call("vgchange -ay vg1"),
            mock.call("vgchange -an vg1")
            ])

    @mock.patch('crmsh.utils.get_pe_number')
    @mock.patch('crmsh.utils.gen_unused_id')
    @mock.patch('crmsh.utils.get_all_vg_name')
    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    @mock.patch('crmsh.log.LoggerUtils.status_long')
    def test_create_lv(self, mock_long, mock_run, mock_all_vg, mock_unused, mock_pe_num):
        mock_all_vg.return_value = []
        mock_unused.return_value = "vg1"
        mock_pe_num.return_value = 1234
        res = self.ocfs2_inst3._create_lv()
        self.assertEqual(res, "/dev/vg1/ocfs2-lv")
        mock_run.assert_has_calls([
            mock.call("pvcreate /dev/sdb2 /dev/sdc2 -y"),
            mock.call("vgcreate --shared vg1 /dev/sdb2 /dev/sdc2 -y"),
            mock.call("lvcreate -l 1234 vg1 -n ocfs2-lv -y")
            ])

    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_ra_scripts')
    def test_gen_group_and_clone_scripts(self, mock_gen):
        mock_gen.side_effect = [("id1", "group_script\n"), ("id2", "clone_script\n")]
        res = self.ocfs2_inst3._gen_group_and_clone_scripts(["ra1", "ra2"])
        self.assertEqual(res, "group_script\nclone_script\n")
        mock_gen.assert_has_calls([
            mock.call('GROUP', {'id': 'ocfs2-group', 'ra_string': 'ra1 ra2'}),
            mock.call('CLONE', {'id': 'ocfs2-clone', 'group_id': 'id1'})
            ])

    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_ra_scripts')
    def test_gen_fs_scripts(self, mock_gen):
        mock_gen.return_value = "scripts"
        self.ocfs2_inst3.mount_point = "/data"
        self.ocfs2_inst3.target_device = "/dev/sda1"
        res = self.ocfs2_inst3._gen_fs_scripts()
        self.assertEqual(res, "scripts")
        mock_gen.assert_called_once_with("Filesystem", {'id': 'ocfs2-clusterfs', 'mnt_point': '/data', 'fs_type': 'ocfs2', 'device': '/dev/sda1'})

    @mock.patch('crmsh.bootstrap.wait_for_resource')
    @mock.patch('crmsh.utils.append_res_to_group')
    @mock.patch('crmsh.bootstrap.crm_configure_load')
    def test_load_append_and_wait(self, mock_load, mock_append, mock_wait):
        self.ocfs2_inst3.group_id = "g1"
        self.ocfs2_inst3._load_append_and_wait("scripts", "res_id", "messages data")
        mock_load.assert_called_once_with("update", "scripts")
        mock_append.assert_called_once_with("g1", "res_id")
        mock_wait.assert_called_once_with("messages data", "res_id")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._load_append_and_wait')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_group_and_clone_scripts')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_ra_scripts')
    def test_config_dlm(self, mock_gen_ra, mock_gen_group, mock_load_wait):
        mock_gen_ra.return_value = ("dlm_id", "dlm_scripts\n")
        mock_gen_group.return_value = "group_scripts\n"
        self.ocfs2_inst3._config_dlm()
        mock_gen_ra.assert_called_once_with("DLM", {"id": "ocfs2-dlm"})
        mock_gen_group.assert_called_once_with(["dlm_id"])
        mock_load_wait.assert_called_once_with("dlm_scripts\ngroup_scripts\n", "dlm_id", "  Wait for DLM(dlm_id) start", need_append=False)

    @mock.patch('crmsh.ocfs2.OCFS2Manager._load_append_and_wait')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_ra_scripts')
    def test_config_lvmlockd(self, mock_gen_ra, mock_load_wait):
        mock_gen_ra.return_value = ("ra_id", "ra_scripts\n")
        self.ocfs2_inst3._config_lvmlockd()
        mock_gen_ra.assert_called_once_with("LVMLockd", {"id": "ocfs2-lvmlockd"})
        mock_load_wait.assert_called_once_with("ra_scripts\n", "ra_id", "  Wait for LVMLockd(ra_id) start")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._load_append_and_wait')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_ra_scripts')
    def test_config_lvmactivate(self, mock_gen_ra, mock_load_wait):
        mock_gen_ra.return_value = ("ra_id", "ra_scripts\n")
        self.ocfs2_inst3.vg_id = "vg1"
        self.ocfs2_inst3._config_lvmactivate()
        mock_gen_ra.assert_called_once_with("LVMActivate", {"id": "ocfs2-lvmactivate", "vgname": "vg1"})
        mock_load_wait.assert_called_once_with("ra_scripts\n", "ra_id", "  Wait for LVMActivate(ra_id) start")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._load_append_and_wait')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._gen_fs_scripts')
    @mock.patch('crmsh.utils.mkdirp')
    def test_config_fs(self, mock_mkdir, mock_gen_fs, mock_load_wait):
        mock_gen_fs.return_value = ("ra_id", "ra_scripts\n")
        self.ocfs2_inst3.mount_point = "/data"
        self.ocfs2_inst3._config_fs()
        mock_mkdir.assert_called_once_with("/data")
        mock_gen_fs.assert_called_once_with()
        mock_load_wait.assert_called_once_with("ra_scripts\n", "ra_id", "  Wait for Filesystem(ra_id) start")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_fs')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_lvmactivate')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._mkfs')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._vg_change')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._create_lv')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_lvmlockd')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_dlm')
    def test_config_resource_stack_lvm2(self, mock_dlm, mock_lvmlockd, mock_lv, mock_vg, mock_mkfs, mock_lvmactivate, mock_fs):
        mock_lv.return_value = "/dev/sda1"
        self.ocfs2_inst3._config_resource_stack_lvm2()
        mock_dlm.assert_called_once_with()
        mock_lvmlockd.assert_called_once_with()
        mock_lv.assert_called_once_with()
        mock_mkfs.assert_called_once_with("/dev/sda1")
        mock_lvmactivate.assert_called_once_with()
        mock_fs.assert_called_once_with()

    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_fs')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._mkfs')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_dlm')
    def test_config_resource_stack_ocfs2_along(self, mock_dlm, mock_mkfs, mock_fs):
        self.ocfs2_inst3._config_resource_stack_ocfs2_along()
        mock_dlm.assert_called_once_with()
        mock_mkfs.assert_called_once_with("/dev/sdb2")
        mock_fs.assert_called_once_with()

    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_resource_stack_lvm2')
    @mock.patch('crmsh.utils.set_property')
    @mock.patch('crmsh.utils.get_property')
    @mock.patch('crmsh.utils.all_exist_id')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._dynamic_verify')
    @mock.patch('logging.Logger.info')
    def test_init_ocfs2_lvm2(self, mock_status, mock_dynamic_verify, mock_all_id, mock_get, mock_set, mock_lvm2):
        mock_all_id.return_value = []
        mock_get.return_value = None
        self.ocfs2_inst7.mount_point = "/data"
        self.ocfs2_inst7.target_device = "/dev/vg1/lv1"
        self.ocfs2_inst7.init_ocfs2()
        mock_status.assert_has_calls([
            mock.call("Configuring OCFS2"),
            mock.call('  \'no-quorum-policy\' is changed to "freeze"'),
            mock.call('  OCFS2 device %s mounted on %s', '/dev/vg1/lv1', '/data')
            ])
        mock_dynamic_verify.assert_called_once_with()
        mock_all_id.assert_called_once_with()
        mock_lvm2.assert_called_once_with()

    @mock.patch('crmsh.ocfs2.OCFS2Manager._config_resource_stack_ocfs2_along')
    @mock.patch('crmsh.utils.set_property')
    @mock.patch('crmsh.utils.get_property')
    @mock.patch('crmsh.utils.all_exist_id')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._dynamic_verify')
    @mock.patch('logging.Logger.info')
    def test_init_ocfs2(self, mock_status, mock_dynamic_verify, mock_all_id, mock_get, mock_set, mock_ocfs2):
        mock_all_id.return_value = []
        mock_get.return_value = None
        self.ocfs2_inst3.mount_point = "/data"
        self.ocfs2_inst3.target_device = "/dev/sda1"
        self.ocfs2_inst3.init_ocfs2()
        mock_status.assert_has_calls([
            mock.call("Configuring OCFS2"),
            mock.call('  \'no-quorum-policy\' is changed to "freeze"'),
            mock.call('  OCFS2 device %s mounted on %s', '/dev/sda1', '/data')
            ])
        mock_dynamic_verify.assert_called_once_with()
        mock_all_id.assert_called_once_with()
        mock_ocfs2.assert_called_once_with()

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_find_target_on_join_none(self, mock_run):
        mock_run.return_value = "data"
        res = self.ocfs2_inst3._find_target_on_join("node1")
        assert res is None
        mock_run.assert_called_once_with("crm configure show", "node1")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_find_target_on_join_exception(self, mock_run):
        mock_run.return_value = """
params directory="/srv/clusterfs" fstype=ocfs2
        """
        with self.assertRaises(ValueError) as err:
            self.ocfs2_inst3._find_target_on_join("node1")
        self.assertEqual("Filesystem require configure device", str(err.exception))
        mock_run.assert_called_once_with("crm configure show", "node1")

    @mock.patch('crmsh.sh.ClusterShell.get_stdout_or_raise_error')
    def test_find_target_on_join(self, mock_run):
        mock_run.return_value = """
params directory="/srv/clusterfs" fstype=ocfs2 device="/dev/sda2"
        """
        res = self.ocfs2_inst3._find_target_on_join("node1")
        self.assertEqual(res, "/dev/sda2")
        mock_run.assert_called_once_with("crm configure show", "node1")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._find_target_on_join')
    def test_join_ocfs2_return(self, mock_find):
        mock_find.return_value = None
        self.ocfs2_inst3.join_ocfs2("node1")
        mock_find.assert_called_once_with("node1")

    @mock.patch('crmsh.utils.compare_uuid_with_peer_dev')
    @mock.patch('crmsh.utils.is_dev_a_plain_raw_disk_or_partition')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._verify_packages')
    @mock.patch('crmsh.xmlutil.CrmMonXmlParser')
    @mock.patch('crmsh.log.LoggerUtils.status_long')
    @mock.patch('crmsh.ocfs2.OCFS2Manager._find_target_on_join')
    def test_join_ocfs2(self, mock_find, mock_long, mock_parser, mock_verify_packages, mock_is_mapper, mock_compare):
        mock_find.return_value = "/dev/sda2"
        mock_parser("node1").is_resource_configured.return_value = False
        mock_is_mapper.return_value = True
        self.ocfs2_inst3.join_ocfs2("node1")
        mock_find.assert_called_once_with("node1")
        mock_verify_packages.assert_called_once_with(False)
        mock_is_mapper.assert_called_once_with("/dev/sda2", "node1")
        mock_compare.assert_called_once_with(["/dev/sda2"], "node1")

    @mock.patch('crmsh.ocfs2.OCFS2Manager._static_verify')
    def test_verify_ocfs2(self, mock_static_verify):
        context1 = mock.Mock(ocfs2_devices=[])
        ocfs2.OCFS2Manager.verify_ocfs2(context1)
        mock_static_verify.assert_called_once_with()