Release SyncVito 1.0.7

tests/test_file_locks.py

@@ -0,0 +1,74 @@
+import ctypes
+import os
+from contextlib import contextmanager
+from ctypes import wintypes
+from pathlib import Path
+
+import pytest
+
+from syncvito.file_locks import is_file_locked
+from syncvito.models import SyncTask
+from syncvito.state_store import StateStore
+from syncvito.sync_engine import SyncEngine
+
+pytestmark = pytest.mark.skipif(os.name != "nt", reason="Windows-Dateisperren")
+
+
+@contextmanager
+def exclusive_lock(path: Path):
+    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel32.CreateFileW.argtypes = [
+        wintypes.LPCWSTR,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        ctypes.c_void_p,
+        wintypes.DWORD,
+        wintypes.DWORD,
+        wintypes.HANDLE,
+    ]
+    kernel32.CreateFileW.restype = wintypes.HANDLE
+    kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
+    kernel32.CloseHandle.restype = wintypes.BOOL
+    handle = kernel32.CreateFileW(str(path), 0x80000000, 0, None, 3, 0x80, None)
+    assert handle != ctypes.c_void_p(-1).value, ctypes.get_last_error()
+    try:
+        yield
+    finally:
+        assert kernel32.CloseHandle(handle)
+
+
+def test_detects_real_lock_and_release(tmp_path: Path) -> None:
+    path = tmp_path / "gesperrt.txt"
+    path.write_text("Inhalt")
+    assert not is_file_locked(path)
+    with exclusive_lock(path):
+        assert is_file_locked(path)
+    assert not is_file_locked(path)
+    assert not is_file_locked(tmp_path / "fehlt.txt")
+
+
+@pytest.mark.parametrize("locked_side", ["source", "target"])
+def test_sync_preserves_backup_and_continues_past_lock(tmp_path: Path, locked_side: str) -> None:
+    source = tmp_path / "source"
+    target = tmp_path / "target"
+    source.mkdir()
+    target.mkdir()
+    (source / "a.txt").write_text("neue Version")
+    (target / "a.txt").write_text("alt")
+    (source / "b.txt").write_text("lesbar")
+    locked_path = tmp_path / locked_side / "a.txt"
+    task = SyncTask(name="AppData", source=str(source), target=str(target))
+    engine = SyncEngine(StateStore(tmp_path / "state.json"))
+
+    with exclusive_lock(locked_path):
+        result = engine.synchronize(task)
+        assert not result.successful
+        assert [item.path for item in result.blocked] == [locked_path]
+        assert not result.errors
+        assert result.copied == 1
+        assert (target / "b.txt").read_text() == "lesbar"
+
+    assert (target / "a.txt").read_text() == "alt"
+    assert not list(target.glob("*.tmp"))
+    assert engine.synchronize(task).successful
+    assert (target / "a.txt").read_text() == "neue Version"