mgr/volumes: Tests amended and added to ensure subvolume trash functionality

Amended a few test cases to ensure created subvolumes and snaps
are removed, and trash stays empty at the end of the test.

Further added one test case for create errors in a retained
v2 subvolume, to ensure metadata is sane, and created incarnation
is not present.

Fixes: https://tracker.ceph.com/issues/47154
Signed-off-by: Shyamsundar Ranganathan <srangana@redhat.com>
(cherry picked from commit 88cffe6ea97733f2f0169362ffbde3e3827dc861)

diff --git a/qa/tasks/cephfs/test_volumes.py b/qa/tasks/cephfs/test_volumes.py
index 42c430c978f48c08e95e4977e0793b1c8d42bd20..42d0c9a9a68cd67b0a9b4d61ac419619dec61727 100644 (file)
--- a/qa/tasks/cephfs/test_volumes.py
+++ b/qa/tasks/cephfs/test_volumes.py
@@ -534,6 +534,12 @@ class TestVolumes(CephFSTestCase):
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, nsize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_shrink(self):
         """
         That a subvolume can be shrinked in size and its quota matches the expected size.
@@ -556,6 +562,12 @@ class TestVolumes(CephFSTestCase):
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, nsize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_resize_fail_invalid_size(self):
         """
         That a subvolume cannot be resized to an invalid size and the quota did not change
@@ -575,15 +587,20 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "resize", self.volname, subvolname, str(nsize))
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.EINVAL:
-                raise
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on resize of subvolume with invalid size")
         else:
-            raise RuntimeError("expected the 'fs subvolume resize' command to fail")
+            self.fail("expected the 'fs subvolume resize' command to fail")
 
         # verify the quota did not change
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, osize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_resize_fail_zero_size(self):
         """
         That a subvolume cannot be resized to a zero size and the quota did not change
@@ -603,15 +620,20 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "resize", self.volname, subvolname, str(nsize))
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.EINVAL:
-                raise
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on resize of subvolume with invalid size")
         else:
-            raise RuntimeError("expected the 'fs subvolume resize' command to fail")
+            self.fail("expected the 'fs subvolume resize' command to fail")
 
         # verify the quota did not change
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, osize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_resize_quota_lt_used_size(self):
         """
         That a subvolume can be resized to a size smaller than the current used size
@@ -645,12 +667,18 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "resize", self.volname, subvolname, str(nsize))
         except CommandFailedError:
-            raise RuntimeError("expected the 'fs subvolume resize' command to succeed")
+            self.fail("expected the 'fs subvolume resize' command to succeed")
 
         # verify the quota
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, nsize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
 
     def test_subvolume_resize_fail_quota_lt_used_size_no_shrink(self):
         """
@@ -685,15 +713,20 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "resize", self.volname, subvolname, str(nsize), "--no_shrink")
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.EINVAL:
-                raise
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on resize of subvolume with invalid size")
         else:
-            raise RuntimeError("expected the 'fs subvolume resize' command to fail")
+            self.fail("expected the 'fs subvolume resize' command to fail")
 
         # verify the quota did not change
         size = int(self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes"))
         self.assertEqual(size, osize)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_resize_expand_on_full_subvolume(self):
         """
         That the subvolume can be expanded from a full subvolume and future writes succeed.
@@ -731,12 +764,18 @@ class TestVolumes(CephFSTestCase):
             try:
                 self.mount_a.write_n_mb(os.path.join(subvolpath, filename), file_size)
             except CommandFailedError:
