refactoring

zfs-backup.py

@@ -30,6 +30,7 @@ SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are
 DISK_BY_UUID = Path("/dev/disk/by-uuid")
 MAPPER_PATH = Path("/dev/mapper")
 
+# noinspection SpellCheckingInspection
 """
 Erstes Backup muss für alle notwendigen Datasets durchgeführt werden mit
 # zfs send -Rv rpool/ROOT@zfs-auto-snap_daily-2018-06-14-1648 | zfs receive -dvF backup1
@@ -119,7 +120,7 @@ class Manager:
             local_snap
             for local_snap in list(cls._find_snapshots_with_backup_tag(local_dataset))
             for remote_snap in list(cls._find_snapshots_with_backup_tag(remote_dataset))
-            if local_snap.snapname == remote_snap.snapname
+            if local_snap.snapshot_name == remote_snap.snapshot_name
         ]
         common_snapshots.sort()
         return common_snapshots[-1]
@@ -131,38 +132,38 @@ def _get_regex_matching_snapshots_with_tags(tags: list[str]):
 
 
 class ZFSPath:
-    def __init__(self, elements: list[str], snapname: str | None = None):
+    def __init__(self, elements: list[str], snapshot_name: str | None = None):
         self._elements = elements
-        self._snapname = snapname
+        self._snapshot_name = snapshot_name
         self._is_snapshot = False
-        if snapname:
+        if snapshot_name:
             self._is_snapshot = True
         self._sanity_check()
 
     def __repr__(self):
         path = "/".join(self._elements)
         if self._is_snapshot:
-            return "@".join([path, self._snapname])
+            return "@".join([path, self._snapshot_name])
         return path
 
     def __eq__(self, other: ZFSPath):
         return (
                 self._elements == other._elements
                 and self._is_snapshot == other._is_snapshot
-            and self._snapname == other._snapname
+                and self._snapshot_name == other._snapshot_name
         )
 
     @classmethod
     def from_string(cls, string: str) -> ZFSPath:
-        snapname = None
+        snapshot_name = None
         if "@" in string:
-            string, snapname = string.split("@")
+            string, snapshot_name = string.split("@")
         elements: list[str] = string.split("/")
-        return ZFSPath(elements, snapname)
+        return ZFSPath(elements, snapshot_name)
 
     @property
-    def snapname(self) -> str:
+    def snapshot_name(self) -> str:
-        return self._snapname
+        return self._snapshot_name
 
     @property
     def is_snapshot(self) -> bool:
@@ -177,13 +178,13 @@ class ZFSPath:
         return ZFSPath(self._elements)
 
     def replace_pool(self, pool: str) -> ZFSPath:
-        return ZFSPath([pool] + self._elements[1:].copy(), self.snapname)
+        return ZFSPath([pool] + self._elements[1:].copy(), self.snapshot_name)
 
     @property
     def name_without_pool(self) -> str:
         name = "/".join(self._elements[1:])
         if self.is_snapshot:
-            name = "@".join([name, self.snapname])
+            name = "@".join([name, self.snapshot_name])
         return name
 
     def _sanity_check(self):
@@ -200,17 +201,26 @@ def _assert_type(instance: Any, object_type: Any, message: str) -> None:
 class ZPOOL:
     _ZPOOL = Path("/usr/bin/zpool")
 
+    class _Subcommands:
+        # -N: no mount
+        # -d: directory to search the pool in
+        import_ = ["import", "-N", "-d"]
+        export = ["export"]
+        scrub = ["scrub"]
+        status = ["status"]
+        wait = ["wait", "-t", "scrub"]
+
     @classmethod
     def import_from_directory(cls, pool: Pool, directory: Path) -> None:
         _assert_type(pool, Pool, "Can only import Pool objects")
-        # -N: no mount
+        CommandRunner.run(
-        # -d: directory to search the pool in
+            [str(cls._ZPOOL), *cls._Subcommands.import_, directory, pool.name]
-        CommandRunner.run([cls._ZPOOL, "import", "-N", "-d", directory, pool.name])
+        )
 
     @classmethod
     def export(cls, pool: Pool) -> None:
         _assert_type(pool, Pool, "Can only export Pool objects")
-        CommandRunner.run([cls._ZPOOL, "export", pool.name])
+        CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.export, pool.name])
 
     @classmethod
     def scrub(
@@ -223,19 +233,22 @@ class ZPOOL:
         _assert_type(pool, Pool, "Can only scrub pools.")
         cls._scrub_if_necessary(pool, skip_if_recently_scrubbed)
         if wait_for_finish:
-            CommandRunner.run([cls._ZPOOL, "wait", "-t", "scrub", pool.name])
+            CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.wait, pool.name])
         if not cls._healthy(pool):
             raise IOError(f"Pool {pool.name} is not healthy.")
 
     @classmethod
     def _scrub_if_necessary(cls, pool: Pool, skip_if_recently_scrubbed: bool) -> None:
+        def do_scrub():
+            CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.scrub, pool.name])
+
         if cls._scrub_in_progress(pool):
             return
         if not skip_if_recently_scrubbed:
-            CommandRunner.run([cls._ZPOOL, "scrub", pool.name])
+            do_scrub()
         else:
             if not cls._recently_scrubbed(pool):
-                CommandRunner.run([cls._ZPOOL, "scrub", pool.name])
+                do_scrub()
 
     @classmethod
     def _recently_scrubbed(cls, pool: Pool) -> bool:
@@ -249,9 +262,9 @@ class ZPOOL:
     def _healthy(cls, pool: Pool) -> bool:
         return "ONLINE" in cls._get_pool_status_output(pool)
 
-    @staticmethod
+    @classmethod
-    def _get_pool_status_output(pool: Pool) -> str:
+    def _get_pool_status_output(cls, pool: Pool) -> str:
-        return CommandRunner.get_output([ZFS, "status", pool.name])
+        return CommandRunner.get_output([ZFS, *cls._Subcommands.status, pool.name])
 
     @classmethod
     def _last_scrub(cls, pool: Pool) -> datetime:
@@ -276,8 +289,8 @@ class ZFS:
         # -R  Replicate filesystem
         # -I  send all intermediary snapshots
         send = ["send", "-R", "-I"]
-        # -d  Discard the first element of the sent snapshot's file system name
+        # -d  Discard the first element of the "send" snapshot's file system name
-        # -F  Force a rollback of the file system to the most recent snapshot before performing the receive operation.
+        # -F  Force a rollback of the file system to the most recent snapshot before performing the "receive".
         # -u  File system that is associated with the received stream is not mounted.
         receive = ["receive", "-d", "-F", "-u"]
 
@@ -411,6 +424,7 @@ class Cryptsetup:
         close = "close"
         status = "status"
         open = "open"
+
     _CRYPTSETUP = Path("/usr/bin/cryptsetup")
 
     @classmethod
@@ -578,14 +592,14 @@ class Snapshot:
         return cls(ZFSPath.from_string(qualified_name))
 
     def __lt__(self, other: Snapshot) -> bool:
-        return self.snapname < other.snapname
+        return self.snapshot_name < other.snapshot_name
 
     def __repr__(self):
         return str(self._path)
 
     @property
-    def snapname(self) -> str:
+    def snapshot_name(self) -> str:
-        return self._path.snapname
+        return self._path.snapshot_name
 
     def matches_regex(self, regex: str) -> bool:
         return bool(search(regex, str(self._path)))