R-179: the uninstall removes the NAS network-storage units

Units carrying the agent's network-storage marker (mnt-*.automount first, then mnt-*.mount) are
disabled --now, reset-failed and removed before the drive umount loop; a share that will not stop is
not forced — its unit is kept and named in KEPT with the commands. Enrolled-drive and foreign units
are never touched. scripts/test_hostinstall.py: test_net_units_* (5).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
2026-10-05 21:21:46 +02:00
parent 7f3944eff9
commit c11d4fbf2e
2 changed files with 141 additions and 2 deletions
+96 -2
View File
@@ -11,7 +11,7 @@ functions act on are variables the test points into the temp directory. Nothing
Where behaviour cannot be isolated, a STATIC test checks the wiring (the function is CALLED, from the
right place) — a helper defined and never called is the seam-built-never-wired shape.
Rows: R-275, R-276, R-881 (uninstall residue); R-306 (--preflight-only writes no state); R-130 (lvm minimum wording); R-180 (archive storage outside the ACL set).
Rows: R-275, R-276, R-881 (uninstall residue); R-306 (--preflight-only writes no state); R-130 (lvm minimum wording); R-180 (archive storage outside the ACL set); R-179 (NAS units).
Run: python3 scripts/test_hostinstall.py
"""
import os
@@ -97,7 +97,7 @@ class Sandbox:
def stub(self, name, body="exit 0"):
p = os.path.join(self.bin, name)
with open(p, "w") as f:
f.write('#!/bin/sh\necho "%s $*" >> "%s"\n%s\n' % (name, self.calls, body))
f.write('#!/bin/sh\nprintf "%%s\\n" "%s $*" >> "%s"\n%s\n' % (name, self.calls, body))
os.chmod(p, 0o755)
def path(self, *parts):
@@ -421,6 +421,100 @@ def test_archive_storage_check_is_wired_into_preflight():
assert pre and tok and max(pre) < tok.start(), "preflight no longer runs before the token step"
# ── R-179: the NAS network-storage units ─────────────────────────────────────────────────────────
AGENT_NETMOUNT = os.path.join(os.path.dirname(os.path.dirname(HERE)), "felhom-agent", "internal", "storage", "netmount.go")
NET_MARK = "# Managed by felhom-agent (network storage) — do not edit by hand.\n"
SHARE = r"mnt-felhom\x2ddrives-Felhom\x2dShare"
NET_SYSTEMCTL = """
for u in "$@"; do last="$u"; done
case "$1" in
is-active) [ -e "$SB/active/$last" ]; exit $? ;;
disable) [ -e "$SB/sticky/$last" ] || rm -f "$SB/active/$last"; exit 0 ;;
esac
exit 0"""
def _net_sandbox(sticky=()):
sb = Sandbox()
os.mkdir(sb.path("units")); os.mkdir(sb.path("active")); os.mkdir(sb.path("sticky"))
sb.stub("systemctl", NET_SYSTEMCTL)
units = {SHARE + ".automount": NET_MARK + "[Automount]\n",
SHARE + ".mount": NET_MARK + "[Mount]\n",
r"mnt-felhom\x2ddrives-disk1.mount": "# Managed by felhom-agent — do not edit by hand.\n[Mount]\n",
"mnt-other.mount": "[Mount]\nWhere=/mnt/other\n"}
for n, c in units.items():
open(sb.path("units", n), "w").write(c)
open(sb.path("active", SHARE + ".mount"), "w").close()
for n in sticky:
open(sb.path("sticky", n), "w").close()
return sb
NET_RUN = ('NET_UNIT_DIR="$SB/units"; NET_UNIT_MARKER="Managed by felhom-agent (network storage)"\n'
'_NET_UNITS_REMOVED=(); _NET_UNITS_BUSY=()\n_remove_network_storage_units\n'
'echo "REMOVED=${#_NET_UNITS_REMOVED[@]} BUSY=${_NET_UNITS_BUSY[*]}"\n')
def test_net_units_removed_and_only_ours():
sb = _net_sandbox()
try:
rc, out = sb.run(prelude() + func("_remove_network_storage_units") + NET_RUN)
assert rc == 0, out
left = sorted(os.listdir(sb.path("units")))
assert left == sorted([r"mnt-felhom\x2ddrives-disk1.mount", "mnt-other.mount"]), \
"wrong units left after the uninstall: %s" % left
log = sb.logged()
a = log.find("disable --now -- %s.automount" % SHARE)
m = log.find("disable --now -- %s.mount" % SHARE)
assert a >= 0 and m >= 0 and a < m, "automount must be stopped before its mount:\n%s" % log
assert "disk1" not in log and "mnt-other" not in log, "touched a unit that is not a network share:\n%s" % log
assert "REMOVED=2 BUSY=" in out, out
finally:
sb.close()
def test_net_units_busy_share_is_not_forced():
sb = _net_sandbox(sticky=(SHARE + ".mount",))
try:
rc, out = sb.run(prelude() + func("_remove_network_storage_units") + NET_RUN)
assert rc == 0, out
assert os.path.exists(sb.path("units", SHARE + ".mount")), "a busy share's unit was removed"
assert not os.path.exists(sb.path("units", SHARE + ".automount"))
assert "BUSY=%s.mount" % SHARE in out and "NOT forcing" in out, out
assert "umount" not in sb.logged(), sb.logged()
finally:
sb.close()
def test_net_units_dry_run_changes_nothing():
sb = _net_sandbox()
try:
rc, out = sb.run(prelude(dry=True) + func("_remove_network_storage_units") + NET_RUN)
assert rc == 0, out
assert len(os.listdir(sb.path("units"))) == 4, os.listdir(sb.path("units"))
assert "disable" not in sb.logged(), sb.logged()
finally:
sb.close()
def test_net_units_wired_before_the_umount_loop():
body = func("run_uninstall")
call = body.find("_remove_network_storage_units")
loop = body.find("findmnt -rn -o TARGET")
assert call > 0, "run_uninstall does not call _remove_network_storage_units (R-179)"
assert call < loop, "the share units must be stopped before the drive umount loop"
def test_net_unit_marker_matches_the_agent():
m = re.search(r'^NET_UNIT_MARKER="([^"]+)"', SRC, re.M)
assert m, "NET_UNIT_MARKER not found"
if not os.path.exists(AGENT_NETMOUNT):
print(" note: %s absent (CI) — marker currency not checked" % AGENT_NETMOUNT)
return
a = re.search(r'netUnitMarker\s*=\s*"([^"]+)"', open(AGENT_NETMOUNT, encoding="utf-8").read())
assert a and a.group(1) == m.group(1), "installer marker %r != agent netUnitMarker %r" % (m.group(1), a and a.group(1))
def main():
tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)]
fails = 0