SyncVito/tests/test_sync_engine.py

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"))


@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()