inline all messages

zfs_backup/commands/cryptsetup.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from pathlib import Path
 
 from ._core import CommandInterface
-from ..messages import MAPPER_ENTRY_ALREADY_EXISTS, CANNOT_FIND_DEVICE_PATH
 from ..misc import MAPPER_PATH, DISK_BY_UUID
 
 
@@ -31,8 +30,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)
+            raise FileExistsError("mapper entry already exists")
-            raise FileExistsError()
         cls._run_command(cls._Subcommands.open, str(path), mapper_entry)
         return False
 
@@ -71,8 +69,9 @@ class Cryptsetup(CommandInterface):
                 device = Path(line.split(":")[-1].strip())
                 break
         else:
-            raise FileNotFoundError(CANNOT_FIND_DEVICE_PATH)
+            raise FileNotFoundError(
-        # noinspection PyUnboundLocalVariable
+                "Cannot find device path from output of `cryptsetup status`"
+            )
         return device
 
     @classmethod

zfs_backup/commands/zfs.py

@@ -6,7 +6,6 @@ from subprocess import Popen
 from .. import Pool
 from ._core import CommandInterface
 from ..dataset import Dataset
-from ..messages import SAME_DATASET
 from ..misc import assert_type
 from ..snapshot import Snapshot
 from ..zfs_path import ZFSPath
@@ -90,7 +89,9 @@ 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(
+                "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset"
+            )
         if old_snapshot.newer_than(new_snapshot):
             raise ValueError("Old snapshot is newer than new snapshot.")
         if old_snapshot == new_snapshot:

zfs_backup/disk.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 from pathlib import Path
 from uuid import UUID
 
-from .messages import ALREADY_DECRYPTED, NOT_DECRYPTED
 from .commands.cryptsetup import Cryptsetup
 from .misc import DISK_BY_UUID
 
@@ -18,7 +17,7 @@ class Disk:
     def decrypt(self) -> None:
         """Open a LUKS-encrypted disk with `cryptsetup`. Silently pass if it was already decrypted."""
         if self._decrypted:
-            raise ValueError(ALREADY_DECRYPTED, self._name)
+            raise ValueError("Cannot decrypt, already decrypted", self._name)
         self._was_already_decrypted = Cryptsetup.decrypt(self._path, self._mapper_entry)
         self._decrypted = True
 
@@ -35,7 +34,7 @@ class Disk:
     def encrypt(self) -> None:
         """Close a LUKS-encrypted container with `cryptsetup`."""
         if not self._decrypted:
-            raise ValueError(NOT_DECRYPTED, self._name)
+            raise ValueError("Cannot encrypt, not decrypted", self._name)
         if self._was_already_decrypted:
             return
         Cryptsetup.encrypt(self._mapper_entry)

zfs_backup/messages.py

@@ -1,13 +0,0 @@
-from __future__ import annotations
-
-
-ALREADY_DECRYPTED = "Cannot decrypt, already decrypted"
-ALREADY_IMPORTED = "Cannot import, already imported"
-CANNOT_FIND_DEVICE_PATH = "Cannot find device path from output of `cryptsetup status`"
-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"
-SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset"

zfs_backup/pool.py

@@ -4,7 +4,6 @@ from datetime import timedelta
 from uuid import UUID
 
 from .disk import Disk
-from .messages import CANNOT_FIND_BACKUP_DRIVE, ALREADY_IMPORTED, NOT_IMPORTED
 from .misc import MAPPER_PATH, DISK_BY_UUID
 from .commands.zfs import ZFS
 from .dataset import Dataset
@@ -39,7 +38,7 @@ class ExternalPool(Pool):
     def import_(self) -> None:
         """Import a pool. Raise an error if the pool was previously imported."""
         if self._imported:
-            raise ValueError(ALREADY_IMPORTED, self.name)
+            raise ValueError("Cannot import, already imported", self.name)
         self._disk.decrypt()
         ZPool.import_from_directory(self, MAPPER_PATH)
         self._imported = True
@@ -47,7 +46,7 @@ class ExternalPool(Pool):
     def export(self) -> None:
         """Export a previously imported pool. Raise an error if the pool was not previously imported."""
         if not self._imported:
-            raise ValueError(NOT_IMPORTED, self.name)
+            raise ValueError("Cannot export, not imported", self.name)
         ZPool.export(self)
         self._disk.encrypt()
         self._imported = False
@@ -73,7 +72,7 @@ class ExternalPool(Pool):
                 print(f"Found disk {pool_name}, {disk_uuid}")
                 return cls(pool_name, disk)
         else:
-            raise FileNotFoundError(CANNOT_FIND_BACKUP_DRIVE)
+            raise FileNotFoundError("Could not find a backup drive")
 
     def clean_old_snapshots(self) -> None:
         print("Cleaning old snapshots")

zfs_backup/zfs_path.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from zfs_backup.messages import NOT_VALID_ZFS_PATH
-
 
 class ZFSPath:
     """Represents a path in ZFS.
@@ -79,4 +77,4 @@ class ZFSPath:
     def _sanity_check(elements: list[str]) -> None:
         for item in elements:
             if len(item) == 0:
-                raise ValueError(NOT_VALID_ZFS_PATH)
+                raise ValueError("Not a valid ZFSPath")