272 lines
9.3 KiB
Python
272 lines
9.3 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from syncvito.models import ConflictMode, DeleteMode, SyncDirection, SyncTask
|
|
from syncvito.state_store import StateStore
|
|
from syncvito.sync_engine import SyncEngine
|
|
|
|
|
|
def _engine(tmp_path: Path) -> SyncEngine:
|
|
return SyncEngine(StateStore(tmp_path / "state.json"))
|
|
|
|
|
|
def test_progress_reports_scan_before_copying(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from itertools import count
|
|
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
(source / "datei.txt").write_text("Inhalt")
|
|
ticks = count()
|
|
monkeypatch.setattr("syncvito.sync_engine.monotonic", lambda: float(next(ticks)))
|
|
messages = []
|
|
|
|
def progress(value: int, detail: str) -> None:
|
|
messages.append((value, detail))
|
|
if "eingelesen" in detail or "Dateien gefunden" in detail:
|
|
assert value == 0
|
|
assert not (target / "datei.txt").exists()
|
|
|
|
result = _engine(tmp_path).synchronize(SyncTask(name="Test", source=str(source), target=str(target)), progress)
|
|
|
|
assert result.successful
|
|
assert "Ordner werden geprüft" in messages[0][1]
|
|
assert any("Quelle wird eingelesen" in detail for _, detail in messages)
|
|
assert any("Ziel wird eingelesen" in detail for _, detail in messages)
|
|
assert any("1 Dateien gefunden" in detail for _, detail in messages)
|
|
assert messages[-1][0] == 100
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "nt", reason="Windows-Junctions")
|
|
def test_inventory_does_not_follow_windows_junctions(tmp_path: Path) -> None:
|
|
import _winapi
|
|
|
|
source = tmp_path / "source"
|
|
outside = tmp_path / "outside"
|
|
source.mkdir()
|
|
outside.mkdir()
|
|
(outside / "extern.txt").write_text("Nicht synchronisieren")
|
|
(source / "lokal.txt").write_text("Synchronisieren")
|
|
junction = source / "verknuepfung"
|
|
_winapi.CreateJunction(str(outside), str(junction))
|
|
try:
|
|
files, directories = SyncEngine._inventory(source, [])
|
|
assert set(files) == {"lokal.txt"}
|
|
assert "verknuepfung" not in directories
|
|
finally:
|
|
junction.rmdir()
|
|
|
|
|
|
def test_inventory_prunes_unselected_subtrees(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
source = tmp_path / "source"
|
|
(source / "auswahl").mkdir(parents=True)
|
|
(source / "ignorieren" / "unterordner").mkdir(parents=True)
|
|
(source / "auswahl" / "ja.txt").write_text("ja")
|
|
visited = []
|
|
original_walk = os.walk
|
|
|
|
def walk(*args, **kwargs):
|
|
for entry in original_walk(*args, **kwargs):
|
|
visited.append(Path(entry[0]))
|
|
yield entry
|
|
|
|
monkeypatch.setattr("syncvito.sync_engine.os.walk", walk)
|
|
files, _ = SyncEngine._inventory(source, ["auswahl"])
|
|
assert set(files) == {str(Path("auswahl/ja.txt"))}
|
|
assert visited == [source, source / "auswahl"]
|
|
|
|
|
|
@pytest.mark.parametrize("winerror", [None, 5])
|
|
def test_access_denied_is_not_reported_as_file_lock(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, winerror: int | None
|
|
) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
(source / "geschuetzt.txt").write_text("geschützt")
|
|
(source / "lesbar.txt").write_text("lesbar")
|
|
from syncvito import sync_engine
|
|
|
|
copy = sync_engine._atomic_copy
|
|
|
|
def copy_with_denied_file(source_path: Path, target_path: Path) -> None:
|
|
if source_path.name == "geschuetzt.txt":
|
|
error = PermissionError(13, "Zugriff verweigert", str(source_path))
|
|
if winerror is not None:
|
|
error.winerror = winerror
|
|
raise error
|
|
copy(source_path, target_path)
|
|
|
|
monkeypatch.setattr(sync_engine, "_atomic_copy", copy_with_denied_file)
|
|
result = _engine(tmp_path).synchronize(SyncTask(name="AppData", source=str(source), target=str(target)))
|
|
|
|
assert not result.successful
|
|
assert not result.blocked
|
|
assert len(result.errors) == 1
|
|
assert str(source / "geschuetzt.txt") in result.errors[0]
|
|
assert (target / "lesbar.txt").read_text() == "lesbar"
|
|
|
|
|
|
def test_one_way_copy_and_mirror_delete(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
target.mkdir()
|
|
(source / "neu.txt").write_text("aktuell", encoding="utf-8")
|
|
(target / "alt.txt").write_text("veraltet", encoding="utf-8")
|
|
task = SyncTask(
|
|
name="Einweg",
|
|
source=str(source),
|
|
target=str(target),
|
|
delete_mode=DeleteMode.MIRROR,
|
|
)
|
|
|
|
result = _engine(tmp_path).synchronize(task)
|
|
|
|
assert result.successful
|
|
assert (target / "neu.txt").read_text(encoding="utf-8") == "aktuell"
|
|
assert not (target / "alt.txt").exists()
|
|
|
|
|
|
def test_selection_limits_synced_files(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
(source / "auswahl").mkdir(parents=True)
|
|
target.mkdir()
|
|
(source / "auswahl" / "ja.txt").write_text("ja", encoding="utf-8")
|
|
(source / "nein.txt").write_text("nein", encoding="utf-8")
|
|
task = SyncTask(
|
|
name="Auswahl",
|
|
source=str(source),
|
|
target=str(target),
|
|
selected_paths=["auswahl"],
|
|
)
|
|
|
|
_engine(tmp_path).synchronize(task)
|
|
|
|
assert (target / "auswahl" / "ja.txt").exists()
|
|
assert not (target / "nein.txt").exists()
|
|
|
|
|
|
def test_bidirectional_newer_file_wins(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
target.mkdir()
|
|
left = source / "datei.txt"
|
|
right = target / "datei.txt"
|
|
left.write_text("links", encoding="utf-8")
|
|
right.write_text("rechts", encoding="utf-8")
|
|
os.utime(left, ns=(2_000_000_000, 2_000_000_000))
|
|
os.utime(right, ns=(1_000_000_000, 1_000_000_000))
|
|
task = SyncTask(
|
|
name="Beide",
|
|
source=str(source),
|
|
target=str(target),
|
|
direction=SyncDirection.BIDIRECTIONAL,
|
|
)
|
|
|
|
result = _engine(tmp_path).synchronize(task)
|
|
|
|
assert result.successful
|
|
assert right.read_text(encoding="utf-8") == "links"
|
|
|
|
|
|
def test_conflict_copy_preserves_replaced_version(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
target.mkdir()
|
|
left = source / "plan.txt"
|
|
right = target / "plan.txt"
|
|
left.write_text("neu", encoding="utf-8")
|
|
right.write_text("alt", encoding="utf-8")
|
|
os.utime(left, ns=(3_000_000_000, 3_000_000_000))
|
|
os.utime(right, ns=(1_000_000_000, 1_000_000_000))
|
|
task = SyncTask(
|
|
name="Konflikt",
|
|
source=str(source),
|
|
target=str(target),
|
|
direction=SyncDirection.BIDIRECTIONAL,
|
|
conflict_mode=ConflictMode.CONFLICT_COPY,
|
|
)
|
|
|
|
result = _engine(tmp_path).synchronize(task)
|
|
|
|
conflict_files = list(target.glob("plan_conflict_*.txt"))
|
|
assert result.successful
|
|
assert result.conflicts == 1
|
|
assert right.read_text(encoding="utf-8") == "neu"
|
|
assert conflict_files[0].read_text(encoding="utf-8") == "alt"
|
|
|
|
|
|
def test_bidirectional_mirrors_known_deletion(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
target.mkdir()
|
|
(source / "entfernen.txt").write_text("inhalt", encoding="utf-8")
|
|
task = SyncTask(
|
|
name="Löschung",
|
|
source=str(source),
|
|
target=str(target),
|
|
direction=SyncDirection.BIDIRECTIONAL,
|
|
delete_mode=DeleteMode.MIRROR,
|
|
)
|
|
engine = _engine(tmp_path)
|
|
engine.synchronize(task)
|
|
(target / "entfernen.txt").unlink()
|
|
|
|
result = engine.synchronize(task)
|
|
|
|
assert result.successful
|
|
assert not (source / "entfernen.txt").exists()
|
|
|
|
|
|
def test_stage_progress_reporting_for_multiple_folders(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
(source / "AppData").mkdir(parents=True)
|
|
(source / "Documents").mkdir(parents=True)
|
|
(source / "AppData" / "config.json").write_text("{}", encoding="utf-8")
|
|
(source / "Documents" / "doc.txt").write_text("text", encoding="utf-8")
|
|
task = SyncTask(name="Backup", source=str(source), target=str(target))
|
|
|
|
messages: list[tuple[int, str]] = []
|
|
result = _engine(tmp_path).synchronize(task, progress=lambda v, d: messages.append((v, d)))
|
|
|
|
assert result.successful
|
|
assert any("[1/2 AppData]" in detail for _, detail in messages)
|
|
assert any("[2/2 Documents]" in detail for _, detail in messages)
|
|
|
|
|
|
def test_pause_and_resume_continues_where_interrupted(tmp_path: Path) -> None:
|
|
source = tmp_path / "source"
|
|
target = tmp_path / "target"
|
|
source.mkdir()
|
|
(source / "a.txt").write_text("a", encoding="utf-8")
|
|
(source / "b.txt").write_text("b", encoding="utf-8")
|
|
(source / "c.txt").write_text("c", encoding="utf-8")
|
|
task = SyncTask(name="ResumeTest", source=str(source), target=str(target))
|
|
engine = _engine(tmp_path)
|
|
|
|
# 1. Unterbrechung nach der ersten Datei
|
|
def cancel_after_one() -> bool:
|
|
return (target / "a.txt").exists()
|
|
|
|
result_paused = engine.synchronize(task, cancel_check=cancel_after_one)
|
|
assert result_paused.paused
|
|
assert not result_paused.successful
|
|
assert (target / "a.txt").exists()
|
|
assert not (target / "c.txt").exists()
|
|
|
|
# 2. Fortsetzen: Bereits kopierte Datei a.txt wird übersprungen, b und c werden kopiert
|
|
result_resumed = engine.synchronize(task)
|
|
assert result_resumed.successful
|
|
assert result_resumed.copied == 2
|
|
assert (target / "a.txt").exists()
|
|
assert (target / "b.txt").exists()
|
|
assert (target / "c.txt").exists()
|