61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""Test the generated storage readiness script in a private mount namespace.
|
|
|
|
Usage: sudo python3 scripts/media-storage-check.py <media-storage-ready-unit>
|
|
Uses temporary tmpfs mounts and NSS fixtures; host media and mount state are untouched.
|
|
"""
|
|
|
|
import os, re, subprocess as sp, sys, tempfile
|
|
from pathlib import Path
|
|
|
|
original = os.environ.get("FRED_TEST_ORIGINAL_MNTNS")
|
|
if not original:
|
|
os.execvpe(
|
|
"unshare",
|
|
["unshare", "-m", sys.executable, *sys.argv],
|
|
{**os.environ, "FRED_TEST_ORIGINAL_MNTNS": os.readlink("/proc/self/ns/mnt")},
|
|
)
|
|
assert original != os.readlink("/proc/self/ns/mnt")
|
|
sp.run(["mount", "--make-rprivate", "/"], check=True)
|
|
sp.run(["mount", "-t", "tmpfs", "tmpfs", "/mnt"], check=True)
|
|
for name in ["storage", "disk1", "disk2", "disk3"]:
|
|
Path("/mnt", name).mkdir()
|
|
unit = (Path(sys.argv[1]) / "media-storage-ready.service").read_text()
|
|
script = re.search(r"^ExecStart=(\S+)", unit, re.M)[1]
|
|
|
|
|
|
def ready(expected):
|
|
r = sp.run([script], capture_output=True, text=True)
|
|
assert (r.returncode == 0) == expected, r.stderr
|
|
print(r.stderr.strip() or "Readiness succeeded", flush=True)
|
|
|
|
|
|
ready(False)
|
|
assert list(Path("/mnt/storage").iterdir()) == []
|
|
for name in ["storage", "disk1", "disk2"]:
|
|
sp.run(["mount", "-t", "tmpfs", "tmpfs", "/mnt/" + name], check=True)
|
|
ready(False)
|
|
assert list(Path("/mnt/storage").iterdir()) == []
|
|
# Isolated NSS fixtures let tmpfiles resolve the real configured names.
|
|
root = Path(tempfile.mkdtemp(prefix="fred-storage-users."))
|
|
(root / "passwd").write_text(
|
|
"root:x:0:0:root:/root:/bin/sh\n"
|
|
+ "".join(
|
|
f"{name}:x:0:0::{root}:/bin/sh\n"
|
|
for name in ["fred", "qbittorrent", "sonarr", "radarr", "sabnzbd"]
|
|
)
|
|
)
|
|
(root / "group").write_text("root:x:0:\nmedia:x:0:\n")
|
|
for name in ["passwd", "group"]:
|
|
sp.run(["mount", "--bind", str(root / name), "/etc/" + name], check=True)
|
|
sp.run(["mount", "-t", "tmpfs", "tmpfs", "/mnt/disk3"], check=True)
|
|
ready(True)
|
|
assert Path("/mnt/storage/torrents/shows").is_dir()
|
|
sp.run(["umount", "/mnt/disk2"], check=True)
|
|
ready(False)
|
|
sp.run(["umount", "/mnt/storage"], check=True)
|
|
assert list(Path("/mnt/storage").iterdir()) == [], (
|
|
"wrote into underlying root directory"
|
|
)
|
|
print(
|
|
"Storage guard: missing pool/branch and lost branch refused; mounted pool prepared; underlying directory untouched"
|
|
)
|