Florian WildeandClaude Opus 5.5
8276ce2afb
CIFS lieferte per stat() sekundenlang eine veraltete Größe einer noch wachsenden Datei; erst open() erzwingt frische Attribute. Gesperrte Dateien gelten als in Benutzung. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
782 lines
29 KiB
Python
782 lines
29 KiB
Python
"""Mover-Engine mit lokalen Zielen – Fallstricke 1, 3, 4, 10, 11, 14 und Quarantäne."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import errno
|
||
import os
|
||
import time
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from sqlalchemy import select
|
||
from sqlalchemy.orm import Session, sessionmaker
|
||
|
||
from netmover.db import session_scope
|
||
from netmover.models import (
|
||
CollisionPolicy,
|
||
CopiedFile,
|
||
ItemStatus,
|
||
PatternType,
|
||
Rule,
|
||
RuleAction,
|
||
RunStatus,
|
||
)
|
||
from netmover.mover import destinations as dest_mod
|
||
from netmover.mover.destinations import LocalDestination, TransferError
|
||
from netmover.mover.engine import Engine, RuleSnapshot, RunResult
|
||
from netmover.mover.ledger import Ledger
|
||
from netmover.mover.naming import collision_candidates
|
||
from netmover.mover.scan import Matcher
|
||
from netmover.timeouts import CancelToken
|
||
from tests.conftest import MountFactory
|
||
|
||
|
||
@pytest.fixture
|
||
def rule_id(make_mount: MountFactory, session_factory: sessionmaker[Session]) -> int:
|
||
mid = make_mount()
|
||
with session_scope(session_factory) as s:
|
||
r = Rule(name="r", src_mount_id=mid, dst_path="/x")
|
||
s.add(r)
|
||
s.flush()
|
||
return r.id
|
||
|
||
|
||
@pytest.fixture
|
||
def ledger(session_factory: sessionmaker[Session]) -> Ledger:
|
||
return Ledger(session_factory)
|
||
|
||
|
||
Sleeper = Callable[[float], None]
|
||
|
||
|
||
@pytest.fixture
|
||
def hooks() -> list[Callable[[], None]]:
|
||
"""Aktionen, die während des Stabilitätsfensters ausgeführt werden."""
|
||
return []
|
||
|
||
|
||
@pytest.fixture
|
||
def engine(ledger: Ledger, hooks: list[Callable[[], None]]) -> Engine:
|
||
def sleep(_: float) -> None:
|
||
for h in hooks:
|
||
h()
|
||
|
||
return Engine(ledger, sleep=sleep)
|
||
|
||
|
||
@pytest.fixture
|
||
def cross_device(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
"""tmp_path liegt auf einem Gerät – für Kopier-Tests den Schnellweg abschalten."""
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
|
||
|
||
@pytest.fixture
|
||
def src(tmp_path: Path) -> Path:
|
||
p = tmp_path / "src"
|
||
p.mkdir()
|
||
return p
|
||
|
||
|
||
@pytest.fixture
|
||
def dst(tmp_path: Path) -> Path:
|
||
p = tmp_path / "dst"
|
||
p.mkdir()
|
||
return p
|
||
|
||
|
||
def snap(rule_id: int, **kw: object) -> RuleSnapshot:
|
||
return RuleSnapshot(id=rule_id, name="r", **kw) # type: ignore[arg-type]
|
||
|
||
|
||
def run(
|
||
engine: Engine, rule: RuleSnapshot, src: Path, dst: Path | LocalDestination, **kw: object
|
||
) -> RunResult:
|
||
d = dst if isinstance(dst, LocalDestination) else LocalDestination(dst)
|
||
return engine.execute(rule, src, d, CancelToken(), **kw) # type: ignore[arg-type]
|
||
|
||
|
||
def write(path: Path, data: str = "inhalt") -> Path:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(data)
|
||
return path
|
||
|
||
|
||
def statuses(result: RunResult) -> dict[str, ItemStatus]:
|
||
return {i.src: i.status for i in result.items}
|
||
|
||
|
||
# -- Fallstrick 1: halb geschriebene Dateien --------------------------------------------
|
||
|
||
|
||
def test_growing_file_is_not_touched(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]]
|
||
) -> None:
|
||
write(src / "fertig.pdf", "a" * 100)
|
||
growing = write(src / "wachsend.pdf", "b" * 10)
|
||
hooks.append(lambda: growing.write_text("b" * 20)) # wächst während des Fensters
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"fertig.pdf": ItemStatus.MOVED, "wachsend.pdf": ItemStatus.SKIPPED}
|
||
assert growing.read_text() == "b" * 20
|
||
assert not (dst / "wachsend.pdf").exists()
|
||
assert (dst / "fertig.pdf").read_text() == "a" * 100
|
||
|
||
|
||
def test_mtime_change_alone_counts_as_unstable(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, hooks: list[Callable[[], None]]
|
||
) -> None:
|
||
f = write(src / "scan.pdf")
|
||
st = f.stat()
|
||
hooks.append(lambda: os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000)))
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"scan.pdf": ItemStatus.SKIPPED}
|
||
assert f.exists()
|
||
|
||
|
||
def test_stability_window_is_waited(ledger: Ledger, rule_id: int, src: Path, dst: Path) -> None:
|
||
waited: list[float] = []
|
||
eng = Engine(ledger, sleep=waited.append)
|
||
write(src / "a.pdf")
|
||
run(eng, snap(rule_id, stability_seconds=7), src, dst)
|
||
assert waited == [7]
|
||
waited.clear()
|
||
run(eng, snap(rule_id, stability_seconds=7), src, dst) # nichts mehr da → kein Warten
|
||
assert waited == []
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"name",
|
||
[".versteckt.pdf", "scan.pdf.part", "x.tmp", "dl.crdownload", "f.filepart", "t.!ut", "A.PART"],
|
||
)
|
||
def test_hidden_and_partial_files_are_ignored(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, name: str
|
||
) -> None:
|
||
write(src / name)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.seen == 0
|
||
assert (src / name).exists()
|
||
|
||
|
||
def test_min_age_and_max_size(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
write(src / "neu.pdf")
|
||
old = write(src / "alt.pdf")
|
||
os.utime(old, (time.time() - 3600, time.time() - 3600))
|
||
write(src / "gross.pdf", "x" * (2 * 1024 * 1024))
|
||
os.utime(src / "gross.pdf", (time.time() - 3600, time.time() - 3600))
|
||
result = run(engine, snap(rule_id, min_age_seconds=600, max_size_mb=1), src, dst)
|
||
assert statuses(result) == {
|
||
"alt.pdf": ItemStatus.MOVED,
|
||
"gross.pdf": ItemStatus.SKIPPED,
|
||
"neu.pdf": ItemStatus.SKIPPED,
|
||
}
|
||
|
||
|
||
# -- Fallstrick 3: Atomarität ---------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.usefixtures("cross_device")
|
||
def test_final_name_never_shows_partial_content(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
data = "x" * (3 * dest_mod.CHUNK + 17)
|
||
write(src / "gross.pdf", data)
|
||
seen_during_write: list[list[str]] = []
|
||
original = LocalDestination._upload
|
||
|
||
def spy(self: LocalDestination, *a: object, **kw: object) -> list[str]:
|
||
notes = original(self, *a, **kw) # type: ignore[arg-type]
|
||
seen_during_write.append(sorted(p.name for p in dst.iterdir()))
|
||
return notes
|
||
|
||
monkeypatch.setattr(LocalDestination, "_upload", spy)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"gross.pdf": ItemStatus.MOVED}
|
||
# während/nach dem Schreiben existiert nur die .part-Datei, nicht der Endname
|
||
(names,) = seen_during_write
|
||
assert len(names) == 1 and names[0].startswith(".netmover-") and names[0].endswith(".part")
|
||
assert (dst / "gross.pdf").read_text() == data
|
||
assert not (src / "gross.pdf").exists()
|
||
|
||
|
||
STAGES = ["_upload", "_apply_meta", "_link_noreplace", "verify"]
|
||
|
||
|
||
@pytest.mark.usefixtures("cross_device")
|
||
@pytest.mark.parametrize("stage", STAGES)
|
||
def test_abort_at_any_stage_keeps_source_and_leaves_no_part(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, stage: str
|
||
) -> None:
|
||
write(src / "scan.pdf", "wichtig")
|
||
if stage == "verify":
|
||
# Quelle wird zwischen Kopie und Umbenennen verändert
|
||
original = LocalDestination._apply_meta
|
||
|
||
def touch(self: LocalDestination, rel: str, meta: object) -> list[str]:
|
||
(src / "scan.pdf").write_text("wichtig!!")
|
||
return original(self, rel, meta) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr(LocalDestination, "_apply_meta", touch)
|
||
else:
|
||
|
||
def boom(self: object, *a: object, **kw: object) -> None:
|
||
raise OSError(errno.EIO if stage == "_upload" else errno.ENOSPC, "Fehler")
|
||
|
||
monkeypatch.setattr(LocalDestination, stage, boom)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert (src / "scan.pdf").exists()
|
||
assert list(dst.iterdir()) == []
|
||
item = result.items[0] if result.items else None
|
||
if stage == "_upload": # EIO → Anbindung gestört, ganzer Lauf abgebrochen
|
||
assert result.status is RunStatus.ERROR
|
||
elif stage == "verify":
|
||
assert item is not None and item.status is ItemStatus.SKIPPED
|
||
else:
|
||
assert item is not None and item.status is ItemStatus.FAILED
|
||
|
||
|
||
@pytest.mark.parametrize("path", ["same-device", "copy"])
|
||
def test_source_not_deletable_keeps_target_and_prevents_duplicate(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch, path: str
|
||
) -> None:
|
||
if path == "copy":
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
else:
|
||
# Dateisystem ohne renameat2 → link + unlink, unlink scheitert
|
||
monkeypatch.setattr("netmover.mover.destinations.rename_noreplace", lambda *a: False)
|
||
write(src / "scan.pdf")
|
||
real_unlink = os.unlink
|
||
|
||
def deny(path: object, *a: object, **kw: object) -> None:
|
||
if str(path).startswith(str(src)):
|
||
raise PermissionError(errno.EACCES, "Permission denied", str(path))
|
||
real_unlink(path) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr("netmover.mover.engine.os.unlink", deny)
|
||
first = run(engine, snap(rule_id), src, dst)
|
||
assert first.items[0].status is ItemStatus.FAILED
|
||
assert "nicht löschbar" in (first.items[0].message or "")
|
||
assert (dst / "scan.pdf").exists() and (src / "scan.pdf").exists()
|
||
second = run(engine, snap(rule_id), src, dst)
|
||
assert second.items == [] and second.message == "1 bereits übertragen"
|
||
assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"]
|
||
|
||
|
||
def test_stale_part_files_are_cleaned(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
old = write(dst / ".netmover-dead.part", "rest")
|
||
young = write(dst / ".netmover-busy.part", "rest")
|
||
os.utime(old, (time.time() - 7200, time.time() - 7200))
|
||
run(engine, snap(rule_id), src, dst)
|
||
assert not old.exists()
|
||
assert young.exists()
|
||
|
||
|
||
def test_same_device_move_uses_link_without_copy(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
write(src / "a.pdf", "x")
|
||
ino = (src / "a.pdf").stat().st_ino
|
||
|
||
def no_copy(*a: object, **kw: object) -> None:
|
||
raise AssertionError("darf auf demselben Gerät nicht kopieren")
|
||
|
||
monkeypatch.setattr(LocalDestination, "_upload", no_copy)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
|
||
assert (dst / "a.pdf").stat().st_ino == ino
|
||
assert not (src / "a.pdf").exists()
|
||
|
||
|
||
def test_cross_device_path_is_used_when_devices_differ(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
write(src / "a.pdf", "x")
|
||
ino = (src / "a.pdf").stat().st_ino
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
|
||
assert (dst / "a.pdf").stat().st_ino != ino
|
||
|
||
|
||
def test_filesystem_without_hardlinks_falls_back_to_rename(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
|
||
def no_link(*a: object, **kw: object) -> None:
|
||
raise OSError(errno.EPERM, "Operation not permitted")
|
||
|
||
monkeypatch.setattr("netmover.mover.destinations.os.link", no_link)
|
||
write(src / "a.pdf")
|
||
write(dst / "a.pdf", "alt")
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.items[0].dst == str(dst / "a-1.pdf")
|
||
assert (dst / "a.pdf").read_text() == "alt"
|
||
|
||
|
||
# -- Fallstrick 4: Namenskollisionen ---------------------------------------------------------
|
||
|
||
|
||
def test_collision_candidates() -> None:
|
||
gen = collision_candidates("sub/scan001.pdf")
|
||
assert [next(gen) for _ in range(3)] == [
|
||
"sub/scan001.pdf",
|
||
"sub/scan001-1.pdf",
|
||
"sub/scan001-2.pdf",
|
||
]
|
||
gen = collision_candidates("README")
|
||
assert [next(gen) for _ in range(2)] == ["README", "README-1"]
|
||
|
||
|
||
def test_rename_counts_up_against_filesystem(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path
|
||
) -> None:
|
||
write(dst / "scan001.pdf", "alt")
|
||
for i in range(3):
|
||
write(src / "scan001.pdf", f"neu{i}")
|
||
run(engine, snap(rule_id), src, dst)
|
||
assert sorted(p.name for p in dst.iterdir()) == [
|
||
"scan001-1.pdf",
|
||
"scan001-2.pdf",
|
||
"scan001-3.pdf",
|
||
"scan001.pdf",
|
||
]
|
||
assert (dst / "scan001.pdf").read_text() == "alt"
|
||
assert (dst / "scan001-3.pdf").read_text() == "neu2"
|
||
|
||
|
||
def test_rename_rechecks_filesystem_each_time(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Zwischen Plan und Umbenennen belegt ein Dritter den Namen – kein Überschreiben."""
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
write(dst / "scan.pdf", "alt")
|
||
write(src / "scan.pdf", "neu")
|
||
original = LocalDestination._apply_meta
|
||
|
||
def intruder(self: LocalDestination, rel: str, meta: object) -> list[str]:
|
||
write(dst / "scan-1.pdf", "fremd")
|
||
return original(self, rel, meta) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr(LocalDestination, "_apply_meta", intruder)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.items[0].dst == str(dst / "scan-2.pdf")
|
||
assert (dst / "scan-1.pdf").read_text() == "fremd"
|
||
assert (dst / "scan-2.pdf").read_text() == "neu"
|
||
|
||
|
||
def test_skip_policy(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
write(dst / "a.pdf", "alt")
|
||
write(src / "a.pdf", "neu")
|
||
result = run(engine, snap(rule_id, on_collision=CollisionPolicy.SKIP), src, dst)
|
||
assert statuses(result) == {"a.pdf": ItemStatus.SKIPPED}
|
||
assert (dst / "a.pdf").read_text() == "alt"
|
||
assert (src / "a.pdf").exists()
|
||
|
||
|
||
def test_overwrite_policy(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
write(dst / "a.pdf", "alt")
|
||
write(src / "a.pdf", "neu")
|
||
result = run(engine, snap(rule_id, on_collision=CollisionPolicy.OVERWRITE), src, dst)
|
||
assert statuses(result) == {"a.pdf": ItemStatus.MOVED}
|
||
assert (dst / "a.pdf").read_text() == "neu"
|
||
|
||
|
||
# -- Fallstrick 11: Kopiermodus mit Gedächtnis ---------------------------------------------
|
||
|
||
|
||
def test_copy_mode_copies_each_version_once(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session]
|
||
) -> None:
|
||
rule = snap(rule_id, action=RuleAction.COPY)
|
||
f = write(src / "scan.pdf", "v1")
|
||
for _ in range(3):
|
||
run(engine, rule, src, dst)
|
||
assert sorted(p.name for p in dst.iterdir()) == ["scan.pdf"]
|
||
assert f.exists()
|
||
|
||
# neue Version → läuft durch die Kollisionsregel
|
||
f.write_text("version 2")
|
||
os.utime(f, ns=(f.stat().st_atime_ns, f.stat().st_mtime_ns + 5_000_000_000))
|
||
result = run(engine, rule, src, dst)
|
||
assert statuses(result) == {"scan.pdf": ItemStatus.COPIED}
|
||
run(engine, rule, src, dst)
|
||
assert sorted(p.name for p in dst.iterdir()) == ["scan-1.pdf", "scan.pdf"]
|
||
assert (dst / "scan-1.pdf").read_text() == "version 2"
|
||
|
||
# Quelle verschwindet → Buchführung wird bereinigt
|
||
f.unlink()
|
||
run(engine, rule, src, dst)
|
||
with session_factory() as s:
|
||
assert s.scalars(select(CopiedFile)).all() == []
|
||
|
||
|
||
def test_copy_mode_preserves_mtime(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
f = write(src / "a.pdf")
|
||
os.utime(f, ns=(1_600_000_000_000_000_000, 1_600_000_000_123_456_789))
|
||
run(engine, snap(rule_id, action=RuleAction.COPY), src, dst)
|
||
assert (dst / "a.pdf").stat().st_mtime_ns == 1_600_000_000_123_456_789
|
||
|
||
|
||
# -- Fallstrick 14: Platz ---------------------------------------------------------------------
|
||
|
||
|
||
def test_insufficient_space_skips_file_without_partial_copy(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
write(src / "a.pdf", "x" * 1000)
|
||
monkeypatch.setattr(LocalDestination, "free_bytes", lambda self: 1040) # < 1000 * 1.05
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.items[0].status is ItemStatus.FAILED
|
||
assert "zu wenig Platz" in (result.items[0].message or "")
|
||
assert list(dst.iterdir()) == []
|
||
assert (src / "a.pdf").exists()
|
||
|
||
|
||
def test_metadata_failure_is_only_a_hint(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
|
||
def deny(*a: object, **kw: object) -> None:
|
||
raise PermissionError(errno.EPERM, "Operation not permitted")
|
||
|
||
monkeypatch.setattr("netmover.mover.destinations.os.utime", deny)
|
||
write(src / "a.pdf")
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.items[0].status is ItemStatus.MOVED
|
||
assert "Änderungszeit nicht übernommen" in (result.items[0].message or "")
|
||
|
||
|
||
# -- Fallstrick 10: Berechtigungen ---------------------------------------------------------
|
||
|
||
|
||
def test_set_mode_applied(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
write(src / "a.pdf").chmod(0o600)
|
||
run(engine, snap(rule_id, set_mode=0o640), src, dst)
|
||
assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o640
|
||
|
||
|
||
def test_unchanged_mode_is_copied_from_source(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
write(src / "a.pdf").chmod(0o604)
|
||
run(engine, snap(rule_id), src, dst)
|
||
assert (dst / "a.pdf").stat().st_mode & 0o777 == 0o604
|
||
|
||
|
||
def test_chown_without_permission_is_clear_message(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path
|
||
) -> None:
|
||
if os.geteuid() == 0:
|
||
pytest.skip("als root ist chown erlaubt")
|
||
write(src / "a.pdf")
|
||
result = run(engine, snap(rule_id, set_owner=(0, 0)), src, dst)
|
||
assert result.items[0].status is ItemStatus.MOVED
|
||
assert "keine Berechtigung" in (result.items[0].message or "")
|
||
|
||
|
||
@pytest.mark.parametrize(("kind", "squash"), [("cifs", None), ("nfs", True), ("nfs", None)])
|
||
def test_chown_skipped_on_cifs_and_squashed_nfs_warns_once(
|
||
engine: Engine,
|
||
rule_id: int,
|
||
src: Path,
|
||
dst: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
kind: str,
|
||
squash: bool | None,
|
||
) -> None:
|
||
called: list[object] = []
|
||
monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a))
|
||
write(src / "a.pdf")
|
||
write(src / "b.pdf")
|
||
d = LocalDestination(dst, kind) # type: ignore[arg-type]
|
||
result = run(engine, snap(rule_id, set_owner=(1000, 1000), dst_root_squash=squash), src, d)
|
||
assert called == []
|
||
msgs = [i.message for i in result.items if i.message and "chown übersprungen" in i.message]
|
||
assert len(msgs) == 1 # einmalig warnen
|
||
|
||
|
||
def test_chown_on_nfs_without_squash_is_tried(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
called: list[object] = []
|
||
monkeypatch.setattr("netmover.mover.destinations.os.chown", lambda *a: called.append(a))
|
||
write(src / "a.pdf")
|
||
run(
|
||
engine,
|
||
snap(rule_id, set_owner=(1000, 100), dst_root_squash=False),
|
||
src,
|
||
LocalDestination(dst, "nfs"),
|
||
)
|
||
assert called and called[0][1:] == (1000, 100) # type: ignore[index]
|
||
|
||
|
||
# -- Filter, Rekursion, Probelauf, Quarantäne ---------------------------------------------
|
||
|
||
|
||
def test_matcher() -> None:
|
||
m = Matcher("*.pdf; *.JPG", PatternType.GLOB)
|
||
assert m("a/Scan.PDF") and m("b.jpg") and not m("c.txt")
|
||
r = Matcher(r"^rechnungen/.*\.pdf$", PatternType.REGEX)
|
||
assert r("rechnungen/2026.pdf") and not r("scan/2026.pdf")
|
||
|
||
|
||
def test_recursive_preserves_structure_and_skips_synology_dirs(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path
|
||
) -> None:
|
||
write(src / "max" / "a.pdf")
|
||
write(src / "@eaDir" / "a.pdf" / "SYNOFILE_THUMB_M.jpg")
|
||
write(src / "#recycle" / "alt.pdf")
|
||
write(src / "top.pdf")
|
||
flat = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(flat) == {"top.pdf": ItemStatus.MOVED}
|
||
rec = run(engine, snap(rule_id, recursive=True), src, dst)
|
||
assert statuses(rec) == {"max/a.pdf": ItemStatus.MOVED}
|
||
assert (dst / "max" / "a.pdf").exists()
|
||
assert (src / "#recycle" / "alt.pdf").exists()
|
||
|
||
|
||
def test_symlinks_are_ignored(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, tmp_path: Path
|
||
) -> None:
|
||
secret = write(tmp_path / "geheim.txt")
|
||
(src / "link.txt").symlink_to(secret)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.seen == 0 and secret.exists()
|
||
|
||
|
||
def test_dry_run_touches_nothing(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, session_factory: sessionmaker[Session]
|
||
) -> None:
|
||
write(src / "a.pdf")
|
||
write(dst / "a.pdf", "alt")
|
||
write(src / "b.pdf")
|
||
result = run(engine, snap(rule_id, action=RuleAction.COPY), src, dst, dry_run=True)
|
||
assert {i.src: (i.status, i.dst) for i in result.items} == {
|
||
"a.pdf": (ItemStatus.PLANNED, str(dst / "a-1.pdf")),
|
||
"b.pdf": (ItemStatus.PLANNED, str(dst / "b.pdf")),
|
||
}
|
||
assert sorted(p.name for p in dst.iterdir()) == ["a.pdf"]
|
||
assert sorted(p.name for p in src.iterdir()) == ["a.pdf", "b.pdf"]
|
||
with session_factory() as s:
|
||
assert s.scalars(select(CopiedFile)).all() == []
|
||
|
||
|
||
def test_quarantine_after_repeated_failures(
|
||
engine: Engine,
|
||
rule_id: int,
|
||
src: Path,
|
||
dst: Path,
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
|
||
def broken(self: object, *a: object, **kw: object) -> None:
|
||
raise TransferError("kaputt")
|
||
|
||
monkeypatch.setattr(LocalDestination, "_link_noreplace", broken)
|
||
write(src / "defekt.pdf")
|
||
quarantine = src / "quarantaene"
|
||
rule = snap(rule_id, quarantine_after=3, recursive=True)
|
||
msgs = []
|
||
for _ in range(3):
|
||
r = run(engine, rule, src, dst, quarantine_root=quarantine)
|
||
msgs.append(r.items[0].message)
|
||
assert msgs[-1] == "kaputt (Versuch 3)"
|
||
r = run(engine, rule, src, dst, quarantine_root=quarantine)
|
||
assert statuses(r) == {"defekt.pdf": ItemStatus.QUARANTINED}
|
||
assert (quarantine / "defekt.pdf").exists()
|
||
# Quarantäne liegt in der Quelle, wird aber trotz Rekursion nicht erneut gescannt
|
||
r = run(engine, rule, src, dst, quarantine_root=quarantine)
|
||
assert r.seen == 0
|
||
|
||
|
||
def test_mount_error_mid_run_aborts_instead_of_failing_every_file(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr(LocalDestination, "move_same_device", lambda *a, **k: None)
|
||
|
||
def stale(self: object, *a: object, **kw: object) -> None:
|
||
raise OSError(errno.ESTALE, "Stale file handle")
|
||
|
||
monkeypatch.setattr(LocalDestination, "_upload", stale)
|
||
write(src / "a.pdf")
|
||
write(src / "b.pdf")
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.status is RunStatus.ERROR
|
||
assert "gestört" in (result.message or "")
|
||
assert (src / "a.pdf").exists() and (src / "b.pdf").exists()
|
||
|
||
|
||
def test_cancel_stops_before_next_file(engine: Engine, rule_id: int, src: Path, dst: Path) -> None:
|
||
write(src / "a.pdf")
|
||
token = CancelToken()
|
||
token.cancel()
|
||
result = engine.execute(snap(rule_id), src, LocalDestination(dst), token)
|
||
assert result.status is RunStatus.ERROR
|
||
assert (src / "a.pdf").exists()
|
||
|
||
|
||
def test_missing_source_dir(engine: Engine, rule_id: int, tmp_path: Path, dst: Path) -> None:
|
||
result = run(engine, snap(rule_id), tmp_path / "fehlt", dst)
|
||
assert result.status is RunStatus.ERROR and "existiert nicht" in (result.message or "")
|
||
|
||
|
||
def test_local_destination_must_exist(
|
||
engine: Engine, rule_id: int, src: Path, tmp_path: Path
|
||
) -> None:
|
||
write(src / "a.pdf")
|
||
result = run(engine, snap(rule_id), src, LocalDestination(tmp_path / "fehlt"))
|
||
assert result.status is RunStatus.ERROR
|
||
assert (src / "a.pdf").exists()
|
||
|
||
|
||
@pytest.mark.usefixtures("cross_device")
|
||
def test_fsync_before_final_rename(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
from netmover.mover import fsops
|
||
|
||
events: list[str] = []
|
||
real_fsync, real_rename = os.fsync, fsops.rename_noreplace
|
||
|
||
def fsync(fd: int) -> None:
|
||
events.append("fsync")
|
||
real_fsync(fd)
|
||
|
||
def rename(a: object, b: object) -> bool:
|
||
events.append(f"rename:{Path(str(b)).name}")
|
||
return real_rename(a, b) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr("netmover.mover.destinations.os.fsync", fsync)
|
||
monkeypatch.setattr("netmover.mover.destinations.rename_noreplace", rename)
|
||
write(src / "a.pdf")
|
||
run(engine, snap(rule_id), src, dst)
|
||
assert events.index("fsync") < events.index("rename:a.pdf")
|
||
|
||
|
||
@pytest.mark.usefixtures("cross_device")
|
||
def test_name_taken_between_check_and_link_is_never_overwritten(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
"""Enges Wettlauf-Fenster: exists() sagt „frei“, link() meldet EEXIST."""
|
||
write(dst / "scan.pdf", "fremd")
|
||
write(src / "scan.pdf", "neu")
|
||
real_exists = LocalDestination.exists
|
||
lied: list[str] = []
|
||
|
||
def stale_exists(self: LocalDestination, rel: str) -> bool:
|
||
if rel == "scan.pdf" and not lied and self.root == dst:
|
||
lied.append(rel)
|
||
return False # veraltete Antwort
|
||
return real_exists(self, rel)
|
||
|
||
monkeypatch.setattr(LocalDestination, "plan", lambda self, rel, policy: rel)
|
||
monkeypatch.setattr(LocalDestination, "exists", stale_exists)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert lied
|
||
assert result.items[0].dst == str(dst / "scan-1.pdf")
|
||
assert (dst / "scan.pdf").read_text() == "fremd"
|
||
assert (dst / "scan-1.pdf").read_text() == "neu"
|
||
|
||
|
||
# -- renameat2(RENAME_NOREPLACE) ----------------------------------------------------------
|
||
|
||
|
||
def test_rename_noreplace(tmp_path: Path) -> None:
|
||
from netmover.mover.fsops import rename_noreplace
|
||
|
||
a, b = write(tmp_path / "a", "neu"), write(tmp_path / "b", "alt")
|
||
ino = a.stat().st_ino
|
||
if not rename_noreplace(a, tmp_path / "c"):
|
||
pytest.skip("Dateisystem des Testrechners kann kein renameat2")
|
||
assert (tmp_path / "c").stat().st_ino == ino and not a.exists() # echtes rename
|
||
with pytest.raises(FileExistsError):
|
||
rename_noreplace(tmp_path / "c", b)
|
||
assert b.read_text() == "alt" and (tmp_path / "c").exists()
|
||
|
||
|
||
@pytest.mark.usefixtures("cross_device")
|
||
@pytest.mark.parametrize("supported", [True, False])
|
||
def test_final_step_is_a_real_rename_when_possible(
|
||
engine: Engine,
|
||
rule_id: int,
|
||
src: Path,
|
||
dst: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
supported: bool,
|
||
) -> None:
|
||
"""Watcher wie Paperless sehen ein rename (IN_MOVED_TO); ohne Unterstützung
|
||
bleibt es beim link-Rückfallweg – überschrieben wird in keinem Fall."""
|
||
links: list[object] = []
|
||
real_link = os.link
|
||
|
||
def spy_link(a: object, b: object, *args: object, **kw: object) -> None:
|
||
links.append(b)
|
||
real_link(a, b) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr("netmover.mover.destinations.os.link", spy_link)
|
||
if not supported:
|
||
monkeypatch.setattr("netmover.mover.destinations.rename_noreplace", lambda *a: False)
|
||
write(dst / "a.pdf", "alt")
|
||
write(src / "a.pdf", "neu")
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert result.items[0].dst == str(dst / "a-1.pdf")
|
||
assert (dst / "a.pdf").read_text() == "alt" and (dst / "a-1.pdf").read_text() == "neu"
|
||
assert bool(links) is not supported
|
||
assert [p.name for p in dst.iterdir() if p.name.endswith(".part")] == []
|
||
|
||
|
||
# -- Praxisbefund: veralteter Attribut-Cache (CIFS/NFS) -------------------------------------
|
||
|
||
|
||
def test_stability_uses_fresh_attributes_not_stat_cache(
|
||
engine: Engine,
|
||
rule_id: int,
|
||
src: Path,
|
||
dst: Path,
|
||
hooks: list[Callable[[], None]],
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""Auf dem Testhost lieferte CIFS per stat() sekundenlang eine veraltete Größe,
|
||
während die Datei wuchs. Simuliert: os.stat liefert für die Datei immer den
|
||
alten Stand (Cache), nur öffnen + fstat sieht die Wahrheit."""
|
||
growing = write(src / "langsam.pdf", "a" * 10)
|
||
stale = os.stat(growing)
|
||
real_stat = os.stat
|
||
|
||
def cached_stat(path: object, *a: object, **kw: object) -> os.stat_result:
|
||
if str(path) == str(growing):
|
||
return stale
|
||
return real_stat(path, *a, **kw) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr(os, "stat", cached_stat)
|
||
hooks.append(lambda: growing.write_text("a" * 20))
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"langsam.pdf": ItemStatus.SKIPPED}
|
||
assert "wird noch geschrieben" in (result.items[0].message or "")
|
||
assert growing.exists() and not (dst / "langsam.pdf").exists()
|
||
|
||
|
||
def test_locked_file_counts_as_in_use(
|
||
engine: Engine, rule_id: int, src: Path, dst: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
f = write(src / "gesperrt.pdf")
|
||
real_open = os.open
|
||
|
||
def locked(path: object, flags: int, *a: object, **kw: object) -> int:
|
||
if str(path) == str(f):
|
||
raise PermissionError(errno.EACCES, "Permission denied", str(path))
|
||
return real_open(path, flags, *a, **kw) # type: ignore[arg-type]
|
||
|
||
monkeypatch.setattr("netmover.mover.scan.os.open", locked)
|
||
result = run(engine, snap(rule_id), src, dst)
|
||
assert statuses(result) == {"gesperrt.pdf": ItemStatus.SKIPPED}
|
||
assert result.items[0].message == "Datei ist gerade in Benutzung"
|
||
assert f.exists()
|