-                raise RuntimeError("expected filling subvolume {0} with {1} file of size {2}MB"
+                self.fail("expected filling subvolume {0} with {1} file of size {2}MB"
                                    "to succeed".format(subvolname, number_of_files, file_size))
         else:
-            raise RuntimeError("expected filling subvolume {0} with {1} file of size {2}MB"
+            self.fail("expected filling subvolume {0} with {1} file of size {2}MB"
                                "to fail".format(subvolname, number_of_files, file_size))
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_create_idempotence(self):
         # create subvolume
         subvolume = self._generate_random_subvolume_name()
@@ -782,6 +821,12 @@ class TestVolumes(CephFSTestCase):
         self._get_subtrees(status=status, rank=1)
         self._wait_subtrees([(path, 1)], status=status)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolumegroup_pin_distributed(self):
         self.fs.set_max_mds(2)
         status = self.fs.wait_for_daemons()
@@ -797,6 +842,13 @@ class TestVolumes(CephFSTestCase):
             self._fs_cmd("subvolume", "create", self.volname, subvolume, "--group_name", group)
         self._wait_distributed_subtrees(10, status=status)
 
+        # remove subvolumes
+        for subvolume in subvolumes:
+            self._fs_cmd("subvolume", "rm", self.volname, subvolume, group)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_pin_random(self):
         self.fs.set_max_mds(2)
         self.fs.wait_for_daemons()
@@ -807,6 +859,12 @@ class TestVolumes(CephFSTestCase):
         self._fs_cmd("subvolume", "pin", self.volname, subvolume, "random", ".01")
         # no verification
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_create_isolated_namespace(self):
         """
         Create subvolume in separate rados namespace
@@ -834,10 +892,12 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "create", self.volname, subvolume, "--pool_layout", data_pool)
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.EINVAL:
-                raise
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on create of subvolume with invalid pool layout")
         else:
-            raise RuntimeError("expected the 'fs subvolume create' command to fail")
+            self.fail("expected the 'fs subvolume create' command to fail")
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
 
     def test_subvolume_rm_force(self):
         # test removing non-existing subvolume with --force
@@ -845,7 +905,7 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--force")
         except CommandFailedError:
-            raise RuntimeError("expected the 'fs subvolume rm --force' command to succeed")
+            self.fail("expected the 'fs subvolume rm --force' command to succeed")
 
     def test_subvolume_create_with_auto_cleanup_on_fail(self):
         subvolume = self._generate_random_subvolume_name()
@@ -858,10 +918,12 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "getpath", self.volname, subvolume)
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.ENOENT:
-                raise
+            self.assertEqual(ce.exitstatus, errno.ENOENT, "invalid error code on getpath of non-existent subvolume")
         else:
-            raise RuntimeError("expected the 'fs subvolume getpath' command to fail")
+            self.fail("expected the 'fs subvolume getpath' command to fail")
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
 
     def test_subvolume_create_with_invalid_size(self):
         # create subvolume with an invalid size -1
@@ -869,10 +931,12 @@ class TestVolumes(CephFSTestCase):
         try:
             self._fs_cmd("subvolume", "create", self.volname, subvolume, "--size", "-1")
         except CommandFailedError as ce:
-            if ce.exitstatus != errno.EINVAL:
-                raise
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on create of subvolume with invalid size")
         else:
-            raise RuntimeError("expected the 'fs subvolume create' command to fail")
+            self.fail("expected the 'fs subvolume create' command to fail")
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
 
     def test_nonexistent_subvolume_rm(self):
         # remove non-existing subvolume
@@ -917,6 +981,9 @@ class TestVolumes(CephFSTestCase):
         # remove subvolume
         self._fs_cmd("subvolume", "rm", self.volname, subvolume)
 
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_ls(self):
         # tests the 'fs subvolume ls' command
 
@@ -930,11 +997,18 @@ class TestVolumes(CephFSTestCase):
         # list subvolumes
         subvolumels = json.loads(self._fs_cmd('subvolume', 'ls', self.volname))
         if len(subvolumels) == 0:
-            raise RuntimeError("Expected the 'fs subvolume ls' command to list the created subvolumes.")
+            self.fail("Expected the 'fs subvolume ls' command to list the created subvolumes.")
         else:
             subvolnames = [subvolume['name'] for subvolume in subvolumels]
             if collections.Counter(subvolnames) != collections.Counter(subvolumes):
-                raise RuntimeError("Error creating or listing subvolumes")
+                self.fail("Error creating or listing subvolumes")
+
+        # remove subvolume
+        for subvolume in subvolumes:
+            self._fs_cmd("subvolume", "rm", self.volname, subvolume)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
 
     def test_subvolume_ls_for_notexistent_default_group(self):
         # tests the 'fs subvolume ls' command when the default group '_nogroup' doesn't exist
@@ -967,6 +1041,12 @@ class TestVolumes(CephFSTestCase):
         size = self.mount_a.getfattr(subvolpath, "ceph.quota.max_bytes")
         self.assertEqual(size, None)
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_resize_infinite_size_future_writes(self):
         """
         That a subvolume can be resized to an infinite size and the future writes succeed.
@@ -999,9 +1079,15 @@ class TestVolumes(CephFSTestCase):
         try:
             self.mount_a.write_n_mb(os.path.join(subvolpath, filename), file_size)
         except CommandFailedError:
-            raise RuntimeError("expected filling subvolume {0} with {1} file of size {2}MB "
+            self.fail("expected filling subvolume {0} with {1} file of size {2}MB "
                                "to succeed".format(subvolname, number_of_files, file_size))
 
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolname)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_info(self):
         # tests the 'fs subvolume info' command
 
@@ -1211,6 +1297,9 @@ class TestVolumes(CephFSTestCase):
         self._fs_cmd("subvolume", "rm", self.volname, subvol1, group)
         self._fs_cmd("subvolumegroup", "rm", self.volname, group)
 
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_group_create_with_desired_mode(self):
         group1, group2 = self._generate_random_group_name(2)
         # default mode
@@ -1294,6 +1383,9 @@ class TestVolumes(CephFSTestCase):
         self._fs_cmd("subvolume", "rm", self.volname, subvol3, group)
         self._fs_cmd("subvolumegroup", "rm", self.volname, group)
 
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_create_with_desired_uid_gid(self):
         """
         That the subvolume can be created with the desired uid and gid and its uid and gid matches the
@@ -1319,6 +1411,9 @@ class TestVolumes(CephFSTestCase):
         # remove subvolume
         self._fs_cmd("subvolume", "rm", self.volname, subvolname)
 
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_nonexistent_subvolume_group_rm(self):
         group = "non_existent_group"
 
@@ -1541,11 +1636,21 @@ class TestVolumes(CephFSTestCase):
 
         subvolsnapshotls = json.loads(self._fs_cmd('subvolume', 'snapshot', 'ls', self.volname, subvolume))
         if len(subvolsnapshotls) == 0:
-            raise RuntimeError("Expected the 'fs subvolume snapshot ls' command to list the created subvolume snapshots")
+            self.fail("Expected the 'fs subvolume snapshot ls' command to list the created subvolume snapshots")
         else:
             snapshotnames = [snapshot['name'] for snapshot in subvolsnapshotls]
             if collections.Counter(snapshotnames) != collections.Counter(snapshots):
-                raise RuntimeError("Error creating or listing subvolume snapshots")
+                self.fail("Error creating or listing subvolume snapshots")
+
+        # remove snapshot
+        for snapshot in snapshots:
+            self._fs_cmd("subvolume", "snapshot", "rm", self.volname, subvolume, snapshot)
+
+        # remove subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
 
     def test_subvolume_group_snapshot_create_and_rm(self):
         subvolume = self._generate_random_subvolume_name()
@@ -2099,6 +2204,50 @@ class TestVolumes(CephFSTestCase):
         # verify trash dir is clean
         self._wait_for_trash_empty()
 
+    def test_subvolume_retain_snapshot_invalid_recreate(self):
+        """
+        ensure retained subvolume recreate does not leave any incarnations in the subvolume and trash
+        """
+        subvolume = self._generate_random_subvolume_name()
+        snapshot = self._generate_random_snapshot_name()
+
+        # create subvolume
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+
+        # snapshot subvolume
+        self._fs_cmd("subvolume", "snapshot", "create", self.volname, subvolume, snapshot)
+
+        # remove with snapshot retention
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--retain-snapshots")
+
+        # recreate subvolume with an invalid pool
+        data_pool = "invalid_pool"
+        try:
+            self._fs_cmd("subvolume", "create", self.volname, subvolume, "--pool_layout", data_pool)
+        except CommandFailedError as ce:
+            self.assertEqual(ce.exitstatus, errno.EINVAL, "invalid error code on recreate of subvolume with invalid poolname")
+        else:
+            self.fail("expected recreate of subvolume with invalid poolname to fail")
+
+        # fetch info
+        subvol_info = json.loads(self._fs_cmd("subvolume", "info", self.volname, subvolume))
+        self.assertEqual(subvol_info["state"], "snapshot-retained",
+                         msg="expected state to be 'snapshot-retained', found '{0}".format(subvol_info["state"]))
+
+        # getpath
+        try:
+            self._fs_cmd("subvolume", "getpath", self.volname, subvolume)
+        except CommandFailedError as ce:
+            self.assertEqual(ce.exitstatus, errno.ENOENT, "invalid error code on getpath of subvolume with retained snapshots")
+        else:
+            self.fail("expected getpath of subvolume with retained snapshots to fail")
+
+        # remove snapshot (should remove volume)
+        self._fs_cmd("subvolume", "snapshot", "rm", self.volname, subvolume, snapshot)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
     def test_subvolume_retain_snapshot_recreate_subvolume(self):
         """
         ensure a retained subvolume can be recreated and further snapshotted