refactor: ZFS

zfs-backup.py

@@ -191,13 +191,78 @@ class ZFSPath:
                 raise ValueError(NOT_VALID_ZFS_PATH)
 
 
-class ZFS:
+class ZFSBase:
-    # TODO introduce command constants
     @staticmethod
     def _assert_type(instance: Any, object_type: Any, message: str) -> None:
         if not isinstance(instance, object_type):
             raise TypeError(message)
 
+
+class ZPOOL(ZFSBase):
+    _ZPOOL = Path("/usr/bin/zpool")
+
+    @classmethod
+    def import_from_directory(cls, pool: Pool, directory: Path) -> None:
+        cls._assert_type(pool, Pool, "Can only import Pool objects")
+        # -N: no mount
+        # -d: directory to search the pool in
+        Command.run([cls._ZPOOL, "import", "-N", "-d", directory, pool.name])
+
+    @classmethod
+    def export(cls, pool: Pool) -> None:
+        cls._assert_type(pool, Pool, "Can only export Pool objects")
+        Command.run([cls._ZPOOL, "export", pool.name])
+
+    @classmethod
+    def scrub(
+        cls,
+        pool: Pool,
+        *,
+        skip_if_recently_scrubbed: bool,
+        wait_for_finish: bool = True,
+    ) -> None:
+        cls._assert_type(pool, Pool, "Can only scrub pools.")
+        cls._scrub_if_necessary(pool, skip_if_recently_scrubbed)
+        if wait_for_finish:
+            Command.run([cls._ZPOOL, "wait", "-t", "scrub", 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:
+        if cls._scrub_in_progress(pool):
+            return
+        if not skip_if_recently_scrubbed:
+            Command.run([cls._ZPOOL, "scrub", pool.name])
+        else:
+            if not cls._recently_scrubbed(pool):
+                Command.run([cls._ZPOOL, "scrub", pool.name])
+
+    @classmethod
+    def _recently_scrubbed(cls, pool: Pool) -> bool:
+        return datetime.now() - cls._last_scrub(pool) < config.recent_scrub_timedelta
+
+    @classmethod
+    def _scrub_in_progress(cls, pool: Pool) -> bool:
+        return "scrub in progress" in cls._get_pool_status_output(pool)
+
+    @classmethod
+    def _healthy(cls, pool: Pool) -> bool:
+        return "ONLINE" in cls._get_pool_status_output(pool)
+
+    @staticmethod
+    def _get_pool_status_output(pool: Pool) -> str:
+        return Command.get_output([ZFS, "status", pool.name])
+
+    @classmethod
+    def _last_scrub(cls, pool: Pool) -> datetime:
+        return Time.get_last_scrub_from_status_output(cls._get_pool_status_output(pool))
+
+
+class ZFS(ZFSBase):
+    _ZFS = Path("/usr/bin/zfs")
+    # TODO introduce command constants
+
     @classmethod
     def get_first_level_datasets(cls, pool: Pool) -> list[Dataset]:
         cls._assert_type(pool, Pool, "Can only get datasets from Pool objects")
@@ -219,18 +284,6 @@ class ZFS:
         datasets_list = datasets_str.split("\n")[1:]
         return [Dataset(ZFSPath.from_string(d)) for d in datasets_list]
 
-    @classmethod
-    def import_pool_from_directory(cls, pool: Pool, directory: Path) -> None:
-        cls._assert_type(pool, Pool, "Can only import Pool objects")
-        # -N: no mount
-        # -d: directory to search the pool in
-        Command.run([Binaries.ZPOOL, "import", "-N", "-d", directory, pool.name])
-
-    @classmethod
-    def export_pool(cls, pool: Pool) -> None:
-        cls._assert_type(pool, Pool, "Can only export Pool objects")
-        Command.run([Binaries.ZPOOL, "export", pool.name])
-
     @classmethod
     def get_snapshots(cls, dataset: Dataset) -> list[Snapshot]:
         cls._assert_type(
@@ -292,64 +345,17 @@ class ZFS:
         if old_snapshot == new_snapshot:
             raise ValueError("Cannot send stream, snapshots are identical.")
 
-    @staticmethod
+    @classmethod
-    def receive(stream: Popen[str], pool: Pool):
+    def receive(cls, stream: Popen[str], pool: Pool):
         # -d  Discard the first element of the sent snapshot's file system name
         # -F  Force a rollback of the file system to the most recent snapshot before performing the receive operation.
         # -u  File system that is associated with the received stream is not mounted.
-        if not isinstance(pool, Pool):
+        cls._assert_type(pool, Pool, "Can only receive into pools.")
-            raise TypeError("Can only receive into pools.")
         Command.receive_from_stream(
             [ZFS, "receive", "-d", "-F", "-u", pool.name],
             stream,
         )
 
-    @classmethod
-    def scrub(
-        cls,
-        pool: Pool,
-        *,
-        skip_if_recently_scrubbed: bool,
-        wait_for_finish: bool = True,
-    ) -> None:
-        if not isinstance(pool, Pool):
-            raise TypeError("Can only scrub pools.")
-        cls._scrub_if_necessary(pool, skip_if_recently_scrubbed)
-        if wait_for_finish:
-            Command.run([Binaries.ZPOOL, "wait", "-t", "scrub", 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:
-        if cls._scrub_in_progress(pool):
-            return
-        if not skip_if_recently_scrubbed:
-            Command.run([Binaries.ZPOOL, "scrub", pool.name])
-        else:
-            if not cls._recently_scrubbed(pool):
-                Command.run([Binaries.ZPOOL, "scrub", pool.name])
-
-    @classmethod
-    def _recently_scrubbed(cls, pool: Pool) -> bool:
-        return datetime.now() - cls._last_scrub(pool) < config.recent_scrub_timedelta
-
-    @classmethod
-    def _scrub_in_progress(cls, pool: Pool) -> bool:
-        return "scrub in progress" in cls._get_pool_status_output(pool)
-
-    @classmethod
-    def _healthy(cls, pool: Pool) -> bool:
-        return "ONLINE" in cls._get_pool_status_output(pool)
-
-    @staticmethod
-    def _get_pool_status_output(pool: Pool) -> str:
-        return Command.get_output([ZFS, "status", pool.name])
-
-    @classmethod
-    def _last_scrub(cls, pool: Pool) -> datetime:
-        return Time.get_last_scrub_from_status_output(cls._get_pool_status_output(pool))
-
 
 class Time:
     @staticmethod
@@ -410,14 +416,20 @@ class Disk:
             raise ValueError(NOT_DECRYPTED, self._name)
         if self._was_already_decrypted:
             return
-        Command.run([Binaries.CRYPTSETUP, "close", self._mapper_entry])
+        Cryptsetup.encrypt(self._mapper_entry)
         self._decrypted = False
 
 
 class Cryptsetup:
+    _CRYPTSETUP = Path("/usr/bin/cryptsetup")
+
+    @classmethod
+    def encrypt(cls, mapper_entry: str) -> None:
+        Command.run([cls._CRYPTSETUP, "close", mapper_entry])
+
     @classmethod
     def _status(cls, name: str) -> str:
-        return Command.get_output([Binaries.CRYPTSETUP, "status", name])
+        return Command.get_output([cls._CRYPTSETUP, "status", name])
 
     @classmethod
     def decrypt(cls, path: Path, mapper_entry: str) -> bool:
@@ -426,7 +438,7 @@ class Cryptsetup:
                 return True
             print(MAPPER_ENTRY_ALREADY_EXISTS)
             exit(EXIT_ERROR)
-        Command.run([Binaries.CRYPTSETUP, "open", str(path), mapper_entry])
+        Command.run([cls._CRYPTSETUP, "open", str(path), mapper_entry])
         return False
 
     @classmethod
@@ -478,7 +490,7 @@ class ExternalPool(Pool):
         if self._imported:
             raise ValueError(ALREADY_IMPORTED, self.name)
         self._disk.decrypt()
-        ZFS.import_pool_from_directory(self, MAPPER_PATH)
+        ZPOOL.import_from_directory(self, MAPPER_PATH)
         self._imported = True
 
     def export(self) -> None:
@@ -490,11 +502,11 @@ class ExternalPool(Pool):
         print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.")
 
     def _export_pool(self) -> None:
-        ZFS.export_pool(self)
+        ZPOOL.export(self)
 
     def scrub(self, skip_if_recently_scrubbed: bool) -> None:
         print(f"Scrubbing {self._name}")
-        ZFS.scrub(self, skip_if_recently_scrubbed=skip_if_recently_scrubbed)
+        ZPOOL.scrub(self, skip_if_recently_scrubbed=skip_if_recently_scrubbed)
 
     @classmethod
     def find_from_dict(cls, backup_pools: dict[str, UUID]) -> ExternalPool:
@@ -620,11 +632,5 @@ def verify_running_as_root():
         exit(EXIT_ERROR)
 
 
-class Binaries:
-    ZFS = Path("/usr/bin/zfs")
-    ZPOOL = Path("/usr/bin/zpool")
-    CRYPTSETUP = Path("/usr/bin/cryptsetup")
-
-
 if __name__ == "__main__":
     main()