From 1c895e463fe5f2c9301b4f3c73e9b31a8199c7bb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 20:38:18 +0200 Subject: [PATCH 1/9] test: Rename a shell-interface test file --- .../tests/{test_shell_interface.py => test_run_cmd.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename projects/shell-interface/tests/{test_shell_interface.py => test_run_cmd.py} (100%) diff --git a/projects/shell-interface/tests/test_shell_interface.py b/projects/shell-interface/tests/test_run_cmd.py similarity index 100% rename from projects/shell-interface/tests/test_shell_interface.py rename to projects/shell-interface/tests/test_run_cmd.py From fbd303c1da138c0d9528e266f5d4c8ce0f3e7c68 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:02:50 +0200 Subject: [PATCH 2/9] refactor: Move ensure_directory to shell-interface --- .../butter-backup/src/butter_backup/cli.py | 2 +- .../src/shell_interface/__init__.py | 2 + .../src/shell_interface/shell_interface.py | 25 ++++++++ .../tests/test_ensure_directory.py | 61 +++++++++++++++++++ .../src/storage_device_managers/__init__.py | 28 +-------- .../tests/test_ensure_directory.py | 59 ------------------ 6 files changed, 90 insertions(+), 87 deletions(-) create mode 100644 projects/shell-interface/tests/test_ensure_directory.py delete mode 100644 projects/storage-device-managers/tests/test_ensure_directory.py diff --git a/projects/butter-backup/src/butter_backup/cli.py b/projects/butter-backup/src/butter_backup/cli.py index 57c32973..9d5ebd54 100644 --- a/projects/butter-backup/src/butter_backup/cli.py +++ b/projects/butter-backup/src/butter_backup/cli.py @@ -145,7 +145,7 @@ def _open_device( topmost_created_ancestor = None try: _refresh_sudo(sudo_pass_cmd) - topmost_created_ancestor = sdm.ensure_directory(mount_dir) + topmost_created_ancestor = sh.ensure_directory(mount_dir) decrypted = sdm.open_encrypted_device(cfg.device(), cfg.DevicePassCmd) sdm.mount_device(decrypted, mount_dir=mount_dir, compression=cfg.compression()) except Exception: diff --git a/projects/shell-interface/src/shell_interface/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index 8fc8af1b..6d6840d3 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -4,6 +4,7 @@ PassCmdError, ShellInterfaceError, StrPathList, + ensure_directory, get_group, get_user, pipe_pass_cmd_to_real_cmd, @@ -15,6 +16,7 @@ "PassCmdError", "ShellInterfaceError", "StrPathList", + "ensure_directory", "get_group", "get_user", "pipe_pass_cmd_to_real_cmd", diff --git a/projects/shell-interface/src/shell_interface/shell_interface.py b/projects/shell-interface/src/shell_interface/shell_interface.py index 9897aaa4..431d03c2 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -134,6 +134,31 @@ def pipe_pass_cmd_to_real_cmd( return completed_process +def ensure_directory(directory: Path) -> Path | None: + """Ensure a directory exists, creating it with root privileges if needed. + + Parameters: + ----------- + directory + directory that should exist + + Returns: + -------- + Path | None + The first missing ancestor that had to be created, or ``None`` if the + directory already existed + """ + if directory.is_dir(): + return None + first_created = next( + (parent for parent in reversed(directory.parents) if not parent.is_dir()), + directory, + ) + cmd: StrPathList = ["sudo", "mkdir", "-p", directory] + run_cmd(cmd=cmd) + return first_created + + def get_user() -> str: """Get user who started ButterBackup diff --git a/projects/shell-interface/tests/test_ensure_directory.py b/projects/shell-interface/tests/test_ensure_directory.py new file mode 100644 index 00000000..ab27b55a --- /dev/null +++ b/projects/shell-interface/tests/test_ensure_directory.py @@ -0,0 +1,61 @@ +import typing as t +from pathlib import Path + +import pytest + +import shell_interface as sh + + +def sudo_mkdir(path: Path) -> None: + """ + Create a directory with sudo privileges + """ + sudo_mkdir_cmd: sh.StrPathList = ["sudo", "mkdir", "-p", path] + sh.run_cmd(cmd=sudo_mkdir_cmd) + + +@pytest.fixture +def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: + """ + Create a temporary directory owned by root for testing + """ + root_owned_path = tmp_path / "root_owned" + root_owned_path.mkdir() + current_user = sh.get_user() + current_group = sh.get_group(current_user) + chown_to_root: sh.StrPathList = ["sudo", "chown", "root:root", root_owned_path] + chown_to_user: sh.StrPathList = [ + "sudo", + "chown", + f"{current_user}:{current_group}", + root_owned_path, + ] + sh.run_cmd(cmd=chown_to_root) + try: + yield root_owned_path + finally: + sh.run_cmd(cmd=chown_to_user) + + +@pytest.mark.parametrize("destination", ["destDir", "dest dir with spaces"]) +@pytest.mark.parametrize( + "first_created", + [None, "subdir", "sub dir/with spaces/", "very/deeply/nested/subdir"], +) +def test_ensure_directory_creates_directory( + root_owned_tmp_path: Path, first_created: str | None, destination: str +) -> None: + if first_created is None: + first_created_p = root_owned_tmp_path / destination + destination_p = root_owned_tmp_path / destination + else: + first_created_p = root_owned_tmp_path / first_created + destination_p = root_owned_tmp_path / first_created / destination + sudo_mkdir(first_created_p.parent) + + result_created = sh.ensure_directory(destination_p) + result_noop = sh.ensure_directory(destination_p) + assert destination_p is not None + assert destination_p.is_dir() + assert result_noop is None + assert result_created == first_created_p diff --git a/projects/storage-device-managers/src/storage_device_managers/__init__.py b/projects/storage-device-managers/src/storage_device_managers/__init__.py index 6ad55f32..95bcae79 100644 --- a/projects/storage-device-managers/src/storage_device_managers/__init__.py +++ b/projects/storage-device-managers/src/storage_device_managers/__init__.py @@ -36,7 +36,6 @@ "close_decrypted_device", "decrypted_device", "encrypt_device", - "ensure_directory", "generate_passcmd", "get_filesystem", "get_mounted_devices", @@ -166,31 +165,6 @@ def decrypted_device(device: Path, pass_cmd: str) -> Iterator[Path]: ) -def ensure_directory(directory: Path) -> Path | None: - """Ensure a directory exists, creating it with root privileges if needed. - - Parameters: - ----------- - directory - directory that should exist - - Returns: - -------- - Path | None - The first missing ancestor that had to be created, or ``None`` if the - directory already existed - """ - if directory.is_dir(): - return None - first_created = next( - (parent for parent in reversed(directory.parents) if not parent.is_dir()), - directory, - ) - cmd: sh.StrPathList = ["sudo", "mkdir", "-p", directory] - sh.run_cmd(cmd=cmd) - return first_created - - @contextlib.contextmanager def mounted_device( device: Path, @@ -234,7 +208,7 @@ def mounted_device( if is_mounted(device): unmount_device(device) if destination is not None: - ensure_directory(destination) + sh.ensure_directory(destination) ctx: contextlib.AbstractContextManager[Path] = ( _temporary_directory() if destination is None diff --git a/projects/storage-device-managers/tests/test_ensure_directory.py b/projects/storage-device-managers/tests/test_ensure_directory.py deleted file mode 100644 index aa413f23..00000000 --- a/projects/storage-device-managers/tests/test_ensure_directory.py +++ /dev/null @@ -1,59 +0,0 @@ -import typing as t -from pathlib import Path - -import pytest -import shell_interface as sh - -import storage_device_managers as sdm - - -@pytest.fixture -def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: - """ - Create a temporary directory owned by root for testing - """ - root_owned_path = tmp_path / "root_owned" - root_owned_path.mkdir() - current_user = sh.get_user() - current_group = sh.get_group(current_user) - chown_to_root: sh.StrPathList = ["sudo", "chown", "root:root", root_owned_path] - chown_to_user: sh.StrPathList = [ - "sudo", - "chown", - f"{current_user}:{current_group}", - root_owned_path, - ] - sh.run_cmd(cmd=chown_to_root) - try: - yield root_owned_path - finally: - sh.run_cmd(cmd=chown_to_user) - - -def test_ensure_directory_creates_missing_directory(tmp_path: Path) -> None: - destination = tmp_path / "mount" - assert not destination.exists() - assert sdm.ensure_directory(destination) == destination - assert destination.exists() - - -def test_ensure_directory_returns_first_created_ancestor(tmp_path: Path) -> None: - expected = tmp_path / "grandparent" - destination = expected / "parent" / "child" - assert not destination.exists() - assert sdm.ensure_directory(destination) == expected - assert destination.exists() - - -def test_ensure_directory_creates_missing_directory_in_root_owned_path( - root_owned_tmp_path: Path, -) -> None: - expected = root_owned_tmp_path / "nested" - destination = expected / "mount" - assert not destination.exists() - assert sdm.ensure_directory(destination) == expected - assert destination.exists() - - -def test_ensure_directory_skips_existing_directory(tmp_path: Path) -> None: - assert sdm.ensure_directory(tmp_path) is None From 87f1b27223d1941ce907f62a2f23f651fdb2e370 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:02:50 +0200 Subject: [PATCH 3/9] refactor: Move chown to shell-interface --- .../src/butter_backup/backup_backends.py | 5 +- .../src/butter_backup/device_managers.py | 7 ++- .../test_btrfs_rsync_backend.py | 5 +- .../tests/cli/test_sudo_pass_cmd.py | 11 +---- .../src/shell_interface/__init__.py | 2 + .../src/shell_interface/shell_interface.py | 47 ++++++++++++++++++ .../tests/test_chown.py | 11 ++--- .../tests/test_ensure_directory.py | 13 ++--- .../src/storage_device_managers/__init__.py | 48 ------------------- .../tests/test_mount_unmount.py | 2 +- 10 files changed, 68 insertions(+), 83 deletions(-) rename projects/{storage-device-managers => shell-interface}/tests/test_chown.py (89%) diff --git a/projects/butter-backup/src/butter_backup/backup_backends.py b/projects/butter-backup/src/butter_backup/backup_backends.py index a57ca468..ee7b9572 100644 --- a/projects/butter-backup/src/butter_backup/backup_backends.py +++ b/projects/butter-backup/src/butter_backup/backup_backends.py @@ -5,7 +5,6 @@ from typing import overload import shell_interface as sh -import storage_device_managers as sdm from loguru import logger from . import config_parser as cp @@ -90,7 +89,7 @@ def adapt_ownership(snapshot_root: Path) -> None: # correct ownership. # Therefore, it is believed that writing test that fails if `recursive=True` is # currently impossible. - sdm.chown(snapshot_root, user, group, recursive=False) + sh.chown(snapshot_root, user, group, recursive=False) def snapshot(self, *, src: Path, backup_repository: Path) -> Path: timestamp = dt.datetime.now() @@ -165,7 +164,7 @@ def adapt_ownership(backup_repository: Path) -> None: user=user, group=group, ) - sdm.chown(backup_repository, user, group, recursive=True) + sh.chown(backup_repository, user, group, recursive=True) def copy_files(self, backup_repository: Path) -> None: restic_cmd: sh.StrPathList = [ diff --git a/projects/butter-backup/src/butter_backup/device_managers.py b/projects/butter-backup/src/butter_backup/device_managers.py index 928410a1..b7963f35 100644 --- a/projects/butter-backup/src/butter_backup/device_managers.py +++ b/projects/butter-backup/src/butter_backup/device_managers.py @@ -35,7 +35,7 @@ def prepare_device_for_butterbackend(device: Path) -> cp.BtrFSRsyncConfig: initial_subvol, ] sh.run_cmd(cmd=subvol_cmd) - sdm.chown(mounted, user, group, recursive=True) + sh.chown(mounted, user, group, recursive=True) config = cp.BtrFSRsyncConfig( BackupRepositoryFolder=backup_repository_folder, @@ -63,7 +63,6 @@ def prepare_device_for_resticbackend( sdm.mkfs(decrypted, file_system) with sdm.mounted_device(decrypted) as mounted: backup_repo = mounted / backup_repository_folder - mkdir_repo: sh.StrPathList = ["sudo", "mkdir", backup_repo] restic_init: sh.StrPathList = [ "sudo", "restic", @@ -71,9 +70,9 @@ def prepare_device_for_resticbackend( "-r", backup_repo, ] - sh.run_cmd(cmd=mkdir_repo) + sh.ensure_directory(backup_repo) sh.pipe_pass_cmd_to_real_cmd(repository_passcmd, restic_init) - sdm.chown(mounted, user, group, recursive=True) + sh.chown(mounted, user, group, recursive=True) config = cp.ResticConfig( BackupRepositoryFolder=backup_repository_folder, DevicePassCmd=device_passcmd, diff --git a/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py b/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py index cf1597ac..caa8a5a6 100644 --- a/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py +++ b/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py @@ -8,7 +8,6 @@ import pytest import shell_interface as sh -import storage_device_managers as sdm from butter_backup import backup_backends as bb from butter_backup import config_parser as cp @@ -65,7 +64,7 @@ def test_btrfs_backend_gracefully_handles_existing_snapshots_owned_by_root( latest_snapshot = sorted(snapshot_root.iterdir())[-1] for cur in itertools.chain(snapshot_root.glob("*"), snapshot_root.glob("*/*")): print(f"Changing ownership of {cur} to root:root") - sdm.chown(cur, "root", "root", recursive=False) + sh.chown(cur, "root", "root", recursive=False) sh.run_cmd(cmd=["sudo", "rm", "-rf", latest_snapshot / first_config.FilesDest]) second_config = run_backup_cycle(empty_config, second_source, device) @@ -140,7 +139,7 @@ def test_do_backup_for_btrfs_rsync_preserves_ownership_of_source_files( source_dir = tmp_path / cur.name shutil.copytree(cur, source_dir) rm_cmd: sh.StrPathList = ["sudo", "rm", "-r", source_dir] - sdm.chown(source_dir, f"{test_owner_uid}", f"{test_group_gid}", recursive=True) + sh.chown(source_dir, f"{test_owner_uid}", f"{test_group_gid}", recursive=True) config = run_backup_cycle(empty_config, source_dir, device) # remove source dir to avoid permission issues with pytest and "user" 1337 sh.run_cmd(cmd=rm_cmd) diff --git a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py index dc9879ce..f4cdf8bc 100644 --- a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py +++ b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py @@ -33,18 +33,11 @@ def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: root_owned_path.mkdir() current_user = sh.get_user() current_group = sh.get_group(current_user) - chown_to_root: sh.StrPathList = ["sudo", "chown", "root:root", root_owned_path] - chown_to_user: sh.StrPathList = [ - "sudo", - "chown", - f"{current_user}:{current_group}", - root_owned_path, - ] - sh.run_cmd(cmd=chown_to_root) + sh.chown(root_owned_path, "root", "root", recursive=False) try: yield root_owned_path finally: - sh.run_cmd(cmd=chown_to_user) + sh.chown(root_owned_path, current_user, current_group, recursive=False) def _invalidate_sudo_session() -> None: diff --git a/projects/shell-interface/src/shell_interface/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index 6d6840d3..f460a0e3 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -4,6 +4,7 @@ PassCmdError, ShellInterfaceError, StrPathList, + chown, ensure_directory, get_group, get_user, @@ -16,6 +17,7 @@ "PassCmdError", "ShellInterfaceError", "StrPathList", + "chown", "ensure_directory", "get_group", "get_user", diff --git a/projects/shell-interface/src/shell_interface/shell_interface.py b/projects/shell-interface/src/shell_interface/shell_interface.py index 431d03c2..f56025c5 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -186,3 +186,50 @@ def get_group(user: str) -> str: raw_group = run_cmd(cmd=["id", "-gn", user], capture_output=True) group = raw_group.stdout.decode().splitlines()[0] return group + + +def chown( + file_or_folder: Path, + /, + user: int | str, + group: int | str | None = None, + *, + recursive: bool, +) -> None: + """Change user and group of a device or folder + + This function will change the ownership as specified. It requires root + privileges and will ask for them if not available. If no group is given, + only the owner is changed. + + If recursive is true, ownership information of all files and folders + contained by `file_or_folder` will be adapted. + + If `file_or_folder` points to a file, `recursive` must be `False`. + Otherwise a ValueError will be raised. + + + Parameters: + ----------- + user + user ID, either as name or as UID + group + group ID, either as name or as GID + recursive + whether or not to change ownership for content + + Raises: + -------- + ValueError + if `file_or_folder` is a file but `recursive` is `True` + """ + if file_or_folder.is_file() and recursive: + raise ValueError( + "First argument must point to a directory if `recursive` is `True`!" + ) + + user_spec = str(user) if group is None else f"{user}:{group}" + chown_cmd: StrPathList = ["sudo", "chown", user_spec, file_or_folder] + if recursive: + chown_cmd.append("--recursive") + run_cmd(cmd=chown_cmd) diff --git a/projects/storage-device-managers/tests/test_chown.py b/projects/shell-interface/tests/test_chown.py similarity index 89% rename from projects/storage-device-managers/tests/test_chown.py rename to projects/shell-interface/tests/test_chown.py index 0fc3a92e..e3518999 100644 --- a/projects/storage-device-managers/tests/test_chown.py +++ b/projects/shell-interface/tests/test_chown.py @@ -4,9 +4,8 @@ from tempfile import NamedTemporaryFile import pytest -import shell_interface as sh -import storage_device_managers as sdm +import shell_interface as sh @pytest.fixture @@ -30,14 +29,14 @@ def test_chown_raises_valueerror(): NamedTemporaryFile() as temp_file, ): file = Path(temp_file.name) - sdm.chown(file, file.owner(), recursive=True) + sh.chown(file, file.owner(), recursive=True) def test_chown_file(directory_with_content): _, file = directory_with_content assert file.owner() == "root" expected_user = sh.get_user() - sdm.chown(file, expected_user, recursive=False) + sh.chown(file, expected_user, recursive=False) result_user = file.owner() assert result_user == expected_user @@ -53,7 +52,7 @@ def test_chown_recursive(directory_with_content): expected_user = sh.get_user() expected_group = sh.get_group(expected_user) - sdm.chown(directory, expected_user, expected_group, recursive=True) + sh.chown(directory, expected_user, expected_group, recursive=True) result_users = {cur.owner() for cur in all_files_and_folders} result_group = {cur.group() for cur in all_files_and_folders} @@ -70,7 +69,7 @@ def test_chown_directory_not_recursive(directory_with_content): initial_nested_owners = {cur.owner() for cur in nested_items} assert initial_nested_owners == expected_nested_owners - sdm.chown(directory, expected_user, recursive=False) + sh.chown(directory, expected_user, recursive=False) result_root_owner = directory.owner() result_nested_owners = {cur.owner() for cur in nested_items} diff --git a/projects/shell-interface/tests/test_ensure_directory.py b/projects/shell-interface/tests/test_ensure_directory.py index ab27b55a..1267bcea 100644 --- a/projects/shell-interface/tests/test_ensure_directory.py +++ b/projects/shell-interface/tests/test_ensure_directory.py @@ -23,18 +23,13 @@ def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: root_owned_path.mkdir() current_user = sh.get_user() current_group = sh.get_group(current_user) - chown_to_root: sh.StrPathList = ["sudo", "chown", "root:root", root_owned_path] - chown_to_user: sh.StrPathList = [ - "sudo", - "chown", - f"{current_user}:{current_group}", - root_owned_path, - ] - sh.run_cmd(cmd=chown_to_root) + sh.chown(root_owned_path, user="root", group="root", recursive=False) try: yield root_owned_path finally: - sh.run_cmd(cmd=chown_to_user) + sh.chown( + root_owned_path, user=current_user, group=current_group, recursive=False + ) @pytest.mark.parametrize("destination", ["destDir", "dest dir with spaces"]) diff --git a/projects/storage-device-managers/src/storage_device_managers/__init__.py b/projects/storage-device-managers/src/storage_device_managers/__init__.py index 95bcae79..a3e009b8 100644 --- a/projects/storage-device-managers/src/storage_device_managers/__init__.py +++ b/projects/storage-device-managers/src/storage_device_managers/__init__.py @@ -32,7 +32,6 @@ "UnmountError", "ValidCompressions", "ValidFileSystems", - "chown", "close_decrypted_device", "decrypted_device", "encrypt_device", @@ -625,50 +624,3 @@ def generate_passcmd() -> str: alphabet = string.ascii_letters + string.digits passphrase = "".join(secrets.choice(alphabet) for _ in range(n_chars)) return f"echo {passphrase}" - - -def chown( - file_or_folder: Path, - /, - user: int | str, - group: int | str | None = None, - *, - recursive: bool, -) -> None: - """Change user and group of a device or folder - - This function will change the ownership as specified. It requires root - privileges and will ask for them if not available. If no group is given, - only the owner is changed. - - If recursive is true, ownership information of all files and folders - contained by `file_or_folder` will be adapted. - - If `file_or_folder` points to a file, `recursive` must be `False`. - Otherwise a ValueError will be raised. - - - Parameters: - ----------- - user - user ID, either as name or as UID - group - group ID, either as name or as GID - recursive - whether or not to change ownership for content - - Raises: - -------- - ValueError - if `file_or_folder` is a file but `recursive` is `True` - """ - if file_or_folder.is_file() and recursive: - raise ValueError( - "First argument must point to a directory if `recursive` is `True`!" - ) - - user_spec = str(user) if group is None else f"{user}:{group}" - chown_cmd: sh.StrPathList = ["sudo", "chown", user_spec, file_or_folder] - if recursive: - chown_cmd.append("--recursive") - sh.run_cmd(cmd=chown_cmd) diff --git a/projects/storage-device-managers/tests/test_mount_unmount.py b/projects/storage-device-managers/tests/test_mount_unmount.py index 096e5ee3..3c399557 100644 --- a/projects/storage-device-managers/tests/test_mount_unmount.py +++ b/projects/storage-device-managers/tests/test_mount_unmount.py @@ -158,7 +158,7 @@ def test_mounted_device_does_not_delete_content_on_umount_error( sdm.mounted_device(device, **compression_kwargs) as md, ): sentinel = md / "sentinel-file" - sdm.chown(md, user, recursive=True) + sh.chown(md, user, recursive=True) sentinel.write_text("This file should not be deleted.") assert sentinel.exists(), "Sentinel file was deleted after unmount error." assert sentinel.read_text() == sentinel_text From aabca2c3b3c0d54fb36a658b682974f444b8846c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:19:28 +0200 Subject: [PATCH 4/9] refactor: Replace two manual mkdir with ensure_directory --- projects/butter-backup/src/butter_backup/device_managers.py | 3 +-- projects/shell-interface/tests/test_chown.py | 3 +-- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/projects/butter-backup/src/butter_backup/device_managers.py b/projects/butter-backup/src/butter_backup/device_managers.py index b7963f35..c52e5e6d 100644 --- a/projects/butter-backup/src/butter_backup/device_managers.py +++ b/projects/butter-backup/src/butter_backup/device_managers.py @@ -21,8 +21,7 @@ def prepare_device_for_butterbackend(device: Path) -> cp.BtrFSRsyncConfig: sdm.mkfs(decrypted, "btrfs") with sdm.mounted_device(decrypted) as mounted: backup_repository = mounted / backup_repository_folder - mkdir_cmd: sh.StrPathList = ["sudo", "mkdir", backup_repository] - sh.run_cmd(cmd=mkdir_cmd) + sh.ensure_directory(backup_repository) initial_subvol = backup_repository / date.today().strftime( cp.BtrFSRsyncConfig.SubvolTimestampFmt diff --git a/projects/shell-interface/tests/test_chown.py b/projects/shell-interface/tests/test_chown.py index e3518999..803e4ed2 100644 --- a/projects/shell-interface/tests/test_chown.py +++ b/projects/shell-interface/tests/test_chown.py @@ -11,8 +11,7 @@ @pytest.fixture def directory_with_content(tmp_path): subfolder = tmp_path / "subfolder" - mkdir_cmd: sh.StrPathList = ["sudo", "mkdir", subfolder] - sh.run_cmd(cmd=mkdir_cmd) + sh.ensure_directory(subfolder) file = tmp_path / "important-file" touch_cmd: sh.StrPathList = ["sudo", "touch", file] From 713506b5e28b22a9decfd1954101b35f08fec0cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:34:34 +0200 Subject: [PATCH 5/9] refactor: Move refresh_sudo to shell_interface Moving this helper caused some mock based tests to fail. They were rewritten to stop using mocking. Instead, they use a spy to ensure the expected refresh commands were issued. Working on this revealed an edge-case bug. When no files or folders were configured to be backed up, the restic backend crashed due to a crash of restic. This is prevented by a guard now. --- .../src/butter_backup/backup_backends.py | 23 ++-- .../butter-backup/src/butter_backup/cli.py | 15 +-- .../test_btrfs_rsync_backend.py | 2 +- .../backup_backends/test_restic_backend.py | 2 +- .../tests/cli/test_sudo_pass_cmd.py | 104 +++++++++++------- .../src/shell_interface/__init__.py | 2 + .../src/shell_interface/shell_interface.py | 20 ++++ 7 files changed, 105 insertions(+), 63 deletions(-) diff --git a/projects/butter-backup/src/butter_backup/backup_backends.py b/projects/butter-backup/src/butter_backup/backup_backends.py index ee7b9572..acbc8571 100644 --- a/projects/butter-backup/src/butter_backup/backup_backends.py +++ b/projects/butter-backup/src/butter_backup/backup_backends.py @@ -10,13 +10,6 @@ from . import config_parser as cp -def _refresh_sudo(sudo_pass_cmd: str | None) -> None: - if sudo_pass_cmd is not None: - sh.pipe_pass_cmd_to_real_cmd( - sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True - ) - - class BackupBackend(abc.ABC): @abc.abstractmethod def do_backup(self, mount_dir: Path, sudo_pass_cmd: str | None = None) -> None: ... @@ -51,12 +44,12 @@ def do_backup(self, mount_dir: Path, sudo_pass_cmd: str | None = None) -> None: backup_root = self.snapshot( src=src_snapshot, backup_repository=backup_repository ) - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) self.adapt_ownership(backup_root) for src, dest_name in self.config.Folders.items(): dest = backup_root / dest_name - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) self.rsync_folder(src, dest, self.config.ExcludePatternsFile) files_dest = backup_root / self.config.FilesDest @@ -64,7 +57,7 @@ def do_backup(self, mount_dir: Path, sudo_pass_cmd: str | None = None) -> None: files_dest.unlink() files_dest.mkdir(parents=True, exist_ok=True) for src in self.config.Files: - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) self.rsync_file(src, files_dest) @staticmethod @@ -149,9 +142,9 @@ class ResticBackend(BackupBackend): def do_backup(self, mount_dir: Path, sudo_pass_cmd: str | None = None) -> None: logger.info(f"Beginne mit Restic-Backup für Speichermedium {self.config.Name}.") backup_repository = mount_dir / self.config.BackupRepositoryFolder - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) self.copy_files(backup_repository) - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) self.adapt_ownership(backup_repository) @staticmethod @@ -167,6 +160,12 @@ def adapt_ownership(backup_repository: Path) -> None: sh.chown(backup_repository, user, group, recursive=True) def copy_files(self, backup_repository: Path) -> None: + if len(self.config.FilesAndFolders) == 0: + logger.warning( + "Es wurden keine Dateien oder Ordner zum Sichern angegeben. " + "Es wird kein Backup durchgeführt." + ) + return restic_cmd: sh.StrPathList = [ "sudo", "restic", diff --git a/projects/butter-backup/src/butter_backup/cli.py b/projects/butter-backup/src/butter_backup/cli.py index 9d5ebd54..1dd7ef77 100644 --- a/projects/butter-backup/src/butter_backup/cli.py +++ b/projects/butter-backup/src/butter_backup/cli.py @@ -99,13 +99,6 @@ def _get_default_file_system(backend: ValidBackends) -> ValidFileSystems: t.assert_never(backend) -def _refresh_sudo(sudo_pass_cmd: str | None) -> None: - if sudo_pass_cmd is not None: - sh.pipe_pass_cmd_to_real_cmd( - sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True - ) - - def _skip_device( config: cp.DeviceConfiguration, *, @@ -144,7 +137,7 @@ def _open_device( mount_dir = base_dir / cfg.Name topmost_created_ancestor = None try: - _refresh_sudo(sudo_pass_cmd) + sh.refresh_sudo(sudo_pass_cmd) topmost_created_ancestor = sh.ensure_directory(mount_dir) decrypted = sdm.open_encrypted_device(cfg.device(), cfg.DevicePassCmd) sdm.mount_device(decrypted, mount_dir=mount_dir, compression=cfg.compression()) @@ -245,7 +238,7 @@ def close( device=cfg.Name, ) continue - _refresh_sudo(parsed_config.SudoPassCmd) + sh.refresh_sudo(parsed_config.SudoPassCmd) sdm.unmount_device(map_name) sdm.close_decrypted_device(map_name) @@ -287,7 +280,7 @@ def backup( ): continue backend = bb.BackupBackend.from_config(cfg) - _refresh_sudo(parsed_config.SudoPassCmd) + sh.refresh_sudo(parsed_config.SudoPassCmd) open_dir = parsed_config.OpenDirectory dest = open_dir / cfg.Name if open_dir is not None else None with ( @@ -300,7 +293,7 @@ def backup( # A backup could take so long that the sudo session expires. In this # case the user would have to enter the password again to unmount and # close the device. To prevent this, the sudo session is refreshed. - _refresh_sudo(parsed_config.SudoPassCmd) + sh.refresh_sudo(parsed_config.SudoPassCmd) @app.command() diff --git a/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py b/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py index caa8a5a6..9a071d03 100644 --- a/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py +++ b/projects/butter-backup/tests/backup_backends/test_btrfs_rsync_backend.py @@ -104,7 +104,7 @@ def test_btrfs_backend_refreshes_sudo_session_in_do_backup( UUID=uuid4(), ) backend = bb.BtrFSRsyncBackend(config=config) - mock_refresh = mocker.patch("butter_backup.backup_backends._refresh_sudo") + mock_refresh = mocker.patch("butter_backup.backup_backends.sh.refresh_sudo") mocker.patch.object( bb.BtrFSRsyncBackend, "get_source_snapshot", return_value=tmp_path ) diff --git a/projects/butter-backup/tests/backup_backends/test_restic_backend.py b/projects/butter-backup/tests/backup_backends/test_restic_backend.py index 8e743d7d..51128b40 100644 --- a/projects/butter-backup/tests/backup_backends/test_restic_backend.py +++ b/projects/butter-backup/tests/backup_backends/test_restic_backend.py @@ -60,7 +60,7 @@ def test_restic_backend_refreshes_sudo_session_in_do_backup( UUID=uuid4(), ) backend = bb.ResticBackend(config=config) - mock_refresh = mocker.patch("butter_backup.backup_backends._refresh_sudo") + mock_refresh = mocker.patch("butter_backup.backup_backends.sh.refresh_sudo") mocker.patch.object(bb.ResticBackend, "copy_files") mocker.patch.object(bb.ResticBackend, "adapt_ownership") diff --git a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py index f4cdf8bc..d2021719 100644 --- a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py +++ b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py @@ -1,4 +1,5 @@ import os +import subprocess import typing as t from pathlib import Path @@ -40,10 +41,33 @@ def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: sh.chown(root_owned_path, current_user, current_group, recursive=False) +def _assert_sudo_refresh_occurred_before_privileged_cmd( + refresh_idx: list[int], other_sudo_idx: list[int] +) -> None: + # Ensure that the sudo refresh and other privileged commands occurred alternating. + # Ideally, we would check that no two refreshes occurred without a privileged + # command in between, but this is believed to be too strict. In the test setup, some + # privileged commands may be skipped, e.g. because there is nothing to back up. + assert len(refresh_idx) > 0 + assert len(other_sudo_idx) > 0 + assert min(refresh_idx) < min(other_sudo_idx) + assert max(refresh_idx) < max(other_sudo_idx) + + def _invalidate_sudo_session() -> None: sh.run_cmd(cmd=["sudo", "-k"]) +def _is_sudo_refresh(call: t.Any) -> bool: + ret: bool = call.args and call.args[0][:2] == ["sudo", "-Sv"] + return ret + + +def _is_non_refresh_sudo_cmd(call: t.Any) -> bool: + ret: bool = call.args and call.args[0][0] == "sudo" and not _is_sudo_refresh(call) + return ret + + def test_sudo_pass_cmd_is_used_in_open( runner: CliRunner, encrypted_device: cp.DeviceConfiguration, @@ -61,18 +85,19 @@ def test_sudo_pass_cmd_is_used_in_open( config_file = tmp_path / "config.json" config_file.write_text(wrapped_config.model_dump_json()) - mock_pipe = mocker.patch("shell_interface.pipe_pass_cmd_to_real_cmd") - mocker.patch( - "storage_device_managers.open_encrypted_device", - return_value=Path("/dev/mapper/test"), - ) - mocker.patch("storage_device_managers.mount_device") + spy = mocker.spy(subprocess, "run") - runner.invoke(app, ["open", "--config", str(config_file)]) + open_result = runner.invoke(app, ["open", "--config", str(config_file)]) + assert open_result.exit_code == 0, open_result.output - mock_pipe.assert_called_once_with( - sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True - ) + sudo_refresh_calls = [ + i for i, c in enumerate(spy.call_args_list) if _is_sudo_refresh(c) + ] + other_privileged_calls = [ + i for i, c in enumerate(spy.call_args_list) if _is_non_refresh_sudo_cmd(c) + ] + assert sudo_refresh_calls and other_privileged_calls + assert max(sudo_refresh_calls) < min(other_privileged_calls) def test_open_uses_sudo_to_create_mount_dir( @@ -95,7 +120,7 @@ def test_open_uses_sudo_to_create_mount_dir( runner.invoke(app, ["close", "--config", str(config_file)]) -def test_sudo_pass_cmd_is_used_in_backup( +def test_sudo_session_is_refreshed_around_backup( runner: CliRunner, encrypted_device: cp.DeviceConfiguration, mocker, @@ -105,28 +130,32 @@ def test_sudo_pass_cmd_is_used_in_backup( wrapped_config = cp.Configuration( DeviceConfigurations=[encrypted_device], SudoPassCmd=sudo_pass_cmd ) + + match encrypted_device: + case cp.BtrFSRsyncConfig(): + expected_nof_refreshes = 3 + case cp.ResticConfig(): + expected_nof_refreshes = 4 + case _: + t.assert_never(encrypted_device) + config_file = tmp_path / "config.json" config_file.write_text(wrapped_config.model_dump_json()) - mock_pipe = mocker.patch("shell_interface.pipe_pass_cmd_to_real_cmd") - mocker.patch("storage_device_managers.decrypted_device") - mocker.patch("storage_device_managers.mounted_device") - mocker.patch( - "butter_backup.backup_backends.BackupBackend.from_config", - return_value=mocker.MagicMock(), - ) + spy = mocker.spy(subprocess, "run") + result = runner.invoke(app, ["backup", "--config", str(config_file)]) + assert result.exit_code == 0, result.output - runner.invoke(app, ["backup", "--config", str(config_file)]) + refresh_idx = [i for i, c in enumerate(spy.call_args_list) if _is_sudo_refresh(c)] + other_idx = [ + i for i, c in enumerate(spy.call_args_list) if _is_non_refresh_sudo_cmd(c) + ] - expected_nof_calls = 2 # One before opening the device and one post backup - result_calls = mock_pipe.call_args_list - expected_calls = [ - ((sudo_pass_cmd, ["sudo", "-Sv"]), {"capture_output": True}) - ] * expected_nof_calls - assert result_calls == expected_calls + assert len(refresh_idx) == expected_nof_refreshes + _assert_sudo_refresh_occurred_before_privileged_cmd(refresh_idx, other_idx) -def test_sudo_pass_cmd_is_used_in_close( +def test_sudo_session_is_refreshed_before_close( runner: CliRunner, encrypted_device: cp.DeviceConfiguration, mocker, @@ -138,21 +167,20 @@ def test_sudo_pass_cmd_is_used_in_close( ) config_file = tmp_path / "config.json" config_file.write_text(wrapped_config.model_dump_json()) - map_name = str(encrypted_device.map_name()) - mock_pipe = mocker.patch("shell_interface.pipe_pass_cmd_to_real_cmd") - mocker.patch( - "storage_device_managers.get_mounted_devices", - return_value={map_name: [tmp_path / "mnt"]}, - ) - mocker.patch("storage_device_managers.unmount_device") - mocker.patch("storage_device_managers.close_decrypted_device") + open_result = runner.invoke(app, ["open", "--config", str(config_file)]) + assert open_result.exit_code == 0, open_result.output + spy = mocker.spy(subprocess, "run") + close_result = runner.invoke(app, ["close", "--config", str(config_file)]) + assert close_result.exit_code == 0, close_result.output - runner.invoke(app, ["close", "--config", str(config_file)]) + refresh_idx = [i for i, c in enumerate(spy.call_args_list) if _is_sudo_refresh(c)] + other_idx = [ + i for i, c in enumerate(spy.call_args_list) if _is_non_refresh_sudo_cmd(c) + ] - mock_pipe.assert_called_once_with( - sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True - ) + assert len(refresh_idx) == 1 + _assert_sudo_refresh_occurred_before_privileged_cmd(refresh_idx, other_idx) @_requires_sudo_pass_cmd diff --git a/projects/shell-interface/src/shell_interface/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index f460a0e3..52027d3c 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -9,6 +9,7 @@ get_group, get_user, pipe_pass_cmd_to_real_cmd, + refresh_sudo, run_cmd, ) @@ -22,5 +23,6 @@ "get_group", "get_user", "pipe_pass_cmd_to_real_cmd", + "refresh_sudo", "run_cmd", ] diff --git a/projects/shell-interface/src/shell_interface/shell_interface.py b/projects/shell-interface/src/shell_interface/shell_interface.py index f56025c5..408a7622 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -233,3 +233,23 @@ def chown( if recursive: chown_cmd.append("--recursive") run_cmd(cmd=chown_cmd) + + +def refresh_sudo(sudo_pass_cmd: str | None) -> None: + """ + Refresh sudo credentials, if a password command is providedo + + Some use cases, most notably butter-backup and the library storage-device-managers, + require elevated privileges to run. Passing a password command for each and every + command that requires elevated privileges is cumbersome and error-prone. + + A trade-off is to refresh the sudo cache once after long-running operations, so that + the user does not have to enter their password multiple times. + + This function will run the provided password command and pipe its output to the + `sudo -Sv` command, which refreshes the sudo cache. If no password command is + provided, this function does nothing. Without a password command, the user will be + prompted for their password once needed. + """ + if sudo_pass_cmd is not None: + pipe_pass_cmd_to_real_cmd(sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True) From 2bf2956c857766d3da6c95704d0dc883dd3f5d22 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:45:24 +0200 Subject: [PATCH 6/9] refactor: Move rmdir wrapper to shell-interface --- .../butter-backup/src/butter_backup/cli.py | 19 +-------------- .../src/shell_interface/__init__.py | 2 ++ .../src/shell_interface/shell_interface.py | 18 ++++++++++++++ projects/shell-interface/tests/conftest.py | 24 +++++++++++++++++++ .../tests/test_ensure_directory.py | 19 --------------- .../shell-interface/tests/test_rmdir_up_to.py | 24 +++++++++++++++++++ 6 files changed, 69 insertions(+), 37 deletions(-) create mode 100644 projects/shell-interface/tests/conftest.py create mode 100644 projects/shell-interface/tests/test_rmdir_up_to.py diff --git a/projects/butter-backup/src/butter_backup/cli.py b/projects/butter-backup/src/butter_backup/cli.py index 1dd7ef77..6cd7fde9 100644 --- a/projects/butter-backup/src/butter_backup/cli.py +++ b/projects/butter-backup/src/butter_backup/cli.py @@ -149,28 +149,11 @@ def _open_device( f"Speichermedium {cfg.Name} konnte nicht geöffnet werden. Es wird übersprungen." ) if topmost_created_ancestor is not None: - _rmdir_ancestor_path(start=topmost_created_ancestor, stop=base_dir) + sh.rmdir_up_to(start=topmost_created_ancestor, stop=base_dir) else: typer.echo(f"Speichermedium {cfg.Name} wurde in {mount_dir} geöffnet.") -def _rmdir_ancestor_path(start: Path, stop: Path) -> None: - """ - Remove all directories from `start` to `stop` (inclusive). - - The directories are removed in a bottom-up manner. Execution stops at the first - non-empty directory or directly after removing stop, whatever comes first. - """ - if not start.is_relative_to(stop): - raise ValueError(f"Start path {start} is not a subpath of stop path {stop}.") - current = start - while current.is_relative_to(stop): - with contextlib.suppress(sh.ShellInterfaceError): - cmd: sh.StrPathList = ["sudo", "rmdir", current] - sh.run_cmd(cmd=cmd) - current = current.parent - - @app.command() def open( # noqa: A001 config: t.Annotated[Path | None, CONFIG_OPTION] = None, diff --git a/projects/shell-interface/src/shell_interface/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index 52027d3c..28bb40f6 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -10,6 +10,7 @@ get_user, pipe_pass_cmd_to_real_cmd, refresh_sudo, + rmdir_up_to, run_cmd, ) @@ -24,5 +25,6 @@ "get_user", "pipe_pass_cmd_to_real_cmd", "refresh_sudo", + "rmdir_up_to", "run_cmd", ] diff --git a/projects/shell-interface/src/shell_interface/shell_interface.py b/projects/shell-interface/src/shell_interface/shell_interface.py index 408a7622..5682fcdf 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -1,5 +1,6 @@ from __future__ import annotations +import contextlib import getpass import os import subprocess @@ -253,3 +254,20 @@ def refresh_sudo(sudo_pass_cmd: str | None) -> None: """ if sudo_pass_cmd is not None: pipe_pass_cmd_to_real_cmd(sudo_pass_cmd, ["sudo", "-Sv"], capture_output=True) + + +def rmdir_up_to(start: Path, stop: Path) -> None: + """ + Remove all directories from `start` to `stop` (inclusive). + + The directories are removed in a bottom-up manner. Execution stops at the first + non-empty directory or directly after removing stop, whatever comes first. + """ + if not start.is_relative_to(stop): + raise ValueError(f"Start path {start} is not a subpath of stop path {stop}.") + current = start + while current.is_relative_to(stop): + with contextlib.suppress(ShellInterfaceError): + cmd: StrPathList = ["sudo", "rmdir", current] + run_cmd(cmd=cmd) + current = current.parent diff --git a/projects/shell-interface/tests/conftest.py b/projects/shell-interface/tests/conftest.py new file mode 100644 index 00000000..3be1cda8 --- /dev/null +++ b/projects/shell-interface/tests/conftest.py @@ -0,0 +1,24 @@ +from collections import abc +from pathlib import Path + +import pytest + +import shell_interface as sh + + +@pytest.fixture +def root_owned_tmp_path(tmp_path: Path) -> abc.Iterable[Path]: + """ + Create a temporary directory owned by root for testing + """ + root_owned_path = tmp_path / "root_owned" + root_owned_path.mkdir() + current_user = sh.get_user() + current_group = sh.get_group(current_user) + sh.chown(root_owned_path, user="root", group="root", recursive=False) + try: + yield root_owned_path + finally: + sh.chown( + root_owned_path, user=current_user, group=current_group, recursive=False + ) diff --git a/projects/shell-interface/tests/test_ensure_directory.py b/projects/shell-interface/tests/test_ensure_directory.py index 1267bcea..85bcdac4 100644 --- a/projects/shell-interface/tests/test_ensure_directory.py +++ b/projects/shell-interface/tests/test_ensure_directory.py @@ -1,4 +1,3 @@ -import typing as t from pathlib import Path import pytest @@ -14,24 +13,6 @@ def sudo_mkdir(path: Path) -> None: sh.run_cmd(cmd=sudo_mkdir_cmd) -@pytest.fixture -def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: - """ - Create a temporary directory owned by root for testing - """ - root_owned_path = tmp_path / "root_owned" - root_owned_path.mkdir() - current_user = sh.get_user() - current_group = sh.get_group(current_user) - sh.chown(root_owned_path, user="root", group="root", recursive=False) - try: - yield root_owned_path - finally: - sh.chown( - root_owned_path, user=current_user, group=current_group, recursive=False - ) - - @pytest.mark.parametrize("destination", ["destDir", "dest dir with spaces"]) @pytest.mark.parametrize( "first_created", diff --git a/projects/shell-interface/tests/test_rmdir_up_to.py b/projects/shell-interface/tests/test_rmdir_up_to.py new file mode 100644 index 00000000..f8e4333a --- /dev/null +++ b/projects/shell-interface/tests/test_rmdir_up_to.py @@ -0,0 +1,24 @@ +from pathlib import Path + +import pytest + +import shell_interface as sh + + +@pytest.mark.parametrize( + "start", ["subdir", "sub dir/with spaces/", "very/deeply/nested/subdir", None] +) +@pytest.mark.parametrize( + "stop", ["stopDir", "stop dir with spaces", "deeply/nested/stopDir"] +) +def test_rmdir_up_to_works( + root_owned_tmp_path: Path, start: str | None, stop: str +) -> None: + stop_p = root_owned_tmp_path / stop + start_p = stop_p if start is None else stop_p / start + sh.ensure_directory(start_p) + assert start_p.exists() + sh.rmdir_up_to(start_p, stop_p) + assert not start_p.exists() + assert not stop_p.exists() + assert stop_p.parent.exists() From 3551881162ba5377b1865a4b0d3d378d2d63db95 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:52:27 +0200 Subject: [PATCH 7/9] refactor: Move run_cmd functions to dedicated file This reduces the length of the main shell-interface file, improving maintainability. --- .../butter-backup/src/butter_backup/cli.py | 1 - .../src/shell_interface/__init__.py | 8 +- .../src/shell_interface/run_cmd.py | 131 ++++++++++++++++++ .../src/shell_interface/shell_interface.py | 128 +---------------- 4 files changed, 137 insertions(+), 131 deletions(-) create mode 100644 projects/shell-interface/src/shell_interface/run_cmd.py diff --git a/projects/butter-backup/src/butter_backup/cli.py b/projects/butter-backup/src/butter_backup/cli.py index 6cd7fde9..12693e73 100644 --- a/projects/butter-backup/src/butter_backup/cli.py +++ b/projects/butter-backup/src/butter_backup/cli.py @@ -1,5 +1,4 @@ #!/usr/bin/env python3 -import contextlib import enum import json import os diff --git a/projects/shell-interface/src/shell_interface/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index 28bb40f6..27962eb5 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -1,17 +1,19 @@ from importlib import metadata -from .shell_interface import ( +from .run_cmd import ( PassCmdError, ShellInterfaceError, + pipe_pass_cmd_to_real_cmd, + run_cmd, +) +from .shell_interface import ( StrPathList, chown, ensure_directory, get_group, get_user, - pipe_pass_cmd_to_real_cmd, refresh_sudo, rmdir_up_to, - run_cmd, ) __version__ = metadata.version(__name__) diff --git a/projects/shell-interface/src/shell_interface/run_cmd.py b/projects/shell-interface/src/shell_interface/run_cmd.py new file mode 100644 index 00000000..583fc72d --- /dev/null +++ b/projects/shell-interface/src/shell_interface/run_cmd.py @@ -0,0 +1,131 @@ +import os +import subprocess +from pathlib import Path +from types import SimpleNamespace + +StrPathList = list[str | Path] +_CMD_LIST = list[str] | list[Path] | StrPathList + +try: + from loguru import logger # type: ignore[import, unused-ignore] + + logger.disable("shell_interface") +except ModuleNotFoundError: + logger = SimpleNamespace() # type: ignore[assignment, unused-ignore] + logger.debug = lambda msg: None # type: ignore[assignment, unused-ignore] + logger.error = lambda msg: None # type: ignore[assignment, unused-ignore] + + +class PassCmdError(RuntimeError): + pass + + +class ShellInterfaceError(RuntimeError): + pass + + +def run_cmd( + *, + cmd: _CMD_LIST, + env: dict[str, str] | None = None, + capture_output: bool = False, +) -> subprocess.CompletedProcess[bytes]: + """ + Run provided command in shell + + This function will run the provided command in a shell. The command must be + provided as a list of tokens. + + The main difference to using `subprocess.run` directly is that this function + checks the return code of the command by default, while `subprocess.run` + does not. If a non-zero return code is encountered, a `ShellInterfaceError` + is raised. + + Parameters: + ----------- + cmd + command to run in shell + env + environment variables to set for the command; if `None`, the current + environment is used + capture_output + whether to capture the output of the command; if `True`, the output is + returned as part of the `CompletedProcess` object + + Returns: + -------- + subprocess.CompletedProcess[bytes] + object containing information about the completed process + + Raises: + ------- + ShellInterfaceError + if the command returns a non-zero return code + """ + if env is None: + env = dict(os.environ) + logger.debug(f"Shell-Befehl ist `{cmd}`.") + try: + result = subprocess.run(cmd, capture_output=capture_output, check=True, env=env) + except subprocess.CalledProcessError as e: + errmsg = f"Shell-Befehl `{cmd}` ist fehlgeschlagen." + logger.error(errmsg) + raise ShellInterfaceError(errmsg) from e + return result + + +def pipe_pass_cmd_to_real_cmd( + pass_cmd: str, command: _CMD_LIST, *, capture_output: bool = False +) -> subprocess.CompletedProcess[bytes]: + """ + Pipe result of first command to second command + + This function will run the first command in a shell and pipe its output to + the second command. The first command must be provided as a string, while + the second command must be provided as a list of tokens. This allows to take a plain + command as input from a user without having to deal with tokenization. + + The return code of both commands is checked. If a non-zero return code is + encountered, a `ShellInterfaceError` is raised. + + Parameters: + ----------- + pass_cmd + command to run in shell and whose output is piped to the second command + command + command to run in shell and whose input is piped from the first command + capture_output + whether to capture the output of the real command; if `True`, the output is + returned as part of the `CompletedProcess` object + + Returns: + -------- + subprocess.CompletedProcess[bytes] + object containing information about the completed process of the second + command + + Raises: + ------- + PassCmdError + if the password command returns a non-zero return code + ShellInterfaceError + if the real command returns a non-zero return code + """ + logger.debug(f"Shell-Befehl ist `{command}`.") + try: + pwd_proc = subprocess.run( + pass_cmd, stdout=subprocess.PIPE, shell=True, check=True + ) + except subprocess.CalledProcessError as e: + errmsg = f"Shell-Befehl `{pass_cmd}` ist fehlgeschlagen." + logger.error(errmsg) + raise PassCmdError(errmsg) from e + try: + completed_process = subprocess.run( + command, input=pwd_proc.stdout, check=True, capture_output=capture_output + ) + except subprocess.CalledProcessError as e: + errmsg = f"Shell-Befehl `{command}` ist fehlgeschlagen." + logger.error(errmsg) + raise ShellInterfaceError(errmsg) from e + return completed_process diff --git a/projects/shell-interface/src/shell_interface/shell_interface.py b/projects/shell-interface/src/shell_interface/shell_interface.py index 5682fcdf..977498c3 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -2,137 +2,11 @@ import contextlib import getpass -import os -import subprocess from pathlib import Path -from types import SimpleNamespace -try: - from loguru import logger # type: ignore[import, unused-ignore] - - logger.disable("shell_interface") -except ModuleNotFoundError: - logger = SimpleNamespace() # type: ignore[assignment, unused-ignore] - logger.debug = lambda msg: None # type: ignore[assignment, unused-ignore] - logger.error = lambda msg: None # type: ignore[assignment, unused-ignore] +from .run_cmd import ShellInterfaceError, pipe_pass_cmd_to_real_cmd, run_cmd StrPathList = list[str | Path] -_CMD_LIST = list[str] | list[Path] | StrPathList - - -class PassCmdError(RuntimeError): - pass - - -class ShellInterfaceError(RuntimeError): - pass - - -def run_cmd( - *, - cmd: _CMD_LIST, - env: dict[str, str] | None = None, - capture_output: bool = False, -) -> subprocess.CompletedProcess[bytes]: - """ - Run provided command in shell - - This function will run the provided command in a shell. The command must be - provided as a list of tokens. - - The main difference to using `subprocess.run` directly is that this function - checks the return code of the command by default, while `subprocess.run` - does not. If a non-zero return code is encountered, a `ShellInterfaceError` - is raised. - - Parameters: - ----------- - cmd - command to run in shell - env - environment variables to set for the command; if `None`, the current - environment is used - capture_output - whether to capture the output of the command; if `True`, the output is - returned as part of the `CompletedProcess` object - - Returns: - -------- - subprocess.CompletedProcess[bytes] - object containing information about the completed process - - Raises: - ------- - ShellInterfaceError - if the command returns a non-zero return code - """ - if env is None: - env = dict(os.environ) - logger.debug(f"Shell-Befehl ist `{cmd}`.") - try: - result = subprocess.run(cmd, capture_output=capture_output, check=True, env=env) - except subprocess.CalledProcessError as e: - errmsg = f"Shell-Befehl `{cmd}` ist fehlgeschlagen." - logger.error(errmsg) - raise ShellInterfaceError(errmsg) from e - return result - - -def pipe_pass_cmd_to_real_cmd( - pass_cmd: str, command: _CMD_LIST, *, capture_output: bool = False -) -> subprocess.CompletedProcess[bytes]: - """ - Pipe result of first command to second command - - This function will run the first command in a shell and pipe its output to - the second command. The first command must be provided as a string, while - the second command must be provided as a list of tokens. This allows to take a plain - command as input from a user without having to deal with tokenization. - - The return code of both commands is checked. If a non-zero return code is - encountered, a `ShellInterfaceError` is raised. - - Parameters: - ----------- - pass_cmd - command to run in shell and whose output is piped to the second command - command - command to run in shell and whose input is piped from the first command - capture_output - whether to capture the output of the real command; if `True`, the output is - returned as part of the `CompletedProcess` object - - Returns: - -------- - subprocess.CompletedProcess[bytes] - object containing information about the completed process of the second - command - - Raises: - ------- - PassCmdError - if the password command returns a non-zero return code - ShellInterfaceError - if the real command returns a non-zero return code - """ - logger.debug(f"Shell-Befehl ist `{command}`.") - try: - pwd_proc = subprocess.run( - pass_cmd, stdout=subprocess.PIPE, shell=True, check=True - ) - except subprocess.CalledProcessError as e: - errmsg = f"Shell-Befehl `{pass_cmd}` ist fehlgeschlagen." - logger.error(errmsg) - raise PassCmdError(errmsg) from e - try: - completed_process = subprocess.run( - command, input=pwd_proc.stdout, check=True, capture_output=capture_output - ) - except subprocess.CalledProcessError as e: - errmsg = f"Shell-Befehl `{command}` ist fehlgeschlagen." - logger.error(errmsg) - raise ShellInterfaceError(errmsg) from e - return completed_process def ensure_directory(directory: Path) -> Path | None: From 5f0fccba795f9d5f537ba05eb87a7d8b00b9a24f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 21:58:00 +0200 Subject: [PATCH 8/9] refactor: Drop loguru from both libraries The logging functionality was never used, for years. A major release seems to be the best opportunity to get rid of it. --- .github/workflows/ci-shell-interface.yml | 3 +- .../workflows/ci-storage-device-managers.yml | 3 +- projects/shell-interface/pyproject.toml | 5 ---- .../src/shell_interface/run_cmd.py | 15 ---------- .../storage-device-managers/pyproject.toml | 5 ---- .../src/storage_device_managers/__init__.py | 30 +------------------ uv.lock | 14 --------- 7 files changed, 3 insertions(+), 72 deletions(-) diff --git a/.github/workflows/ci-shell-interface.yml b/.github/workflows/ci-shell-interface.yml index 13e446cd..9445b137 100644 --- a/.github/workflows/ci-shell-interface.yml +++ b/.github/workflows/ci-shell-interface.yml @@ -9,7 +9,6 @@ jobs: strategy: matrix: python-version: ["3.11", "3.12", "3.13", "3.14"] - dependency-extras: ["", "--extra logging"] steps: - uses: actions/checkout@v7 @@ -18,7 +17,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: uv sync --package shell-interface ${{ matrix.dependency-extras }} + run: uv sync --package shell-interface - name: Check formatting with ruff run: uv run --package shell-interface ruff format --check projects/shell-interface - name: Check import ordering with ruff diff --git a/.github/workflows/ci-storage-device-managers.yml b/.github/workflows/ci-storage-device-managers.yml index 35276280..9be49609 100644 --- a/.github/workflows/ci-storage-device-managers.yml +++ b/.github/workflows/ci-storage-device-managers.yml @@ -9,7 +9,6 @@ jobs: strategy: matrix: python-version: ["3.11", "3.12", "3.13", "3.14"] - dependency-extras: ["", "--extra logging"] steps: - uses: actions/checkout@v7 @@ -18,7 +17,7 @@ jobs: with: python-version: ${{ matrix.python-version }} - name: Install dependencies - run: uv sync --package storage-device-managers ${{ matrix.dependency-extras }} + run: uv sync --package storage-device-managers - name: Check formatting with ruff run: uv run --package storage-device-managers ruff format --check projects/storage-device-managers - name: Check import ordering with ruff diff --git a/projects/shell-interface/pyproject.toml b/projects/shell-interface/pyproject.toml index 7e35dda6..d7f5918a 100644 --- a/projects/shell-interface/pyproject.toml +++ b/projects/shell-interface/pyproject.toml @@ -15,11 +15,6 @@ Repository = "https://github.com/MaxG87/shell-interface" Issues = "https://github.com/MaxG87/shell-interface/issues" Changelog = "https://github.com/MaxG87/shell-interface/blob/main/CHANGELOG.md" -[project.optional-dependencies] -logging = [ - "loguru", -] - [dependency-groups] dev = [ "hypothesis>=6.155.7", diff --git a/projects/shell-interface/src/shell_interface/run_cmd.py b/projects/shell-interface/src/shell_interface/run_cmd.py index 583fc72d..ed3cd394 100644 --- a/projects/shell-interface/src/shell_interface/run_cmd.py +++ b/projects/shell-interface/src/shell_interface/run_cmd.py @@ -1,20 +1,10 @@ import os import subprocess from pathlib import Path -from types import SimpleNamespace StrPathList = list[str | Path] _CMD_LIST = list[str] | list[Path] | StrPathList -try: - from loguru import logger # type: ignore[import, unused-ignore] - - logger.disable("shell_interface") -except ModuleNotFoundError: - logger = SimpleNamespace() # type: ignore[assignment, unused-ignore] - logger.debug = lambda msg: None # type: ignore[assignment, unused-ignore] - logger.error = lambda msg: None # type: ignore[assignment, unused-ignore] - class PassCmdError(RuntimeError): pass @@ -64,12 +54,10 @@ def run_cmd( """ if env is None: env = dict(os.environ) - logger.debug(f"Shell-Befehl ist `{cmd}`.") try: result = subprocess.run(cmd, capture_output=capture_output, check=True, env=env) except subprocess.CalledProcessError as e: errmsg = f"Shell-Befehl `{cmd}` ist fehlgeschlagen." - logger.error(errmsg) raise ShellInterfaceError(errmsg) from e return result @@ -111,14 +99,12 @@ def pipe_pass_cmd_to_real_cmd( ShellInterfaceError if the real command returns a non-zero return code """ - logger.debug(f"Shell-Befehl ist `{command}`.") try: pwd_proc = subprocess.run( pass_cmd, stdout=subprocess.PIPE, shell=True, check=True ) except subprocess.CalledProcessError as e: errmsg = f"Shell-Befehl `{pass_cmd}` ist fehlgeschlagen." - logger.error(errmsg) raise PassCmdError(errmsg) from e try: completed_process = subprocess.run( @@ -126,6 +112,5 @@ def pipe_pass_cmd_to_real_cmd( ) except subprocess.CalledProcessError as e: errmsg = f"Shell-Befehl `{command}` ist fehlgeschlagen." - logger.error(errmsg) raise ShellInterfaceError(errmsg) from e return completed_process diff --git a/projects/storage-device-managers/pyproject.toml b/projects/storage-device-managers/pyproject.toml index 7ad3d59d..c612ed21 100644 --- a/projects/storage-device-managers/pyproject.toml +++ b/projects/storage-device-managers/pyproject.toml @@ -20,11 +20,6 @@ Repository = "https://github.com/MaxG87/storage-device-managers" Issues = "https://github.com/MaxG87/storage-device-managers/issues" Changelog = "https://github.com/MaxG87/storage-device-managers/blob/main/CHANGELOG.md" - -[project.optional-dependencies] -logging = ["loguru"] - - [dependency-groups] dev = [ "hypothesis>=6.155.7", diff --git a/projects/storage-device-managers/src/storage_device_managers/__init__.py b/projects/storage-device-managers/src/storage_device_managers/__init__.py index a3e009b8..3ddf300b 100644 --- a/projects/storage-device-managers/src/storage_device_managers/__init__.py +++ b/projects/storage-device-managers/src/storage_device_managers/__init__.py @@ -7,22 +7,12 @@ from collections.abc import Iterator from importlib import metadata from pathlib import Path -from types import SimpleNamespace from uuid import UUID, uuid4 import shell_interface as sh from storage_device_managers._findmnt import MountOptions, get_mounted_devices -try: - from loguru import logger # type: ignore[import, unused-ignore] - - logger.disable("storage_device_managers") -except ModuleNotFoundError: - logger = SimpleNamespace() # type: ignore[assignment, unused-ignore] - logger.success = lambda msg: None # type: ignore[assignment, unused-ignore] - logger.info = lambda msg: None # type: ignore[assignment, unused-ignore] - __version__ = metadata.version(__name__) __all__ = [ @@ -154,14 +144,10 @@ def decrypted_device(device: Path, pass_cmd: str) -> Iterator[Path]: if cryptsetup returns a non-zero exit code """ decrypted = open_encrypted_device(device, pass_cmd) - logger.success(f"Speichermedium {device} erfolgreich entschlüsselt.") try: yield decrypted finally: close_decrypted_device(decrypted) - logger.success( - f"Verschlüsselung des Speichermediums {device} erfolgreich geschlossen." - ) @contextlib.contextmanager @@ -215,16 +201,10 @@ def mounted_device( ) with ctx as mount_dir: mount_device(device, mount_dir, compression) - logger.success( - f"Speichermedium {device} erfolgreich nach {mount_dir} gemountet." - ) try: yield Path(mount_dir) finally: unmount_device(device) - logger.success( - "Speichermedium {device} erfolgreich ausgehangen.", device=device - ) @contextlib.contextmanager @@ -257,7 +237,6 @@ def symbolic_link(src: Path, dest: Path) -> Iterator[Path]: absolute_dest = dest.absolute() ln_cmd: sh.StrPathList = ["sudo", "ln", "-s", src.absolute(), absolute_dest] sh.run_cmd(cmd=ln_cmd) - logger.success(f"Symlink von {src} nach {dest} erfolgreich erstellt.") try: yield absolute_dest finally: @@ -265,7 +244,6 @@ def symbolic_link(src: Path, dest: Path) -> Iterator[Path]: # all, the aimed for state has been reached. rm_cmd: sh.StrPathList = ["sudo", "rm", "-f", absolute_dest] sh.run_cmd(cmd=rm_cmd) - logger.success(f"Symlink von {src} nach {dest} erfolgreich entfernt.") def _mount_btrfs_device( @@ -368,13 +346,7 @@ def is_mounted(device: Path) -> bool: True if `device` is mounted, False otherwise """ device_as_str = str(device) - try: - mount_dest = get_mounted_devices()[device_as_str] - logger.info(f"Mount des Speichermediums {device} in {mount_dest} gefunden.") - except KeyError: - logger.info(f"Kein Mountpunkt für Speichermedium {device} gefunden.") - return False - return True + return device_as_str in get_mounted_devices() def sync_device(device: Path) -> None: diff --git a/uv.lock b/uv.lock index 03680230..81b58176 100644 --- a/uv.lock +++ b/uv.lock @@ -854,11 +854,6 @@ name = "shell-interface" version = "2.0.0" source = { editable = "projects/shell-interface" } -[package.optional-dependencies] -logging = [ - { name = "loguru" }, -] - [package.dev-dependencies] dev = [ { name = "hypothesis" }, @@ -873,8 +868,6 @@ typecheck = [ ] [package.metadata] -requires-dist = [{ name = "loguru", marker = "extra == 'logging'" }] -provides-extras = ["logging"] [package.metadata.requires-dev] dev = [ @@ -915,11 +908,6 @@ dependencies = [ { name = "shell-interface" }, ] -[package.optional-dependencies] -logging = [ - { name = "loguru" }, -] - [package.dev-dependencies] dev = [ { name = "hypothesis" }, @@ -938,11 +926,9 @@ typecheck = [ [package.metadata] requires-dist = [ { name = "loguru", specifier = ">=0.7.3" }, - { name = "loguru", marker = "extra == 'logging'" }, { name = "msgspec", specifier = ">=0.21.1" }, { name = "shell-interface", editable = "projects/shell-interface" }, ] -provides-extras = ["logging"] [package.metadata.requires-dev] dev = [ From c79226f2c689577352741d81530b390f78e14833 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Max=20G=C3=B6rner?= <5477952+MaxG87@users.noreply.github.com> Date: Tue, 21 Jul 2026 22:01:58 +0200 Subject: [PATCH 9/9] refactor: Create top-level conftest.py to share fixtures --- .../tests/conftest.py => conftest.py | 0 .../tests/cli/test_sudo_pass_cmd.py | 16 ---------------- 2 files changed, 16 deletions(-) rename projects/shell-interface/tests/conftest.py => conftest.py (100%) diff --git a/projects/shell-interface/tests/conftest.py b/conftest.py similarity index 100% rename from projects/shell-interface/tests/conftest.py rename to conftest.py diff --git a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py index d2021719..a61adacf 100644 --- a/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py +++ b/projects/butter-backup/tests/cli/test_sudo_pass_cmd.py @@ -25,22 +25,6 @@ def runner(): return CliRunner() -@pytest.fixture -def root_owned_tmp_path(tmp_path: Path) -> t.Iterable[Path]: - """ - Create a temporary directory owned by root for testing - """ - root_owned_path = tmp_path / "root_owned" - root_owned_path.mkdir() - current_user = sh.get_user() - current_group = sh.get_group(current_user) - sh.chown(root_owned_path, "root", "root", recursive=False) - try: - yield root_owned_path - finally: - sh.chown(root_owned_path, current_user, current_group, recursive=False) - - def _assert_sudo_refresh_occurred_before_privileged_cmd( refresh_idx: list[int], other_sudo_idx: list[int] ) -> None: