From 5dac3e7ed0eaaf5ce42afd601fceb15193583197 Mon Sep 17 00:00:00 2001 From: Rishabh Dave Date: Thu, 16 Oct 2025 19:35:41 +0530 Subject: [PATCH] test_cephfs: add unit tests for cptree() in cephfs python bindings Signed-off-by: Rishabh Dave --- src/test/pybind/test_cephfs.py | 663 +++++++++++++++++++++++++++++++++ 1 file changed, 663 insertions(+) diff --git a/src/test/pybind/test_cephfs.py b/src/test/pybind/test_cephfs.py index b081ce528da..ddc88d6336a 100644 --- a/src/test/pybind/test_cephfs.py +++ b/src/test/pybind/test_cephfs.py @@ -1808,3 +1808,666 @@ class TestRmtree: # occur. cephfs.rmtree('dir1', should_cancel, suppress_errors=False) assert_raises(libcephfs.ObjectNotFound, cephfs.stat, 'dir1') + + +class TestCptree: + ''' + Test cptree() method of CephFS python bindings. + ''' + + def test_cptree_on_regfile(self, testdir): + ''' + Test that cptree() copies a regular file too when src path passed to it + is that of a regular file. + ''' + src = 'file1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(dst, 0o755) + + fd = cephfs.open(f'{src}', '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.cptree(src, dst, should_cancel, suppress_errors=False) + + cephfs.stat(dst) + cephfs.stat(f'{dst}/file1') + + # ensure src file was left as it is. + cephfs.stat(src) + + def test_cptree_on_regfile_no_perms(self, testdir): + ''' + Test that cptree() fails when src path passed to it is that of a regular + file and permissions are not granted on it. + ''' + src = 'file1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(dst, 0o755) + + fd = cephfs.open(src, 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + cephfs.chmod(src, 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.cptree, src, dst, + should_cancel, suppress_errors=False) + + # ensure src file was left as it is. + cephfs.stat(src) + + def test_cptree_on_symlink(self, testdir): + ''' + Test that cptree() copies a symbolic link too when src path passed to it + is that of a symbolic link. + ''' + file = 'file1' + slink = 'slink1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(dst, 0o755) + + fd = cephfs.open(file, 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + cephfs.symlink(file, slink) + + # 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.cptree(slink, dst, should_cancel, suppress_errors=False) + + cephfs.stat(f'{dst}/{slink}', follow_symlink=False) + cephfs.stat(file) + cephfs.stat(slink) + + def test_cptree_on_symlink_no_perms(self, testdir): + file = 'file1' + slink = 'slink1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(dst, 0o755) + + fd = cephfs.open(file, 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + cephfs.symlink(file, slink) + + cephfs.chmod(slink, 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.cptree(slink, dst, should_cancel, suppress_errors=False) + + cephfs.stat(f'{dst}/{slink}', follow_symlink=False) + cephfs.stat(file) + cephfs.stat(slink) + + def test_cptree_when_tree_contains_only_regfiles(self, testdir): + ''' + Test cptree() successfully copies entire file hierarchy that contains + only regular files. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + fd = cephfs.open(f'/{src}/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.cptree(src, dst, should_cancel, suppress_errors=False) + + cephfs.stat(src) + cephfs.stat(dst) + for i in range(1, 6): + cephfs.stat(f'{src}/file{i}') + cephfs.stat(f'{dst}/{src}/file{i}') + + def test_cptree_when_tree_contains_dirs_and_regfiles(self, testdir): + ''' + Test that cptree() successfully copies entire file hierarchy that + contains only directories and regular files. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + cephfs.mkdir(f'/{src}/{src}{i}', 0o755) + for j in range(1, 6): + fd = cephfs.open(f'/{src}/{src}{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.cptree(src, dst, should_cancel, suppress_errors=False) + + # verify that files are copied to dst path + for i in range(1, 6): + cephfs.stat(f'{dst}/{src}/{src}{i}') + for j in range(1, 6): + cephfs.stat(f'{src}/{src}{i}/file{j}') + + # verify that files are as it is in src path + for i in range(1, 6): + cephfs.stat(f'{src}/{src}{i}') + for j in range(1, 6): + cephfs.stat(f'{src}/{src}{i}/file{j}') + + def test_cptree_when_tree_contains_dirs_regfiles_and_symlinks(self, testdir): + ''' + Test that cptree() successfully copies entire file hierarchy that + contains directories, regular files as well as symbolic links. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + for i in range(1, 6): + fd = cephfs.open(f'/{src}/file{i}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + file_name = f'file{i}'.encode('utf-8') + slink_name = f'/{src}/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.cptree(src, dst, should_cancel, suppress_errors=False) + + # verify that files are copied to dst path + for i in range(1, 6): + cephfs.stat(f'{dst}/{src}/file{i}') + cephfs.stat(f'{dst}/{src}/slink{i}') + + # verify that files are as it is in src path + for i in range(1, 6): + cephfs.stat(f'{src}/file{i}') + cephfs.stat(f'{src}/slink{i}') + + + def test_cptree_when_symlink_points_to_parent_dir(self, testdir): + ''' + Test that cptree() successfully copies 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.mkdir('dir5', 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.cptree('dir3', 'dir5', should_cancel, suppress_errors=False) + + cephfs.stat('dir3/dir4/slink1') + cephfs.stat('dir5/dir3/dir4/slink1') + + def test_cptree_when_tree_contains_only_empty_dirs(self, testdir): + ''' + Test that cptree() successfully copies entire file hierarchy that contains + only empty directories. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + cephfs.mkdir(f'/{src}/{src}{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.cptree(src, dst, should_cancel, suppress_errors=False) + + for i in range(1, 6): + cephfs.stat(f'/{dst}/{src}/{src}{i}', 0o755) + cephfs.stat(f'/{src}/{src}{i}', 0o755) + + def test_cptree_when_root_is_empty_dir(self, testdir): + ''' + Test that cptree() successfully copies an empty directory too. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 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.cptree(src, dst, should_cancel, suppress_errors=False) + + cephfs.stat(src) + cephfs.stat(f'{dst}/{src}') + + def test_cptree_no_perm_on_nonroot_dir_suppress_errors(self, testdir): + ''' + Test that cptree() successfully copies the entire file hierarchy except + the branch where permission for one of the (non-root) directories is not + granted while suppressing/not raising any errors for the dir for which + permission is not granted. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + cephfs.mkdir(f'{src}/{src}{i}', 0o755) + for j in range(1, 6): + fd = cephfs.open(f'/{src}/{src}{i}/file{j}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + # actual test + cephfs.chmod(f'/{src}/{src}3', 0o000) + # Errors are expected from call to this method. Set suppress_errors to + # True to confirm that this argument works. + cephfs.cptree(src, dst, should_cancel, suppress_errors=True) + + # ensure /dir1/dir3 wasn't copied + for i in range(1, 6): + for j in range(1, 6): + if i == 3: + cephfs.stat(f'{src}/{src}{i}') + cephfs.stat(f'{dst}/{src}/{src}{i}') + assert_raises(libcephfs.ObjectNotFound, cephfs.stat, + f'{dst}/{src}/{src}{i}/file{j}') + else: + cephfs.stat(f'{src}/{src}{i}/file{j}') + cephfs.stat(f'{dst}/{src}/{src}{i}/file{j}') + + cephfs.chmod(f'/{src}/{src}3', 0o755) + cephfs.chmod(f'/{dst}/{src}/{src}3', 0o755) + + def test_cptree_no_perm_on_nonroot_dir_dont_suppress_errors(self, testdir): + ''' + Test that cptree() aborts its attempt to copy the entire file hierarchy + when it finds a non-root directory where permission is not granted. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + cephfs.mkdir(f'{src}/{src}{i}', 0o755) + for j in range(1, 6): + fd = cephfs.open(f'/{src}/{src}{i}/file{j}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + # actual test + cephfs.chmod(f'/{src}/{src}3', 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.cptree, src, dst, + should_cancel, suppress_errors=False) + + cephfs.chmod(f'/{src}/{src}3', 0o755) + + def test_cptree_no_perm_on_root_suppress_errors(self, testdir): + ''' + Test cptree() exits without error when permission is not granted for + the root of the file hierarchy. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 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(src, 0o000) + # Errors are expected from call to this method. Set suppress_errors to + # True to confirm that this argument works. + cephfs.cptree(src, dst, should_cancel, suppress_errors=True) + + # cleanup + cephfs.chmod('/dir1', 0o755) + + def test_cptree_no_perm_on_root_dont_suppress_errors(self, testdir): + ''' + Test cptree() aborts with error when permission is not granted for the + root of the file hierarchy. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 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.cptree, src, dst, + should_cancel, suppress_errors=False) + + # cleanup + cephfs.chmod('/dir1', 0o755) + + def test_cptree_on_tree_with_snaps(self, testdir): + ''' + Test that cptree() successfully copies the entire file hierarchy except + the snapshot. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + cephfs.mkdir('dir1/dir11', 0o755) + for i in range(1, 6): + fd = cephfs.open(f'/dir1/dir11/file{i}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + cephfs.mksnap('/dir1/dir11', '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.cptree(src, dst, should_cancel, suppress_errors=False) + + for i in range(1, 6): + cephfs.stat(f'{dst}/{src}/dir11/file{i}') + # ensure src was left as it was. + cephfs.stat(f'{src}/dir11/file{i}') + + # cleanup + cephfs.rmsnap('/dir1/dir11', 'snap1') + + def test_cptree_on_tree_with_snaps_on_root(self, testdir): + ''' + Test that cptree() successfully copies the entire file hierarchy except + the snapshot. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 6): + fd = cephfs.open(f'/{src}/file{i}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + + cephfs.mksnap(src, '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.cptree(src, dst, should_cancel, suppress_errors=False) + + for i in range(1, 6): + cephfs.stat(f'{dst}/{src}/file{i}') + # ensure src was left as it was. + cephfs.stat(f'{src}/file{i}') + + # cleanup + cephfs.rmsnap('/dir1', 'snap1') + + 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_cptree_aborts_when_should_cancel_is_true(self, testdir): + ''' + Test that cptree() stops copying the file hierarchy when the return + value of "should_cancel" becomes True. + ''' + src = 'dir1' + dst = 'dir2' + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + # populate with enough files that copying doesn't finish before + # cancel fag is set. + cephfs.chdir(src) + for i in range(1, 101): + fd = cephfs.open(f'file{i}', 'w', 0o755) + cephfs.write(fd, b'abcd', 0) + cephfs.close(fd) + cephfs.chdir('/') + + # 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 cptree_assert_wrapper(src, dst, should_cancel, suppress_error=False): + assert_raises(libcephfs.OpCanceled, cephfs.cptree, src, dst, + should_cancel, suppress_error) + + from threading import Event, Thread + cancel_flag = Event() + def should_cancel(): + # this would force libcephfs's cptree() to sleep every time + # should_cancel is called. this is necessary to ensure copying + # won't finish before cancel flag is set. + print('here123') + time.sleep(0.1) + return cancel_flag.is_set() + + # 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=cptree_assert_wrapper, + args=(src, dst, should_cancel, False)).start() + time.sleep(1) + # this will change return value of should_cancel and therefore halt + # execution of cptree() + cancel_flag.set() + + # ensure that copying had begun + cephfs.stat(f'{dst}/{src}') + # ensure that deletion had begun but hadn't finished and was halted + file_count1 = self.get_file_count(f'{dst}/{src}') + assertion_msg = f'file_count1 = {file_count1}' + assert file_count1 > 0 and file_count1 < 100, assertion_msg + + # ensure that deletion has made no progress since it was halted + time.sleep(2) + file_count2 = self.get_file_count(f'{dst}/{src}') + assertion_msg = f'file_count2 = {file_count2}' + assert file_count2 > 0 and file_count2 < 100, assertion_msg + assertion_msg = f'file_count1 = {file_count1} file_count2 = {file_count2}' + assert file_count1 == file_count2, assertion_msg + + def test_cptree_on_a_200_dir_broad_tree(self, testdir): + ''' + Test that cptree() successfully copies a file hierarchy with 200 + subdirectories on the same level. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 201): + cephfs.mkdir(f'{src}/{src}{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.cptree(src, dst, should_cancel, suppress_errors=False) + + for i in range(1, 201): + cephfs.stat(f'{src}/{src}{i}') + cephfs.stat(f'{dst}/{src}/{src}{i}') + + def test_cptree_on_a_2k_dir_broad_tree(self, testdir): + ''' + Test that cptree() successfully copies a file hierarchy with 2000 + subdirectories on the same level. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + for i in range(1, 201): + cephfs.mkdir(f'{src}/{src}{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.cptree(src, dst, should_cancel, suppress_errors=False) + + for i in range(1, 201): + cephfs.stat(f'{src}/{src}{i}') + cephfs.stat(f'{dst}/{src}/{src}{i}') + + def test_cptree_on_a_200_dir_deep_tree(self, testdir): + ''' + Test that cptree() successfully copies a file hierarchy with 200 + levels. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + cephfs.chdir(src) + 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.cptree(src, dst, should_cancel, suppress_errors=False) + + # verify the dst has the entire file hierarchy cloned + cephfs.chdir(dst) + cephfs.chdir(src) + for i in range(1, 201): + dirname = f'dir{i}' + cephfs.stat(dirname, 0o755) + cephfs.chdir(dirname) + cephfs.chdir('/') + + # verify the src has the entire file hierarchy as it was + cephfs.chdir(src) + for i in range(1, 201): + dirname = f'dir{i}' + cephfs.stat(dirname, 0o755) + cephfs.chdir(dirname) + cephfs.chdir('/') + + def test_cptree_on_a_2k_dir_deep_tree(self, testdir): + ''' + Test that cptree() successfully copies a file hierarchy with 2000 + levels. + ''' + src = 'dir1' + dst = 'dir2' + should_cancel = lambda: False + + cephfs.mkdir(src, 0o755) + cephfs.mkdir(dst, 0o755) + + cephfs.chdir(src) + 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.cptree(src, dst, should_cancel, suppress_errors=False) + + # verify the dst has the entire file hierarchy cloned + cephfs.chdir(dst) + cephfs.chdir(src) + for i in range(1, 2001): + dirname = f'dir{i}' + cephfs.stat(dirname, 0o755) + cephfs.chdir(dirname) + cephfs.chdir('/') + + # verify the src has the entire file hierarchy as it was + cephfs.chdir(src) + for i in range(1, 2001): + dirname = f'dir{i}' + cephfs.stat(dirname, 0o755) + cephfs.chdir(dirname) + cephfs.chdir('/') -- 2.39.5