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/conftest.py b/conftest.py new file mode 100644 index 00000000..3be1cda8 --- /dev/null +++ b/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/butter-backup/src/butter_backup/backup_backends.py b/projects/butter-backup/src/butter_backup/backup_backends.py index a57ca468..acbc8571 100644 --- a/projects/butter-backup/src/butter_backup/backup_backends.py +++ b/projects/butter-backup/src/butter_backup/backup_backends.py @@ -5,19 +5,11 @@ 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 -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: ... @@ -52,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 @@ -65,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 @@ -90,7 +82,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() @@ -150,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 @@ -165,9 +157,15 @@ 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: + 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 57c32973..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 @@ -99,13 +98,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,8 +136,8 @@ def _open_device( mount_dir = base_dir / cfg.Name topmost_created_ancestor = None try: - _refresh_sudo(sudo_pass_cmd) - topmost_created_ancestor = sdm.ensure_directory(mount_dir) + 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()) except Exception: @@ -156,28 +148,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, @@ -245,7 +220,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 +262,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 +275,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/src/butter_backup/device_managers.py b/projects/butter-backup/src/butter_backup/device_managers.py index 928410a1..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 @@ -35,7 +34,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 +62,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 +69,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..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 @@ -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) @@ -105,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 ) @@ -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/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 dc9879ce..a61adacf 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 @@ -24,33 +25,33 @@ 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) - 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 _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, @@ -68,18 +69,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( @@ -102,7 +104,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, @@ -112,28 +114,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, @@ -145,21 +151,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/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/__init__.py b/projects/shell-interface/src/shell_interface/__init__.py index 8fc8af1b..27962eb5 100644 --- a/projects/shell-interface/src/shell_interface/__init__.py +++ b/projects/shell-interface/src/shell_interface/__init__.py @@ -1,13 +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, - run_cmd, + refresh_sudo, + rmdir_up_to, ) __version__ = metadata.version(__name__) @@ -15,8 +21,12 @@ "PassCmdError", "ShellInterfaceError", "StrPathList", + "chown", + "ensure_directory", "get_group", "get_user", "pipe_pass_cmd_to_real_cmd", + "refresh_sudo", + "rmdir_up_to", "run_cmd", ] 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..ed3cd394 --- /dev/null +++ b/projects/shell-interface/src/shell_interface/run_cmd.py @@ -0,0 +1,116 @@ +import os +import subprocess +from pathlib import Path + +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) + 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." + 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 + """ + 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." + 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." + 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 9897aaa4..977498c3 100644 --- a/projects/shell-interface/src/shell_interface/shell_interface.py +++ b/projects/shell-interface/src/shell_interface/shell_interface.py @@ -1,137 +1,37 @@ from __future__ import annotations +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. +def ensure_directory(directory: Path) -> Path | None: + """Ensure a directory exists, creating it with root privileges if needed. 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 + directory + directory that should exist Returns: -------- - subprocess.CompletedProcess[bytes] - object containing information about the completed process - - Raises: - ------- - ShellInterfaceError - if the command returns a non-zero return code + Path | None + The first missing ancestor that had to be created, or ``None`` if the + directory already existed """ - 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 + 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: @@ -161,3 +61,87 @@ 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) + + +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) + + +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/storage-device-managers/tests/test_chown.py b/projects/shell-interface/tests/test_chown.py similarity index 85% rename from projects/storage-device-managers/tests/test_chown.py rename to projects/shell-interface/tests/test_chown.py index 0fc3a92e..803e4ed2 100644 --- a/projects/storage-device-managers/tests/test_chown.py +++ b/projects/shell-interface/tests/test_chown.py @@ -4,16 +4,14 @@ from tempfile import NamedTemporaryFile import pytest -import shell_interface as sh -import storage_device_managers as sdm +import shell_interface as sh @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] @@ -30,14 +28,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 +51,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 +68,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 new file mode 100644 index 00000000..85bcdac4 --- /dev/null +++ b/projects/shell-interface/tests/test_ensure_directory.py @@ -0,0 +1,37 @@ +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.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/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() 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 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 6ad55f32..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__ = [ @@ -32,11 +22,9 @@ "UnmountError", "ValidCompressions", "ValidFileSystems", - "chown", "close_decrypted_device", "decrypted_device", "encrypt_device", - "ensure_directory", "generate_passcmd", "get_filesystem", "get_mounted_devices", @@ -156,39 +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." - ) - - -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 @@ -234,7 +193,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 @@ -242,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 @@ -284,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: @@ -292,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( @@ -395,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: @@ -651,50 +596,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_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 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 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 = [