Add priviliged mkdir command

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

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)