mirror of
https://github.com/borgbackup/borg.git
synced 2025-03-04 02:28:34 +00:00
Merge pull request #1756 from ThomasWaldmann/fix-freebsd-fuse-tests
Fix freebsd fuse tests
This commit is contained in:
commit
25ef14e853
2 changed files with 37 additions and 24 deletions
|
@ -60,11 +60,11 @@ class BaseTestCase(unittest.TestCase):
|
||||||
yield
|
yield
|
||||||
self.assert_true(os.path.exists(path), '{} should exist'.format(path))
|
self.assert_true(os.path.exists(path), '{} should exist'.format(path))
|
||||||
|
|
||||||
def assert_dirs_equal(self, dir1, dir2):
|
def assert_dirs_equal(self, dir1, dir2, **kwargs):
|
||||||
diff = filecmp.dircmp(dir1, dir2)
|
diff = filecmp.dircmp(dir1, dir2)
|
||||||
self._assert_dirs_equal_cmp(diff)
|
self._assert_dirs_equal_cmp(diff, **kwargs)
|
||||||
|
|
||||||
def _assert_dirs_equal_cmp(self, diff):
|
def _assert_dirs_equal_cmp(self, diff, ignore_bsdflags=False, ignore_xattrs=False):
|
||||||
self.assert_equal(diff.left_only, [])
|
self.assert_equal(diff.left_only, [])
|
||||||
self.assert_equal(diff.right_only, [])
|
self.assert_equal(diff.right_only, [])
|
||||||
self.assert_equal(diff.diff_files, [])
|
self.assert_equal(diff.diff_files, [])
|
||||||
|
@ -77,7 +77,7 @@ class BaseTestCase(unittest.TestCase):
|
||||||
# Assume path2 is on FUSE if st_dev is different
|
# Assume path2 is on FUSE if st_dev is different
|
||||||
fuse = s1.st_dev != s2.st_dev
|
fuse = s1.st_dev != s2.st_dev
|
||||||
attrs = ['st_mode', 'st_uid', 'st_gid', 'st_rdev']
|
attrs = ['st_mode', 'st_uid', 'st_gid', 'st_rdev']
|
||||||
if has_lchflags:
|
if has_lchflags and not ignore_bsdflags:
|
||||||
attrs.append('st_flags')
|
attrs.append('st_flags')
|
||||||
if not fuse or not os.path.isdir(path1):
|
if not fuse or not os.path.isdir(path1):
|
||||||
# dir nlink is always 1 on our fuse filesystem
|
# dir nlink is always 1 on our fuse filesystem
|
||||||
|
@ -96,11 +96,12 @@ class BaseTestCase(unittest.TestCase):
|
||||||
else:
|
else:
|
||||||
d1.append(round(s1.st_mtime_ns, st_mtime_ns_round))
|
d1.append(round(s1.st_mtime_ns, st_mtime_ns_round))
|
||||||
d2.append(round(s2.st_mtime_ns, st_mtime_ns_round))
|
d2.append(round(s2.st_mtime_ns, st_mtime_ns_round))
|
||||||
d1.append(no_selinux(get_all(path1, follow_symlinks=False)))
|
if not ignore_xattrs:
|
||||||
d2.append(no_selinux(get_all(path2, follow_symlinks=False)))
|
d1.append(no_selinux(get_all(path1, follow_symlinks=False)))
|
||||||
|
d2.append(no_selinux(get_all(path2, follow_symlinks=False)))
|
||||||
self.assert_equal(d1, d2)
|
self.assert_equal(d1, d2)
|
||||||
for sub_diff in diff.subdirs.values():
|
for sub_diff in diff.subdirs.values():
|
||||||
self._assert_dirs_equal_cmp(sub_diff)
|
self._assert_dirs_equal_cmp(sub_diff, ignore_bsdflags=ignore_bsdflags, ignore_xattrs=ignore_xattrs)
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def fuse_mount(self, location, mountpoint, *options):
|
def fuse_mount(self, location, mountpoint, *options):
|
||||||
|
|
|
@ -1040,11 +1040,16 @@ class ArchiverTestCase(ArchiverTestCaseBase):
|
||||||
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
|
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
|
||||||
# mount the whole repository, archive contents shall show up in archivename subdirs of mountpoint:
|
# mount the whole repository, archive contents shall show up in archivename subdirs of mountpoint:
|
||||||
with self.fuse_mount(self.repository_location, mountpoint):
|
with self.fuse_mount(self.repository_location, mountpoint):
|
||||||
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive', 'input'))
|
# bsdflags are not supported by the FUSE mount
|
||||||
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive2', 'input'))
|
# we also ignore xattrs here, they are tested separately
|
||||||
|
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive', 'input'),
|
||||||
|
ignore_bsdflags=True, ignore_xattrs=True)
|
||||||
|
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive2', 'input'),
|
||||||
|
ignore_bsdflags=True, ignore_xattrs=True)
|
||||||
# mount only 1 archive, its contents shall show up directly in mountpoint:
|
# mount only 1 archive, its contents shall show up directly in mountpoint:
|
||||||
with self.fuse_mount(self.repository_location + '::archive', mountpoint):
|
with self.fuse_mount(self.repository_location + '::archive', mountpoint):
|
||||||
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'input'))
|
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'input'),
|
||||||
|
ignore_bsdflags=True, ignore_xattrs=True)
|
||||||
# regular file
|
# regular file
|
||||||
in_fn = 'input/file1'
|
in_fn = 'input/file1'
|
||||||
out_fn = os.path.join(mountpoint, 'input', 'file1')
|
out_fn = os.path.join(mountpoint, 'input', 'file1')
|
||||||
|
@ -1064,20 +1069,6 @@ class ArchiverTestCase(ArchiverTestCaseBase):
|
||||||
# read
|
# read
|
||||||
with open(in_fn, 'rb') as in_f, open(out_fn, 'rb') as out_f:
|
with open(in_fn, 'rb') as in_f, open(out_fn, 'rb') as out_f:
|
||||||
assert in_f.read() == out_f.read()
|
assert in_f.read() == out_f.read()
|
||||||
# list/read xattrs
|
|
||||||
in_fn = 'input/fusexattr'
|
|
||||||
out_fn = os.path.join(mountpoint, 'input', 'fusexattr')
|
|
||||||
if not xattr.XATTR_FAKEROOT and xattr.is_enabled(self.input_path):
|
|
||||||
assert no_selinux(xattr.listxattr(out_fn)) == ['user.foo', ]
|
|
||||||
assert xattr.getxattr(out_fn, 'user.foo') == b'bar'
|
|
||||||
else:
|
|
||||||
assert xattr.listxattr(out_fn) == []
|
|
||||||
try:
|
|
||||||
xattr.getxattr(out_fn, 'user.foo')
|
|
||||||
except OSError as e:
|
|
||||||
assert e.errno == llfuse.ENOATTR
|
|
||||||
else:
|
|
||||||
assert False, "expected OSError(ENOATTR), but no error was raised"
|
|
||||||
# hardlink (to 'input/file1')
|
# hardlink (to 'input/file1')
|
||||||
in_fn = 'input/hardlink'
|
in_fn = 'input/hardlink'
|
||||||
out_fn = os.path.join(mountpoint, 'input', 'hardlink')
|
out_fn = os.path.join(mountpoint, 'input', 'hardlink')
|
||||||
|
@ -1097,6 +1088,27 @@ class ArchiverTestCase(ArchiverTestCaseBase):
|
||||||
out_fn = os.path.join(mountpoint, 'input', 'fifo1')
|
out_fn = os.path.join(mountpoint, 'input', 'fifo1')
|
||||||
sto = os.stat(out_fn)
|
sto = os.stat(out_fn)
|
||||||
assert stat.S_ISFIFO(sto.st_mode)
|
assert stat.S_ISFIFO(sto.st_mode)
|
||||||
|
# list/read xattrs
|
||||||
|
try:
|
||||||
|
in_fn = 'input/fusexattr'
|
||||||
|
out_fn = os.path.join(mountpoint, 'input', 'fusexattr')
|
||||||
|
if not xattr.XATTR_FAKEROOT and xattr.is_enabled(self.input_path):
|
||||||
|
assert no_selinux(xattr.listxattr(out_fn)) == ['user.foo', ]
|
||||||
|
assert xattr.getxattr(out_fn, 'user.foo') == b'bar'
|
||||||
|
else:
|
||||||
|
assert xattr.listxattr(out_fn) == []
|
||||||
|
try:
|
||||||
|
xattr.getxattr(out_fn, 'user.foo')
|
||||||
|
except OSError as e:
|
||||||
|
assert e.errno == llfuse.ENOATTR
|
||||||
|
else:
|
||||||
|
assert False, "expected OSError(ENOATTR), but no error was raised"
|
||||||
|
except OSError as err:
|
||||||
|
if sys.platform.startswith(('freebsd', )) and err.errno == errno.ENOTSUP:
|
||||||
|
# some systems have no xattr support on FUSE
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise
|
||||||
|
|
||||||
@unittest.skipUnless(has_llfuse, 'llfuse not installed')
|
@unittest.skipUnless(has_llfuse, 'llfuse not installed')
|
||||||
def test_fuse_allow_damaged_files(self):
|
def test_fuse_allow_damaged_files(self):
|
||||||
|
|
Loading…
Add table
Reference in a new issue