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

refactor: class Command

Browse Source
This commit is contained in:
timeshifter
2021-12-02 15:12:49 +01:00
parent 20d5fe4282
commit 9aa9205a9d
1 changed files with 19 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zfs-backup.py
+19 -19
View File
@@ -192,7 +192,7 @@ class ZFSPath:
class ZFS: class ZFS:
@staticmethod @staticmethod
def get_datasets(name: str) -> list[Dataset]: def get_datasets(name: str) -> list[Dataset]:
datasets_str = get_output_of_command( datasets_str = Command.get_output(
[ [
config.ZFS, config.ZFS,
"list", "list",
@@ -214,17 +214,17 @@ class ZFS:
def import_pool_from_directory(name: str, directory: Path) -> None: def import_pool_from_directory(name: str, directory: Path) -> None:
# -N: no mount # -N: no mount
# -d: directory to search the pool in # -d: directory to search the pool in
run_command([config.ZPOOL, "import", "-N", "-d", directory, name]) Command.run([config.ZPOOL, "import", "-N", "-d", directory, name])
@staticmethod @staticmethod
def export_pool(name: str) -> None: def export_pool(name: str) -> None:
run_command([config.ZPOOL, "export", name]) Command.run([config.ZPOOL, "export", name])
@staticmethod @staticmethod
def get_snapshots(name: str) -> list[Snapshot]: def get_snapshots(name: str) -> list[Snapshot]:
return [ return [
Snapshot(ZFSPath.from_string(name)) Snapshot(ZFSPath.from_string(name))
for name in get_output_of_command( for name in Command.get_output(
[ [
config.ZFS, config.ZFS,
"list", "list",
@@ -251,7 +251,7 @@ class ZFS:
if recursive: if recursive:
cmdline.append("-r") cmdline.append("-r")
cmdline.append(str(snapshot)) cmdline.append(str(snapshot))
run_command(cmdline) Command.run(cmdline)
@staticmethod @staticmethod
def send_incremental(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]: def send_incremental(old_snap: Snapshot, new_snap: Snapshot) -> Popen[str]:
@@ -283,9 +283,9 @@ class ZFS:
if not isinstance(pool, Pool): if not isinstance(pool, Pool):
raise TypeError("Can only scrub pools.") raise TypeError("Can only scrub pools.")
if not cls._scrub_in_progress(pool): if not cls._scrub_in_progress(pool):
run_command([config.ZPOOL, "scrub", pool.name]) Command.run([config.ZPOOL, "scrub", pool.name])
if wait_for_finish: if wait_for_finish:
run_command([config.ZPOOL, "wait", "-t", "scrub", pool.name]) Command.run([config.ZPOOL, "wait", "-t", "scrub", pool.name])
if not cls._healthy(pool): if not cls._healthy(pool):
raise IOError(f"Pool {pool.name} is not healthy.") raise IOError(f"Pool {pool.name} is not healthy.")
@@ -299,7 +299,7 @@ class ZFS:
@staticmethod @staticmethod
def _pool_status_output(pool): def _pool_status_output(pool):
return get_output_of_command([config.ZFS, "status", pool.name]) return Command.get_output([config.ZFS, "status", pool.name])
class Disk: class Disk:
@@ -316,7 +316,7 @@ class Disk:
self._decrypted = True self._decrypted = True
def _decrypt_with_cryptsetup(self): def _decrypt_with_cryptsetup(self):
run_command([config.CRYPTSETUP, "open", self._path, self._mapper_entry]) Command.run([config.CRYPTSETUP, "open", self._path, self._mapper_entry])
def _verify_mapper_entry_not_in_use(self): def _verify_mapper_entry_not_in_use(self):
for item in MAPPER_PATH.iterdir(): for item in MAPPER_PATH.iterdir():
@@ -336,8 +336,7 @@ class Disk:
def encrypt(self): def encrypt(self):
if not self._decrypted: if not self._decrypted:
raise ValueError(NOT_DECRYPTED, self._name) raise ValueError(NOT_DECRYPTED, self._name)
# TODO use run instead? Command.run([config.CRYPTSETUP, "close", self._mapper_entry])
get_output_of_command([config.CRYPTSETUP, "close", self._mapper_entry])
self._decrypted = False self._decrypted = False
@@ -499,15 +498,16 @@ class Snapshot:
return Dataset(self._path.dataset) return Dataset(self._path.dataset)
# TODO class CommandRunner class Command:
def run_command(cmdline: list[str]): @staticmethod
call(cmdline) def run(cmdline: list[str]):
call(cmdline)
@staticmethod
def get_output_of_command(cmdline: str | list[str]) -> str: def get_output(cmdline: str | list[str]) -> str:
if isinstance(cmdline, str): if isinstance(cmdline, str):
cmdline = [cmdline] cmdline = [cmdline]
return check_output(cmdline).strip().decode() return check_output(cmdline).strip().decode()
def verify_running_as_root(): def verify_running_as_root():