Dev: unittests: Add unit test for previous commit

data-manifest

@@ -188,6 +188,7 @@ test/unittests/test_bootstrap.py
 test/unittests/test_bugs.py
 test/unittests/test_cib.py
 test/unittests/test_cliformat.py
+test/unittests/test_cluster_fs.py
 test/unittests/test.conf
 test/unittests/test_corosync_config_format.py
 test/unittests/test_corosync.py

test/unittests/test_cluster_fs.py

@@ -0,0 +1,107 @@
+import unittest
+from unittest import mock
+
+from crmsh import cluster_fs
+
+
+class TestClusterFSManager(unittest.TestCase):
+
+    @mock.patch("crmsh.cluster_fs.ClusterFSManager._verify_options")
+    def setUp(self, mock_verify_options):
+        ocfs2_context_one_device = mock.Mock(ocfs2_devices=["/dev/sda1"], gfs2_devices=[], use_cluster_lvm2=False)
+        gfs2_context_one_device_clvm2 = mock.Mock(ocfs2_devices=[], gfs2_devices=["/dev/sda1"], use_cluster_lvm2=True)
+        self.ocfs2_instance_one_device = cluster_fs.ClusterFSManager(ocfs2_context_one_device)
+        self.gfs2_instance_one_device_clvm2 = cluster_fs.ClusterFSManager(gfs2_context_one_device_clvm2)
+
+        ocfs2_gfs2_both_context = mock.Mock(ocfs2_devices=["/dev/sda1"], gfs2_devices=["/dev/sda2"])
+        self.instance_both = cluster_fs.ClusterFSManager(ocfs2_gfs2_both_context)
+
+        ocfs2_stage_without_device_context = mock.Mock(ocfs2_devices=[], gfs2_devices=[], stage="ocfs2")
+        self.instance_ocfs2_stage_without_device = cluster_fs.ClusterFSManager(ocfs2_stage_without_device_context)
+
+        gfs2_stage_without_device_context = mock.Mock(ocfs2_devices=[], gfs2_devices=[], stage="gfs2")
+        self.instance_gfs2_stage_without_device = cluster_fs.ClusterFSManager(gfs2_stage_without_device_context)
+
+        clvm2_without_device_context = mock.Mock(ocfs2_devices=[], gfs2_devices=[], use_cluster_lvm2=True)
+        self.instance_clvm2_without_device = cluster_fs.ClusterFSManager(clvm2_without_device_context)
+
+        multi_ocfs2_devices_without_clvm2_context = mock.Mock(ocfs2_devices=["/dev/sda1", "/dev/sda2"], gfs2_devices=[], use_cluster_lvm2=False)
+        self.multi_ocfs2_devices_without_clvm2 = cluster_fs.ClusterFSManager(multi_ocfs2_devices_without_clvm2_context)
+
+        multi_gfs2_devices_without_clvm2_context = mock.Mock(ocfs2_devices=[], gfs2_devices=["/dev/sda1", "/dev/sda2"], use_cluster_lvm2=False)
+        self.multi_gfs2_devices_without_clvm2 = cluster_fs.ClusterFSManager(multi_gfs2_devices_without_clvm2_context)
+
+        gfs2_context_one_device_with_mount_point = mock.Mock(ocfs2_devices=[], gfs2_devices=["/dev/sda1"], use_cluster_lvm2=False, mount_point="/mnt/gfs2")
+        self.gfs2_instance_one_device_with_mount_point = cluster_fs.ClusterFSManager(gfs2_context_one_device_with_mount_point)
+
+    @mock.patch("crmsh.utils.package_is_installed")
+    def test_verify_packages_gfs2(self, mock_package_is_installed):
+        mock_package_is_installed.side_effect = [True, False]
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.gfs2_instance_one_device_clvm2._verify_packages()
+        self.assertIn("Missing required package for configuring GFS2: lvm2-lockd", str(context.exception))
+        mock_package_is_installed.assert_has_calls([mock.call("gfs2-utils"), mock.call("lvm2-lockd")])
+
+    @mock.patch("crmsh.utils.package_is_installed")
+    def test_verify_packages_ocfs2(self, mock_package_is_installed):
+        mock_package_is_installed.return_value = True
+        self.ocfs2_instance_one_device._verify_packages()
+        mock_package_is_installed.assert_called_once_with("ocfs2-tools")
+
+    def test_verify_options(self):
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.instance_both._verify_options()
+        self.assertIn("Can't use -g and -o options together", str(context.exception))
+
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.instance_ocfs2_stage_without_device._verify_options()
+        self.assertIn("ocfs2 stage require -o option", str(context.exception))
+
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.instance_gfs2_stage_without_device._verify_options()
+        self.assertIn("gfs2 stage require -g option", str(context.exception))
+
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.instance_clvm2_without_device._verify_options()
+        self.assertIn("-C option only valid together with -o or -g option", str(context.exception))
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.multi_ocfs2_devices_without_clvm2._verify_options()
+        self.assertIn("Without Cluster LVM2 (-C option), -o option only support one device", str(context.exception))
+
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.multi_gfs2_devices_without_clvm2._verify_options()
+        self.assertIn("Without Cluster LVM2 (-C option), -g option only support one device", str(context.exception))
+
+    @mock.patch("crmsh.utils.has_mount_point_used")
+    def test_verify_options_mount_point(self, mock_has_mount_point_used):
+        mock_has_mount_point_used.return_value = True
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.gfs2_instance_one_device_with_mount_point._verify_options()
+        self.assertIn("Mount point /mnt/gfs2 already mounted", str(context.exception))
+
+    @mock.patch("crmsh.utils.is_block_device")
+    def test_verify_devices_not_block_device(self, mock_is_block_device):
+        mock_is_block_device.return_value = False
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.ocfs2_instance_one_device._verify_devices()
+        self.assertIn("/dev/sda1 doesn't look like a block device", str(context.exception))
+
+    @mock.patch("crmsh.utils.is_dev_used_for_lvm")
+    @mock.patch("crmsh.utils.is_block_device")
+    def test_verify_devices_clvm2_with_lv(self, mock_is_block_device, mock_is_dev_used_for_lvm):
+        mock_is_block_device.return_value = True
+        mock_is_dev_used_for_lvm.return_value = True
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.gfs2_instance_one_device_clvm2._verify_devices()
+        self.assertIn("/dev/sda1 is a Logical Volume, cannot be used with the -C option", str(context.exception))
+
+    @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_already_mounted(self, mock_is_block_device, mock_is_dev_used_for_lvm, mock_has_disk_mounted):
+        mock_is_block_device.return_value = True
+        mock_is_dev_used_for_lvm.return_value = False
+        mock_has_disk_mounted.return_value = True
+        with self.assertRaises(cluster_fs.Error) as context:
+            self.ocfs2_instance_one_device._verify_devices()
+        self.assertIn("/dev/sda1 is already mounted", str(context.exception))