timeshifter/zfs-backup
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

add: Messages

Browse Source
This commit is contained in:
timeshifter
2021-12-04 23:15:21 +01:00
parent d22108d20a
commit 20b9b7eb8b
1 changed files with 23 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zfs-backup.py
+23 -21
View File
@@ -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)