From 20b9b7eb8bb13d3821a776f4a23dcfd2a7cacd96 Mon Sep 17 00:00:00 2001 From: timeshifter Date: Sat, 4 Dec 2021 23:15:21 +0100 Subject: [PATCH] add: Messages --- zfs-backup.py | 44 +++++++++++++++++++++++--------------------- 1 file changed, 23 insertions(+), 21 deletions(-) diff --git a/zfs-backup.py b/zfs-backup.py index 02b594c..313ee97 100755 --- a/zfs-backup.py +++ b/zfs-backup.py @@ -16,17 +16,6 @@ import config EXIT_ERROR = 1 -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") @@ -254,7 +243,7 @@ class ZFSPath: def _sanity_check(self): for item in self._elements: if len(item) == 0: - raise ValueError(NOT_VALID_ZFS_PATH) + raise ValueError(Messages.NOT_VALID_ZFS_PATH) def _assert_type(instance: Any, object_type: Any, message: str) -> None: @@ -401,7 +390,7 @@ class ZFS(CommandInterface): def destroy_snapshot(cls, snapshot: Snapshot): _assert_type(snapshot, Snapshot, "Can only destroy snapshots.") if config.local_pool_to_backup == snapshot.pool: - raise Exception(DELETE_IN_LOCAL_POOL) + raise Exception(Messages.DELETE_IN_LOCAL_POOL) cls._run_command(*cls._Subcommands.destroy, str(snapshot)) @classmethod @@ -425,7 +414,7 @@ class ZFS(CommandInterface): f"Cannot send stream, object is not a Snapshot: {snapshot}", ) if old_snapshot.dataset != new_snapshot.dataset: - raise ValueError(SAME_DATASET) + raise ValueError(Messages.SAME_DATASET) if old_snapshot.newer_than(new_snapshot): raise ValueError("Old snapshot is newer than new snapshot.") if old_snapshot == new_snapshot: @@ -484,7 +473,7 @@ class Disk: def decrypt(self) -> None: if self._decrypted: - raise ValueError(ALREADY_DECRYPTED, self._name) + raise ValueError(Messages.ALREADY_DECRYPTED, self._name) self._was_already_decrypted = Cryptsetup.decrypt(self._path, self._mapper_entry) self._decrypted = True @@ -498,7 +487,7 @@ class Disk: def encrypt(self): if not self._decrypted: - raise ValueError(NOT_DECRYPTED, self._name) + raise ValueError(Messages.NOT_DECRYPTED, self._name) if self._was_already_decrypted: return Cryptsetup.encrypt(self._mapper_entry) @@ -527,7 +516,7 @@ class Cryptsetup(CommandInterface): if cls._mapper_entry_in_use(mapper_entry): if cls._already_decrypted(path, mapper_entry): return True - print(MAPPER_ENTRY_ALREADY_EXISTS) + print(Messages.MAPPER_ENTRY_ALREADY_EXISTS) exit(EXIT_ERROR) cls._run_command(cls._Subcommands.open, str(path), mapper_entry) return False @@ -596,14 +585,14 @@ class ExternalPool(Pool): def import_(self) -> None: if self._imported: - raise ValueError(ALREADY_IMPORTED, self.name) + raise ValueError(Messages.ALREADY_IMPORTED, self.name) self._disk.decrypt() ZPool.import_from_directory(self, MAPPER_PATH) self._imported = True def export(self) -> None: if not self._imported: - raise ValueError(NOT_IMPORTED, self.name) + raise ValueError(Messages.NOT_IMPORTED, self.name) self._export_pool() self._disk.encrypt() self._imported = False @@ -625,7 +614,7 @@ class ExternalPool(Pool): print(f"Found disk {pool_name}, {disk_uuid}") return cls(pool_name, disk) else: - print(CANNOT_FIND_BACKUP_DRIVE) + print(Messages.CANNOT_FIND_BACKUP_DRIVE) exit(EXIT_ERROR) def clean_old_snapshots(self): @@ -759,9 +748,22 @@ class CommandRunner: return Popen(cmdline, stdout=PIPE) +class Messages: + 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" + + def _verify_running_as_root(): if getuid() > 0: - print(RUN_AS_ROOT) + print(Messages.RUN_AS_ROOT) exit(EXIT_ERROR)