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>
370 lines
13 KiB
Python
370 lines
13 KiB
Python
"""Mount-Verwaltung mit FakeMountBackend – Fallstricke 2, 6, 7 und 13."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import errno
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy.orm import Session, sessionmaker
|
||
|
||
from netmover.models import CredentialKind, Mount, MountState, MountType
|
||
from netmover.mounting.backend import MountError
|
||
from netmover.mounting.fake import FakeMountBackend
|
||
from netmover.mounting.manager import (
|
||
CredentialData,
|
||
MountConfig,
|
||
MountManager,
|
||
MountUnavailable,
|
||
build_spec,
|
||
)
|
||
from netmover.validation import ValidationError
|
||
from tests.conftest import MountFactory
|
||
|
||
|
||
def _state(sf: sessionmaker[Session], mount_id: int) -> Mount:
|
||
with sf() as s:
|
||
m = s.get(Mount, mount_id)
|
||
assert m is not None
|
||
s.expunge(m)
|
||
return m
|
||
|
||
|
||
# -- build_spec -------------------------------------------------------------------
|
||
|
||
|
||
def test_build_spec_cifs(tmp_path: Path) -> None:
|
||
cfg = MountConfig(
|
||
"scans",
|
||
MountType.CIFS,
|
||
"nas",
|
||
"scans",
|
||
uid=1000,
|
||
gid=100,
|
||
file_mode="0644",
|
||
extra_options="vers=3.0",
|
||
)
|
||
cred = CredentialData(CredentialKind.PASSWORD, "scanner", "pw", "WG")
|
||
spec = build_spec(cfg, tmp_path, cred)
|
||
assert spec.fstype == "cifs"
|
||
assert spec.source == "//nas/scans"
|
||
assert spec.options == ("vers=3.0", "uid=1000", "gid=100", "file_mode=0644")
|
||
assert spec.credentials is not None and spec.credentials.password == "pw"
|
||
assert "pw" not in repr(spec)
|
||
|
||
|
||
def test_build_spec_nfs() -> None:
|
||
nfs3 = build_spec(MountConfig("a", MountType.NFS3, "nas", "/volume1/a"), Path("/x"), None)
|
||
assert (nfs3.fstype, nfs3.source, nfs3.options) == ("nfs", "nas:/volume1/a", ("vers=3",))
|
||
nfs4 = build_spec(MountConfig("a", MountType.NFS4, "nas", "/volume1/a"), Path("/x"), None)
|
||
assert (nfs4.fstype, nfs4.options) == ("nfs4", ())
|
||
|
||
|
||
def test_build_spec_guest_and_revalidation() -> None:
|
||
spec = build_spec(MountConfig("a", MountType.CIFS, "nas", "pub"), Path("/x"), None)
|
||
assert spec.options == ("guest",)
|
||
# Auch Werte „aus der Datenbank“ laufen erneut durch die Validierung
|
||
with pytest.raises(ValidationError):
|
||
build_spec(MountConfig("a", MountType.CIFS, "nas;id", "pub"), Path("/x"), None)
|
||
with pytest.raises(ValidationError):
|
||
build_spec(
|
||
MountConfig("a", MountType.CIFS, "nas", "pub", extra_options="password=x"),
|
||
Path("/x"),
|
||
None,
|
||
)
|
||
|
||
|
||
# -- Einhängen / Aushängen (Fallstrick 6) --------------------------------------------
|
||
|
||
|
||
def test_mount_creates_mountpoint_under_base_and_unmount_removes_it(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
mount_base: Path,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
remote = fake_backend.export("nas", "scans")
|
||
(remote / "a.pdf").write_text("x")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = mount_base.resolve() / "scans"
|
||
assert fake_backend.is_mounted(mp)
|
||
assert (mp / "a.pdf").read_text() == "x"
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.MOUNTED and m.last_mounted_at is not None
|
||
# Zugangsdaten wurden entschlüsselt übergeben
|
||
spec = fake_backend.calls[0].spec
|
||
assert spec is not None and spec.credentials is not None
|
||
assert spec.credentials.password == "geheim123"
|
||
|
||
manager.unmount(mid)
|
||
assert not mp.exists()
|
||
assert _state(session_factory, mid).state is MountState.UNMOUNTED
|
||
|
||
|
||
def test_mount_refuses_non_empty_mountpoint(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
mount_base: Path,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
mp = mount_base.resolve() / "scans"
|
||
mp.mkdir(parents=True)
|
||
(mp / "verirrt.pdf").write_text("lokal geschrieben")
|
||
with pytest.raises(MountError, match="nicht leer"):
|
||
manager.mount(mid)
|
||
assert not fake_backend.calls
|
||
assert (mp / "verirrt.pdf").exists()
|
||
assert _state(session_factory, mid).state is MountState.ERROR
|
||
|
||
|
||
def test_failed_mount_cleans_up_and_records_error(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
mount_base: Path,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
mid = make_mount() # nicht exportiert → „No such file or directory“
|
||
with pytest.raises(MountError):
|
||
manager.mount(mid)
|
||
assert not (mount_base.resolve() / "scans").exists()
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.ERROR
|
||
assert m.last_error and "No such file" in m.last_error
|
||
|
||
|
||
def test_mountpoint_for_rejects_escape(manager: MountManager) -> None:
|
||
for bad in ["../etc", "a/b", "", "Scans"]:
|
||
with pytest.raises(ValidationError):
|
||
manager.mountpoint_for(bad)
|
||
|
||
|
||
def test_sftp_is_never_mounted(manager: MountManager, make_mount: MountFactory) -> None:
|
||
mid = make_mount("paperless", type=MountType.SFTP, share="/consume")
|
||
with pytest.raises(MountError, match="SFTP"):
|
||
manager.mount(mid)
|
||
|
||
|
||
def test_restore_all_only_auto_mount(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
notifications: list[tuple[str, str, str]],
|
||
) -> None:
|
||
fake_backend.export("nas", "a")
|
||
fake_backend.export("nas", "b")
|
||
a = make_mount("a", share="a")
|
||
make_mount("b", share="b", auto_mount=False)
|
||
make_mount("c", share="fehlt")
|
||
make_mount("d", share="a", enabled=False)
|
||
results = {r.name: r.ok for r in manager.restore_all()}
|
||
assert results == {"a": True, "c": False}
|
||
assert fake_backend.is_mounted(manager.mountpoint_for("a"))
|
||
assert notifications and notifications[0][:2] == ("mount.restore_failed", "c")
|
||
# zweiter Aufruf ist idempotent
|
||
manager.mount(a)
|
||
assert (
|
||
sum(1 for c in fake_backend.calls if c.action == "mount" and c.mountpoint.name == "a") == 1
|
||
)
|
||
|
||
|
||
# -- Test (drei Schritte) -----------------------------------------------------------
|
||
|
||
|
||
def test_check_unsaved_config_mounts_temporarily(
|
||
manager: MountManager, fake_backend: FakeMountBackend, mount_base: Path
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
cfg = MountConfig("neu", MountType.CIFS, "nas", "scans")
|
||
cred = CredentialData(CredentialKind.PASSWORD, "u", "pw")
|
||
report = manager.check(cfg, cred)
|
||
assert [(s.name, s.ok) for s in report.steps] == [
|
||
("mount", True),
|
||
("read", True),
|
||
("write", True),
|
||
]
|
||
assert report.can_write and report.can_delete
|
||
assert report.free_bytes is not None
|
||
assert [r.value for r in report.roles] == ["Quelle mit Löschen", "Ziel"]
|
||
# temporärer Mountpoint wieder weg, Testdatei gelöscht
|
||
assert list(mount_base.iterdir()) == []
|
||
assert list(fake_backend.remote_root.joinpath("nas", "scans").iterdir()) == []
|
||
assert not fake_backend.mounted
|
||
|
||
|
||
def test_check_mount_failure_reports_steps(manager: MountManager, mount_base: Path) -> None:
|
||
report = manager.check(MountConfig("x", MountType.CIFS, "nas", "fehlt"), None)
|
||
assert [(s.name, s.ok) for s in report.steps] == [
|
||
("mount", False),
|
||
("read", None),
|
||
("write", None),
|
||
]
|
||
assert not report.ok
|
||
assert list(mount_base.iterdir()) == []
|
||
|
||
|
||
def test_check_read_only_share_is_source_copy_only(
|
||
manager: MountManager, fake_backend: FakeMountBackend
|
||
) -> None:
|
||
remote = fake_backend.export("nas", "ro")
|
||
remote.chmod(0o555)
|
||
try:
|
||
report = manager.check(MountConfig("ro", MountType.CIFS, "nas", "ro"), None)
|
||
finally:
|
||
remote.chmod(0o755)
|
||
assert report.can_read and not report.can_write
|
||
assert [r.value for r in report.roles] == ["nur Quelle, kopieren"]
|
||
assert report.steps[2].ok is False
|
||
|
||
|
||
def test_check_stores_result_on_saved_mount(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
m = _state(session_factory, mid)
|
||
report = manager.check(MountConfig.from_model(m), None, mount_id=mid)
|
||
assert report.steps[0].message == "bereits eingehängt"
|
||
m = _state(session_factory, mid)
|
||
assert m.write_ok is True and m.last_test_result and '"ok": true' in m.last_test_result
|
||
assert fake_backend.is_mounted(manager.mountpoint_for("scans")) # bleibt eingehängt
|
||
|
||
|
||
# -- Fallstrick 2: nicht eingehängt ---------------------------------------------------
|
||
|
||
|
||
def test_check_alive_refuses_unmounted_mountpoint(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = manager.check_alive(mid)
|
||
assert mp is not None
|
||
fake_backend.vanish(mp)
|
||
assert mp.is_dir() and list(mp.iterdir()) == [] # sieht aus wie eine leere Freigabe
|
||
with pytest.raises(MountUnavailable, match="nicht eingehängt"):
|
||
manager.check_alive(mid)
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.ERROR
|
||
# kein Wiederherstellungsversuch, kein Schreiben
|
||
assert [c.action for c in fake_backend.calls] == ["mount"]
|
||
assert list(mp.iterdir()) == []
|
||
|
||
|
||
# -- Fallstrick 13: gestörte und hängende Mounts ---------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("code", [errno.ESTALE, errno.EIO, errno.ENOTCONN])
|
||
def test_stale_mount_is_lazily_unmounted_and_remounted(
|
||
code: int,
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = manager.mountpoint_for("scans")
|
||
fake_backend.inject_fault(mp, code)
|
||
assert manager.check_alive(mid) == mp
|
||
actions = [(c.action, c.lazy) for c in fake_backend.calls]
|
||
assert actions == [("mount", False), ("unmount", True), ("mount", False)]
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.MOUNTED and m.last_error is None
|
||
|
||
|
||
def test_recovery_gives_up_after_three_attempts_with_growing_delay(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
sleeps: list[float],
|
||
notifications: list[tuple[str, str, str]],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = manager.mountpoint_for("scans")
|
||
fake_backend.inject_fault(mp, errno.ESTALE)
|
||
fake_backend.fail_next_mount(times=3)
|
||
with pytest.raises(MountUnavailable, match="3 Versuchen"):
|
||
manager.check_alive(mid)
|
||
assert sleeps == [1, 2] # 1. Versuch sofort, dann wachsender Abstand
|
||
assert sum(1 for c in fake_backend.calls if c.action == "mount") == 4
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.ERROR
|
||
assert notifications[-1][0] == "mount.error"
|
||
|
||
|
||
def test_hanging_mount_times_out_and_is_recovered(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = manager.mountpoint_for("scans")
|
||
fake_backend.inject_fault(mp, "hang")
|
||
assert manager.check_alive(mid) == mp # blockiert nicht, Zeitlimit 0.5 s
|
||
assert ("unmount", True) in [(c.action, c.lazy) for c in fake_backend.calls]
|
||
assert _state(session_factory, mid).state is MountState.MOUNTED
|
||
|
||
|
||
def test_hanging_mount_without_recovery_shows_since(
|
||
manager: MountManager,
|
||
fake_backend: FakeMountBackend,
|
||
make_mount: MountFactory,
|
||
session_factory: sessionmaker[Session],
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
mp = manager.mountpoint_for("scans")
|
||
fake_backend.heal_on_remount = False
|
||
fake_backend.inject_fault(mp, "hang")
|
||
with pytest.raises(MountUnavailable):
|
||
manager.check_alive(mid)
|
||
m = _state(session_factory, mid)
|
||
assert m.state is MountState.ERROR
|
||
assert m.state_since is not None
|
||
assert m.last_error and "keine Antwort" in m.last_error
|
||
|
||
|
||
def test_non_recoverable_error_does_not_remount(
|
||
manager: MountManager, fake_backend: FakeMountBackend, make_mount: MountFactory
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
manager.mount(mid)
|
||
fake_backend.inject_fault(manager.mountpoint_for("scans"), errno.EACCES)
|
||
with pytest.raises(MountUnavailable):
|
||
manager.check_alive(mid)
|
||
assert [c.action for c in fake_backend.calls] == ["mount"]
|
||
|
||
|
||
def test_usage(
|
||
manager: MountManager, fake_backend: FakeMountBackend, make_mount: MountFactory
|
||
) -> None:
|
||
fake_backend.export("nas", "scans")
|
||
mid = make_mount()
|
||
assert manager.usage(mid) is None
|
||
manager.mount(mid)
|
||
usage = manager.usage(mid)
|
||
assert usage is not None and usage[0] >= usage[1] > 0
|