75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
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"
|