summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rwxr-xr-xgluster/gfapi.py170
-rw-r--r--test/functional/libgfapi-python-tests.py263
-rw-r--r--test/unit/gluster/test_gfapi.py157
3 files changed, 457 insertions, 133 deletions
diff --git a/gluster/gfapi.py b/gluster/gfapi.py
index b346d8d..6186723 100755
--- a/gluster/gfapi.py
+++ b/gluster/gfapi.py
@@ -556,7 +556,7 @@ class DirEntry(object):
def stat(self, follow_symlinks=False):
"""
Returns information equivalent of a lstat() system call on the entry.
- This does not follow symlinks.
+ This does not follow symlinks by default.
"""
if follow_symlinks:
if self._stat is None:
@@ -1285,20 +1285,23 @@ class Volume(object):
raise
if self.islink(path):
raise OSError("Cannot call rmtree on a symbolic link")
- names = []
+
try:
- names = self.listdir(path)
+ for entry in self.scandir(path):
+ fullname = os.path.join(path, entry.name)
+ if entry.is_dir(follow_symlinks=False):
+ self.rmtree(fullname, ignore_errors, onerror)
+ else:
+ try:
+ self.unlink(fullname)
+ except OSError as e:
+ onerror(self.unlink, fullname, e)
except OSError as e:
- onerror(self.listdir, path, e)
- for name in names:
- fullname = os.path.join(path, name)
- if self.isdir(fullname):
- self.rmtree(fullname, ignore_errors, onerror)
- else:
- try:
- self.unlink(fullname)
- except OSError as e:
- onerror(self.unlink, fullname, e)
+ # self.scandir() is not a list and is a true iterator, it can
+ # raise an exception and blow-up. The try-except block here is to
+ # handle it gracefully and return.
+ onerror(self.scandir, path, e)
+
try:
self.rmdir(path)
except OSError as e:
@@ -1464,39 +1467,65 @@ class Volume(object):
def walk(self, top, topdown=True, onerror=None, followlinks=False):
"""
- Directory tree generator. Yields a 3-tuple dirpath, dirnames, filenames
-
- dirpath is the path to the directory, dirnames is a list of the names
- of the subdirectories in dirpath. filenames is a list of the names of
- the non-directiry files in dirpath
-
+ Generate the file names in a directory tree by walking the tree either
+ top-down or bottom-up.
+
+ Slight difference in behaviour in comparison to os.walk():
+ When os.walk() is called with 'followlinks=False' (default), symlinks
+ to directories are included in the 'dirnames' list. When Volume.walk()
+ is called with 'followlinks=False' (default), symlinks to directories
+ are included in 'filenames' list. This is NOT a bug.
+ http://python.6.x6.nabble.com/os-walk-with-followlinks-False-td3559133.html
+
+ :param top: Directory path to walk
+ :param topdown: If topdown is True or not specified, the triple for a
+ directory is generated before the triples for any of
+ its subdirectories. If topdown is False, the triple
+ for a directory is generated after the triples for all
+ of its subdirectories.
+ :param onerror: If optional argument onerror is specified, it should be
+ a function; it will be called with one argument, an
+ OSError instance. It can report the error to continue
+ with the walk, or raise exception to abort the walk.
+ :param followlinks: Set followlinks to True to visit directories
+ pointed to by symlinks.
:raises: OSError on failure if onerror is None
+ :yields: a 3-tuple (dirpath, dirnames, filenames) where dirpath is a
+ string, the path to the directory. dirnames is a list of the
+ names of the subdirectories in dirpath (excluding '.' and
+ '..'). filenames is a list of the names of the non-directory
+ files in dirpath.
"""
- # TODO: Write a more efficient walk by leveraging d_type information
- # returned in readdir.
+ dirs = [] # List of DirEntry objects
+ nondirs = [] # List of names (strings)
+
try:
- names = self.listdir(top)
+ for entry in self.scandir(top):
+ if entry.is_dir(follow_symlinks=followlinks):
+ dirs.append(entry)
+ else:
+ nondirs.append(entry.name)
except OSError as err:
+ # self.scandir() is not a list and is a true iterator, it can
+ # raise an exception and blow-up. The try-except block here is to
+ # handle it gracefully and return.
if onerror is not None:
onerror(err)
return
- dirs, nondirs = [], []
- for name in names:
- if self.isdir(os.path.join(top, name)):
- dirs.append(name)
- else:
- nondirs.append(name)
-
if topdown:
- yield top, dirs, nondirs
- for name in dirs:
- new_path = os.path.join(top, name)
- if followlinks or not self.islink(new_path):
+ yield top, [d.name for d in dirs], nondirs
+
+ for directory in dirs:
+ # NOTE: Both is_dir() and is_symlink() can be true for the same
+ # path when follow_symlinks is set to True
+ if followlinks or not directory.is_symlink():
+ new_path = os.path.join(top, directory.name)
for x in self.walk(new_path, topdown, onerror, followlinks):
yield x
+
if not topdown:
- yield top, dirs, nondirs
+ yield top, [d.name for d in dirs], nondirs
def samefile(self, path1, path2):
"""
@@ -1630,3 +1659,76 @@ class Volume(object):
dst = os.path.join(dst, os.path.basename(src))
self.copyfile(src, dst)
self.copystat(src, dst)
+
+ def copytree(self, src, dst, symlinks=False, ignore=None):
+ """
+ Recursively copy a directory tree using copy2().
+
+ The destination directory must not already exist.
+ If exception(s) occur, an Error is raised with a list of reasons.
+
+ If the optional symlinks flag is true, symbolic links in the
+ source tree result in symbolic links in the destination tree; if
+ it is false, the contents of the files pointed to by symbolic
+ links are copied.
+
+ The optional ignore argument is a callable. If given, it
+ is called with the 'src' parameter, which is the directory
+ being visited by copytree(), and 'names' which is the list of
+ 'src' contents, as returned by os.listdir():
+
+ callable(src, names) -> ignored_names
+
+ Since copytree() is called recursively, the callable will be
+ called once for each directory that is copied. It returns a
+ list of names relative to the 'src' directory that should
+ not be copied.
+ """
+ def _isdir(path, statinfo, follow_symlinks=False):
+ if stat.S_ISDIR(statinfo.st_mode):
+ return True
+ if follow_symlinks and stat.S_ISLNK(statinfo.st_mode):
+ return self.isdir(path)
+ return False
+
+ # Can't used scandir() here to support ignored_names functionality
+ names_with_stat = self.listdir_with_stat(src)
+ if ignore is not None:
+ ignored_names = ignore(src, [n for n, s in names_with_stat])
+ else:
+ ignored_names = set()
+
+ self.makedirs(dst)
+ errors = []
+ for (name, st) in names_with_stat:
+ if name in ignored_names:
+ continue
+ srcpath = os.path.join(src, name)
+ dstpath = os.path.join(dst, name)
+ try:
+ if symlinks and stat.S_ISLNK(st.st_mode):
+ linkto = self.readlink(srcpath)
+ self.symlink(linkto, dstpath)
+ # shutil's copytree() calls os.path.isdir() which will return
+ # true even if it's a symlink pointing to a dir. Mimicking the
+ # same behaviour here with _isdir()
+ elif _isdir(srcpath, st, follow_symlinks=not symlinks):
+ self.copytree(srcpath, dstpath, symlinks)
+ else:
+ # The following is equivalent of copy2(). copy2() is not
+ # invoked directly to avoid multiple duplicate stat calls.
+ with self.fopen(srcpath, 'rb') as fsrc:
+ with self.fopen(dstpath, 'wb') as fdst:
+ self.copyfileobj(fsrc, fdst)
+ self.utime(dstpath, (st.st_atime, st.st_mtime))
+ self.chmod(dstpath, stat.S_IMODE(st.st_mode))
+ except (Error, EnvironmentError, OSError) as why:
+ errors.append((srcpath, dstpath, str(why)))
+
+ try:
+ self.copystat(src, dst)
+ except OSError as why:
+ errors.append((src, dst, str(why)))
+
+ if errors:
+ raise Error(errors)
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)):