mirror/borg
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2025-01-01 04:37:34 +00:00
Code Issues Releases Wiki Activity

deduplicate fuse (u)mount code

Browse source
This commit is contained in:
Thomas Waldmann 2016-07-08 19:18:03 +02:00
parent f1d3fd431c
commit 41348a76ef
2 changed files with 19 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
borg/testsuite/__init__.py
View file

@ -93,6 +93,21 @@ def _assert_dirs_equal_cmp(self, diff):
for sub_diff in diff.subdirs.values():
self._assert_dirs_equal_cmp(sub_diff)
@contextmanager
def fuse_mount(self, location, mountpoint):
os.mkdir(mountpoint)
self.cmd('mount', location, mountpoint, fork=True)
self.wait_for_mount(mountpoint)
yield
if sys.platform.startswith('linux'):
cmd = 'fusermount -u %s' % mountpoint
else:
cmd = 'umount %s' % mountpoint
os.system(cmd)
os.rmdir(mountpoint)
# Give the daemon some time to exit
time.sleep(.2)
def wait_for_mount(self, path, timeout=5):
"""Wait until a filesystem is mounted on `path`
"""

30
borg/testsuite/archiver.py
View file

@ -975,45 +975,23 @@ def test_help(self):
@unittest.skipUnless(has_llfuse, 'llfuse not installed')
def test_fuse_mount_repository(self):
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
os.mkdir(mountpoint)
self.cmd('init', self.repository_location)
self.create_test_files()
self.cmd('create', self.repository_location + '::archive', 'input')
self.cmd('create', self.repository_location + '::archive2', 'input')
try:
self.cmd('mount', self.repository_location, mountpoint, fork=True)
self.wait_for_mount(mountpoint)
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
with self.fuse_mount(self.repository_location, mountpoint):
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive', 'input'))
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'archive2', 'input'))
finally:
if sys.platform.startswith('linux'):
os.system('fusermount -u ' + mountpoint)
else:
os.system('umount ' + mountpoint)
os.rmdir(mountpoint)
# Give the daemon some time to exit
time.sleep(.2)
@unittest.skipUnless(has_llfuse, 'llfuse not installed')
def test_fuse_mount_archive(self):
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
os.mkdir(mountpoint)
self.cmd('init', self.repository_location)
self.create_test_files()
self.cmd('create', self.repository_location + '::archive', 'input')
try:
self.cmd('mount', self.repository_location + '::archive', mountpoint, fork=True)
self.wait_for_mount(mountpoint)
mountpoint = os.path.join(self.tmpdir, 'mountpoint')
with self.fuse_mount(self.repository_location + '::archive', mountpoint):
self.assert_dirs_equal(self.input_path, os.path.join(mountpoint, 'input'))
finally:
if sys.platform.startswith('linux'):
os.system('fusermount -u ' + mountpoint)
else:
os.system('umount ' + mountpoint)
os.rmdir(mountpoint)
# Give the daemon some time to exit
time.sleep(.2)
def verify_aes_counter_uniqueness(self, method):
seen = set() # Chunks already seen