Florian WildeandClaude Opus 5.5
e5ed8c4d88
MountBackend-Interface mit LinuxMountBackend (Argumentliste, kurzlebige Credentials-Datei) und FakeMountBackend; MountManager für Einhängen, Aushängen, Start-Wiederherstellung, Drei-Schritt-Test mit Root-Squash- Erkennung, Lebenszeichen und umount -l/Neu-Einhängen; SFTP-Anbindung mit festgehaltenem Hostschlüssel. Tests inkl. echtem SFTP-Server im Prozess. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
"""Fallstricke 5 und 7 am echten Backend – ``subprocess`` wird nur abgefangen."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import stat
|
||
import subprocess
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from netmover.mounting.backend import CifsCredentials, MountError, MountSpec
|
||
from netmover.mounting.linux import LinuxMountBackend, explain_mount_error
|
||
|
||
|
||
class Recorder:
|
||
def __init__(self, returncode: int = 0, stderr: str = "") -> None:
|
||
self.returncode = returncode
|
||
self.stderr = stderr
|
||
self.calls: list[tuple[list[str], dict[str, Any]]] = []
|
||
self.cred_snapshot: dict[str, Any] = {}
|
||
|
||
def __call__(self, argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
|
||
self.calls.append((argv, kwargs))
|
||
for opt in ",".join(argv).split(","):
|
||
if opt.startswith("credentials="):
|
||
p = Path(opt.removeprefix("credentials="))
|
||
self.cred_snapshot = {
|
||
"path": p,
|
||
"mode": stat.S_IMODE(p.stat().st_mode),
|
||
"content": p.read_text(),
|
||
}
|
||
return subprocess.CompletedProcess(argv, self.returncode, "", self.stderr)
|
||
|
||
|
||
def _spec(tmp_path: Path, password: str = "S3cr3t,pw=x") -> MountSpec:
|
||
return MountSpec(
|
||
fstype="cifs",
|
||
source="//nas/scans",
|
||
mountpoint=tmp_path / "mnt" / "scans",
|
||
options=("vers=3.0", "uid=1000"),
|
||
credentials=CifsCredentials("scanner", password, "WORKGROUP"),
|
||
)
|
||
|
||
|
||
def test_mount_uses_argv_and_credentials_file(tmp_path: Path) -> None:
|
||
rec = Recorder()
|
||
backend = LinuxMountBackend(tmp_path / "run", runner=rec)
|
||
backend.mount(_spec(tmp_path))
|
||
|
||
argv, kwargs = rec.calls[0]
|
||
assert "shell" not in kwargs
|
||
assert argv[:4] == ["mount", "-t", "cifs", "-o"]
|
||
assert argv[-2:] == ["//nas/scans", str(tmp_path / "mnt" / "scans")]
|
||
assert argv[4].startswith("vers=3.0,uid=1000,credentials=")
|
||
# Passwort nie in der Prozessliste
|
||
assert not any("S3cr3t" in a for a in argv)
|
||
assert "password" not in argv[4]
|
||
# Credentials-Datei: 0600, Inhalt korrekt, danach gelöscht
|
||
snap = rec.cred_snapshot
|
||
assert snap["mode"] == 0o600
|
||
assert snap["content"] == "username=scanner\npassword=S3cr3t,pw=x\ndomain=WORKGROUP\n"
|
||
assert snap["path"].parent == tmp_path / "run"
|
||
assert not snap["path"].exists()
|
||
assert kwargs["env"]["LC_ALL"] == "C"
|
||
|
||
|
||
def test_credentials_file_removed_on_failure_and_output_redacted(tmp_path: Path) -> None:
|
||
rec = Recorder(returncode=32, stderr="mount error(13): Permission denied for S3cr3t,pw=x")
|
||
backend = LinuxMountBackend(tmp_path / "run", runner=rec)
|
||
with pytest.raises(MountError) as exc:
|
||
backend.mount(_spec(tmp_path))
|
||
assert "S3cr3t" not in str(exc.value)
|
||
assert "Benutzername/Passwort" in str(exc.value)
|
||
assert list((tmp_path / "run").iterdir()) == []
|
||
|
||
|
||
def test_mount_timeout(tmp_path: Path) -> None:
|
||
def runner(argv: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
|
||
raise subprocess.TimeoutExpired(argv, kwargs["timeout"])
|
||
|
||
backend = LinuxMountBackend(tmp_path / "run", runner=runner)
|
||
with pytest.raises(MountError, match="keine Antwort nach 5 s"):
|
||
backend.mount(_spec(tmp_path), timeout=5)
|
||
assert list((tmp_path / "run").iterdir()) == []
|
||
|
||
|
||
def test_unmount_lazy_argv(tmp_path: Path) -> None:
|
||
rec = Recorder()
|
||
backend = LinuxMountBackend(tmp_path / "run", runner=rec)
|
||
backend.unmount(Path("/mnt/netmover/scans"), lazy=True)
|
||
assert rec.calls[0][0] == ["umount", "-l", "/mnt/netmover/scans"]
|
||
|
||
|
||
def test_is_mounted_reads_mountinfo(tmp_path: Path) -> None:
|
||
info = tmp_path / "mountinfo"
|
||
info.write_text("98 22 0:52 / /mnt/netmover/scans rw - cifs //nas/scans rw\n")
|
||
backend = LinuxMountBackend(tmp_path / "run", mountinfo_path=info, runner=Recorder())
|
||
assert backend.is_mounted(Path("/mnt/netmover/scans"))
|
||
assert not backend.is_mounted(Path("/mnt/netmover/other"))
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("fstype", "msg", "expect"),
|
||
[
|
||
(
|
||
"cifs",
|
||
"mount: wrong fs type, bad option; you might need a /sbin/mount.<type> helper program",
|
||
"cifs-utils",
|
||
),
|
||
("nfs4", "mount.nfs4: access denied by server while mounting nas:/x", "NFS-Berechtigungen"),
|
||
("cifs", "mount error(112): Host is down", "nicht erreichbar"),
|
||
("cifs", "mount: only root can do that", "CAP_SYS_ADMIN"),
|
||
],
|
||
)
|
||
def test_explain(fstype: str, msg: str, expect: str) -> None:
|
||
assert expect in explain_mount_error(fstype, msg)
|