qa/cephfs: update tests for test_volumes & unit-test for earmarking

Signed-off-by: Avan Thakkar <[email protected]>

Author: avanthakkar

qa/tasks/cephfs/test_volumes.py

@@ -2367,6 +2367,124 @@ def test_subvolume_create_and_ls_providing_group_as_nogroup(self):
 
         # verify trash dir is clean.
         self._wait_for_trash_empty()
+    
+    def test_subvolume_create_with_earmark(self):
+        # create subvolume with earmark
+        subvolume = self._gen_subvol_name()
+        earmark = "nfs.test"
+        self._fs_cmd("subvolume", "create", self.volname, subvolume, "--earmark", earmark)
+
+        # make sure it exists
+        subvolpath = self._get_subvolume_path(self.volname, subvolume)
+        self.assertNotEqual(subvolpath, None)
+
+        # verify the earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark.rstrip('\n'), earmark)
+    
+    def test_subvolume_set_and_get_earmark(self):
+        # create subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        # get earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark.rstrip('\n'), earmark)
+    
+    def test_subvolume_clear_earmark(self):
+        # create subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        # remove earmark
+        self._fs_cmd("subvolume", "earmark", "rm", self.volname, subvolume)
+
+        # get earmark
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark, "")
+
+    def test_earmark_on_non_existing_subvolume(self):
+        subvolume = "non_existing_subvol"
+        earmark = "nfs.test"
+        commands = [
+            ("set", earmark),
+            ("get", None),
+            ("rm", None),
+        ]
+
+        for action, arg in commands:
+            try:
+                # Build the command arguments
+                cmd_args = ["subvolume", "earmark", action, self.volname, subvolume]
+                if arg is not None:
+                    cmd_args.extend(["--earmark", arg])
+
+                # Execute the command with built arguments
+                self._fs_cmd(*cmd_args)
+            except CommandFailedError as ce:
+                self.assertEqual(ce.exitstatus, errno.ENOENT)
+
+    def test_get_remove_earmark_when_not_set(self):
+        # Create a subvolume without setting an earmark
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # Attempt to get an earmark when it's not set
+        get_earmark = self._fs_cmd("subvolume", "earmark", "get", self.volname, subvolume)
+        self.assertEqual(get_earmark, "")
+
+        # Attempt to remove an earmark when it's not set
+        self._fs_cmd("subvolume", "earmark", "rm", self.volname, subvolume)
+    
+    def test_set_invalid_earmark(self):
+        # Create a subvolume
+        subvolume = self._gen_subvol_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # Attempt to set an invalid earmark
+        invalid_earmark = "invalid_format"
+        expected_message = (
+            f"Invalid earmark specified: '{invalid_earmark}'. A valid earmark should "
+            "either be empty or start with 'nfs' or 'smb', followed by dot-separated "
+            "non-empty components."
+        )
+        try:
+            self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", invalid_earmark)
+        except CommandFailedError as ce:
+            self.assertEqual(ce.exitstatus, errno.EINVAL, expected_message)
+
+    def test_earmark_on_deleted_subvolume_with_retained_snapshot(self):
+        subvolume = self._gen_subvol_name()
+        snapshot = self._gen_subvol_snap_name()
+
+        # Create subvolume and snapshot
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+        self._fs_cmd("subvolume", "snapshot", "create", self.volname, subvolume, snapshot)
+
+        # Delete subvolume while retaining the snapshot
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--retain-snapshots")
+
+        # Define the expected error message
+        error_message = f'subvolume "{subvolume}" is removed and has only snapshots retained'
+
+        # Test cases for setting, getting, and removing earmarks
+        for operation in ["get", "rm", "set"]:
+            try:
+                extra_arg = "smb" if operation == "set" else None
+                if operation == "set":
+                    self._fs_cmd("subvolume", "earmark", operation, self.volname, subvolume, "--earmark", extra_arg)
+                else:
+                    self._fs_cmd("subvolume", "earmark", operation, self.volname, subvolume)
+            except CommandFailedError as ce:
+                self.assertEqual(ce.exitstatus, errno.ENOENT, error_message)
 
     def test_subvolume_expand(self):
         """
@@ -2440,6 +2558,14 @@ def test_subvolume_info(self):
         for feature in ['snapshot-clone', 'snapshot-autoprotect', 'snapshot-retention']:
             self.assertIn(feature, subvol_info["features"], msg="expected feature '{0}' in subvolume".format(feature))
 
+        # set earmark
+        earmark = "smb.test"
+        self._fs_cmd("subvolume", "earmark", "set", self.volname, subvolume, "--earmark", earmark)
+
+        subvol_info = json.loads(self._get_subvolume_info(self.volname, subvolume))
+
+        self.assertEqual(subvol_info["earmark"], earmark)
+        
         # remove subvolumes
         self._fs_cmd("subvolume", "rm", self.volname, subvolume)
 

src/python-common/ceph/tests/test_earmarking.py

@@ -0,0 +1,86 @@
+import pytest
+import errno
+from unittest import mock
+
+from ceph.fs.earmarking import CephFSVolumeEarmarking, EarmarkException, EarmarkTopScope
+# Mock constants
+XATTR_SUBVOLUME_EARMARK_NAME = 'user.ceph.subvolume.earmark'
+
+
+class TestCephFSVolumeEarmarking:
+
+    @pytest.fixture
+    def mock_fs(self):
+        return mock.Mock()
+
+    @pytest.fixture
+    def earmarking(self, mock_fs):
+        return CephFSVolumeEarmarking(mock_fs, "/test/path")
+
+    def test_get_earmark_success(self, earmarking, mock_fs):
+        mock_fs.getxattr.return_value = b"nfs"
+        result = earmarking.get_earmark()
+        assert result == "nfs"
+        mock_fs.getxattr.assert_called_once_with("/test/path", XATTR_SUBVOLUME_EARMARK_NAME)
+
+    def test_get_earmark_no_earmark_set(self, earmarking, mock_fs):
+        mock_fs.getxattr.return_value = b""
+        result = earmarking.get_earmark()
+
+        assert result == ""
+        mock_fs.getxattr.assert_called_once_with("/test/path", XATTR_SUBVOLUME_EARMARK_NAME)
+
+    def test_get_earmark_error(self, earmarking, mock_fs):
+        mock_fs.getxattr.side_effect = OSError(errno.EIO, "I/O error")
+
+        with pytest.raises(EarmarkException) as excinfo:
+            earmarking.get_earmark()
+
+        assert excinfo.value.errno == -errno.EIO
+        assert "I/O error" in str(excinfo.value)
+
+        # Ensure that the getxattr method was called exactly once
+        mock_fs.getxattr.assert_called_once_with("/test/path", XATTR_SUBVOLUME_EARMARK_NAME)
+
+    def test_set_earmark_success(self, earmarking, mock_fs):
+        earmarking.set_earmark(EarmarkTopScope.NFS.value)
+        mock_fs.setxattr.assert_called_once_with(
+            "/test/path", XATTR_SUBVOLUME_EARMARK_NAME, b"nfs", 0
+        )
+
+    def test_set_earmark_invalid(self, earmarking):
+        with pytest.raises(EarmarkException) as excinfo:
+            earmarking.set_earmark("invalid_scope")
+
+        assert excinfo.value.errno == errno.EINVAL
+        assert "Invalid earmark specified" in str(excinfo.value)
+
+    def test_set_earmark_error(self, earmarking, mock_fs):
+        mock_fs.setxattr.side_effect = OSError(errno.EIO, "I/O error")
+
+        with pytest.raises(EarmarkException) as excinfo:
+            earmarking.set_earmark(EarmarkTopScope.NFS.value)
+
+        assert excinfo.value.errno == -errno.EIO
+        assert "I/O error" in str(excinfo.value)
+        mock_fs.setxattr.assert_called_once_with(
+            "/test/path", XATTR_SUBVOLUME_EARMARK_NAME, b"nfs", 0
+        )
+
+    def test_clear_earmark_success(self, earmarking, mock_fs):
+        earmarking.clear_earmark()
+        mock_fs.setxattr.assert_called_once_with(
+            "/test/path", XATTR_SUBVOLUME_EARMARK_NAME, b"", 0
+        )
+
+    def test_clear_earmark_error(self, earmarking, mock_fs):
+        mock_fs.setxattr.side_effect = OSError(errno.EIO, "I/O error")
+
+        with pytest.raises(EarmarkException) as excinfo:
+            earmarking.clear_earmark()
+
+        assert excinfo.value.errno == -errno.EIO
+        assert "I/O error" in str(excinfo.value)
+        mock_fs.setxattr.assert_called_once_with(
+            "/test/path", XATTR_SUBVOLUME_EARMARK_NAME, b"", 0
+        )