mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 10:51:36 -04:00
910503b641
QA on 10.0.0.8 (mount-issues.md findings 2 and 3) showed the reconciler path missed both NTFS accommodations that mount_drive() and add_fstab_entry() already have: secondary NTFS drives mounted owned by root (writable only via ntfs-3g's permissive default), and a dirty NTFS volume would fail to mount at boot with no cleanup attempted. Inject uid/gid for ntfs/ntfs3 in the reconciler, and run ntfsfix -d in SubprocessMountExecutor before NTFS mounts (translating UUID=/LABEL= specs to /dev/disk/by-* paths, which ntfsfix needs). Also return an error MountResult instead of raising when a binary is missing, per the executor protocol. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
252 lines
9.4 KiB
Python
252 lines
9.4 KiB
Python
"""
|
|
Tests for controller.lib.mount_executor.
|
|
|
|
The fake is a load-bearing test double; if it lies the entire Reconciler
|
|
test suite is meaningless. These tests pin its behaviour without using
|
|
mocks. The Subprocess executor is exercised against its own narrow
|
|
abstraction (a subprocess.run replacement injected via a small seam) so
|
|
the production path is also covered without leaning on real mount(8).
|
|
"""
|
|
|
|
from controller.lib.mount_executor import (
|
|
FakeMountExecutor,
|
|
MountCall,
|
|
MountResult,
|
|
SubprocessMountExecutor,
|
|
UnmountCall,
|
|
)
|
|
|
|
|
|
# --- FakeMountExecutor --------------------------------------------------
|
|
|
|
class TestFakeMountExecutorState:
|
|
|
|
def test_starts_with_initially_mounted(self):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/a", "/media/b"}))
|
|
assert fake.current_mount_points() == {"/media/a", "/media/b"}
|
|
|
|
def test_mount_adds_to_live_set(self):
|
|
fake = FakeMountExecutor()
|
|
r = fake.mount("UUID=1", "/media/x", "ext4", "defaults")
|
|
assert r.ok is True
|
|
assert r.error is None
|
|
assert fake.current_mount_points() == {"/media/x"}
|
|
|
|
def test_unmount_removes_from_live_set(self):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/x"}))
|
|
r = fake.unmount("/media/x")
|
|
assert r.ok is True
|
|
assert fake.current_mount_points() == set()
|
|
|
|
def test_unmount_of_missing_path_is_still_recorded(self):
|
|
# We model umount(8): unmounting an unmounted path is allowed (it's
|
|
# not an error in the fake). The Reconciler is responsible for
|
|
# only calling unmount on currently-mounted paths.
|
|
fake = FakeMountExecutor()
|
|
r = fake.unmount("/media/never-mounted")
|
|
assert r.ok is True
|
|
assert fake.unmount_calls == [UnmountCall("/media/never-mounted")]
|
|
|
|
def test_prefix_filter(self):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset(
|
|
{"/media/a", "/media/b", "/mnt/c"}))
|
|
# The fake stores anything but current_mount_points filters.
|
|
assert fake.current_mount_points("/media") == {"/media/a", "/media/b"}
|
|
assert fake.current_mount_points("/mnt") == {"/mnt/c"}
|
|
|
|
|
|
class TestFakeMountExecutorCallRecording:
|
|
|
|
def test_records_mount_arguments(self):
|
|
fake = FakeMountExecutor()
|
|
fake.mount("UUID=1234", "/media/x", "ext4", "defaults,nofail")
|
|
assert fake.mount_calls == [
|
|
MountCall("UUID=1234", "/media/x", "ext4", "defaults,nofail"),
|
|
]
|
|
|
|
def test_records_unmount_arguments(self):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/x"}))
|
|
fake.unmount("/media/x")
|
|
assert fake.unmount_calls == [UnmountCall("/media/x")]
|
|
|
|
|
|
class TestFakeMountExecutorConfiguredFailures:
|
|
|
|
def test_mount_failure_leaves_set_unchanged(self):
|
|
fake = FakeMountExecutor(fail_for_mount=frozenset({"/media/bad"}))
|
|
r = fake.mount("UUID=1", "/media/bad", "ext4", "defaults")
|
|
assert r.ok is False
|
|
assert "fake" in (r.error or "")
|
|
assert fake.current_mount_points() == set()
|
|
# Call still recorded so tests can assert what was attempted.
|
|
assert len(fake.mount_calls) == 1
|
|
|
|
def test_unmount_failure_leaves_set_unchanged(self):
|
|
fake = FakeMountExecutor(
|
|
initially_mounted=frozenset({"/media/x"}),
|
|
fail_for_unmount=frozenset({"/media/x"}),
|
|
)
|
|
r = fake.unmount("/media/x")
|
|
assert r.ok is False
|
|
assert "/media/x" in fake.current_mount_points()
|
|
|
|
|
|
# --- SubprocessMountExecutor --------------------------------------------
|
|
|
|
class _FakeCompletedProcess:
|
|
def __init__(self, returncode: int, stderr: str = ""):
|
|
self.returncode = returncode
|
|
self.stderr = stderr
|
|
|
|
|
|
class TestSubprocessMountExecutor:
|
|
"""Cover the subprocess-backed executor without ever invoking mount(8).
|
|
|
|
We inject a fake subprocess.run into the executor's module namespace via
|
|
monkeypatch (pytest's built-in feature, not mock.patch). This is the
|
|
closest the suite gets to mocking — it's necessary because subprocess is
|
|
the boundary that real I/O lives on.
|
|
"""
|
|
|
|
def test_mount_success(self, monkeypatch):
|
|
calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return _FakeCompletedProcess(returncode=0)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
executor = SubprocessMountExecutor()
|
|
result = executor.mount("UUID=1", "/media/x", "ext4", "defaults")
|
|
assert result == MountResult(ok=True)
|
|
assert calls == [["mount", "-t", "ext4", "-o", "defaults",
|
|
"UUID=1", "/media/x"]]
|
|
|
|
def test_ntfs_mount_runs_ntfsfix_first(self, monkeypatch):
|
|
"""NTFS mounts run ntfsfix -d first to clear the Windows dirty flag;
|
|
UUID= specs are translated to /dev/disk/by-uuid for ntfsfix."""
|
|
calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return _FakeCompletedProcess(returncode=0)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
executor = SubprocessMountExecutor()
|
|
result = executor.mount("UUID=ABC123", "/media/x", "ntfs", "defaults")
|
|
assert result == MountResult(ok=True)
|
|
assert calls[0] == ["ntfsfix", "-d", "/dev/disk/by-uuid/ABC123"]
|
|
assert calls[1][0] == "mount"
|
|
|
|
def test_ntfsfix_failure_does_not_block_mount(self, monkeypatch):
|
|
def fake_run(cmd, **kwargs):
|
|
if cmd[0] == "ntfsfix":
|
|
return _FakeCompletedProcess(returncode=1, stderr="dirty")
|
|
return _FakeCompletedProcess(returncode=0)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
executor = SubprocessMountExecutor()
|
|
result = executor.mount("/dev/sda1", "/media/x", "ntfs3", "defaults")
|
|
assert result == MountResult(ok=True)
|
|
|
|
def test_ext4_mount_does_not_run_ntfsfix(self, monkeypatch):
|
|
calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return _FakeCompletedProcess(returncode=0)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
executor = SubprocessMountExecutor()
|
|
executor.mount("UUID=1", "/media/x", "ext4", "defaults")
|
|
assert len(calls) == 1
|
|
assert calls[0][0] == "mount"
|
|
|
|
def test_missing_binary_returns_error_instead_of_raising(self, monkeypatch):
|
|
"""The executor protocol forbids raising for I/O failures; a missing
|
|
binary (FileNotFoundError) must come back as MountResult(ok=False)."""
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
raise FileNotFoundError(2, "No such file or directory", cmd[0])
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
executor = SubprocessMountExecutor()
|
|
result = executor.mount("UUID=1", "/media/x", "ext4", "defaults")
|
|
assert result.ok is False
|
|
assert "No such file" in result.error
|
|
|
|
def test_mount_propagates_stderr(self, monkeypatch):
|
|
def fake_run(cmd, **kwargs):
|
|
return _FakeCompletedProcess(returncode=32, stderr="busy")
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
result = SubprocessMountExecutor().mount(
|
|
"UUID=1", "/media/x", "ext4", "defaults")
|
|
assert result.ok is False
|
|
assert "busy" in (result.error or "")
|
|
|
|
def test_mount_timeout(self, monkeypatch):
|
|
import subprocess as _sp
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
raise _sp.TimeoutExpired(cmd=cmd, timeout=30)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
result = SubprocessMountExecutor().mount(
|
|
"UUID=1", "/media/x", "ext4", "defaults")
|
|
assert result.ok is False
|
|
assert "timed out" in (result.error or "")
|
|
|
|
def test_unmount_command_shape(self, monkeypatch):
|
|
calls = []
|
|
|
|
def fake_run(cmd, **kwargs):
|
|
calls.append(cmd)
|
|
return _FakeCompletedProcess(returncode=0)
|
|
|
|
monkeypatch.setattr(
|
|
"controller.lib.mount_executor.subprocess.run", fake_run,
|
|
)
|
|
SubprocessMountExecutor().unmount("/media/x")
|
|
assert calls == [["umount", "/media/x"]]
|
|
|
|
def test_current_mount_points_parses_proc_mounts(self, monkeypatch, tmp_path):
|
|
# Redirect /proc/mounts to a temp file the executor will read.
|
|
proc_mounts = tmp_path / "proc-mounts"
|
|
proc_mounts.write_text(
|
|
"proc /proc proc rw 0 0\n"
|
|
"/dev/sda1 / ext4 rw 0 0\n"
|
|
"/dev/sdb1 /media/wrolpi/usb ext4 rw 0 0\n"
|
|
"tmpfs /tmp tmpfs rw 0 0\n"
|
|
"/dev/sdc1 /media/photos exfat rw 0 0\n"
|
|
)
|
|
|
|
# Patch only the open(...) lookup the executor performs. We do this
|
|
# via monkeypatch on builtins.open scoped to this test only.
|
|
original_open = open
|
|
|
|
def open_redirect(path, *args, **kwargs):
|
|
if path == "/proc/mounts":
|
|
return original_open(proc_mounts, *args, **kwargs)
|
|
return original_open(path, *args, **kwargs)
|
|
|
|
monkeypatch.setattr("builtins.open", open_redirect)
|
|
|
|
executor = SubprocessMountExecutor()
|
|
assert executor.current_mount_points("/media") == {
|
|
"/media/wrolpi/usb", "/media/photos",
|
|
}
|