diff options
Diffstat (limited to 'test')
| -rw-r--r-- | test/functional/libgfapi-python-tests.py | 263 | ||||
| -rw-r--r-- | test/unit/gluster/test_gfapi.py | 157 |
2 files changed, 321 insertions, 99 deletions
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): diff --git a/test/unit/gluster/test_gfapi.py b/test/unit/gluster/test_gfapi.py index d4a9b52..15ce061 100644 --- a/test/unit/gluster/test_gfapi.py +++ b/test/unit/gluster/test_gfapi.py @@ -21,7 +21,7 @@ from gluster.gfapi import File, Dir, Volume, DirEntry from gluster import api from gluster.exceptions import LibgfapiException from nose import SkipTest -from mock import Mock, patch +from mock import Mock, MagicMock, patch from contextlib import nested @@ -961,41 +961,34 @@ class TestVolume(unittest.TestCase): "key1") def test_rmtree_success(self): - dir1_list = ["dir2", "file"] - empty_list = [] - mock_listdir = Mock() - mock_listdir.side_effect = [dir1_list, empty_list] - - mock_isdir = Mock() - mock_isdir.side_effect = [True, False] + s_file = api.Stat() + s_file.st_mode = stat.S_IFREG + d = DirEntry(None, 'dirpath', 'file1', s_file) + mock_scandir = MagicMock() + mock_scandir.return_value = [d] mock_unlink = Mock() - mock_unlink.return_value = 0 - mock_rmdir = Mock() - mock_rmdir.return_value = 0 - - mock_islink = Mock() - mock_islink.return_value = False + mock_islink = Mock(return_value=False) - with nested(patch("gluster.gfapi.Volume.listdir", mock_listdir), - patch("gluster.gfapi.Volume.isdir", mock_isdir), + with nested(patch("gluster.gfapi.Volume.scandir", mock_scandir), patch("gluster.gfapi.Volume.islink", mock_islink), patch("gluster.gfapi.Volume.unlink", mock_unlink), patch("gluster.gfapi.Volume.rmdir", mock_rmdir)): - self.vol.rmtree("dir1") - mock_rmdir.assert_any_call("dir1/dir2") - mock_unlink.assert_called_once_with("dir1/file") - mock_rmdir.assert_called_with("dir1") + self.vol.rmtree("dirpath") + + mock_islink.assert_called_once_with("dirpath") + mock_unlink.assert_called_once_with("dirpath/file1") + mock_rmdir.assert_called_once_with("dirpath") def test_rmtree_listdir_exception(self): - mock_listdir = Mock() - mock_listdir.side_effect = [OSError] + mock_scandir = MagicMock() + mock_scandir.side_effect = [OSError] mock_islink = Mock() mock_islink.return_value = False - with nested(patch("gluster.gfapi.Volume.listdir", mock_listdir), + with nested(patch("gluster.gfapi.Volume.scandir", mock_scandir), patch("gluster.gfapi.Volume.islink", mock_islink)): self.assertRaises(OSError, self.vol.rmtree, "dir1") @@ -1007,32 +1000,25 @@ class TestVolume(unittest.TestCase): self.assertRaises(OSError, self.vol.rmtree, "dir1") def test_rmtree_ignore_unlink_rmdir_exception(self): - dir1_list = ["dir2", "file"] - empty_list = [] - mock_listdir = Mock() - mock_listdir.side_effect = [dir1_list, empty_list] + s_file = api.Stat() + s_file.st_mode = stat.S_IFREG + d = DirEntry(None, 'dirpath', 'file1', s_file) + mock_scandir = MagicMock() + mock_scandir.return_value = [d] - mock_isdir = Mock() - mock_isdir.side_effect = [True, False] - - mock_unlink = Mock() - mock_unlink.side_effect = [OSError] + mock_unlink = Mock(side_effect=OSError) + mock_rmdir = Mock(side_effect=OSError) + mock_islink = Mock(return_value=False) - mock_rmdir = Mock() - mock_rmdir.side_effect = [0, OSError] - - mock_islink = Mock() - mock_islink.return_value = False - - with nested(patch("gluster.gfapi.Volume.listdir", mock_listdir), - patch("gluster.gfapi.Volume.isdir", mock_isdir), + with nested(patch("gluster.gfapi.Volume.scandir", mock_scandir), patch("gluster.gfapi.Volume.islink", mock_islink), patch("gluster.gfapi.Volume.unlink", mock_unlink), patch("gluster.gfapi.Volume.rmdir", mock_rmdir)): - self.vol.rmtree("dir1", True) - mock_rmdir.assert_any_call("dir1/dir2") - mock_unlink.assert_called_once_with("dir1/file") - mock_rmdir.assert_called_with("dir1") + self.vol.rmtree("dirpath", True) + + mock_islink.assert_called_once_with("dirpath") + mock_unlink.assert_called_once_with("dirpath/file1") + mock_rmdir.assert_called_once_with("dirpath") def test_setfsuid_success(self): mock_glfs_setfsuid = Mock() @@ -1093,33 +1079,82 @@ class TestVolume(unittest.TestCase): "filelink") def test_walk_success(self): - dir1_list = ["dir2", "file"] - empty_list = [] - mock_listdir = Mock() - mock_listdir.side_effect = [dir1_list, empty_list] - - mock_isdir = Mock() - mock_isdir.side_effect = [True, False] - - with nested(patch("gluster.gfapi.Volume.listdir", mock_listdir), - patch("gluster.gfapi.Volume.isdir", mock_isdir)): - for (path, dirs, files) in self.vol.walk("dir1"): - self.assertEqual(dirs, ['dir2']) - self.assertEqual(files, ['file']) + s_dir = api.Stat() + s_dir.st_mode = stat.S_IFDIR + d1 = DirEntry(Mock(), 'dirpath', 'dir1', s_dir) + d2 = DirEntry(Mock(), 'dirpath', 'dir2', s_dir) + s_file = api.Stat() + s_file.st_mode = stat.S_IFREG + d3 = DirEntry(Mock(), 'dirpath', 'file1', s_file) + d4 = DirEntry(Mock(), 'dirpath', 'file2', s_file) + mock_scandir = MagicMock() + mock_scandir.return_value = [d1, d3, d2, d4] + + with patch("gluster.gfapi.Volume.scandir", mock_scandir): + for (path, dirs, files) in self.vol.walk("dirpath"): + self.assertEqual(dirs, ['dir1', 'dir2']) + self.assertEqual(files, ['file1', 'file2']) break - def test_walk_listdir_exception(self): - mock_listdir = Mock() - mock_listdir.side_effect = [OSError] + def test_walk_scandir_exception(self): + mock_scandir = Mock() + mock_scandir.side_effect = [OSError] def mock_onerror(err): self.assertTrue(isinstance(err, OSError)) - with patch("gluster.gfapi.Volume.listdir", mock_listdir): + with patch("gluster.gfapi.Volume.scandir", mock_scandir): for (path, dirs, files) in self.vol.walk("dir1", onerror=mock_onerror): pass + def test_copytree_success(self): + d_stat = api.Stat() + d_stat.st_mode = stat.S_IFDIR + f_stat = api.Stat() + f_stat.st_mode = stat.S_IFREG + # Depth = 0 + iter1 = [('dir1', d_stat), ('dir2', d_stat), ('file1', f_stat)] + # Depth = 1, dir1 + iter2 = [('file2', f_stat), ('file3', f_stat)] + # Depth = 1, dir2 + iter3 = [('file4', f_stat), ('dir3', d_stat), ('file5', f_stat)] + # Depth = 2, dir3 + iter4 = [] # Empty directory. + # So there are 5 files in total that should to be copied + # and (3 + 1) directories should be created, including the destination + + m_list_s = Mock(side_effect=[iter1, iter2, iter3, iter4]) + m_makedirs = Mock() + m_fopen = MagicMock() + m_copyfileobj = Mock() + m_utime = Mock() + m_chmod = Mock() + m_copystat = Mock() + with nested(patch("gluster.gfapi.Volume.listdir_with_stat", m_list_s), + patch("gluster.gfapi.Volume.makedirs", m_makedirs), + patch("gluster.gfapi.Volume.fopen", m_fopen), + patch("gluster.gfapi.Volume.copyfileobj", m_copyfileobj), + patch("gluster.gfapi.Volume.utime", m_utime), + patch("gluster.gfapi.Volume.chmod", m_chmod), + patch("gluster.gfapi.Volume.copystat", m_copystat)): + self.vol.copytree('/source', '/destination') + + # Assert that listdir_with_stat() was called on all directories + self.assertEqual(m_list_s.call_count, 3 + 1) + # Assert that fopen() was called 10 times - twice for each file + # i.e once for reading and another time for writing. + self.assertEqual(m_fopen.call_count, 10) + # Assert number of files copied + self.assertEqual(m_copyfileobj.call_count, 5) + # Assert that utime and chmod was called on the files + self.assertEqual(m_utime.call_count, 5) + self.assertEqual(m_chmod.call_count, 5) + # Assert number of directories created + self.assertEqual(m_makedirs.call_count, 3 + 1) + # Assert that copystat() was called on source and destination dir + m_copystat.called_once_with('/source', '/destination') + def test_utime(self): # Test times arg being invalid. for junk in ('a', 1234.1234, (1, 2, 3), (1)): |