more fixes

zfs-backup.py

@@ -3,18 +3,28 @@
 
 from __future__ import annotations
 
-import os
-import re
-import subprocess
 from functools import cache
+from os import getuid
 from pathlib import Path
-from subprocess import PIPE, Popen, check_output
+from re import search
+from subprocess import PIPE, Popen, call, check_output
 from sys import exit
 from typing import Optional
 from uuid import UUID
 
 import config
 
+ALREADY_DECRYPTED = "Cannot decrypt, already decrypted"
+ALREADY_IMPORTED = "Cannot import, already imported"
+CANNOT_FIND_BACKUP_DRIVE = "Could not find a backup drive"
+DELETE_IN_LOCAL_POOL = "ALERT! Tried to delete in local pool!"
+MAPPER_ENTRY_ALREADY_EXISTS = "mapper entry already exists"
+NOT_DECRYPTED = "Cannot encrypt, not decrypted"
+NOT_IMPORTED = "Cannot export, not imported"
+NOT_VALID_ZFS_PATH = "Not a valid ZFSPath"
+RUN_AS_ROOT = "Run as root."
+SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset"
+
 DISK_BY_UUID = Path("/dev/disk/by-uuid")
 MAPPER_PATH = Path("/dev/mapper")
 
@@ -162,7 +172,7 @@ class ZFSPath:
     def _sanity_check(self):
         for item in self._elements:
             if len(item) == 0:
-                raise ValueError("Not a valid ZFSPath")
+                raise ValueError(NOT_VALID_ZFS_PATH)
 
 
 class ZFS:
@@ -220,8 +230,7 @@ class ZFS:
     @staticmethod
     def destroy(full_name, *, recursive=False):
         if config.local_pool_to_backup in full_name:
-            msg = "ALERT! Tried to delete in local pool!"
+            raise Exception(DELETE_IN_LOCAL_POOL)
-            raise Exception(msg)
         cmdline = [config.ZFS, "destroy"]
         if recursive:
             cmdline.append("-r")
@@ -231,9 +240,7 @@ class ZFS:
     @staticmethod
     def send_incremental(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]:
         if old_snap.dataset != new_snap.dataset:
-            raise ValueError(
+            raise ValueError(SAME_DATASET)
-                "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset"
-            )
         sender = Popen(
             [config.ZFS, "send", "-R", "-I", old_snap, new_snap], stdout=PIPE
         )
@@ -264,7 +271,7 @@ class Disk:
 
     def decrypt(self) -> None:
         if self._decrypted:
-            raise ValueError(f"Already decrypted {self._name}")
+            raise ValueError(ALREADY_DECRYPTED, self._name)
         self._verify_mapper_entry_not_in_use()
         self._decrypt_with_cryptsetup()
         self._decrypted = True
@@ -275,7 +282,7 @@ class Disk:
     def _verify_mapper_entry_not_in_use(self):
         for item in MAPPER_PATH.iterdir():
             if self._mapper_entry == item.name:
-                print(f"mapper entry {self._mapper_entry} already exists")
+                print(MAPPER_ENTRY_ALREADY_EXISTS, self._mapper_entry)
                 exit(1)
 
     @property
@@ -288,7 +295,7 @@ class Disk:
 
     def encrypt(self):
         if not self._decrypted:
-            raise ValueError(f"Not decrypted {self._name}")
+            raise ValueError(NOT_DECRYPTED, self._name)
         get_output_of_command([config.CRYPTSETUP, "close", self._name])
         self._decrypted = False
 
@@ -319,14 +326,14 @@ class ExternalPool(Pool):
 
     def import_(self) -> None:
         if self._imported:
-            raise ValueError(f"Already imported {self.name}")
+            raise ValueError(ALREADY_IMPORTED, self.name)
         self._disk.decrypt()
         self._import_from_directory(MAPPER_PATH)
         self._imported = True
 
     def export(self) -> None:
         if not self._imported:
-            raise ValueError(f"Not imported {self.name}")
+            raise ValueError(NOT_IMPORTED, self.name)
         self._export_pool()
         self._disk.encrypt()
         self._imported = False
@@ -349,7 +356,7 @@ class ExternalPool(Pool):
                 print(f"Found disk {pool_name}, {disk_uuid}")
                 return cls(pool_name, disk)
         else:
-            print("Could not find a backup drive")
+            print(CANNOT_FIND_BACKUP_DRIVE)
             exit(1)
 
     # def datasets_to_backup(self, do_not_backup: list[str]) -> list[Dataset]:
@@ -425,7 +432,7 @@ class Snapshot:
         return self._path.snapname
 
     def matches_regex(self, regex: str) -> bool:
-        return bool(re.search(regex, str(self._path)))
+        return bool(search(regex, str(self._path)))
 
     def destroy(self):
         ZFS.destroy(str(self))
@@ -440,18 +447,18 @@ class Snapshot:
 
 
 def run_command(cmdline: list[str]):
-    subprocess.call(cmdline)
+    call(cmdline)
 
 
 def get_output_of_command(cmdline: str | list[str]) -> str:
     if isinstance(cmdline, str):
         cmdline = [cmdline]
-    return subprocess.check_output(cmdline).strip().decode()
+    return check_output(cmdline).strip().decode()
 
 
 def verify_running_as_root():
-    if os.getuid() > 0:
+    if getuid() > 0:
-        print("Run as root.")
+        print(RUN_AS_ROOT)
         exit(1)