From 37ce11939885ac812403edddbae24d80c603ee31 Mon Sep 17 00:00:00 2001 From: Prashanth Pai Date: Fri, 17 Jun 2016 20:14:02 +0530 Subject: Implement copytree() and enhance walk(), rmtree() This change: * Implements copytree() API which is very similar to the one provided by shutils built-in module in Python. * Enhances walk() and rmtree() implementation to leverage scandir() optimization. Change-Id: Iac5aef1a5c558fdeceac4e5128339141a3ebd4d1 Signed-off-by: Prashanth Pai --- test/functional/libgfapi-python-tests.py | 263 ++++++++++++++++++++++++++----- 1 file changed, 225 insertions(+), 38 deletions(-) (limited to 'test/functional') diff --git a/test/functional/libgfapi-python-tests.py b/test/functional/libgfapi-python-tests.py index 947e49d..8c62685 100644 --- a/test/functional/libgfapi-python-tests.py +++ b/test/functional/libgfapi-python-tests.py @@ -47,10 +47,9 @@ class BinFileOpsTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.vol = Volume(HOST, VOLNAME) - ret = cls.vol.mount() - if ret == 0: - # Cleanup volume - cls.vol.rmtree("/", ignore_errors=True) + cls.vol.mount() + # Cleanup volume + cls.vol.rmtree("/", ignore_errors=True) @classmethod def tearDownClass(cls): @@ -81,10 +80,9 @@ class FileOpsTest(unittest.TestCase): @classmethod def setUpClass(cls): cls.vol = Volume(HOST, VOLNAME) - ret = cls.vol.mount() - if ret == 0: - # Cleanup volume - cls.vol.rmtree("/", ignore_errors=True) + cls.vol.mount() + # Cleanup volume + cls.vol.rmtree("/", ignore_errors=True) @classmethod def tearDownClass(cls): @@ -801,60 +799,96 @@ class DirOpsTest(unittest.TestCase): data = None dir_path = None - testfile = None @classmethod def setUpClass(cls): cls.vol = Volume(HOST, VOLNAME) - ret = cls.vol.mount() - if ret == 0: - # Cleanup volume - cls.vol.rmtree("/", ignore_errors=True) - cls.testfile = "testfile" + cls.vol.mount() + # Cleanup volume + cls.vol.rmtree("/", ignore_errors=True) @classmethod def tearDownClass(cls): cls.vol.rmtree("/", ignore_errors=True) cls.vol = None - cls.testfile = None def setUp(self): + # Create a filesystem tree self.data = "gluster is awesome" self.dir_path = self._testMethodName + "_dir" self.vol.mkdir(self.dir_path, 0755) for x in range(0, 3): - f = os.path.join(self.dir_path, self.testfile + str(x)) - with File(self.vol.open(f, os.O_CREAT | os.O_WRONLY | os.O_EXCL, - 0644)) as f: - rc = f.write(self.data) - self.assertEqual(rc, len(self.data)) - f.fdatasync() + d = os.path.join(self.dir_path, 'testdir' + str(x)) + self.vol.mkdir(d) + # Create files inside two of the three directories + if x != 1: + for i in range(0, 2): + f = os.path.join(d, 'nestedfile' + str(i)) + with self.vol.fopen(f, 'w') as f: + rc = f.write(self.data) + self.assertEqual(rc, len(self.data)) + # Create single file in root of directory + if x == 2: + file_path = os.path.join(self.dir_path, "testfile") + with self.vol.fopen(file_path, 'w') as f: + rc = f.write(self.data) + self.assertEqual(rc, len(self.data)) + + # Create symlinks - one pointing to a file and another to a dir + # Beware: rmtree() cannot remove these symlinks + self.vol.symlink("testfile", + os.path.join(self.dir_path, 'test_symlink_file')) + self.vol.symlink("testdir2", + os.path.join(self.dir_path, 'test_symlink_dir')) + + # The dir tree set up for testing now looks like this: + # test_name_here + # |-- testdir0 + # | |-- nestedfile0 + # | |-- nestedfile1 + # |-- testdir1 + # |-- testdir2 + # | |-- nestedfile0 + # | |-- nestedfile1 + # |-- testfile + # |-- testsymlink_file --> testfile + # |-- testsymlink_dir --> testdir2 def tearDown(self): + self._symlinks_cleanup() self.dir_path = None self.data = None def test_isdir(self): - isdir = self.vol.isdir(self.dir_path) - self.assertTrue(isdir) - - def test_isfile_false(self): - isfile = self.vol.isfile(self.dir_path) - self.assertFalse(isfile) + self.assertTrue(self.vol.isdir(self.dir_path)) + self.assertFalse(self.vol.isfile(self.dir_path)) def test_listdir(self): dir_list = self.vol.listdir(self.dir_path) dir_list.sort() - self.assertEqual(dir_list, ["testfile0", "testfile1", "testfile2"]) + self.assertEqual(dir_list, + ["test_symlink_dir", "test_symlink_file", + "testdir0", "testdir1", "testdir2", "testfile"]) def test_listdir_with_stat(self): dir_list = self.vol.listdir_with_stat(self.dir_path) dir_list_sorted = sorted(dir_list, key=lambda tup: tup[0]) + dir_count = 0 + file_count = 0 + symlink_count = 0 for index, (name, stat_info) in enumerate(dir_list_sorted): - self.assertEqual(name, 'testfile%s' % (index)) self.assertTrue(isinstance(stat_info, Stat)) - self.assertTrue(stat.S_ISREG(stat_info.st_mode)) - self.assertEqual(stat_info.st_size, len(self.data)) + if stat.S_ISREG(stat_info.st_mode): + self.assertEqual(stat_info.st_size, len(self.data)) + file_count += 1 + elif stat.S_ISDIR(stat_info.st_mode): + self.assertEqual(stat_info.st_size, 4096) + dir_count += 1 + elif stat.S_ISLNK(stat_info.st_mode): + symlink_count += 1 + self.assertEqual(dir_count, 3) + self.assertEqual(file_count, 1) + self.assertEqual(symlink_count, 2) # Error - path does not exist self.assertRaises(OSError, @@ -866,13 +900,25 @@ class DirOpsTest(unittest.TestCase): self.assertTrue(isinstance(entry, DirEntry)) entries.append(entry) + dir_count = 0 + file_count = 0 + symlink_count = 0 entries_sorted = sorted(entries, key=lambda e: e.name) for index, entry in enumerate(entries_sorted): - self.assertEqual(entry.name, 'testfile%s' % (index)) - self.assertTrue(entry.is_file()) - self.assertFalse(entry.is_dir()) self.assertTrue(isinstance(entry.stat(), Stat)) - self.assertEqual(entry.stat().st_size, len(self.data)) + if entry.is_file(): + self.assertEqual(entry.stat().st_size, len(self.data)) + self.assertFalse(entry.is_dir()) + file_count += 1 + elif entry.is_dir(): + self.assertEqual(entry.stat().st_size, 4096) + self.assertFalse(entry.is_file()) + dir_count += 1 + elif entry.is_symlink(): + symlink_count += 1 + self.assertEqual(dir_count, 3) + self.assertEqual(file_count, 1) + self.assertEqual(symlink_count, 2) def test_makedirs(self): name = self.dir_path + "/subd1/subd2/subd3" @@ -893,10 +939,151 @@ class DirOpsTest(unittest.TestCase): """ by testing rmtree, we are also testing unlink and rmdir """ - f = os.path.join(self.dir_path, self.testfile + "1") - self.vol.rmtree(self.dir_path, True) + f = os.path.join(self.dir_path, "testdir0", "nestedfile0") + self.vol.exists(f) + d = os.path.join(self.dir_path, "testdir0") + self.vol.rmtree(d, True) self.assertRaises(OSError, self.vol.lstat, f) - self.assertRaises(OSError, self.vol.lstat, self.dir_path) + self.assertRaises(OSError, self.vol.lstat, d) + + def test_walk_default(self): + # Default: topdown=True, followlinks=False + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(self.dir_path): + for name in files: + file_list.append(name) + for name in dirs: + dir_list.append(name) + self.assertEqual(len(dir_list), 3) # 3 regular directories + self.assertEqual(len(file_list), 7) # 5 regular files + 2 symlinks + + def test_walk_topdown_and_followinks(self): + # topdown=True, followlinks=True + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(self.dir_path, + followlinks=True): + for name in files: + file_list.append(name) + for name in dirs: + dir_list.append(name) + # 4 = 3 regular directories + + # 1 symlink which is pointing to a directory + self.assertEqual(len(dir_list), 4) + # 8 = 5 regular files + + # 1 symlink that points to a file + + # 2 regular files listed again as they are in a directory which has + # a symlink pointing to it. This results in that directory being + # visited twice. + self.assertEqual(len(file_list), 8) + + def test_walk_no_topdown_no_followlinks(self): + # topdown=False, followlinks=False + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(self.dir_path, topdown=False): + for name in files: + file_list.append(name) + for name in dirs: + dir_list.append(name) + self.assertEqual(len(dir_list), 3) # 3 regular directories + self.assertEqual(len(file_list), 7) # 5 regular files + 2 symlinks + + def test_walk_no_topdown_and_followlinks(self): + # topdown=False, followlinks=True + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(self.dir_path, topdown=False, + followlinks=True): + for name in files: + file_list.append(name) + for name in dirs: + dir_list.append(name) + # 4 = 3 regular directories + + # 1 symlink which is pointing to a directory + self.assertEqual(len(dir_list), 4) + # 8 = 5 regular files + + # 1 symlink that points to a file + + # 2 regular files listed again as they are in a directory which has + # a symlink pointing to it. This results in that directory being + # visited twice. + self.assertEqual(len(file_list), 8) + + def test_walk_error(self): + # Test onerror handling + # + # onerror not set + try: + for root, dirs, files in self.vol.walk("non-existent-path"): + pass + except OSError: + self.fail("No exception should be raised") + + # onerror method is set + def handle_error(err): + raise err + try: + for root, dirs, files in self.vol.walk("non-existent-path", + onerror=handle_error): + pass + except OSError: + pass + else: + self.fail("Expecting OSError exception") + + def _symlinks_cleanup(self): + # rmtree() cannot remove these symlinks, hence removing manually. + symlinks = ('test_symlink_dir', 'test_symlink_file') + for name in symlinks: + try: + self.vol.unlink(os.path.join(self.dir_path, name)) + except OSError as err: + if err.errno != errno.ENOENT: + raise + + def test_copy_tree(self): + dest_path = self.dir_path + '_dest' + + # symlinks = False (contents pointed by symlinks are copied) + self.vol.copytree(self.dir_path, dest_path, symlinks=False) + + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(dest_path): + for name in files: + fullpath = os.path.join(root, name) + s = self.vol.lstat(fullpath) + # Assert that there are no symlinks + self.assertFalse(stat.S_ISLNK(s.st_mode)) + file_list.append(name) + for name in dirs: + fullpath = os.path.join(root, name) + s = self.vol.lstat(fullpath) + # Assert that there are no symlinks + self.assertFalse(stat.S_ISLNK(s.st_mode)) + dir_list.append(name) + self.assertEqual(len(dir_list), 4) # 4 regular directories + self.assertEqual(len(file_list), 8) # 8 regular files + + # Cleanup + self.vol.rmtree(dest_path) + + # symlinks = True (symlinks itself is copied as is) + self.vol.copytree(self.dir_path, dest_path, symlinks=True) + + file_list = [] + dir_list = [] + for root, dirs, files in self.vol.walk(dest_path): + for name in files: + file_list.append(name) + for name in dirs: + dir_list.append(name) + self.assertEqual(len(dir_list), 3) # 3 regular directories + self.assertEqual(len(file_list), 7) # 5 regular files + 2 symlinks + + # Error - The destination directory must not exist + self.assertRaises(OSError, self.vol.copytree, self.dir_path, dest_path) class TestVolumeInit(unittest.TestCase): -- cgit