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')