From e082719423624f53b429c8024655c3f56b4773bd Mon Sep 17 00:00:00 2001 From: Avan Thakkar Date: Tue, 17 Sep 2024 13:09:15 +0530 Subject: [PATCH] qa/cephfs: update tests for test_volumes & unit-test for earmarking Signed-off-by: Avan Thakkar (cherry picked from commit d2f8d1022152b7c5f587d7772ce790c305de3767) --- qa/tasks/cephfs/test_volumes.py | 126 ++++++++++++++++++ .../ceph/tests/test_earmarking.py | 86 ++++++++++++ 2 files changed, 212 insertions(+) create mode 100644 src/python-common/ceph/tests/test_earmarking.py diff --git a/qa/tasks/cephfs/test_volumes.py b/qa/tasks/cephfs/test_volumes.py index 037b046304e..453bab79a0c 100644 --- a/qa/tasks/cephfs/test_volumes.py +++ b/qa/tasks/cephfs/test_volumes.py @@ -2333,6 +2333,124 @@ class TestSubvolumes(TestVolumesHelper): # 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): """ @@ -2406,6 +2524,14 @@ class TestSubvolumes(TestVolumesHelper): 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) diff --git a/src/python-common/ceph/tests/test_earmarking.py b/src/python-common/ceph/tests/test_earmarking.py new file mode 100644 index 00000000000..2add5d62064 --- /dev/null +++ b/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 + ) -- 2.39.5