rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 02:48:38 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Moving WROLPi-managed mounts off /etc/fstab onto a drive-resident fstab.yaml.

Browse source 
A WROLPi drive can move between hosts and across deployment styles
(Portable, RPi, Debian).  /etc/fstab is host-specific and on the Portable
image it's part of a read-only squashfs, so persisting secondary mounts
there was both wrong-host and lost-on-reboot.  This change makes
/media/wrolpi/config/fstab.yaml the single source of truth for WROLPi-
managed mounts and introduces a new wrolpi-mounts.service that reconciles
live state to match.

Architecture:

- The Controller writes intent (fstab.yaml) but never calls mount(8) or
  umount(8) itself.  Every mutation triggers `systemctl restart
  wrolpi-mounts.service`; the Controller then verifies the outcome via
  the same MountExecutor abstraction the service uses.

- wrolpi-mounts.service runs a thin entry point that constructs a real
  Reconciler with a SubprocessMountExecutor and calls apply().  The
  reconciler mounts what's in fstab.yaml but missing from the live state,
  and unmounts paths it previously claimed that are no longer in
  fstab.yaml.  A managed-set file in /run/wrolpi-mounts/managed prevents
  the reconciler from ever touching user-mounted paths it didn't own.

- One-time migration (marker-gated) lifts both /etc/fstab WROLPi entries
  and any controller.yaml drives.mounts entries into fstab.yaml.
  /etc/fstab is backed up to /etc/fstab.wrolpi-migration.backup;
  migrated lines are commented in place with a "migrated to fstab.yaml:"
  prefix so a host admin can hand-recover.

Test approach: dependency injection in place of patching.  The
MountExecutor protocol has SubprocessMountExecutor (production) and
FakeMountExecutor (in-memory, records calls).  The Reconciler accepts an
executor in its constructor, so the 20 reconciler tests + 13 migration
tests + 18 fstab_yaml tests use real files in tmp_path with zero
mock.patch calls.  test_disks_api.py drops to mocks only for the
host-touching read helpers; the mount/unmount paths now run the real
Reconciler against the fake executor.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Roland 2026-05-22 22:02:30 -06:00 • committed by lrnselfreliance
parent 94d56f313a
commit 719905cb77
15 changed files with 2257 additions and 320 deletions
Download patch file Download diff file Expand all files Collapse all files

292
controller/api/disks.py
View file

