mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 02:48:38 -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>
314 lines
12 KiB
Python
314 lines
12 KiB
Python
"""
|
|
Tests for controller.lib.reconciler.
|
|
|
|
The reconciler is pure orchestration: it reads fstab.yaml, asks an injected
|
|
MountExecutor what is live, then issues mount/unmount calls. These tests
|
|
use FakeMountExecutor + real fstab.yaml files in tmp_path. No mocks, no
|
|
patches.
|
|
|
|
Every test reads top-to-bottom:
|
|
1. set up fstab.yaml on disk
|
|
2. construct a FakeMountExecutor with the live state at the start of
|
|
the test
|
|
3. run reconciler.apply()
|
|
4. assert on the result and on the executor's recorded calls
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save as save_fstab
|
|
from controller.lib.mount_executor import FakeMountExecutor, MountCall, UnmountCall
|
|
from controller.lib.reconciler import (
|
|
RESERVED_MOUNT_POINTS,
|
|
ReconcileResult,
|
|
Reconciler,
|
|
)
|
|
|
|
|
|
# --- Helpers --------------------------------------------------------------
|
|
|
|
def make_reconciler(
|
|
tmp_path: Path,
|
|
*,
|
|
fstab: FstabFile = None,
|
|
executor: FakeMountExecutor = None,
|
|
managed: set[str] = None,
|
|
wrolpi_uid: int = 1001,
|
|
wrolpi_gid: int = 1001,
|
|
) -> tuple[Reconciler, FakeMountExecutor]:
|
|
fstab_path = tmp_path / "fstab.yaml"
|
|
managed_path = tmp_path / "managed"
|
|
save_fstab(fstab if fstab is not None else FstabFile(), fstab_path)
|
|
if managed:
|
|
managed_path.write_text("\n".join(sorted(managed)) + "\n")
|
|
fake = executor if executor is not None else FakeMountExecutor()
|
|
rec = Reconciler(
|
|
executor=fake,
|
|
fstab_path=fstab_path,
|
|
managed_path=managed_path,
|
|
wrolpi_uid=wrolpi_uid,
|
|
wrolpi_gid=wrolpi_gid,
|
|
)
|
|
return rec, fake
|
|
|
|
|
|
def _entry(device="UUID=1", mount_point="/media/wrolpi/usb",
|
|
fstype="ext4", options="defaults"):
|
|
return FstabEntry(device, mount_point, fstype, options)
|
|
|
|
|
|
# --- Mount path -----------------------------------------------------------
|
|
|
|
class TestMounts:
|
|
|
|
def test_empty_fstab_is_noop(self, tmp_path):
|
|
rec, fake = make_reconciler(tmp_path)
|
|
result = rec.apply()
|
|
assert result == ReconcileResult()
|
|
assert fake.mount_calls == []
|
|
assert fake.unmount_calls == []
|
|
|
|
def test_mounts_a_missing_entry(self, tmp_path):
|
|
rec, fake = make_reconciler(tmp_path, fstab=FstabFile(mounts=[_entry()]))
|
|
result = rec.apply()
|
|
assert result.mounted == ["/media/wrolpi/usb"]
|
|
assert fake.mount_calls == [MountCall(
|
|
"UUID=1", "/media/wrolpi/usb", "ext4", "defaults",
|
|
)]
|
|
assert fake.unmount_calls == []
|
|
|
|
def test_skips_already_mounted(self, tmp_path):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/wrolpi/usb"}))
|
|
rec, fake = make_reconciler(
|
|
tmp_path, fstab=FstabFile(mounts=[_entry()]), executor=fake)
|
|
result = rec.apply()
|
|
assert result.mounted == []
|
|
assert fake.mount_calls == []
|
|
# Diagnostics explain the "nothing to do" run in the journal.
|
|
assert result.desired_count == 1
|
|
assert result.already_mounted == ["/media/wrolpi/usb"]
|
|
|
|
def test_empty_fstab_reports_zero_desired(self, tmp_path):
|
|
rec, fake = make_reconciler(tmp_path, fstab=FstabFile())
|
|
result = rec.apply()
|
|
assert result.desired_count == 0
|
|
assert result.already_mounted == []
|
|
|
|
def test_mount_failure_is_recorded(self, tmp_path):
|
|
fake = FakeMountExecutor(fail_for_mount=frozenset({"/media/wrolpi/usb"}))
|
|
rec, fake = make_reconciler(
|
|
tmp_path, fstab=FstabFile(mounts=[_entry()]), executor=fake)
|
|
result = rec.apply()
|
|
assert result.mounted == []
|
|
assert len(result.mount_failures) == 1
|
|
assert result.mount_failures[0][0] == "/media/wrolpi/usb"
|
|
assert result.all_succeeded is False
|
|
|
|
def test_continues_after_individual_failure(self, tmp_path):
|
|
# First entry fails; second still gets mounted — never block the
|
|
# boot because of one bad drive.
|
|
fake = FakeMountExecutor(fail_for_mount=frozenset({"/media/wrolpi/usb"}))
|
|
fstab = FstabFile(mounts=[
|
|
_entry(device="UUID=1", mount_point="/media/wrolpi/usb"),
|
|
_entry(device="UUID=2", mount_point="/media/wrolpi/2tb"),
|
|
])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab, executor=fake)
|
|
result = rec.apply()
|
|
assert result.mounted == ["/media/wrolpi/2tb"]
|
|
assert len(result.mount_failures) == 1
|
|
|
|
|
|
# --- Unmount path ---------------------------------------------------------
|
|
|
|
class TestUnmounts:
|
|
|
|
def test_unmounts_removed_managed_path(self, tmp_path):
|
|
# Previously mounted by us, still live, but no longer in fstab.yaml
|
|
# — must be unmounted.
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/wrolpi/usb"}))
|
|
rec, fake = make_reconciler(
|
|
tmp_path,
|
|
fstab=FstabFile(),
|
|
managed={"/media/wrolpi/usb"},
|
|
executor=fake,
|
|
)
|
|
result = rec.apply()
|
|
assert result.unmounted == ["/media/wrolpi/usb"]
|
|
assert fake.unmount_calls == [UnmountCall("/media/wrolpi/usb")]
|
|
|
|
def test_does_not_unmount_unmanaged_path(self, tmp_path):
|
|
# User has /media/photos mounted manually — fstab.yaml is empty and
|
|
# managed set is empty. Reconciler must NOT touch /media/photos.
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/photos"}))
|
|
rec, fake = make_reconciler(tmp_path, fstab=FstabFile(), executor=fake)
|
|
result = rec.apply()
|
|
assert result.unmounted == []
|
|
assert fake.unmount_calls == []
|
|
|
|
def test_does_not_unmount_when_still_in_fstab(self, tmp_path):
|
|
# In managed AND in fstab.yaml AND live — leave alone.
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/wrolpi/usb"}))
|
|
rec, fake = make_reconciler(
|
|
tmp_path,
|
|
fstab=FstabFile(mounts=[_entry()]),
|
|
managed={"/media/wrolpi/usb"},
|
|
executor=fake,
|
|
)
|
|
result = rec.apply()
|
|
assert result.unmounted == []
|
|
assert fake.unmount_calls == []
|
|
|
|
def test_unmount_failure_keeps_path_in_managed_set(self, tmp_path):
|
|
fake = FakeMountExecutor(
|
|
initially_mounted=frozenset({"/media/wrolpi/usb"}),
|
|
fail_for_unmount=frozenset({"/media/wrolpi/usb"}),
|
|
)
|
|
rec, fake = make_reconciler(
|
|
tmp_path,
|
|
fstab=FstabFile(),
|
|
managed={"/media/wrolpi/usb"},
|
|
executor=fake,
|
|
)
|
|
result = rec.apply()
|
|
assert result.unmounted == []
|
|
assert len(result.unmount_failures) == 1
|
|
# Path still appears managed so a future retry can unmount it.
|
|
managed_after = (tmp_path / "managed").read_text().splitlines()
|
|
assert "/media/wrolpi/usb" in managed_after
|
|
|
|
def test_stale_managed_path_not_live_is_dropped(self, tmp_path):
|
|
# Reconciler restarts after a reboot; managed file says we owned
|
|
# /media/wrolpi/usb but it's not currently mounted (live-boot
|
|
# didn't restore it). Just forget it; don't try to unmount.
|
|
fake = FakeMountExecutor()
|
|
rec, fake = make_reconciler(
|
|
tmp_path,
|
|
fstab=FstabFile(),
|
|
managed={"/media/wrolpi/usb"},
|
|
executor=fake,
|
|
)
|
|
result = rec.apply()
|
|
assert result.unmounted == []
|
|
assert fake.unmount_calls == []
|
|
managed_after = (tmp_path / "managed").read_text()
|
|
assert "/media/wrolpi/usb" not in managed_after
|
|
|
|
|
|
# --- Reserved + outside-/media protection --------------------------------
|
|
|
|
class TestSkips:
|
|
|
|
def test_reserved_mount_point_is_skipped(self, tmp_path):
|
|
# /media/wrolpi appearing in fstab.yaml (e.g. via migration of the
|
|
# primary line) must be ignored, not re-mounted.
|
|
for mp in RESERVED_MOUNT_POINTS:
|
|
fstab = FstabFile(mounts=[_entry(mount_point=mp)])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab)
|
|
result = rec.apply()
|
|
assert fake.mount_calls == [], mp
|
|
assert (mp, "reserved mount point") in result.skipped, mp
|
|
|
|
def test_outside_media_is_skipped(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry(mount_point="/mnt/elsewhere")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab)
|
|
result = rec.apply()
|
|
assert fake.mount_calls == []
|
|
assert any(
|
|
mp == "/mnt/elsewhere" and "outside /media" in reason
|
|
for mp, reason in result.skipped
|
|
)
|
|
|
|
def test_missing_device_or_mount_point_is_skipped(self, tmp_path):
|
|
fstab = FstabFile(mounts=[
|
|
FstabEntry(device="", mount_point="/media/wrolpi/x", fstype="ext4"),
|
|
FstabEntry(device="UUID=1", mount_point="", fstype="ext4"),
|
|
])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab)
|
|
result = rec.apply()
|
|
assert fake.mount_calls == []
|
|
assert len(result.skipped) == 2
|
|
|
|
|
|
# --- exfat/vfat option injection -----------------------------------------
|
|
|
|
class TestUidGidInjection:
|
|
|
|
def test_injects_uid_gid_for_exfat(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry(fstype="exfat")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab,
|
|
wrolpi_uid=2000, wrolpi_gid=2000)
|
|
rec.apply()
|
|
opts = fake.mount_calls[0].options
|
|
assert "uid=2000" in opts
|
|
assert "gid=2000" in opts
|
|
|
|
def test_injects_uid_gid_for_ntfs(self, tmp_path):
|
|
"""ntfs/ntfs3 have no POSIX permissions either; secondary NTFS drives
|
|
must mount owned by the wrolpi user like exfat/vfat do."""
|
|
fstab = FstabFile(mounts=[_entry(fstype="ntfs")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab,
|
|
wrolpi_uid=2000, wrolpi_gid=2000)
|
|
rec.apply()
|
|
opts = fake.mount_calls[0].options
|
|
assert "uid=2000" in opts
|
|
assert "gid=2000" in opts
|
|
|
|
def test_does_not_double_inject_if_user_already_supplied(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry(fstype="vfat",
|
|
options="defaults,uid=1234,gid=1234")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab,
|
|
wrolpi_uid=2000, wrolpi_gid=2000)
|
|
rec.apply()
|
|
opts = fake.mount_calls[0].options
|
|
assert opts.count("uid=") == 1
|
|
assert "uid=1234" in opts
|
|
|
|
def test_ext4_options_untouched(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry(fstype="ext4", options="defaults,noatime")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab,
|
|
wrolpi_uid=2000, wrolpi_gid=2000)
|
|
rec.apply()
|
|
opts = fake.mount_calls[0].options
|
|
assert opts == "defaults,noatime"
|
|
|
|
def test_no_uid_gid_when_unknown(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry(fstype="exfat")])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab,
|
|
wrolpi_uid=None, wrolpi_gid=None)
|
|
rec.apply()
|
|
opts = fake.mount_calls[0].options
|
|
assert "uid=" not in opts
|
|
|
|
|
|
# --- Managed-set persistence ---------------------------------------------
|
|
|
|
class TestManagedSetPersistence:
|
|
|
|
def test_first_mount_writes_managed_entry(self, tmp_path):
|
|
rec, fake = make_reconciler(tmp_path, fstab=FstabFile(mounts=[_entry()]))
|
|
rec.apply()
|
|
managed_after = (tmp_path / "managed").read_text().splitlines()
|
|
assert "/media/wrolpi/usb" in managed_after
|
|
|
|
def test_successful_unmount_removes_from_managed(self, tmp_path):
|
|
fake = FakeMountExecutor(initially_mounted=frozenset({"/media/wrolpi/usb"}))
|
|
rec, fake = make_reconciler(
|
|
tmp_path,
|
|
fstab=FstabFile(),
|
|
managed={"/media/wrolpi/usb"},
|
|
executor=fake,
|
|
)
|
|
rec.apply()
|
|
managed_after = (tmp_path / "managed").read_text()
|
|
assert "/media/wrolpi/usb" not in managed_after
|
|
|
|
def test_idempotent_runs_keep_managed_stable(self, tmp_path):
|
|
fstab = FstabFile(mounts=[_entry()])
|
|
rec, fake = make_reconciler(tmp_path, fstab=fstab)
|
|
rec.apply()
|
|
first = (tmp_path / "managed").read_text()
|
|
rec.apply()
|
|
second = (tmp_path / "managed").read_text()
|
|
assert first == second
|
|
# Mount called exactly once (second run sees it live and skips).
|
|
assert len(fake.mount_calls) == 1
|