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


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


def test_concurrently_deleted_source_file_is_skipped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source = tmp_path / "source"
    target = tmp_path / "target"
    source.mkdir()
    target.mkdir()
    (source / "ephemeral.txt").write_text("wird gelöscht")
    task = SyncTask(name="Transient", source=str(source), target=str(target))
    engine = _engine(tmp_path)

    # Simuliere, dass die Datei kurz vor der Ausführung gelöscht wurde
    def delete_before_execute(operation, result):
        if operation.source and operation.source.name == "ephemeral.txt":
            operation.source.unlink(missing_ok=True)
        return SyncEngine._execute(operation, result)

    monkeypatch.setattr(engine, "_execute", delete_before_execute)
    result = engine.synchronize(task)

    assert result.successful
    assert result.skipped == 1
    assert not result.errors
    assert not (target / "ephemeral.txt").exists()


def test_ignored_paths_are_completely_skipped(tmp_path: Path) -> None:
    source = tmp_path / "source"
    target = tmp_path / "target"
    source.mkdir()
    target.mkdir()
    (source / "normal.txt").write_text("kopieren", encoding="utf-8")
    (source / "locked.db").write_text("gesperrt", encoding="utf-8")
    (source / "subdir").mkdir()
    (source / "subdir" / "ignored_file.txt").write_text("in ordner", encoding="utf-8")

    task = SyncTask(
        name="IgnoredTest",
        source=str(source),
        target=str(target),
        ignored_paths=["locked.db", "subdir/ignored_file.txt"],
    )
    engine = _engine(tmp_path)
    result = engine.synchronize(task)

    assert result.successful
    assert result.copied == 1
    assert (target / "normal.txt").exists()
    assert not (target / "locked.db").exists()
    assert not (target / "subdir" / "ignored_file.txt").exists()


def test_failed_files_added_to_failed_relatives(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source = tmp_path / "source"
    target = tmp_path / "target"
    source.mkdir()
    target.mkdir()
    (source / "ok.txt").write_text("ok", encoding="utf-8")
    (source / "fail.txt").write_text("fail", encoding="utf-8")
    task = SyncTask(name="FailTest", source=str(source), target=str(target))
    engine = _engine(tmp_path)

    real_execute = SyncEngine._execute

    def mock_execute(operation, result):
        if operation.relative == "fail.txt":
            raise PermissionError("Access denied")
        return real_execute(operation, result)

    monkeypatch.setattr(engine, "_execute", mock_execute)
    result = engine.synchronize(task)

    assert not result.successful
    assert "fail.txt" in result.failed_relatives
    assert "ok.txt" not in result.failed_relatives
    assert (target / "ok.txt").exists()