@ -1,45 +1,67 @@
"""
Disk management API endpoints for WROLPi Controller.
The Controller defines what should be mounted (writes fstab.yaml on the
WROLPi drive) but never calls mount(8) or umount(8) itself. After every
write to fstab.yaml the Controller triggers wrolpi-mounts.service via
``systemctl restart``; the service reconciles live state to match. The
Controller verifies the outcome by reading /proc/mounts through the same
MountExecutor abstraction the service uses.
"""
import logging
import subprocess
from typing import Optional
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from controller.lib.config import (
get_config_value,
is_docker_mode,
save_config,
update_config,
)
from controller.lib.config import is_docker_mode
from controller.lib.disks import (
check_shadowed_data,
get_block_devices,
get_mounts,
mount_drive,
unmount_drive,
get_uuid,
validate_mount_point,
)
from controller.lib.fstab import (
add_fstab_entry,
get_wrolpi_fstab_entries,
remove_fstab_entry,
from controller.lib.fstab_yaml import (
FstabEntry,
load as load_fstab,
save as save_fstab,
)
from controller.lib.mount_executor import (
MountExecutor,
SubprocessMountExecutor,
)
from controller.lib.smart import (
get_all_smart_status,
is_smart_available,
)
from controller.lib.wrol_mode import require_normal_mode
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/disks", tags=["disks"])
# Module-level executor — production uses the real subprocess-backed one.
# Tests rebind this attribute (and ``_trigger_reconcile``) via monkeypatch
# to inject a FakeMountExecutor without touching subprocess or /proc.
_executor: MountExecutor = SubprocessMountExecutor()
RECONCILE_TIMEOUT_SECONDS = 60
class MountRequest(BaseModel):
"""Request model for mounting a drive."""
device: str = Field(description="Device path (e.g., /dev/sda1)")
mount_point: str = Field(description="Where to mount (must be under /media)")
fstype: Optional[str] = Field(default=None, description="Filesystem type (auto-detected if not specified)")
fstype: Optional[str] = Field(default=None,
description="Filesystem type (auto-detected if not specified)")
options: str = Field(default="defaults", description="Mount options")
persist: bool = Field(default=False, description="Add to fstab for persistent mounting")
# `persist` is accepted for API compatibility but no longer
# meaningful: every Controller-initiated mount goes through fstab.yaml
# and the Reconciler. A future UI cleanup will remove the field.
persist: bool = Field(default=True, description="(Deprecated; always persistent.)")
force_shadowed: bool = Field(
default=False,
description="Proceed even if existing data at the mount target would be hidden by the new mount",
@ -47,17 +69,17 @@ class MountRequest(BaseModel):
class UnmountRequest(BaseModel):
"""Request model for unmounting a drive."""
mount_point: str = Field(description="Mount point to unmount")
lazy: bool = Field(default=False, description="Use lazy unmount if busy")
lazy: bool = Field(default=False,
description="(Deprecated; reconciler chooses the strategy.)")
class FstabRequest(BaseModel):
"""Request model for adding an fstab entry."""
device: str = Field(description="Device path")
mount_point: str = Field(description="Where to mount")
fstype: str = Field(description="Filesystem type")
options: str = Field(default="defaults,nofail,x-systemd.device-timeout=10s", description="Mount options")
options: str = Field(default="defaults",
description="Mount options")
def _check_native_mode():
@ -70,11 +92,43 @@ def _check_native_mode():
)
def _to_uuid_spec(device: str) -> str:
"""Convert a /dev/sdX path to UUID=... if possible; passthrough otherwise.
Persistent mount entries must be UUID-keyed so they travel with the
drive between hosts. Already-formatted UUID=/PARTUUID=/LABEL= specs
pass through unchanged.
"""
if not device.startswith("/dev/"):
return device
uuid = get_uuid(device)
return f"UUID={uuid}" if uuid else device
def _trigger_reconcile() -> tuple[bool, Optional[str]]:
"""Ask systemd to re-run wrolpi-mounts.service. Blocks until the
oneshot completes; returns (ok, error_or_none)."""
cmd = ["systemctl", "restart", "wrolpi-mounts.service"]
try:
result = subprocess.run(
cmd, capture_output=True, text=True,
timeout=RECONCILE_TIMEOUT_SECONDS,
)
except subprocess.TimeoutExpired:
return False, f"systemctl restart wrolpi-mounts.service timed out " \
f"after {RECONCILE_TIMEOUT_SECONDS}s"
if result.returncode != 0:
return False, (result.stderr.strip()
or f"systemctl exit {result.returncode}")
return True, None
# --- Read endpoints (no fstab.yaml mutation) ------------------------------
@router.get("")
async def list_disks():
"""List all detected disks/partitions."""
_check_native_mode()
devices = get_block_devices()
return [
{
@ -98,13 +152,15 @@ async def list_mounts():
return get_mounts()
# --- Mount / unmount ------------------------------------------------------
@router.post("/mount")
async def disk_mount(request: MountRequest):
"""Mount a partition."""
"""Add a mount to fstab.yaml and reconcile."""
_check_native_mode()
# Soft-block if the mount target already contains user data that would
# be hidden by the new mount. The UI re-submits with force_shadowed=True
# be hidden by the new mount. UI re-submits with force_shadowed=True
# to override.
if not request.force_shadowed:
shadowed = check_shadowed_data(request.mount_point)
@ -121,116 +177,165 @@ async def disk_mount(request: MountRequest):
),
}
result = mount_drive(
device=request.device,
try:
require_normal_mode("mount")
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
try:
validate_mount_point(request.mount_point)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
device_spec = _to_uuid_spec(request.device)
entry = FstabEntry(
device=device_spec,
mount_point=request.mount_point,
fstype=request.fstype,
options=request.options,
fstype=request.fstype or "auto",
options=request.options or "defaults",
)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Mount failed"))
data = load_fstab()
data.add_or_replace(entry)
save_fstab(data)
# Add to fstab if requested
if request.persist:
fstab_result = add_fstab_entry(
device=request.device,
mount_point=request.mount_point,
fstype=request.fstype or "auto",
ok, err = _trigger_reconcile()
if not ok:
raise HTTPException(status_code=500,
detail=f"Failed to reconcile mounts: {err}")
if request.mount_point not in _executor.current_mount_points():
# Entry is in fstab.yaml and will be retried on next reconcile, but
# this attempt did not produce a live mount — surface the failure.
raise HTTPException(
status_code=500,
detail=f"Reconciler ran but {request.mount_point} did not mount. "
f"Check journalctl -u wrolpi-mounts.service.",
)
result["fstab"] = fstab_result
# Save to controller.yaml to track WROLPi-managed mounts
if fstab_result.get("success"):
mount_entry = {
"device": fstab_result.get("device", request.device),
"mount_point": request.mount_point,
"fstype": request.fstype or "auto",
"options": request.options,
}
# Get current mounts, filter out any existing entry for same mount_point or device
current_mounts = get_config_value("drives.mounts", [])
current_mounts = [
m for m in current_mounts
if m.get("mount_point") != request.mount_point
and m.get("device") != mount_entry["device"]
]
current_mounts.append(mount_entry)
update_config("drives.mounts", current_mounts)
try:
save_config()
except RuntimeError:
pass # Primary drive not mounted - can't save yet
return result
return {
"success": True,
"device": device_spec,
"mount_point": request.mount_point,
}
@router.post("/unmount")
async def disk_unmount(request: UnmountRequest):
"""Unmount a partition."""
"""Remove a mount from fstab.yaml and reconcile."""
_check_native_mode()
result = unmount_drive(
mount_point=request.mount_point,
lazy=request.lazy,
)
try:
require_normal_mode("unmount")
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Unmount failed"))
try:
validate_mount_point(request.mount_point)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return result
data = load_fstab()
removed = data.remove_by_mount_point(request.mount_point)
save_fstab(data)
ok, err = _trigger_reconcile()
if not ok:
raise HTTPException(status_code=500,
detail=f"Failed to reconcile mounts: {err}")
if request.mount_point in _executor.current_mount_points():
raise HTTPException(
status_code=500,
detail=f"Reconciler ran but {request.mount_point} is still mounted. "
f"Check journalctl -u wrolpi-mounts.service.",
)
return {
"success": True,
"mount_point": request.mount_point,
"removed_from_fstab": removed,
}
# --- fstab.yaml read/write endpoints --------------------------------------
@router.get("/fstab")
async def list_fstab():
"""Get WROLPi-related fstab entries."""
"""Return the WROLPi-managed mount table (fstab.yaml)."""
_check_native_mode()
return get_wrolpi_fstab_entries()
data = load_fstab()
return [
{
"type": "mount",
"device": e.device,
"mount_point": e.mount_point,
"fstype": e.fstype,
"options": e.options,
"dump": "0",
"pass": "2",
}
for e in data.mounts
]
@router.post("/fstab")
async def add_fstab(request: FstabRequest):
"""Add an fstab entry for persistent mounting."""
"""Add a row to fstab.yaml without triggering a reconcile.
Use disk_mount() to add AND apply. This endpoint is for callers that
want to stage changes without immediately mounting.
"""
_check_native_mode()
result = add_fstab_entry(
device=request.device,
try:
require_normal_mode("add fstab entry")
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
try:
validate_mount_point(request.mount_point)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
device_spec = _to_uuid_spec(request.device)
data = load_fstab()
data.add_or_replace(FstabEntry(
device=device_spec,
mount_point=request.mount_point,
fstype=request.fstype,
options=request.options,
)
options=request.options or "defaults",
))
save_fstab(data)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Failed"))
return result
return {"success": True, "device": device_spec,
"mount_point": request.mount_point}
@router.delete("/fstab/{mount_point:path}")
async def delete_fstab(mount_point: str):
"""Remove an fstab entry."""
"""Remove a row from fstab.yaml without triggering a reconcile."""
_check_native_mode()
# URL decode the mount point
mount_point = f"/{mount_point}"
result = remove_fstab_entry(mount_point)
if not result.get("success"):
raise HTTPException(status_code=500, detail=result.get("error", "Failed"))
# Remove from controller.yaml - mount is no longer persistent
current_mounts = get_config_value("drives.mounts", [])
current_mounts = [m for m in current_mounts if m.get("mount_point") != mount_point]
update_config("drives.mounts", current_mounts)
try:
save_config()
except RuntimeError:
pass # Primary drive not mounted - can't save
require_normal_mode("remove fstab entry")
except PermissionError as e:
raise HTTPException(status_code=403, detail=str(e))
return result
try:
validate_mount_point(mount_point)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
data = load_fstab()
if not data.remove_by_mount_point(mount_point):
raise HTTPException(status_code=404,
detail=f"No entry found for {mount_point}")
save_fstab(data)
return {"success": True, "mount_point": mount_point}
# --- SMART ---
@ -241,7 +346,8 @@ async def list_smart():
_check_native_mode()
if not is_smart_available():
return {"available": False, "reason": "pySMART not installed or not supported"}
return {"available": False,
"reason": "pySMART not installed or not supported"}
return {
"available": True,

229
controller/lib/fstab_migration.py Normal file
View file

@ -0,0 +1,229 @@
"""
One-time migration of WROLPi-managed mount entries into fstab.yaml.
Two legacy sources exist:
1. ``/etc/fstab`` — older WROLPi versions wrote WROLPi-managed mounts as
ordinary fstab lines so systemd-fstab-generator would mount them at
boot. On Portable this is read-only squashfs (writes go to RAM and
evaporate), and on any host an /etc/fstab entry is tied to *that* host,
not the drive.
2. ``controller.yaml drives.mounts`` — a brief intermediate iteration
stored the mount table alongside Controller settings in
``/media/wrolpi/config/controller.yaml``. Lives on the drive but
conflates "definitions" with "applier"; we split them out into a
dedicated file.
Both sources feed into ``/media/wrolpi/config/fstab.yaml``. Each source
is migrated at most once (gated by a marker file). /etc/fstab is backed
up; migrated lines are commented in place rather than deleted so the host
admin can hand-recover if something goes wrong.
"""
import logging
import shutil
from pathlib import Path
import yaml
from controller.lib.fstab_yaml import (
DEFAULT_PATH as DEFAULT_FSTAB_YAML_PATH,
FstabEntry,
load as load_fstab_yaml,
save as save_fstab_yaml,
)
logger = logging.getLogger(__name__)
FSTAB_PATH = Path("/etc/fstab")
CONTROLLER_YAML_PATH = Path("/media/wrolpi/config/controller.yaml")
MARKER_PATH = Path("/media/wrolpi/config/.fstab-migrated")
BACKUP_PATH = Path("/etc/fstab.wrolpi-migration.backup")
# Mount points that legitimately appear in /etc/fstab or drives.mounts but
# must not be lifted into fstab.yaml: the primary's mount is host-specific
# (live-boot, /etc/fstab installer entry, or repair.sh sets it up directly)
# and the temp onboarding mount belongs to the Controller's onboarding flow.
EXCLUDED_MIGRATION_MOUNT_POINTS = frozenset({
"/media/wrolpi",
"/media/wrolpi_temp_onboarding",
})
MIGRATED_PREFIX = "# migrated to fstab.yaml:"
# --- /etc/fstab parser ----------------------------------------------------
def _parse_fstab_text(text: str) -> list[dict]:
"""Parse the contents of an fstab file into structured entries."""
entries: list[dict] = []
for raw in text.splitlines(keepends=False):
stripped = raw.strip()
if not stripped or stripped.startswith("#"):
entries.append({"type": "comment", "line": raw})
continue
parts = stripped.split()
if len(parts) >= 4:
entries.append({
"type": "mount",
"device": parts[0],
"mount_point": parts[1],
"fstype": parts[2],
"options": parts[3],
"line": raw,
})
else:
entries.append({"type": "comment", "line": raw})
return entries
def _is_migration_candidate(entry: dict) -> bool:
if entry.get("type") != "mount":
return False
mp = entry.get("mount_point", "")
if not mp.startswith("/media"):
return False
if mp in EXCLUDED_MIGRATION_MOUNT_POINTS:
return False
return True
# --- controller.yaml helpers ----------------------------------------------
def _read_controller_yaml(path: Path) -> dict:
if not path.exists():
return {}
try:
return yaml.safe_load(path.read_text()) or {}
except (OSError, yaml.YAMLError) as e:
logger.warning("Could not parse %s, treating as empty: %s", path, e)
return {}
def _save_controller_yaml(path: Path, data: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(yaml.safe_dump(data, default_flow_style=False, sort_keys=True))
# --- Main entry point -----------------------------------------------------
def migrate_etc_fstab(
*,
fstab_path: Path = FSTAB_PATH,
controller_yaml_path: Path = CONTROLLER_YAML_PATH,
fstab_yaml_path: Path = None,
marker_path: Path = MARKER_PATH,
backup_path: Path = BACKUP_PATH,
) -> dict:
"""Lift WROLPi-managed mounts from /etc/fstab and
controller.yaml drives.mounts into fstab.yaml.
Returns ``{migrated, reason?, count}`` where ``count`` is the number of
new entries added to fstab.yaml.
"""
if marker_path.exists():
return {"migrated": False, "reason": "already-migrated", "count": 0}
fstab_yaml_path = fstab_yaml_path or DEFAULT_FSTAB_YAML_PATH
# --- Gather candidates from both sources ----------------------------
etc_fstab_candidates: list[dict] = []
if fstab_path.exists():
try:
etc_fstab_text = fstab_path.read_text()
except OSError as e:
logger.warning("Could not read %s: %s", fstab_path, e)
etc_fstab_text = ""
etc_fstab_entries = _parse_fstab_text(etc_fstab_text)
etc_fstab_candidates = [
e for e in etc_fstab_entries if _is_migration_candidate(e)
]
else:
etc_fstab_text = ""
etc_fstab_entries = []
controller_data = _read_controller_yaml(controller_yaml_path)
drives_mounts_candidates = []
for m in (controller_data.get("drives") or {}).get("mounts") or []:
mp = m.get("mount_point", "")
if not mp.startswith("/media") or mp in EXCLUDED_MIGRATION_MOUNT_POINTS:
continue
drives_mounts_candidates.append(m)
# --- Backup /etc/fstab before any modification ----------------------
if etc_fstab_candidates:
try:
shutil.copy2(fstab_path, backup_path)
except OSError as e:
logger.warning("Could not back up %s: %s", fstab_path, e)
# --- Merge into fstab.yaml (dedup by mount_point) -------------------
fstab_data = load_fstab_yaml(fstab_yaml_path)
existing = fstab_data.mount_points()
added = 0
for cand in etc_fstab_candidates:
if cand["mount_point"] in existing:
continue
fstab_data.add_or_replace(FstabEntry(
device=cand["device"],
mount_point=cand["mount_point"],
fstype=cand["fstype"],
options=cand.get("options") or "defaults",
))
existing.add(cand["mount_point"])
added += 1
for cand in drives_mounts_candidates:
mp = cand["mount_point"]
if mp in existing:
continue
fstab_data.add_or_replace(FstabEntry(
device=cand.get("device", ""),
mount_point=mp,
fstype=cand.get("fstype") or "auto",
options=cand.get("options") or "defaults",
))
existing.add(mp)
added += 1
if etc_fstab_candidates or drives_mounts_candidates:
save_fstab_yaml(fstab_data, fstab_yaml_path)
# --- Comment out migrated entries in /etc/fstab ---------------------
if etc_fstab_candidates:
candidate_mps = {e["mount_point"] for e in etc_fstab_candidates}
new_lines = []
for entry in etc_fstab_entries:
if (entry["type"] == "mount"
and entry["mount_point"] in candidate_mps):
new_lines.append(f"{MIGRATED_PREFIX} {entry['line']}")
else:
new_lines.append(entry["line"])
suffix = "\n" if etc_fstab_text.endswith("\n") else ""
try:
fstab_path.write_text("\n".join(new_lines) + suffix)
except OSError as e:
logger.warning("Could not rewrite %s: %s", fstab_path, e)
# --- Strip drives.mounts from controller.yaml -----------------------
if drives_mounts_candidates and (controller_data.get("drives") or {}).get("mounts"):
drives = controller_data.get("drives") or {}
drives.pop("mounts", None)
if not drives:
controller_data.pop("drives", None)
else:
controller_data["drives"] = drives
_save_controller_yaml(controller_yaml_path, controller_data)
# --- Mark done ------------------------------------------------------
marker_path.parent.mkdir(parents=True, exist_ok=True)
marker_path.touch()
logger.info(
"Migration complete: %d /etc/fstab candidate(s), %d drives.mounts "
"candidate(s), %d new entry/entries written to fstab.yaml",
len(etc_fstab_candidates), len(drives_mounts_candidates), added,
)
return {"migrated": True, "count": added}

146
controller/lib/fstab_yaml.py Normal file
View file

@ -0,0 +1,146 @@
"""
Reader/writer for /media/wrolpi/config/fstab.yaml.
This file is the WROLPi-managed mount table. It lives on the drive itself
(so it travels between hosts) and is the sole input to the Reconciler.
Schema::
version: 1
mounts:
- device: UUID=... # filesystem UUID; the entry is portable
mount_point: /media/... # must be under /media
fstype: ext4
options: defaults # mount(8) -o options
The Controller defines what should be mounted by editing this file; the
Reconciler (invoked by wrolpi-mounts.service) applies the diff between this
file and the live mount state.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
import yaml
logger = logging.getLogger(__name__)
DEFAULT_PATH = Path("/media/wrolpi/config/fstab.yaml")
SCHEMA_VERSION = 1
@dataclass
class FstabEntry:
"""A single row in fstab.yaml."""
device: str
mount_point: str
fstype: str
options: str = "defaults"
def to_dict(self) -> dict:
return {
"device": self.device,
"mount_point": self.mount_point,
"fstype": self.fstype,
"options": self.options,
}
@classmethod
def from_dict(cls, data: dict) -> "FstabEntry":
return cls(
device=data.get("device", ""),
mount_point=data.get("mount_point", ""),
fstype=data.get("fstype", "auto"),
options=data.get("options") or "defaults",
)
@dataclass
class FstabFile:
"""In-memory representation of fstab.yaml."""
version: int = SCHEMA_VERSION
mounts: list[FstabEntry] = field(default_factory=list)
# --- Lookup helpers ------------------------------------------------
def find_by_mount_point(self, mount_point: str) -> Optional[FstabEntry]:
for m in self.mounts:
if m.mount_point == mount_point:
return m
return None
def mount_points(self) -> set[str]:
return {m.mount_point for m in self.mounts}
# --- Mutation -------------------------------------------------------
def add_or_replace(self, entry: FstabEntry) -> None:
"""Insert ``entry``, replacing any prior row that shares its
mount_point OR its device. The device clause handles the "remount
the same drive to a different path" case so we don't accumulate
stale rows."""
self.mounts = [
m for m in self.mounts
if m.mount_point != entry.mount_point and m.device != entry.device
]
self.mounts.append(entry)
def remove_by_mount_point(self, mount_point: str) -> bool:
"""Remove the entry at ``mount_point``. Returns True if a row was
removed, False if none matched."""
before = len(self.mounts)
self.mounts = [m for m in self.mounts if m.mount_point != mount_point]
return len(self.mounts) < before
# --- File I/O -------------------------------------------------------------
def load(path: Optional[Path] = None) -> FstabFile:
"""Read fstab.yaml. Returns an empty FstabFile if the file is absent
or malformed; both are normal startup conditions on a fresh drive.
``path`` defaults to the module-level ``DEFAULT_PATH`` at call time
(resolved on each invocation, so tests can redirect via monkeypatch).
"""
if path is None:
path = DEFAULT_PATH
if not path.exists():
return FstabFile()
try:
raw = yaml.safe_load(path.read_text()) or {}
except yaml.YAMLError as e:
logger.warning("fstab.yaml at %s is malformed (%s); treating as empty", path, e)
return FstabFile()
version = raw.get("version", SCHEMA_VERSION)
mounts = [FstabEntry.from_dict(m) for m in (raw.get("mounts") or [])]
return FstabFile(version=version, mounts=mounts)
def save(data: FstabFile, path: Optional[Path] = None) -> None:
"""Write fstab.yaml atomically (write-then-rename). Creates the parent
directory if needed. Sorted by mount_point so diffs stay tidy.
``path`` defaults to the module-level ``DEFAULT_PATH`` at call time so
tests can redirect via monkeypatch.
"""
if path is None:
path = DEFAULT_PATH
path.parent.mkdir(parents=True, exist_ok=True)
sorted_mounts = sorted(data.mounts, key=lambda m: m.mount_point)
serialised = yaml.safe_dump(
{
"version": data.version,
"mounts": [m.to_dict() for m in sorted_mounts],
},
default_flow_style=False,
sort_keys=True,
)
tmp = path.with_suffix(path.suffix + ".tmp")
tmp.write_text(serialised)
tmp.replace(path)

182
controller/lib/mount_executor.py Normal file
View file

@ -0,0 +1,182 @@
"""
The narrow interface between WROLPi's mount reconciler and the operating
system. Pulling these three operations behind a single abstraction lets the
reconciler stay pure-orchestration: tests construct a FakeMountExecutor that
records calls + holds an in-memory mount-table, with no patches against
subprocess, /proc/mounts, or pwd/grp.
Two implementations live here:
- SubprocessMountExecutor — production. Calls mount(8) / umount(8) via
subprocess and reads /proc/mounts.
- FakeMountExecutor — tests. Tracks a mount-point set in memory, records
every mount/unmount call, and accepts predetermined per-call failures via
`fail_for`.
"""
from __future__ import annotations
import logging
import subprocess
from dataclasses import dataclass, field
from typing import Optional, Protocol
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class MountResult:
"""Outcome of a single mount or unmount call."""
ok: bool
error: Optional[str] = None
class MountExecutor(Protocol):
"""The contract every executor must satisfy.
Implementations should never raise for I/O failures; instead, return a
MountResult(ok=False, error="…"). Raising is reserved for programmer
errors (bad arguments, etc.).
"""
def mount(
self,
device: str,
mount_point: str,
fstype: str,
options: str,
) -> MountResult: ...
def unmount(self, mount_point: str) -> MountResult: ...
def current_mount_points(self, prefix: str = "/media") -> set[str]: ...
# --- Production implementation -------------------------------------------
MOUNT_TIMEOUT_SECONDS = 30
class SubprocessMountExecutor:
"""Executes mount/umount via subprocess and parses /proc/mounts."""
def __init__(self, timeout_seconds: int = MOUNT_TIMEOUT_SECONDS):
self._timeout = timeout_seconds
def mount(
self,
device: str,
mount_point: str,
fstype: str,
options: str,
) -> MountResult:
cmd = ["mount", "-t", fstype, "-o", options, device, mount_point]
return self._run(cmd, label=f"mount {device} {mount_point}")
def unmount(self, mount_point: str) -> MountResult:
# Plain umount. If the mount is busy, the reconciler will see it
# still present on the next pass and retry; we deliberately do not
# add `-l` (lazy) here because lazy unmounts can hide bugs.
return self._run(["umount", mount_point], label=f"umount {mount_point}")
def current_mount_points(self, prefix: str = "/media") -> set[str]:
out: set[str] = set()
try:
with open("/proc/mounts") as f:
for line in f:
parts = line.split()
if len(parts) >= 2 and parts[1].startswith(prefix):
out.add(parts[1])
except OSError as e:
logger.error("Could not read /proc/mounts: %s", e)
return out
def _run(self, cmd: list[str], *, label: str) -> MountResult:
logger.info("Running: %s", " ".join(cmd))
try:
result = subprocess.run(
cmd, capture_output=True, text=True, timeout=self._timeout,
)
except subprocess.TimeoutExpired:
return MountResult(
ok=False,
error=f"{label}: timed out after {self._timeout}s",
)
if result.returncode != 0:
return MountResult(
ok=False,
error=f"{label}: {result.stderr.strip() or 'non-zero exit'}",
)
return MountResult(ok=True)
# --- Test fake ------------------------------------------------------------
@dataclass
class MountCall:
"""Recorded mount invocation, for test assertions."""
device: str
mount_point: str
fstype: str
options: str
@dataclass
class UnmountCall:
"""Recorded unmount invocation, for test assertions."""
mount_point: str
@dataclass
class FakeMountExecutor:
"""In-memory MountExecutor for tests.
Behaviour:
- mount(...) adds mount_point to the live set and records the call,
unless the call matches an entry in ``fail_for_mount`` (matched by
mount_point), in which case it returns a failure and the set is
unchanged.
- unmount(...) removes mount_point from the live set and records the
call, unless the call matches ``fail_for_unmount``.
- current_mount_points(prefix) returns a copy of the live set filtered
by prefix.
Tests can seed the live set via the ``initially_mounted`` argument.
"""
initially_mounted: frozenset[str] = frozenset()
fail_for_mount: frozenset[str] = frozenset()
fail_for_unmount: frozenset[str] = frozenset()
mount_calls: list[MountCall] = field(default_factory=list)
unmount_calls: list[UnmountCall] = field(default_factory=list)
_live: set[str] = field(init=False)
def __post_init__(self) -> None:
self._live = set(self.initially_mounted)
def mount(
self,
device: str,
mount_point: str,
fstype: str,
options: str,
) -> MountResult:
self.mount_calls.append(MountCall(device, mount_point, fstype, options))
if mount_point in self.fail_for_mount:
return MountResult(ok=False, error=f"fake: configured to fail mount {mount_point}")
self._live.add(mount_point)
return MountResult(ok=True)
def unmount(self, mount_point: str) -> MountResult:
self.unmount_calls.append(UnmountCall(mount_point))
if mount_point in self.fail_for_unmount:
return MountResult(ok=False, error=f"fake: configured to fail umount {mount_point}")
self._live.discard(mount_point)
return MountResult(ok=True)
def current_mount_points(self, prefix: str = "/media") -> set[str]:
return {mp for mp in self._live if mp.startswith(prefix)}

67
controller/lib/mount_runner.py Normal file
View file

@ -0,0 +1,67 @@
"""
Entry point for wrolpi-mounts.service.
This module is intentionally tiny: it wires up production dependencies
(real subprocess executor, real wrolpi uid/gid lookup, real fstab.yaml on
the drive) and asks the Reconciler to apply. All real work lives in
controller.lib.reconciler; tests for the orchestration live in
test_reconciler.py using FakeMountExecutor.
"""
from __future__ import annotations
import logging
import sys
from controller.lib.disks import get_wrolpi_uid_gid
from controller.lib.fstab_migration import migrate_etc_fstab
from controller.lib.mount_executor import SubprocessMountExecutor
from controller.lib.reconciler import Reconciler
logger = logging.getLogger(__name__)
def main(argv=None) -> int:
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s wrolpi-mounts: %(message)s",
)
# Migration is idempotent (marker-gated). Doing it here means a freshly
# upgraded WROLPi picks up legacy /etc/fstab entries on the first boot
# of the new mount system, before we try to apply them.
try:
migrate_etc_fstab()
except Exception as e: # noqa: BLE001 - never fail boot on migration error
logger.error("fstab migration raised, continuing without it: %s", e)
try:
wrolpi_uid, wrolpi_gid = get_wrolpi_uid_gid()
except Exception as e: # noqa: BLE001 - best-effort enrichment
logger.warning("Could not resolve wrolpi uid/gid (%s)", e)
wrolpi_uid, wrolpi_gid = None, None
reconciler = Reconciler(
executor=SubprocessMountExecutor(),
wrolpi_uid=wrolpi_uid,
wrolpi_gid=wrolpi_gid,
)
result = reconciler.apply()
logger.info(
"Reconcile complete: %d mounted, %d unmounted, %d mount failures, %d unmount failures",
len(result.mounted), len(result.unmounted),
len(result.mount_failures), len(result.unmount_failures),
)
for mp, err in result.mount_failures:
logger.error("Failed to mount %s: %s", mp, err)
for mp, err in result.unmount_failures:
logger.error("Failed to unmount %s: %s", mp, err)
# Always return 0: the service must not block boot if a single mount
# is broken. Failures are logged and surfaced via the journal.
return 0
if __name__ == "__main__":
sys.exit(main())

5
controller/lib/onboarding.py
View file

@ -297,8 +297,9 @@ def commit_onboarding(
mount_result = mount_drive(mount_device, mount_point, fstype=mount_fstype, options=mount_options)
if mount_result.get("success"):
mounted_mounts.append(mount_point)
# Add fstab entry
add_fstab_entry(mount_device, mount_point, mount_fstype, options=mount_options)
# No /etc/fstab write — entry already lives in drives.mounts
# (it's the source we just read), so wrolpi-mounts.service
# will re-apply it on every subsequent boot.
else:
logger.warning(
"Failed to mount secondary drive %s at %s: %s",

183
controller/lib/reconciler.py Normal file
View file

@ -0,0 +1,183 @@
"""
Reconcile the live mount state under /media/ with fstab.yaml.
The Reconciler reads the desired mount table from fstab.yaml and asks an
injected MountExecutor what is currently mounted. Then::
mount each entry in fstab.yaml that is missing from the live set
unmount each previously-managed path that has been removed from fstab.yaml
"Previously-managed" is the load-bearing concept: we never unmount paths
that WROLPi has never claimed. The set of managed mount points is
persisted to ``managed_path`` (a small newline-separated file) so the
reconciler can tell apart its own mounts from the user's manual ones
across reboots. On Portable this file is on tmpfs (overlay upper layer)
which is acceptable — each boot the reconciler starts from a clean managed
set and re-discovers everything from fstab.yaml.
The reconciler never raises for I/O failures. Per-entry failures are
collected into the ReconcileResult and reported, but the run continues so
a single broken drive does not block the rest of the boot.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
from controller.lib.fstab_yaml import DEFAULT_PATH as DEFAULT_FSTAB_PATH
from controller.lib.fstab_yaml import FstabEntry, FstabFile, load as load_fstab
from controller.lib.mount_executor import MountExecutor
logger = logging.getLogger(__name__)
DEFAULT_MANAGED_PATH = Path("/run/wrolpi-mounts/managed")
# Mount points the reconciler will never touch, even if they appear in
# fstab.yaml: the primary mount belongs to live-boot persistence (Portable)
# or the host's installer (/etc/fstab on RPi/Debian), and the temp mount
# belongs to the Controller's onboarding flow.
RESERVED_MOUNT_POINTS = frozenset({
"/media/wrolpi",
"/media/wrolpi_temp_onboarding",
})
@dataclass
class ReconcileResult:
"""Per-run summary suitable for logging or surfacing back to an API."""
mounted: list[str] = field(default_factory=list)
unmounted: list[str] = field(default_factory=list)
mount_failures: list[tuple[str, str]] = field(default_factory=list)
unmount_failures: list[tuple[str, str]] = field(default_factory=list)
skipped: list[tuple[str, str]] = field(default_factory=list)
@property
def all_succeeded(self) -> bool:
return not self.mount_failures and not self.unmount_failures
class Reconciler:
"""Apply fstab.yaml to the live mount state via an injected executor."""
def __init__(
self,
*,
executor: MountExecutor,
fstab_path: Path = DEFAULT_FSTAB_PATH,
managed_path: Path = DEFAULT_MANAGED_PATH,
wrolpi_uid: Optional[int] = None,
wrolpi_gid: Optional[int] = None,
):
self._executor = executor
self._fstab_path = fstab_path
self._managed_path = managed_path
self._wrolpi_uid = wrolpi_uid
self._wrolpi_gid = wrolpi_gid
# --- Public entry point ------------------------------------------
def apply(self) -> ReconcileResult:
"""Reconcile once. Idempotent."""
result = ReconcileResult()
desired = load_fstab(self._fstab_path)
live = self._executor.current_mount_points()
managed = self._read_managed()
# 1. Mount each entry in fstab.yaml that isn't already live.
for entry in desired.mounts:
skip = self._skip_reason(entry)
if skip:
result.skipped.append((entry.mount_point, skip))
continue
if entry.mount_point in live:
# Already mounted — claim ownership so a later removal can
# safely unmount it.
managed.add(entry.mount_point)
continue
options = self._inject_uid_gid(entry)
self._ensure_dir(entry.mount_point)
r = self._executor.mount(
entry.device, entry.mount_point, entry.fstype, options,
)
if r.ok:
result.mounted.append(entry.mount_point)
managed.add(entry.mount_point)
else:
result.mount_failures.append((entry.mount_point, r.error or "unknown"))
# 2. Unmount each previously-managed mount that is no longer in
# fstab.yaml. We only consider paths in the managed set, which
# protects user-mounted /media/* paths the reconciler never owned.
desired_set = desired.mount_points()
for mp in sorted(managed - desired_set):
if mp in RESERVED_MOUNT_POINTS:
continue
if mp not in live:
managed.discard(mp)
continue
r = self._executor.unmount(mp)
if r.ok:
result.unmounted.append(mp)
managed.discard(mp)
else:
result.unmount_failures.append((mp, r.error or "unknown"))
self._write_managed(managed)
return result
# --- Internals -----------------------------------------------------
def _skip_reason(self, entry: FstabEntry) -> Optional[str]:
if not entry.device or not entry.mount_point:
return "missing required field"
if entry.mount_point in RESERVED_MOUNT_POINTS:
return "reserved mount point"
if not entry.mount_point.startswith("/media/"):
return "mount point outside /media/"
return None
def _inject_uid_gid(self, entry: FstabEntry) -> str:
"""exfat/vfat have no POSIX permissions; without uid/gid the
mounted directory belongs to root and the wrolpi user cannot write
to it."""
options = entry.options or "defaults"
if entry.fstype not in ("exfat", "vfat"):
return options
if "uid=" in options or "gid=" in options:
return options
if self._wrolpi_uid is None or self._wrolpi_gid is None:
return options
return f"{options},uid={self._wrolpi_uid},gid={self._wrolpi_gid}"
def _ensure_dir(self, mount_point: str) -> None:
try:
Path(mount_point).mkdir(parents=True, exist_ok=True)
except OSError as e:
logger.warning("Could not create %s: %s", mount_point, e)
def _read_managed(self) -> set[str]:
if not self._managed_path.exists():
return set()
try:
return {
line.strip()
for line in self._managed_path.read_text().splitlines()
if line.strip()
}
except OSError as e:
logger.warning("Could not read %s: %s", self._managed_path, e)
return set()
def _write_managed(self, managed: set[str]) -> None:
try:
self._managed_path.parent.mkdir(parents=True, exist_ok=True)
self._managed_path.write_text(
"\n".join(sorted(managed)) + ("\n" if managed else "")
)
except OSError as e:
logger.warning("Could not write %s: %s", self._managed_path, e)

452
controller/test/test_disks_api.py
View file

@ -1,11 +1,63 @@
"""
Integration tests for Controller disks API endpoints.
The API writes fstab.yaml and asks wrolpi-mounts.service to reconcile.
Tests inject:
- a FakeMountExecutor in place of the production SubprocessMountExecutor,
- a fake ``_trigger_reconcile`` that simulates the service by running the
real Reconciler against the fake executor + a tmp_path fstab.yaml.
The only mocks are ``mock.patch`` for the read helpers (get_block_devices,
get_mounts, get_uuid, SMART) that touch the host's real disks. The
mount/unmount/fstab paths use real I/O against tmp_path and the
FakeMountExecutor.
"""
from pathlib import Path
from unittest import mock
import pytest
from controller.lib import fstab_yaml
from controller.lib.mount_executor import FakeMountExecutor
from controller.lib.reconciler import Reconciler
@pytest.fixture
def disks_env(monkeypatch, tmp_path):
"""Replace production wiring with a real Reconciler driving a fake
executor against a tmp_path fstab.yaml. Returns ``(fake, fstab_path)``.
Calling the API mutates the real fstab.yaml on disk and then runs the
real Reconciler.apply() — verifying the same code path that boot uses.
"""
fstab_path = tmp_path / "fstab.yaml"
managed_path = tmp_path / "managed"
monkeypatch.setattr(fstab_yaml, "DEFAULT_PATH", fstab_path)
fake = FakeMountExecutor()
def fake_trigger():
# Run the real reconciler. This is exactly the work the service
# would do when systemctl restarts it.
Reconciler(
executor=fake,
fstab_path=fstab_path,
managed_path=managed_path,
wrolpi_uid=1001,
wrolpi_gid=1001,
).apply()
return True, None
from controller.api import disks as disks_module
monkeypatch.setattr(disks_module, "_executor", fake)
monkeypatch.setattr(disks_module, "_trigger_reconcile", fake_trigger)
return {"fake": fake, "fstab_path": fstab_path}
# --- Docker mode ----------------------------------------------------------
class TestDockerModeRejectsDisksEndpoints:
"""All /api/disks/* endpoints should return 501 in Docker mode."""
@ -20,18 +72,19 @@ class TestDockerModeRejectsDisksEndpoints:
("delete", "/api/disks/fstab/media/test", None),
("get", "/api/disks/smart", None),
])
def test_disks_endpoint_rejected_in_docker(self, test_client_docker_mode, method, endpoint, payload):
"""Docker mode rejects every disks endpoint with 501."""
def test_disks_endpoint_rejected_in_docker(
self, test_client_docker_mode, method, endpoint, payload,
):
client_call = getattr(test_client_docker_mode, method)
response = client_call(endpoint, json=payload) if payload else client_call(endpoint)
assert response.status_code == 501
# --- Read-only endpoints --------------------------------------------------
class TestDisksListEndpoint:
"""Tests for /api/disks endpoint."""
def test_returns_list_in_native_mode(self, test_client):
"""Should return list of disks in native mode."""
with mock.patch("controller.api.disks.get_block_devices", return_value=[]):
response = test_client.get("/api/disks")
assert response.status_code == 200
@ -39,10 +92,8 @@ class TestDisksListEndpoint:
class TestMountsEndpoint:
"""Tests for /api/disks/mounts endpoint."""
def test_returns_mounts(self, test_client):
"""Should return current mounts."""
mock_mounts = [{"mount_point": "/media/wrolpi", "device": "/dev/sda1"}]
with mock.patch("controller.api.disks.get_mounts", return_value=mock_mounts):
response = test_client.get("/api/disks/mounts")
@ -50,266 +101,217 @@ class TestMountsEndpoint:
assert response.json() == mock_mounts
# --- /api/disks/mount -----------------------------------------------------
class TestMountEndpoint:
"""Tests for /api/disks/mount endpoint."""
def test_mounts_drive_successfully(self, test_client):
"""Should mount drive successfully."""
with mock.patch(
"controller.api.disks.mount_drive",
return_value={"success": True, "mount_point": "/media/test"}
):
def test_mounts_drive_successfully(self, test_client, disks_env):
with mock.patch("controller.api.disks.check_shadowed_data", return_value=None), \
mock.patch("controller.api.disks.get_uuid", return_value="abc-123"):
response = test_client.post(
"/api/disks/mount",
json={"device": "/dev/sda1", "mount_point": "/media/test"}
json={"device": "/dev/sda1", "mount_point": "/media/wrolpi/usb",
"fstype": "ext4"},
)
assert response.status_code == 200
assert response.json()["success"] is True
assert response.status_code == 200
body = response.json()
assert body["success"] is True
assert body["device"] == "UUID=abc-123"
assert body["mount_point"] == "/media/wrolpi/usb"
# Drive is now in fstab.yaml AND live in the fake executor.
assert disks_env["fake"].current_mount_points() == {"/media/wrolpi/usb"}
data = fstab_yaml.load(disks_env["fstab_path"])
assert data.mount_points() == {"/media/wrolpi/usb"}
def test_returns_500_on_failure(self, test_client):
"""Should return 500 on mount failure."""
with mock.patch(
"controller.api.disks.mount_drive",
return_value={"success": False, "error": "Mount failed"}
):
def test_returns_500_when_reconciler_fails(self, test_client, disks_env, monkeypatch):
# Force the fake executor to refuse the mount; reconciler runs but
# the live state will not include the mount point.
disks_env["fake"].fail_for_mount = frozenset({"/media/wrolpi/usb"})
with mock.patch("controller.api.disks.check_shadowed_data", return_value=None), \
mock.patch("controller.api.disks.get_uuid", return_value="abc-123"):
response = test_client.post(
"/api/disks/mount",
json={"device": "/dev/sda1", "mount_point": "/media/test"}
json={"device": "/dev/sda1", "mount_point": "/media/wrolpi/usb",
"fstype": "ext4"},
)
assert response.status_code == 500
assert response.status_code == 500
assert "did not mount" in response.json()["detail"]
# The entry is still in fstab.yaml — it'll be retried next reconcile.
assert fstab_yaml.load(disks_env["fstab_path"]).mount_points() == {"/media/wrolpi/usb"}
def test_shadowed_data_soft_blocks_mount(self, test_client):
"""Existing data at the mount target soft-blocks the mount."""
def test_replaces_existing_entry_for_same_device(self, test_client, disks_env):
# Seed fstab.yaml with an existing entry, then mount the same device
# to a different path. The old row should drop, not accumulate.
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save
save(FstabFile(mounts=[FstabEntry(
"UUID=abc-123", "/media/wrolpi/old", "ext4", "defaults")]),
disks_env["fstab_path"])
with mock.patch("controller.api.disks.check_shadowed_data", return_value=None), \
mock.patch("controller.api.disks.get_uuid", return_value="abc-123"):
response = test_client.post(
"/api/disks/mount",
json={"device": "/dev/sda1", "mount_point": "/media/wrolpi/new",
"fstype": "ext4"},
)
assert response.status_code == 200
data = fstab_yaml.load(disks_env["fstab_path"])
assert data.mount_points() == {"/media/wrolpi/new"}
def test_shadowed_data_soft_blocks_mount(self, test_client, disks_env):
shadowed = {"size_bytes": 1024, "entries": ["videos"]}
with mock.patch("controller.api.disks.check_shadowed_data", return_value=shadowed), \
mock.patch("controller.api.disks.mount_drive") as mock_mount:
with mock.patch("controller.api.disks.check_shadowed_data", return_value=shadowed):
response = test_client.post(
"/api/disks/mount",
json={"device": "/dev/sda1", "mount_point": "/media/test"},
json={"device": "/dev/sda1", "mount_point": "/media/wrolpi/usb"},
)
assert response.status_code == 200
data = response.json()
assert data["success"] is False
assert data["needs_force"] == "shadowed"
assert data["shadowed_data"] == shadowed
mock_mount.assert_not_called()
assert response.status_code == 200
body = response.json()
assert body["success"] is False
assert body["needs_force"] == "shadowed"
# No fstab.yaml mutation.
assert disks_env["fstab_path"].exists() is False
assert disks_env["fake"].mount_calls == []
def test_force_shadowed_overrides_soft_block(self, test_client):
"""force_shadowed=True bypasses the shadowed-data check."""
with mock.patch("controller.api.disks.check_shadowed_data", return_value={"size_bytes": 1, "entries": ["a"]}) as mock_check, \
mock.patch("controller.api.disks.mount_drive", return_value={"success": True, "mount_point": "/media/test"}):
def test_force_shadowed_overrides_soft_block(self, test_client, disks_env):
shadowed = {"size_bytes": 1, "entries": ["a"]}
with mock.patch("controller.api.disks.check_shadowed_data", return_value=shadowed) as chk, \
mock.patch("controller.api.disks.get_uuid", return_value="abc-123"):
response = test_client.post(
"/api/disks/mount",
json={
"device": "/dev/sda1",
"mount_point": "/media/test",
"mount_point": "/media/wrolpi/usb",
"fstype": "ext4",
"force_shadowed": True,
},
)
assert response.status_code == 200
assert response.json()["success"] is True
# Check skipped entirely when force is set.
mock_check.assert_not_called()
assert response.status_code == 200
chk.assert_not_called()
assert "/media/wrolpi/usb" in disks_env["fake"].current_mount_points()
# --- /api/disks/unmount ---------------------------------------------------
class TestUnmountEndpoint:
"""Tests for /api/disks/unmount endpoint."""
def test_unmounts_drive_successfully(self, test_client):
"""Should unmount drive successfully."""
with mock.patch(
"controller.api.disks.unmount_drive",
return_value={"success": True, "mount_point": "/media/test"}
):
response = test_client.post(
"/api/disks/unmount",
json={"mount_point": "/media/test"}
)
assert response.status_code == 200
def test_unmounts_and_removes_from_fstab(self, test_client, disks_env):
# Seed: drive is live AND in fstab.yaml AND managed by reconciler.
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save
save(FstabFile(mounts=[FstabEntry(
"UUID=abc-123", "/media/wrolpi/usb", "ext4", "defaults")]),
disks_env["fstab_path"])
disks_env["fake"]._live.add("/media/wrolpi/usb")
# Tell reconciler we own it (so a removal triggers an unmount).
(disks_env["fstab_path"].parent / "managed").write_text(
"/media/wrolpi/usb\n")
response = test_client.post(
"/api/disks/unmount",
json={"mount_point": "/media/wrolpi/usb"},
)
assert response.status_code == 200
body = response.json()
assert body["success"] is True
assert body["removed_from_fstab"] is True
# Live unmounted, fstab.yaml empty.
assert disks_env["fake"].current_mount_points() == set()
assert fstab_yaml.load(disks_env["fstab_path"]).mounts == []
def test_unmount_returns_500_when_still_mounted(self, test_client, disks_env):
# Reconciler refuses the unmount — surface the failure.
disks_env["fake"]._live.add("/media/wrolpi/usb")
disks_env["fake"].fail_for_unmount = frozenset({"/media/wrolpi/usb"})
(disks_env["fstab_path"].parent / "managed").write_text(
"/media/wrolpi/usb\n")
response = test_client.post(
"/api/disks/unmount",
json={"mount_point": "/media/wrolpi/usb"},
)
assert response.status_code == 500
assert "still mounted" in response.json()["detail"]
class TestFstabEndpoints:
"""Tests for /api/disks/fstab endpoints."""
# --- /api/disks/fstab read endpoint ---------------------------------------
def test_list_fstab_returns_entries(self, test_client):
"""Should return fstab entries."""
mock_entries = [{"mount_point": "/media/wrolpi", "device": "UUID=1234"}]
with mock.patch("controller.api.disks.get_wrolpi_fstab_entries", return_value=mock_entries):
response = test_client.get("/api/disks/fstab")
assert response.status_code == 200
assert response.json() == mock_entries
class TestFstabReadEndpoint:
def test_add_fstab_entry(self, test_client):
"""Should add fstab entry."""
with mock.patch(
"controller.api.disks.add_fstab_entry",
return_value={"success": True, "mount_point": "/media/test"}
):
def test_returns_empty_when_no_fstab_yaml(self, test_client, disks_env):
response = test_client.get("/api/disks/fstab")
assert response.status_code == 200
assert response.json() == []
def test_returns_entries_in_legacy_shape(self, test_client, disks_env):
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save
save(FstabFile(mounts=[
FstabEntry("UUID=abc", "/media/wrolpi/usb", "ext4", "defaults"),
]), disks_env["fstab_path"])
response = test_client.get("/api/disks/fstab")
assert response.status_code == 200
body = response.json()
assert len(body) == 1
assert body[0]["type"] == "mount"
assert body[0]["device"] == "UUID=abc"
assert body[0]["mount_point"] == "/media/wrolpi/usb"
# --- /api/disks/fstab POST / DELETE (no reconcile) ------------------------
class TestFstabMutateEndpoints:
"""POST and DELETE on /fstab modify fstab.yaml without triggering a
reconcile — used to stage changes without immediately applying."""
def test_post_adds_entry_to_fstab_only(self, test_client, disks_env):
with mock.patch("controller.api.disks.get_uuid", return_value="abc-123"):
response = test_client.post(
"/api/disks/fstab",
json={"device": "/dev/sda1", "mount_point": "/media/test", "fstype": "ext4"}
json={"device": "/dev/sda1",
"mount_point": "/media/wrolpi/usb",
"fstype": "ext4"},
)
assert response.status_code == 200
assert response.status_code == 200
assert response.json()["device"] == "UUID=abc-123"
# fstab.yaml updated, but the executor was NOT asked to mount.
assert fstab_yaml.load(disks_env["fstab_path"]).mount_points() == {"/media/wrolpi/usb"}
assert disks_env["fake"].mount_calls == []
def test_delete_removes_entry_from_fstab_only(self, test_client, disks_env):
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save
save(FstabFile(mounts=[
FstabEntry("UUID=abc", "/media/wrolpi/usb", "ext4", "defaults"),
FstabEntry("UUID=def", "/media/wrolpi/other", "ext4", "defaults"),
]), disks_env["fstab_path"])
response = test_client.delete("/api/disks/fstab/media/wrolpi/usb")
assert response.status_code == 200
# Other entry preserved.
remaining = fstab_yaml.load(disks_env["fstab_path"]).mount_points()
assert remaining == {"/media/wrolpi/other"}
# No unmount issued — DELETE is config-only.
assert disks_env["fake"].unmount_calls == []
def test_delete_returns_404_for_missing_entry(self, test_client, disks_env):
response = test_client.delete("/api/disks/fstab/media/missing")
assert response.status_code == 404
# --- SMART ----------------------------------------------------------------
class TestSmartEndpoints:
"""Tests for /api/disks/smart endpoints."""
def test_list_smart_when_not_available(self, test_client):
"""Should return unavailable when SMART not available."""
with mock.patch("controller.api.disks.is_smart_available", return_value=False):
response = test_client.get("/api/disks/smart")
assert response.status_code == 200
data = response.json()
assert data["available"] is False
assert response.json()["available"] is False
def test_list_smart_returns_drives(self, test_client):
"""Should return SMART data for all drives."""
mock_drives = [{"device": "sda", "assessment": "PASS"}]
with mock.patch("controller.api.disks.is_smart_available", return_value=True):
with mock.patch("controller.api.disks.get_all_smart_status", return_value=mock_drives):
response = test_client.get("/api/disks/smart")
assert response.status_code == 200
data = response.json()
assert data["available"] is True
assert len(data["drives"]) == 1
class TestMountPersistSavesConfig:
"""Tests for mount with persist=true saving to controller.yaml."""
def test_mount_with_persist_saves_config(self, test_client):
"""Mount with persist=true should save to controller.yaml."""
with mock.patch(
"controller.api.disks.mount_drive",
return_value={"success": True, "mount_point": "/media/test"}
):
with mock.patch(
"controller.api.disks.add_fstab_entry",
return_value={"success": True, "device": "UUID=1234", "mount_point": "/media/test"}
):
with mock.patch("controller.api.disks.save_config") as mock_save:
with mock.patch("controller.api.disks.get_config_value", return_value=[]):
with mock.patch("controller.api.disks.update_config") as mock_update:
response = test_client.post(
"/api/disks/mount",
json={
"device": "/dev/sda1",
"mount_point": "/media/test",
"fstype": "ext4",
"options": "defaults",
"persist": True
}
)
assert response.status_code == 200
# Check that update_config was called with correct mount entry
mock_update.assert_called_once()
call_args = mock_update.call_args
assert call_args[0][0] == "drives.mounts"
mounts = call_args[0][1]
assert len(mounts) == 1
assert mounts[0]["device"] == "UUID=1234"
assert mounts[0]["mount_point"] == "/media/test"
# Check that save_config was called
mock_save.assert_called_once()
def test_mount_without_persist_does_not_save_config(self, test_client):
"""Mount without persist=true should not save to controller.yaml."""
with mock.patch(
"controller.api.disks.mount_drive",
return_value={"success": True, "mount_point": "/media/test"}
):
with mock.patch("controller.api.disks.save_config") as mock_save:
response = test_client.post(
"/api/disks/mount",
json={
"device": "/dev/sda1",
"mount_point": "/media/test",
"persist": False
}
)
assert response.status_code == 200
mock_save.assert_not_called()
def test_mount_replaces_existing_entry_for_same_device(self, test_client):
"""Mount should replace existing entry for same device UUID."""
existing_mounts = [
{"device": "UUID=1234", "mount_point": "/media/old", "fstype": "ext4", "options": "defaults"}
]
with mock.patch(
"controller.api.disks.mount_drive",
return_value={"success": True}
):
with mock.patch(
"controller.api.disks.add_fstab_entry",
return_value={"success": True, "device": "UUID=1234", "mount_point": "/media/new"}
):
with mock.patch("controller.api.disks.save_config"):
with mock.patch("controller.api.disks.get_config_value", return_value=existing_mounts):
with mock.patch("controller.api.disks.update_config") as mock_update:
response = test_client.post(
"/api/disks/mount",
json={
"device": "/dev/sda1",
"mount_point": "/media/new",
"fstype": "ext4",
"persist": True
}
)
assert response.status_code == 200
# Should have replaced the old entry
call_args = mock_update.call_args
mounts = call_args[0][1]
assert len(mounts) == 1
assert mounts[0]["mount_point"] == "/media/new"
class TestDeleteFstabRemovesConfig:
"""Tests for delete_fstab removing from controller.yaml."""
def test_delete_fstab_removes_from_config(self, test_client):
"""Delete fstab entry should remove from controller.yaml."""
existing_mounts = [
{"device": "UUID=1234", "mount_point": "/media/test", "fstype": "ext4", "options": "defaults"}
]
with mock.patch(
"controller.api.disks.remove_fstab_entry",
return_value={"success": True, "mount_point": "/media/test"}
):
with mock.patch("controller.api.disks.save_config") as mock_save:
with mock.patch("controller.api.disks.get_config_value", return_value=existing_mounts):
with mock.patch("controller.api.disks.update_config") as mock_update:
response = test_client.delete("/api/disks/fstab/media/test")
assert response.status_code == 200
# Check that update_config was called with empty list
mock_update.assert_called_once()
call_args = mock_update.call_args
assert call_args[0][0] == "drives.mounts"
mounts = call_args[0][1]
assert len(mounts) == 0
# Check that save_config was called
mock_save.assert_called_once()
def test_delete_fstab_preserves_other_mounts(self, test_client):
"""Delete fstab should only remove the specified mount."""
existing_mounts = [
{"device": "UUID=1234", "mount_point": "/media/test", "fstype": "ext4", "options": "defaults"},
{"device": "UUID=5678", "mount_point": "/media/other", "fstype": "ext4", "options": "defaults"},
]
with mock.patch(
"controller.api.disks.remove_fstab_entry",
return_value={"success": True, "mount_point": "/media/test"}
):
with mock.patch("controller.api.disks.save_config"):
with mock.patch("controller.api.disks.get_config_value", return_value=existing_mounts):
with mock.patch("controller.api.disks.update_config") as mock_update:
response = test_client.delete("/api/disks/fstab/media/test")
assert response.status_code == 200
# Should preserve the other mount
call_args = mock_update.call_args
mounts = call_args[0][1]
assert len(mounts) == 1
assert mounts[0]["mount_point"] == "/media/other"
# OpenAPI endpoint-presence tests are consolidated in test_api.py::TestOpenAPI.
with mock.patch("controller.api.disks.is_smart_available", return_value=True), \
mock.patch("controller.api.disks.get_all_smart_status", return_value=mock_drives):
response = test_client.get("/api/disks/smart")
assert response.status_code == 200
body = response.json()
assert body["available"] is True
assert len(body["drives"]) == 1

343
controller/test/test_fstab_migration.py Normal file
View file

@ -0,0 +1,343 @@
"""
Tests for controller.lib.fstab_migration — one-time migration of
WROLPi-managed mount entries into fstab.yaml.
Sources:
- /etc/fstab (older WROLPi versions)
- /media/wrolpi/config/controller.yaml drives.mounts (intermediate)
Migration semantics:
- Runs at most once per drive (gated by .fstab-migrated marker).
- All /media/* entries (except /media/wrolpi and the temp onboarding
mount) are candidates.
- fstab.yaml gets each missing entry; existing entries are not touched.
- /etc/fstab candidates are commented in place with a "migrated to
fstab.yaml: " prefix.
- controller.yaml has drives.mounts removed after migration.
- /etc/fstab is backed up before modification.
"""
from pathlib import Path
import yaml
from controller.lib.fstab_migration import migrate_etc_fstab
from controller.lib.fstab_yaml import load as load_fstab_yaml
def _write(path: Path, content: str) -> Path:
path.write_text(content)
return path
def _write_yaml(path: Path, data: dict) -> Path:
path.write_text(yaml.safe_dump(data))
return path
class TestMigrationGating:
"""Marker file behaviour and basic short-circuits."""
def test_marker_present_is_a_no_op(self, tmp_path):
fstab = _write(tmp_path / "fstab", "UUID=1 /media/x ext4 defaults 0 2\n")
fstab_yaml = tmp_path / "fstab.yaml"
marker = tmp_path / ".fstab-migrated"
marker.touch()
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=fstab_yaml,
marker_path=marker,
backup_path=tmp_path / "backup",
)
assert result == {"migrated": False, "reason": "already-migrated", "count": 0}
# /etc/fstab untouched, fstab.yaml not written.
assert "UUID=1 /media/x" in fstab.read_text()
assert not fstab_yaml.exists()
def test_no_sources_writes_marker_only(self, tmp_path):
marker = tmp_path / ".fstab-migrated"
result = migrate_etc_fstab(
fstab_path=tmp_path / "nonexistent-fstab",
controller_yaml_path=tmp_path / "missing.yaml",
fstab_yaml_path=tmp_path / "fstab.yaml",
marker_path=marker,
backup_path=tmp_path / "backup",
)
assert result["migrated"] is True
assert result["count"] == 0
assert marker.exists()
def test_idempotent_when_called_twice(self, tmp_path):
_write(tmp_path / "fstab",
"UUID=1 /media/wrolpi/usb ext4 defaults 0 2\n")
fstab_yaml_p = tmp_path / "fstab.yaml"
marker = tmp_path / ".fstab-migrated"
first = migrate_etc_fstab(
fstab_path=tmp_path / "fstab",
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=fstab_yaml_p,
marker_path=marker,
backup_path=tmp_path / "backup",
)
assert first["migrated"] is True
# Manually re-add an entry to simulate a stray write between runs.
# The marker should still short-circuit and leave fstab.yaml alone.
before = fstab_yaml_p.read_text()
second = migrate_etc_fstab(
fstab_path=tmp_path / "fstab",
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=fstab_yaml_p,
marker_path=marker,
backup_path=tmp_path / "backup",
)
assert second["migrated"] is False
assert second["reason"] == "already-migrated"
assert fstab_yaml_p.read_text() == before
class TestEtcFstabMigration:
def test_only_primary_entry_is_left_alone(self, tmp_path):
fstab = _write(tmp_path / "fstab", (
"UUID=root / ext4 defaults 0 1\n"
"UUID=primary /media/wrolpi ext4 defaults,nofail 0 2\n"
))
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
assert result == {"migrated": True, "count": 0}
# Primary still in fstab, intact.
assert "UUID=primary /media/wrolpi ext4" in fstab.read_text()
# fstab.yaml not written when no candidates.
assert not fstab_yaml_p.exists()
def test_secondary_entry_is_migrated(self, tmp_path):
fstab = _write(tmp_path / "fstab", (
"UUID=root / ext4 defaults 0 1\n"
"UUID=primary /media/wrolpi ext4 defaults 0 2\n"
"# WROLPi managed mount - added 2025-01-01T00:00:00\n"
"UUID=secondary /media/wrolpi/2TB ext4 defaults,nofail 0 2\n"
))
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
assert result["count"] == 1
data = load_fstab_yaml(fstab_yaml_p)
assert len(data.mounts) == 1
e = data.mounts[0]
assert e.device == "UUID=secondary"
assert e.mount_point == "/media/wrolpi/2TB"
assert e.fstype == "ext4"
new_fstab = fstab.read_text()
active_lines = [
ln.strip() for ln in new_fstab.splitlines()
if ln.strip() and not ln.strip().startswith("#")
]
# Migrated line no longer active.
assert not any("/media/wrolpi/2TB" in ln for ln in active_lines)
# Commented form preserved.
assert "migrated to fstab.yaml" in new_fstab
def test_dedup_against_existing_fstab_yaml(self, tmp_path):
from controller.lib.fstab_yaml import FstabEntry, FstabFile, save
save(FstabFile(mounts=[
FstabEntry("UUID=secondary", "/media/wrolpi/2TB", "ext4", "defaults"),
]), tmp_path / "fstab.yaml")
fstab = _write(tmp_path / "fstab", (
"UUID=primary /media/wrolpi ext4 defaults 0 2\n"
"UUID=secondary /media/wrolpi/2TB ext4 defaults 0 2\n"
))
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=tmp_path / "fstab.yaml",
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
# Same mount point already in fstab.yaml; no new entry but /etc/fstab
# line is still commented out (cleanup).
assert result["count"] == 0
data = load_fstab_yaml(tmp_path / "fstab.yaml")
assert len(data.mounts) == 1
assert "migrated to fstab.yaml" in fstab.read_text()
def test_non_media_entries_untouched(self, tmp_path):
fstab = _write(tmp_path / "fstab", (
"UUID=root / ext4 defaults 0 1\n"
"UUID=home /home ext4 defaults 0 2\n"
"tmpfs /tmp tmpfs defaults 0 0\n"
))
migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=tmp_path / "fstab.yaml",
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
new_fstab = fstab.read_text()
for ln in ("UUID=root / ext4", "UUID=home /home ext4", "tmpfs /tmp"):
assert ln in new_fstab
assert f"migrated to fstab.yaml: {ln}" not in new_fstab
def test_excluded_mount_points_skipped(self, tmp_path):
fstab = _write(
tmp_path / "fstab",
"UUID=temp /media/wrolpi_temp_onboarding ext4 defaults 0 2\n")
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=tmp_path / "fstab.yaml",
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
assert result["count"] == 0
# /etc/fstab not commented out — never claimed.
assert "/media/wrolpi_temp_onboarding" in fstab.read_text()
def test_backup_written_when_changes_made(self, tmp_path):
fstab_content = "UUID=secondary /media/wrolpi/2TB ext4 defaults 0 2\n"
fstab = _write(tmp_path / "fstab", fstab_content)
backup = tmp_path / "backup"
migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=tmp_path / "controller.yaml",
fstab_yaml_path=tmp_path / "fstab.yaml",
marker_path=tmp_path / "marker",
backup_path=backup,
)
assert backup.exists()
assert backup.read_text() == fstab_content
class TestControllerYamlMigration:
"""drives.mounts in controller.yaml gets lifted into fstab.yaml and
the key is stripped from the controller.yaml file."""
def test_drives_mounts_migrated_to_fstab_yaml(self, tmp_path):
controller_yaml = _write_yaml(tmp_path / "controller.yaml", {
"port": 80,
"drives": {
"mounts": [{
"device": "UUID=secondary",
"mount_point": "/media/wrolpi/2TB",
"fstype": "ext4",
"options": "defaults",
}],
},
})
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=tmp_path / "nonexistent-fstab",
controller_yaml_path=controller_yaml,
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
assert result["count"] == 1
data = load_fstab_yaml(fstab_yaml_p)
assert data.mount_points() == {"/media/wrolpi/2TB"}
# drives.mounts stripped, other keys preserved.
post = yaml.safe_load(controller_yaml.read_text())
assert post.get("port") == 80
assert "mounts" not in (post.get("drives") or {})
def test_drives_mounts_with_excluded_paths(self, tmp_path):
# The reserved mount points should not be migrated; they should
# remain in drives.mounts? — actually no, drives.mounts is being
# retired entirely. We migrate all valid candidates and drop the
# key regardless. Reserved entries are silently filtered out.
controller_yaml = _write_yaml(tmp_path / "controller.yaml", {
"drives": {
"mounts": [
{"device": "UUID=primary",
"mount_point": "/media/wrolpi",
"fstype": "ext4",
"options": "defaults"},
{"device": "UUID=secondary",
"mount_point": "/media/wrolpi/2TB",
"fstype": "ext4",
"options": "defaults"},
],
},
})
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=tmp_path / "nonexistent-fstab",
controller_yaml_path=controller_yaml,
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
# Only the secondary made it across.
assert result["count"] == 1
data = load_fstab_yaml(fstab_yaml_p)
assert data.mount_points() == {"/media/wrolpi/2TB"}
def test_no_drives_key_is_a_noop(self, tmp_path):
controller_yaml = _write_yaml(tmp_path / "controller.yaml", {"port": 80})
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=tmp_path / "nonexistent-fstab",
controller_yaml_path=controller_yaml,
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
assert result["count"] == 0
# controller.yaml untouched.
assert yaml.safe_load(controller_yaml.read_text()) == {"port": 80}
class TestCombinedSources:
"""Both /etc/fstab AND drives.mounts contain entries — merge dedup."""
def test_merges_both_sources_dedup_by_mount_point(self, tmp_path):
fstab = _write(tmp_path / "fstab",
"UUID=a /media/wrolpi/a ext4 defaults 0 2\n"
"UUID=both /media/wrolpi/both ext4 defaults 0 2\n")
controller_yaml = _write_yaml(tmp_path / "controller.yaml", {
"drives": {
"mounts": [
{"device": "UUID=both",
"mount_point": "/media/wrolpi/both",
"fstype": "ext4",
"options": "defaults"},
{"device": "UUID=b",
"mount_point": "/media/wrolpi/b",
"fstype": "ext4",
"options": "defaults"},
],
},
})
fstab_yaml_p = tmp_path / "fstab.yaml"
result = migrate_etc_fstab(
fstab_path=fstab,
controller_yaml_path=controller_yaml,
fstab_yaml_path=fstab_yaml_p,
marker_path=tmp_path / "marker",
backup_path=tmp_path / "backup",
)
# a (fstab only), both (both sources, dedup), b (controller only)
assert result["count"] == 3
data = load_fstab_yaml(fstab_yaml_p)
assert data.mount_points() == {
"/media/wrolpi/a", "/media/wrolpi/both", "/media/wrolpi/b",
}

167
controller/test/test_fstab_yaml.py Normal file
View file

@ -0,0 +1,167 @@
"""
Tests for controller.lib.fstab_yaml — the on-drive fstab.yaml reader/writer.
All tests use real files in tmp_path; no mocks anywhere. The file format is
a stable on-disk contract so the round-trip tests are the load-bearing ones.
"""
import yaml
from controller.lib.fstab_yaml import (
FstabEntry,
FstabFile,
SCHEMA_VERSION,
load,
save,
)
class TestFstabEntry:
def test_to_dict_round_trip(self):
entry = FstabEntry("UUID=abc", "/media/x", "ext4", "defaults,noatime")
d = entry.to_dict()
assert d == {
"device": "UUID=abc",
"mount_point": "/media/x",
"fstype": "ext4",
"options": "defaults,noatime",
}
assert FstabEntry.from_dict(d) == entry
def test_from_dict_supplies_defaults(self):
entry = FstabEntry.from_dict({"device": "UUID=abc", "mount_point": "/media/x"})
assert entry.fstype == "auto"
assert entry.options == "defaults"
def test_empty_options_falls_back_to_defaults(self):
# Migration data sometimes has options=None; treat as "defaults".
entry = FstabEntry.from_dict({
"device": "UUID=abc",
"mount_point": "/media/x",
"fstype": "ext4",
"options": None,
})
assert entry.options == "defaults"
class TestFstabFile:
def test_mount_points(self):
f = FstabFile(mounts=[
FstabEntry("UUID=1", "/media/a", "ext4"),
FstabEntry("UUID=2", "/media/b", "ext4"),
])
assert f.mount_points() == {"/media/a", "/media/b"}
def test_find_by_mount_point(self):
e = FstabEntry("UUID=1", "/media/a", "ext4")
f = FstabFile(mounts=[e])
assert f.find_by_mount_point("/media/a") is e
assert f.find_by_mount_point("/media/missing") is None
def test_add_or_replace_same_mount_point(self):
f = FstabFile(mounts=[FstabEntry("UUID=1", "/media/a", "ext4", "defaults")])
f.add_or_replace(FstabEntry("UUID=1", "/media/a", "ext4", "noatime"))
assert len(f.mounts) == 1
assert f.mounts[0].options == "noatime"
def test_add_or_replace_same_device_different_mount_point(self):
# User remounts the same drive to a new directory; old row should
# drop, not accumulate.
f = FstabFile(mounts=[FstabEntry("UUID=1", "/media/old", "ext4")])
f.add_or_replace(FstabEntry("UUID=1", "/media/new", "ext4"))
assert len(f.mounts) == 1
assert f.mounts[0].mount_point == "/media/new"
def test_add_or_replace_preserves_unrelated(self):
f = FstabFile(mounts=[
FstabEntry("UUID=1", "/media/a", "ext4"),
FstabEntry("UUID=2", "/media/b", "ext4"),
])
f.add_or_replace(FstabEntry("UUID=3", "/media/c", "ext4"))
assert {m.mount_point for m in f.mounts} == {"/media/a", "/media/b", "/media/c"}
def test_remove_by_mount_point_returns_true_when_removed(self):
f = FstabFile(mounts=[FstabEntry("UUID=1", "/media/a", "ext4")])
assert f.remove_by_mount_point("/media/a") is True
assert f.mounts == []
def test_remove_by_mount_point_returns_false_when_absent(self):
f = FstabFile()
assert f.remove_by_mount_point("/media/a") is False
class TestLoadSave:
def test_load_missing_file_is_empty(self, tmp_path):
f = load(tmp_path / "missing.yaml")
assert f.version == SCHEMA_VERSION
assert f.mounts == []
def test_load_empty_file_is_empty(self, tmp_path):
p = tmp_path / "fstab.yaml"
p.write_text("")
f = load(p)
assert f.mounts == []
def test_load_malformed_yaml_is_empty(self, tmp_path):
p = tmp_path / "fstab.yaml"
p.write_text("this: is: not: valid: yaml: [")
f = load(p)
assert f.mounts == []
def test_load_skips_unrecognised_fields(self, tmp_path):
# Forward-compat: future entries may carry extra fields; load should
# silently ignore them.
p = tmp_path / "fstab.yaml"
p.write_text(yaml.safe_dump({
"version": 1,
"mounts": [{
"device": "UUID=1",
"mount_point": "/media/a",
"fstype": "ext4",
"options": "defaults",
"future_field": "ignored",
}],
}))
f = load(p)
assert len(f.mounts) == 1
def test_save_creates_parent_directory(self, tmp_path):
nested = tmp_path / "config" / "fstab.yaml"
save(FstabFile(mounts=[FstabEntry("UUID=1", "/media/a", "ext4")]), nested)
assert nested.exists()
def test_save_is_atomic(self, tmp_path):
# We can't directly observe atomicity in a unit test, but we can
# confirm the temp file is gone after a successful write.
p = tmp_path / "fstab.yaml"
save(FstabFile(mounts=[FstabEntry("UUID=1", "/media/a", "ext4")]), p)
assert p.exists()
assert not (tmp_path / "fstab.yaml.tmp").exists()
def test_round_trip(self, tmp_path):
original = FstabFile(mounts=[
FstabEntry("UUID=1", "/media/b", "ext4", "defaults"),
FstabEntry("UUID=2", "/media/a", "exfat", "defaults,uid=1001"),
])
p = tmp_path / "fstab.yaml"
save(original, p)
loaded = load(p)
# Mount points sorted on save:
assert [m.mount_point for m in loaded.mounts] == ["/media/a", "/media/b"]
# Other fields preserved:
a = loaded.find_by_mount_point("/media/a")
assert a.device == "UUID=2"
assert a.fstype == "exfat"
assert a.options == "defaults,uid=1001"
def test_save_sorts_mounts(self, tmp_path):
p = tmp_path / "fstab.yaml"
save(FstabFile(mounts=[
FstabEntry("UUID=2", "/media/zeta", "ext4"),
FstabEntry("UUID=1", "/media/alpha", "ext4"),
]), p)
raw = yaml.safe_load(p.read_text())
assert [m["mount_point"] for m in raw["mounts"]] == ["/media/alpha", "/media/zeta"]

191
controller/test/test_mount_executor.py Normal file
View file

@ -0,0 +1,191 @@
"""
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_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",
}

294
controller/test/test_reconciler.py Normal file
View file

@ -0,0 +1,294 @@
"""
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 == []
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_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

22
etc/raspberrypios/wrolpi-mounts.service Normal file
View file

@ -0,0 +1,22 @@
[Unit]
Description=WROLPi secondary-drive mounts
Documentation=https://github.com/lrnselfreliance/wrolpi
# Read drives.mounts from controller.yaml and apply each entry that isn't
# already mounted. Replaces /etc/fstab entries for secondary drives so the
# mount table travels with the drive itself.
After=local-fs.target wrolpi-portable-bootstrap.service
RequiresMountsFor=/media/wrolpi
# Run before WROLPi services so secondary drives are ready when they start.
Before=wrolpi-api.service wrolpi-app.service wrolpi-kiwix.service wrolpi.target
[Service]
Type=oneshot
RemainAfterExit=yes
WorkingDirectory=/opt/wrolpi
ExecStart=/opt/wrolpi/controller/venv/bin/python -m controller.lib.mount_runner
StandardOutput=journal
StandardError=journal
TimeoutStartSec=300
[Install]
WantedBy=multi-user.target

1
repair.sh
View file

@ -79,6 +79,7 @@ systemctl enable wrolpi-api.service
systemctl enable wrolpi-app.service
systemctl enable wrolpi-kiwix.service
systemctl enable wrolpi-help.service
systemctl enable wrolpi-mounts.service
systemctl enable wrolpi-cert-renew.timer
# Repair Controller (offline-safe - no pip install)

3
wrolpi-portable/config/hooks/normal/9999-wrolpi.hook.chroot
View file

@ -167,6 +167,9 @@ systemctl enable iperf3
systemctl enable wrolpi-first-startup.service
# Enable Controller service to start on boot.
systemctl enable wrolpi-controller.service
# Apply controller.yaml drives.mounts at boot (replaces /etc/fstab for
# secondary mounts so the mount table travels with the drive).
systemctl enable wrolpi-mounts.service
# Enable weekly certificate renewal check.
systemctl enable wrolpi-cert-renew.timer