From: Rishabh Dave Date: Mon, 5 May 2025 16:38:19 +0000 (+0530) Subject: qa/cephfs: move tests for "snapshot getpath" cmd to a separate class X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=c1761436d1a48ba75562fa7619ba3b498d4a1d1c;p=ceph.git qa/cephfs: move tests for "snapshot getpath" cmd to a separate class Signed-off-by: Rishabh Dave --- diff --git a/qa/suites/fs/volumes/tasks/volumes/test/snapshot.yaml b/qa/suites/fs/volumes/tasks/volumes/test/snapshot.yaml index d682011370aaa..f27c6d2781647 100644 --- a/qa/suites/fs/volumes/tasks/volumes/test/snapshot.yaml +++ b/qa/suites/fs/volumes/tasks/volumes/test/snapshot.yaml @@ -4,3 +4,4 @@ tasks: modules: - tasks.cephfs.test_volumes.TestSubvolumeGroupSnapshots - tasks.cephfs.test_volumes.TestSubvolumeSnapshots + - tasks.cephfs.test_volumes.TestSubvolumeSnapshotGetpath diff --git a/qa/tasks/cephfs/test_volumes.py b/qa/tasks/cephfs/test_volumes.py index b8622540baf6a..6fb829b913b16 100644 --- a/qa/tasks/cephfs/test_volumes.py +++ b/qa/tasks/cephfs/test_volumes.py @@ -5214,123 +5214,6 @@ class TestSubvolumeSnapshots(TestVolumesHelper): # verify trash dir is clean self._wait_for_trash_empty() - def get_subvol_uuid(self, subvol_name, group_name=None): - ''' - Return the UUID directory component obtained from the path of - subvolume. - ''' - if group_name: - cmd = (f'fs subvolume getpath {self.volname} {subvol_name} ' - f'{group_name}') - else: - cmd = f'fs subvolume getpath {self.volname} {subvol_name}' - - subvol_path = self.get_ceph_cmd_stdout(cmd).strip() - - subvol_uuid = os.path.basename(subvol_path) - return subvol_uuid - - def test_snapshot_getpath(self): - ''' - Test that "ceph fs subvolume snapshot getpath" command returns path to - the specified snapshot in the specified subvolume. - ''' - subvol_name = self._gen_subvol_name() - snap_name = self._gen_subvol_snap_name() - - self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name}') - sv_uuid = self.get_subvol_uuid(subvol_name) - self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' - f'{subvol_name} {snap_name}') - - snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' - f'{self.volname} {subvol_name} ' - f'{snap_name}').strip() - # expected snapshot path - exp_snap_path = os.path.join('/volumes', '_nogroup', subvol_name, - '.snap', snap_name, sv_uuid) - self.assertEqual(snap_path, exp_snap_path) - - def test_snapshot_getpath_in_group(self): - ''' - Test that "ceph fs subvolume snapshot getpath" command returns path to - the specified snapshot in the specified subvolume in the specified - group. - ''' - subvol_name = self._gen_subvol_name() - group_name = self._gen_subvol_grp_name() - snap_name = self._gen_subvol_snap_name() - - self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} {group_name}') - self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} ' - f'{group_name}') - sv_uuid = self.get_subvol_uuid(subvol_name, group_name) - self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' - f'{subvol_name} {snap_name} {group_name}') - - snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' - f'{self.volname} {subvol_name} ' - f'{snap_name} {group_name}')\ - .strip() - # expected snapshot path - exp_snap_path = os.path.join('/volumes', group_name, subvol_name, - '.snap', snap_name, sv_uuid) - self.assertEqual(snap_path, exp_snap_path) - - def test_snapshot_getpath_on_retained_subvol(self): - ''' - Test that "ceph fs subvolume snapshot getpath" command returns path to - the specified snapshot in the specified subvolume that was deleted but - snapshots on which is retained. - ''' - subvol_name = self._gen_subvol_name() - snap_name = self._gen_subvol_snap_name() - - self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name}') - sv_uuid = self.get_subvol_uuid(subvol_name) - self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' - f'{subvol_name} {snap_name}') - self.run_ceph_cmd(f'fs subvolume rm {self.volname} {subvol_name} ' - '--retain-snapshots') - - snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' - f'{self.volname} {subvol_name} ' - f'{snap_name}').strip() - - # expected snapshot path - exp_snap_path = os.path.join('/volumes', '_nogroup', subvol_name, - '.snap', snap_name, sv_uuid) - self.assertEqual(snap_path, exp_snap_path) - - def test_snapshot_getpath_on_retained_subvol_in_group(self): - ''' - Test that "ceph fs subvolume snapshot getpath" command returns path to - the specified snapshot in the specified subvolume that was deleted but - snapshots on which is retained. And the deleted subvolume is located on - a non-default group. - ''' - subvol_name = self._gen_subvol_name() - group_name = self._gen_subvol_grp_name() - snap_name = self._gen_subvol_snap_name() - - self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} {group_name}') - self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} ' - f'{group_name}') - sv_uuid = self.get_subvol_uuid(subvol_name, group_name) - self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' - f'{subvol_name} {snap_name} {group_name}') - self.run_ceph_cmd(f'fs subvolume rm {self.volname} {subvol_name} ' - f'{group_name} --retain-snapshots') - - snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' - f'{self.volname} {subvol_name} ' - f'{snap_name} {group_name}')\ - .strip() - # expected snapshot path - exp_snap_path = os.path.join('/volumes', group_name, subvol_name, - '.snap', snap_name, sv_uuid) - self.assertEqual(snap_path, exp_snap_path) - def test_subvolume_snapshot_info(self): """ @@ -6603,6 +6486,126 @@ class TestSubvolumeSnapshots(TestVolumesHelper): self.mount_a.run_shell(['sudo', 'rm', '-f', tmp_meta_path], omit_sudo=False) +class TestSubvolumeSnapshotGetpath(TestVolumesHelper): + + def get_subvol_uuid(self, subvol_name, group_name=None): + ''' + Return the UUID directory component obtained from the path of + subvolume. + ''' + if group_name: + cmd = (f'fs subvolume getpath {self.volname} {subvol_name} ' + f'{group_name}') + else: + cmd = f'fs subvolume getpath {self.volname} {subvol_name}' + + subvol_path = self.get_ceph_cmd_stdout(cmd).strip() + + subvol_uuid = os.path.basename(subvol_path) + return subvol_uuid + + def test_snapshot_getpath(self): + ''' + Test that "ceph fs subvolume snapshot getpath" command returns path to + the specified snapshot in the specified subvolume. + ''' + subvol_name = self._gen_subvol_name() + snap_name = self._gen_subvol_snap_name() + + self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name}') + sv_uuid = self.get_subvol_uuid(subvol_name) + self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' + f'{subvol_name} {snap_name}') + + snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' + f'{self.volname} {subvol_name} ' + f'{snap_name}').strip() + # expected snapshot path + exp_snap_path = os.path.join('/volumes', '_nogroup', subvol_name, + '.snap', snap_name, sv_uuid) + self.assertEqual(snap_path, exp_snap_path) + + def test_snapshot_getpath_in_group(self): + ''' + Test that "ceph fs subvolume snapshot getpath" command returns path to + the specified snapshot in the specified subvolume in the specified + group. + ''' + subvol_name = self._gen_subvol_name() + group_name = self._gen_subvol_grp_name() + snap_name = self._gen_subvol_snap_name() + + self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} {group_name}') + self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} ' + f'{group_name}') + sv_uuid = self.get_subvol_uuid(subvol_name, group_name) + self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' + f'{subvol_name} {snap_name} {group_name}') + + snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' + f'{self.volname} {subvol_name} ' + f'{snap_name} {group_name}')\ + .strip() + # expected snapshot path + exp_snap_path = os.path.join('/volumes', group_name, subvol_name, + '.snap', snap_name, sv_uuid) + self.assertEqual(snap_path, exp_snap_path) + + def test_snapshot_getpath_on_retained_subvol(self): + ''' + Test that "ceph fs subvolume snapshot getpath" command returns path to + the specified snapshot in the specified subvolume that was deleted but + snapshots on which is retained. + ''' + subvol_name = self._gen_subvol_name() + snap_name = self._gen_subvol_snap_name() + + self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name}') + sv_uuid = self.get_subvol_uuid(subvol_name) + self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' + f'{subvol_name} {snap_name}') + self.run_ceph_cmd(f'fs subvolume rm {self.volname} {subvol_name} ' + '--retain-snapshots') + + snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' + f'{self.volname} {subvol_name} ' + f'{snap_name}').strip() + + # expected snapshot path + exp_snap_path = os.path.join('/volumes', '_nogroup', subvol_name, + '.snap', snap_name, sv_uuid) + self.assertEqual(snap_path, exp_snap_path) + + def test_snapshot_getpath_on_retained_subvol_in_group(self): + ''' + Test that "ceph fs subvolume snapshot getpath" command returns path to + the specified snapshot in the specified subvolume that was deleted but + snapshots on which is retained. And the deleted subvolume is located on + a non-default group. + ''' + subvol_name = self._gen_subvol_name() + group_name = self._gen_subvol_grp_name() + snap_name = self._gen_subvol_snap_name() + + self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} {group_name}') + self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} ' + f'{group_name}') + sv_uuid = self.get_subvol_uuid(subvol_name, group_name) + self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} ' + f'{subvol_name} {snap_name} {group_name}') + self.run_ceph_cmd(f'fs subvolume rm {self.volname} {subvol_name} ' + f'{group_name} --retain-snapshots') + + snap_path = self.get_ceph_cmd_stdout(f'fs subvolume snapshot getpath ' + f'{self.volname} {subvol_name} ' + f'{snap_name} {group_name}')\ + .strip() + # expected snapshot path + exp_snap_path = os.path.join('/volumes', group_name, subvol_name, + '.snap', snap_name, sv_uuid) + self.assertEqual(snap_path, exp_snap_path) + + class TestSubvolumeSnapshotClones(TestVolumesHelper): """ Tests for FS subvolume snapshot clone operations.""" def test_clone_subvolume_info(self):