127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
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_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()
|