mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2025-02-02 14:16:38 +00:00
refactor(backups): clean up unused mounting tools
This commit is contained in:
parent
c89f9cf89d
commit
d621ca6449
|
@ -127,34 +127,6 @@ class ResticBackupper(AbstractBackupper):
|
|||
output,
|
||||
)
|
||||
|
||||
def mount_repo(self, mount_directory: str) -> subprocess.Popen:
|
||||
if not exists(mount_directory):
|
||||
raise FileNotFoundError("no such directory to mount at: ", mount_directory)
|
||||
mount_command = self.restic_command("mount", mount_directory)
|
||||
mount_command.insert(0, "nohup")
|
||||
handle = subprocess.Popen(
|
||||
mount_command,
|
||||
stdout=subprocess.DEVNULL,
|
||||
shell=False,
|
||||
)
|
||||
sleep(2)
|
||||
if "ids" not in listdir(mount_directory):
|
||||
raise IOError("failed to mount dir ", mount_directory)
|
||||
return handle
|
||||
|
||||
def unmount_repo(self, mount_directory: str) -> None:
|
||||
mount_command = ["umount", "-l", mount_directory]
|
||||
with subprocess.Popen(
|
||||
mount_command, stdout=subprocess.PIPE, shell=False
|
||||
) as handle:
|
||||
output = handle.communicate()[0].decode("utf-8")
|
||||
# TODO: check for exit code?
|
||||
if "error" in output.lower():
|
||||
raise IOError("failed to unmount dir ", mount_directory, ": ", output)
|
||||
|
||||
if not listdir(mount_directory) == []:
|
||||
raise IOError("failed to unmount dir ", mount_directory)
|
||||
|
||||
@staticmethod
|
||||
def __flatten_list(list_to_flatten):
|
||||
"""string-aware list flattener"""
|
||||
|
|
|
@ -727,36 +727,6 @@ def test_sync_nonexistent_src(dummy_service):
|
|||
sync(src, dst)
|
||||
|
||||
|
||||
# Restic lowlevel
|
||||
def test_mount_umount(backups, dummy_service, tmpdir):
|
||||
Backups.back_up(dummy_service)
|
||||
backupper = Backups.provider().backupper
|
||||
assert isinstance(backupper, ResticBackupper)
|
||||
|
||||
mountpoint = tmpdir / "mount"
|
||||
makedirs(mountpoint)
|
||||
assert path.exists(mountpoint)
|
||||
assert len(listdir(mountpoint)) == 0
|
||||
|
||||
handle = backupper.mount_repo(mountpoint)
|
||||
assert len(listdir(mountpoint)) != 0
|
||||
|
||||
backupper.unmount_repo(mountpoint)
|
||||
# handle.terminate()
|
||||
assert len(listdir(mountpoint)) == 0
|
||||
|
||||
|
||||
def test_mount_nonexistent(backups, dummy_service, tmpdir):
|
||||
backupper = Backups.provider().backupper
|
||||
assert isinstance(backupper, ResticBackupper)
|
||||
|
||||
mountpoint = tmpdir / "nonexistent"
|
||||
assert not path.exists(mountpoint)
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
handle = backupper.mount_repo(mountpoint)
|
||||
|
||||
|
||||
def test_move_blocks_backups(backups, dummy_service, restore_strategy):
|
||||
snap = Backups.back_up(dummy_service)
|
||||
job = Jobs.add(
|
||||
|
|
Loading…
Reference in a new issue