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 bbbb912..717fac7 100644 --- a/projects/storage-device-managers/src/storage_device_managers/__init__.py +++ b/projects/storage-device-managers/src/storage_device_managers/__init__.py @@ -35,6 +35,7 @@ "close_decrypted_device", "decrypted_device", "encrypt_device", + "ensure_directory", "generate_passcmd", "get_filesystem", "get_mounted_devices", @@ -165,6 +166,27 @@ def decrypted_device(device: Path, pass_cmd: str) -> Iterator[Path]: ) +def ensure_directory(directory: Path) -> bool: + """Ensure a directory exists, creating it with root privileges if needed. + + Parameters: + ----------- + directory + directory that should exist + + Returns: + -------- + bool + ``True`` if the directory had to be created, ``False`` if it already + existed + """ + if directory.is_dir(): + return False + cmd: sh.StrPathList = ["sudo", "mkdir", "-p", directory] + sh.run_cmd(cmd=cmd) + return True + + @contextlib.contextmanager def mounted_device( device: Path, @@ -182,7 +204,8 @@ def mounted_device( the directory is left intact after the context exits. If `destination` is ``None`` (the default), a temporary directory is created for the mount and removed again upon exit. If `destination` is given but does not exist, - it is created before mounting **but not removed again** after unmounting. + it is created with root privileges before mounting **but not removed + again** after unmounting. If `compression` is provided, a mount option specifying the transparent file system compression is set. Compression is only supported for BtrFS @@ -207,7 +230,7 @@ def mounted_device( if is_mounted(device): unmount_device(device) if destination is not None: - destination.mkdir(parents=True, exist_ok=True) + 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 new file mode 100644 index 0000000..af36f7f --- /dev/null +++ b/projects/storage-device-managers/tests/test_ensure_directory.py @@ -0,0 +1,50 @@ +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 / "nested" / "mount" + assert not destination.exists() + assert sdm.ensure_directory(destination) + assert destination.exists() + + +def test_ensure_directory_creates_missing_directory_in_root_owned_path( + root_owned_tmp_path: Path, +) -> None: + destination = root_owned_tmp_path / "nested" / "mount" + assert not destination.exists() + assert sdm.ensure_directory(destination) + assert destination.exists() + + +def test_ensure_directory_skips_existing_directory(tmp_path: Path) -> None: + assert not sdm.ensure_directory(tmp_path)