feature(backups): mounting a repo

selfprivacy_api/backup/backuppers/restic_backupper.py

@@ -6,6 +6,8 @@ from typing import List
 from collections.abc import Iterable
 from json.decoder import JSONDecodeError
 from os.path import exists
+from os import listdir
+from time import sleep
 
 from selfprivacy_api.backup.util import output_yielder
 from selfprivacy_api.backup.backuppers import AbstractBackupper
@@ -52,7 +54,7 @@ class ResticBackupper(AbstractBackupper):
     def _password_command(self):
         return f"echo {LocalBackupSecret.get()}"
 
-    def restic_command(self, *args, tag: str = ""):
+    def restic_command(self, *args, tag: str = "") -> List[str]:
         command = [
             "restic",
             "-o",
@@ -73,6 +75,28 @@ class ResticBackupper(AbstractBackupper):
             command.extend(ResticBackupper.__flatten_list(args))
         return command
 
+    def mount_repo(self, dir):
+        mount_command = self.restic_command("mount", dir)
+        mount_command.insert(0, "nohup")
+        handle = subprocess.Popen(mount_command, stdout=subprocess.DEVNULL, shell=False)
+        sleep(2)
+        if not "ids" in listdir(dir):
+            raise IOError("failed to mount dir ", dir)
+        return handle
+
+    def unmount_repo(self, dir):
+        mount_command = ["umount", "-l", dir]
+        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():
+                return IOError("failed to unmount dir ", dir, ": ", output)
+
+        if not listdir(dir) == []:
+            return IOError("failed to unmount dir ", dir)
+
     @staticmethod
     def __flatten_list(list):
         """string-aware list flattener"""

tests/test_graphql/test_backup.py

@@ -5,6 +5,7 @@ from os import remove
 from os import listdir
 from os import urandom
 from datetime import datetime, timedelta, timezone
+from subprocess import Popen
 
 import selfprivacy_api.services as services
 from selfprivacy_api.services import Service
@@ -19,6 +20,7 @@ import selfprivacy_api.backup.providers as providers
 from selfprivacy_api.backup.providers import AbstractBackupProvider
 from selfprivacy_api.backup.providers.backblaze import Backblaze
 from selfprivacy_api.backup.util import sync
+from selfprivacy_api.backup.backuppers.restic_backupper import ResticBackupper
 
 
 from selfprivacy_api.backup.tasks import start_backup, restore_snapshot
@@ -552,3 +554,22 @@ def test_sync_nonexistent_src(dummy_service):
 
     with pytest.raises(ValueError):
         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