]> git.apps.os.sepia.ceph.com Git - ceph-ci.git/commitdiff
test/pybind: add unit tests for rmtree() in cephfs python bindings
authorRishabh Dave <ridave@redhat.com>
Thu, 3 Jul 2025 06:34:39 +0000 (12:04 +0530)
committerRishabh Dave <ridave@redhat.com>
Tue, 7 Oct 2025 16:27:20 +0000 (21:57 +0530)
Signed-off-by: Rishabh Dave <ridave@redhat.com>
(cherry picked from commit 05082a932984bb6329481c14ee76ae033c019f4e)

src/test/pybind/test_cephfs.py

index 577cb9e4171536fca1a03802409170716aa2e42e..093792694513fb1b767f911c4fb58a9aaea6535a 100644 (file)
@@ -940,3 +940,511 @@ def test_multi_target_command():
     if isinstance(mds_status, list): # if multi target command result
         for mds_sessions in session_map:
             assert(list(mds_sessions.keys())[0].startswith('mds.'))
+
+
+class TestRmtree:
+    '''
+    Test rmtree() method of CephFS python bindings.
+    '''
+
+    def test_rmtree_on_regfile(self, testdir):
+        should_cancel = lambda: False
+
+        fd = cephfs.open(f'/file1', 'w', 0o755)
+        cephfs.write(fd, b'abcd', 0)
+        cephfs.close(fd)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('/file1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'file1')
+
+    def test_rmtree_on_regfile_no_perms(self, testdir):
+        should_cancel = lambda: False
+
+        fd = cephfs.open(f'/file1', 'w', 0o755)
+        cephfs.write(fd, b'abcd', 0)
+        cephfs.close(fd)
+
+        cephfs.chmod('/file1', 0o000)
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('/file1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'file1')
+
+    def test_rmtree_on_symlink(self, testdir):
+        should_cancel = lambda: False
+
+        fd = cephfs.open(f'/file1', 'w', 0o755)
+        cephfs.write(fd, b'abcd', 0)
+        cephfs.close(fd)
+        cephfs.symlink('file1', '/slink1')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('/slink1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'slink1')
+        cephfs.stat('file1')
+
+    def test_rmtree_on_symlink_no_perms(self, testdir):
+        should_cancel = lambda: False
+
+        fd = cephfs.open(f'/file1', 'w', 0o755)
+        cephfs.write(fd, b'abcd', 0)
+        cephfs.close(fd)
+        cephfs.symlink('file1', '/slink1')
+
+        cephfs.chmod('/file1', 0o000)
+        cephfs.chmod('/slink1', 0o000)
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('/slink1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'slink1')
+        cephfs.stat('file1')
+
+    def test_rmtree_when_tree_contains_only_regfiles(self, testdir):
+        '''
+        Test rmtree() successfully deletes the entire file hierarchy that contains
+        only regular files.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_when_tree_contains_dirs_and_regfiles(self, testdir):
+        '''
+        Test that rmtree() successfully deletes the entire file hierarchy that
+        contains only directories and regular files.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir2', 0o755)
+        for i in range(1, 6):
+            cephfs.mkdir(f'/dir2/dir2{i}', 0o755)
+            for j in range(1, 6):
+                fd = cephfs.open(f'/dir2/dir2{i}/file{j}', 'w', 0o755)
+                cephfs.write(fd, b'abcd', 0)
+                cephfs.close(fd)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir2', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir2')
+
+    def test_rmtree_when_tree_contains_dirs_regfiles_and_symlinks(self, testdir):
+        '''
+        Test that rmtree() successfully deletes entire file hierarchy that
+        contains directories, regular files as well as symbolic links.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir3', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir3/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+            file_name = f'/dir3/file{i}'.encode('utf-8')
+            slink_name = f'slink{i}'.encode('utf-8')
+            cephfs.symlink(file_name, slink_name)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir3', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir3')
+
+    def test_rmtree_when_symlink_points_to_parent_dir(self, testdir):
+        '''
+        Test that rmtree() successfully deletes entire file hierarchy that
+        contains directories, regular files as well as symbolic links.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir3', 0o755)
+        cephfs.mkdir('dir3/dir4', 0o755)
+        cephfs.symlink('../dir4', 'dir3/dir4/slink1')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir3', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir3')
+
+    def test_rmtree_when_tree_contains_only_empty_dirs(self, testdir):
+        '''
+        Test that rmtree() successfully deletes entire file hierarchy that contains
+        only empty directories.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir4', 0o755)
+        for i in range(1, 6):
+            cephfs.mkdir(f'/dir4/dir4{i}', 0o755)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir4', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir4')
+
+    def test_rmtree_when_root_is_empty_dir(self, testdir):
+        '''
+        Test that rmtree() successfully deletes entire file hierarchy when it is
+        only an empty directory.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir5', 0o755)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir5', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir5')
+
+    def test_rmtree_no_perm_on_nonroot_dir_suppress_errors(self, testdir):
+        '''
+        Test that rmtree() successfully deletes the entire file hierarchy except the
+        branch where permission for one of the (non-root) directories is not
+        granted.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+
+        cephfs.mkdir('dir1/dir2', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir2/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.mkdir('dir1/dir3', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir3/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.mkdir('dir1/dir4', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir4/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        # actual test
+        cephfs.chmod('/dir1/dir3', 0o000)
+        # Errors are expected from call to this method. Set suppress_errors to
+        # True to confirm that this argument works.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=True)
+        # ensure /dir1/dir3 wasn't deleted
+        cephfs.stat('dir1/dir3')
+        cephfs.chmod('/dir1/dir3', 0o755)
+        for i in range(1, 6):
+            cephfs.stat(f'dir1/dir3/file{i}')
+
+        # cleanup
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_no_perm_on_nonroot_dir_dont_suppress_errors(self, testdir):
+        '''
+        Test that rmtree() successfully deletes the entire file hierarchy except the
+        branch where permission for one of the (non-root) directories is not
+        granted.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+
+        cephfs.mkdir('dir1/dir2', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir2/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.mkdir('dir1/dir3', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir3/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.mkdir('dir1/dir4', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir4/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        # actual test
+        cephfs.chmod('/dir1/dir3', 0o000)
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        assert_raises(libcephfs.PermissionDenied, cephfs.rmtree, 'dir1',
+                      should_cancel, suppress_errors=False)
+        # ensure /dir1/dir3 wasn't deleted
+        cephfs.stat('dir1/dir3')
+
+        # cleanup
+        cephfs.chmod('/dir1/dir3', 0o755)
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_no_perm_on_root_suppress_errors(self, testdir):
+        '''
+        Test rmtree() exits when permission is not granted for the root of the file
+        hierarchy.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.chmod('/dir1', 0o000)
+        # Errors are expected from call to this method. Set suppress_errors to
+        # True to confirm that this argument works.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=True)
+        # ensure /dir1 wasn't deleted
+        cephfs.stat('dir1')
+
+        # cleanup
+        cephfs.chmod('/dir1', 0o755)
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_no_perm_on_root_dont_suppress_errors(self, testdir):
+        '''
+        Test rmtree() exits when permission is not granted for the root of the file
+        hierarchy.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        cephfs.chmod('/dir1', 0o000)
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        assert_raises(libcephfs.PermissionDenied, cephfs.rmtree, 'dir1',
+                      should_cancel, suppress_errors=False)
+        # ensure /dir1 wasn't deleted
+        cephfs.stat('dir1')
+
+        # cleanup
+        cephfs.chmod('/dir1', 0o755)
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_tree_with_snaps(self, testdir):
+        '''
+        Test that rmtree() successfully deletes the entire file hierarchy except
+        the branch where one of the directories contains one or many snapshots.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        cephfs.mkdir('dir1/dir2', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/dir2/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+        cephfs.mksnap('/dir1/dir2', 'snap1', 0o755)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        # ensure dir1 wasn't deleted
+        cephfs.stat('dir1')
+
+        # cleanup
+        cephfs.rmsnap('/dir1/dir2', 'snap1')
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_tree_with_snaps_on_root(self, testdir):
+        '''
+        Test that rmtree() successfully deletes the entire file hierarchy except
+        the branch where one of the directories contains one or many snapshots.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        for i in range(1, 6):
+            fd = cephfs.open(f'/dir1/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+        cephfs.mksnap('/dir1', 'snap1', 0o755)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        # ensure dir1 wasn't deleted
+        cephfs.stat('dir1')
+
+        # cleanup
+        cephfs.rmsnap('/dir1', 'snap1')
+        cephfs.rmtree('dir1', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def get_file_count(self, dir_path):
+        '''
+        Return the number of files present in the given directory.
+        '''
+        i = 0
+        with cephfs.opendir(dir_path) as dir_handle:
+            de = cephfs.readdir(dir_handle)
+            while de:
+                if de.d_name not in (b'.', b'..'):
+                    i += 1
+                de = cephfs.readdir(dir_handle)
+        return i
+
+    def test_rmtree_aborts_when_should_cancel_is_true(self, testdir):
+        '''
+        Test that rmtree() stops deleting the file hierarchy when the return
+        value of "should_cancel" becomes True.
+        '''
+        from threading import Event, Thread
+        cancel_flag = Event()
+        def should_cancel():
+            time.sleep(0.1)
+            return cancel_flag.is_set()
+
+        # NOTE: this method is just a wrapper to provide an appropriate location
+        # to catch the exception OpCanceled. If left uncaught the test passes
+        # but pytest fails citing this exception.
+        def rmtree(path, should_cancel, suppress_error=False):
+            assert_raises(libcephfs.OpCanceled, cephfs.rmtree, path,
+                          should_cancel, suppress_error)
+
+        cephfs.mkdir('dir6', 0o755)
+        for i in range(1, 101):
+            fd = cephfs.open(f'/dir6/file{i}', 'w', 0o755)
+            cephfs.write(fd, b'abcd', 0)
+            cephfs.close(fd)
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        Thread(target=rmtree, args=('dir6', should_cancel, False)).start()
+        time.sleep(1)
+
+        # this will change return value of should_cancel and therefore halt
+        # execution of rmtree()
+        cancel_flag.set()
+        # ensure dir6 wasn't deleted
+        cephfs.stat('dir6')
+        # ensure that deletion had begun but hadn't finished and was halted
+        file_count = self.get_file_count('dir6')
+        assert file_count > 0 and file_count < 100
+
+        # ensure that deletion has made no progress since it was halted
+        time.sleep(2)
+        file_count = self.get_file_count('dir6')
+        assert file_count > 0 and file_count < 100
+
+        # cleanup
+        cancel_flag.clear()
+        cephfs.rmtree('dir6', should_cancel)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_a_very_broad_tree(self, testdir):
+        '''
+        Test that rmtree() successfully deletes a file hierarchy with 200
+        subdirectories on the same level.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        cephfs.chdir('dir1')
+        for i in range(1, 201):
+            dirname = f'dir{i}'
+            cephfs.mkdir(dirname, 0o755)
+        cephfs.chdir('/')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_a_very_very_broad_tree(self, testdir):
+        '''
+        Test that rmtree() successfully deletes a file hierarchy with 2000
+        subdirectories on the same level.
+        '''
+        should_cancel = lambda: False
+
+        cephfs.mkdir('dir1', 0o755)
+        cephfs.chdir('dir1')
+        for i in range(1, 2001):
+            dirname = f'dir{i}'
+            cephfs.mkdir(dirname, 0o755)
+        cephfs.chdir('/')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_a_very_deep_tree(self, testdir):
+        '''
+        Test that rmtree() successfully deletes a file hierarchy with 2000
+        levels.
+        '''
+        should_cancel = lambda: False
+
+        for i in range(1, 201):
+            dirname = f'dir{i}'
+            cephfs.mkdir(dirname, 0o755)
+            cephfs.chdir(dirname)
+        cephfs.chdir('/')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')
+
+    def test_rmtree_on_a_very_very_deep_tree(self, testdir):
+        '''
+        Test that rmtree() successfully deletes a file hierarchy with 2000
+        levels.
+        '''
+        should_cancel = lambda: False
+
+        for i in range(1, 2001):
+            dirname = f'dir{i}'
+            cephfs.mkdir(dirname, 0o755)
+            cephfs.chdir(dirname)
+        cephfs.chdir('/')
+
+        # Errors are not expected from the call to this method. Therefore, set
+        # suppress_errors to False so that tests abort as soon as any errors
+        # occur.
+        cephfs.rmtree('dir1', should_cancel, suppress_errors=False)
+        assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1')