Merge PR ceph#53651 into main

* refs/pull/53651/head:
	qa: add test_subvolume_group_rm_when_its_not_empty
	mgr/volumes: fix `subvolume group rm` error message

Reviewed-by: Kotresh Hiremath Ravishankar <[email protected]>
Reviewed-by: Venky Shankar <[email protected]>

Author: vshankar

qa/tasks/cephfs/test_volumes.py

@@ -1871,6 +1871,32 @@ def test_subvolume_group_exists_without_subvolumegroup_and_with_subvolume(self):
         ret = self._fs_cmd("subvolumegroup", "exist", self.volname)
         self.assertEqual(ret.strip('\n'), "no subvolumegroup exists")
 
+    def test_subvolume_group_rm_when_its_not_empty(self):
+        group = self._generate_random_group_name()
+        subvolume = self._generate_random_subvolume_name()
+
+        # create subvolumegroup
+        self._fs_cmd("subvolumegroup", "create", self.volname, group)
+        # create subvolume in group
+        self._fs_cmd("subvolume", "create", self.volname, subvolume, "--group_name", group)
+        # try, remove subvolume group
+        try:
+            self._fs_cmd("subvolumegroup", "rm", self.volname, group)
+        except CommandFailedError as ce:
+            self.assertEqual(ce.exitstatus, errno.ENOTEMPTY, "invalid error code on deleting "
+                             "subvolumegroup when it is not empty")
+        else:
+            self.fail("expected the 'fs subvolumegroup rm' command to fail")
+        
+        # delete subvolume
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--group_name", group)
+
+        # delete subvolumegroup
+        self._fs_cmd("subvolumegroup", "rm", self.volname, group)
+
+        # verify trash dir is clean
+        self._wait_for_trash_empty()
+
 
 class TestSubvolumes(TestVolumesHelper):
     """Tests for FS subvolume operations, except snapshot and snapshot clone."""

src/pybind/mgr/volumes/fs/operations/group.py

@@ -269,6 +269,9 @@ def remove_group(fs, vol_spec, groupname):
     except cephfs.Error as e:
         if e.args[0] == errno.ENOENT:
             raise VolumeException(-errno.ENOENT, "subvolume group '{0}' does not exist".format(groupname))
+        elif e.args[0] == errno.ENOTEMPTY:
+            raise VolumeException(-errno.ENOTEMPTY, f"subvolume group {groupname} contains subvolume(s) "
+                                  "or retained snapshots of deleted subvolume(s)")
         raise VolumeException(-e.args[0], e.args[1])