5 Commits
Author SHA1 Message Date
Davito a04db4ee96 Release SyncVito 1.0.7 2026-09-24 16:04:51 +02:00
Davito 3d6ab87ce3 Release SyncVito 1.0.6 2026-09-24 13:17:02 +02:00
Davito bb2a0db609 Release SyncVito 1.0.5 2026-09-24 13:10:07 +02:00
Davito 30982e8a4d Release SyncVito 1.0.3 2026-09-24 13:01:39 +02:00
Davito 8eb8cc1d0d Release SyncVito 1.0.2 2026-09-24 12:50:54 +02:00
16 changed files with 543 additions and 44 deletions

No files matched your search

+41
View File
@@ -1,5 +1,46 @@
# Änderungsprotokoll
## 1.0.7 – 2026-09-24
### Behoben
- Bei gesperrten Dateien, etwa in AppData, zeigt SyncVito eine kompakte Zusammenfassung und kennzeichnet die Sicherung als unvollständig. Vollständige Dateipfade und bekannte Sperrprozesse erscheinen im Protokoll.
- Fehlende Zugriffsrechte werden von tatsächlichen Dateisperren unterschieden; zusätzliche Fehler bleiben auch bei teilweise blockierten Sicherungen sichtbar.
- Die Sperrerkennung berücksichtigt die tatsächlich betroffene Quell- oder Zieldatei und verarbeitet Windows-Dateihandles und Fehlercodes korrekt.
- Automatische Wiederholungsversuche starten bereits nach Freigabe einzelner Dateien und berücksichtigen alle bereitstehenden Aufgaben.
- Während eines erneuten Synchronisationslaufs bleibt dessen Aktivität im Gesamtstatus sichtbar.
## 1.0.6 – 2026-09-24
### Behoben
- Dateianzeige und Fortschrittsleiste besitzen nun feste Breiten und verschieben sich beim Wechsel zwischen unterschiedlich langen Dateipfaden nicht mehr.
- Lange Dateipfade werden platzsparend gekürzt; der vollständige Text bleibt als Tooltip verfügbar.
## 1.0.5 – 2026-09-24
### Neu
- SyncVito prüft bei jedem Programmstart und während des Betriebs alle 30 Minuten automatisch auf Updates.
- Verfügbare Updates werden zusätzlich zum Hinweisfenster mit einer auffälligen „UPDATE“-Kennzeichnung am Eintrag „Einstellungen“ angezeigt.
### Behoben
- Wird eine automatische Prüfung durch eine laufende Synchronisation oder Updateaktion verhindert, wird sie nach einer Minute erneut versucht.
## 1.0.3 – 2026-09-24
### Behoben
- Der automatische Neustart nach einem internen Update setzt nun die geerbte PyInstaller-Laufzeitumgebung zurück und verhindert dadurch den Sicherheitsfehler beim Start der aktualisierten Anwendung.
## 1.0.2 – 2026-09-24
### Behoben
- Die Versionsanzeige in der Seitenleiste verwendet nun immer die zentrale Anwendungsversion.
- Unter der Versionsnummer wird jetzt „by Davito“ statt „Lokale Konfiguration“ angezeigt.
## 1.0.1 – 2026-09-23
### Behoben
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "syncvito"
version = "1.0.1"
version = "1.0.7"
description = "Windows-Desktopwerkzeug für lokale und NAS-Synchronisation"
readme = "README.md"
requires-python = ">=3.11"
+25 -16
View File
@@ -25,7 +25,7 @@ def derive_overall_status(
if TaskRuntimeState.ERROR in states:
return "error", "Fehler"
if TaskRuntimeState.WAITING_FOR_RELEASE in states:
return "error", "Wartet auf Dateifreigabe"
return "error", "Synchronisiert" if running else "Unvollständig – Dateien gesperrt"
if globally_paused:
return "paused", "Pausiert"
if TaskRuntimeState.ACTIVE not in states:
@@ -205,25 +205,31 @@ class SyncCoordinator(QObject):
if result.blocked:
paths = [blocked.path for blocked in result.blocked]
self._retry_tasks[task_id] = paths
descriptions: list[str] = []
for blocked in result.blocked:
names = ", ".join(f"{process.name} (PID {process.pid})" for process in blocked.processes)
descriptions.append(f"{blocked.path.name}: {names or 'sperrender Prozess nicht ermittelbar'}")
message = "Zurückgestellt – " + "; ".join(descriptions)
message = (
f"{result.copied} kopiert, {len(result.blocked)} Dateien gesperrt – Sicherung unvollständig. "
"Andere Dateien wurden weiterverarbeitet. Automatischer Wiederholungsversuch nach Freigabe; "
"ggf. betroffene Programme schließen. Details im Protokoll."
)
self._lock_messages[task_id] = message
self._emit_lock_summary()
self.log_created.emit(task.name, "Blockiert", message)
for blocked in result.blocked:
names = ", ".join(f"{process.name} (PID {process.pid})" for process in blocked.processes)
self.log_created.emit(
task.name, "Blockiert", f"{blocked.path}: {names or 'sperrender Prozess nicht ermittelbar'}"
)
self._set_task_state(task_id, TaskRuntimeState.WAITING_FOR_RELEASE)
elif result.errors:
self._retry_tasks[task_id] = []
self._lock_messages.pop(task_id, None)
self._emit_lock_summary()
self.log_created.emit(task.name, "Fehler", " | ".join(result.errors[:5]))
self._set_task_state(task_id, TaskRuntimeState.ERROR)
else:
self._retry_tasks.pop(task_id, None)
self._lock_messages.pop(task_id, None)
self._emit_lock_summary()
if result.errors:
# Auch zusätzliche Fehler eines teilweise blockierten Laufs sichtbar halten.
self._retry_tasks[task_id] = []
for error in result.errors:
self.log_created.emit(task.name, "Fehler", error)
self._set_task_state(task_id, TaskRuntimeState.ERROR)
elif not result.blocked:
self._retry_tasks.pop(task_id, None)
summary = f"{result.copied} kopiert, {result.deleted} gelöscht, {result.conflicts} Konfliktkopien"
self.log_created.emit(task.name, "Erfolg", summary)
self._set_task_state(task_id, TaskRuntimeState.ACTIVE)
@@ -273,13 +279,12 @@ class SyncCoordinator(QObject):
if self.config.settings.paused or self.is_running:
return
for task_id, paths in list(self._retry_tasks.items()):
if paths and any(is_file_locked(path) for path in paths):
if paths and all(is_file_locked(path) for path in paths):
continue
self.log_created.emit(
"Automatischer Versuch", "Hinweis", "Blockierte oder nicht erreichbare Datei wird erneut geprüft."
)
self.enqueue(task_id)
break
def _task(self, task_id: str) -> SyncTask | None:
return next((task for task in self.config.tasks if task.id == task_id), None)
@@ -296,4 +301,8 @@ class SyncCoordinator(QObject):
self.overall_status_changed.emit(status, text)
def _emit_lock_summary(self) -> None:
self.lock_changed.emit("\n".join(self._lock_messages.values()))
messages = list(self._lock_messages.values())
summary = "\n".join(messages[:2])
if len(messages) > 2:
summary += f"\nWeitere {len(messages) - 2} Aufgaben mit gesperrten Dateien. Details im Protokoll."
self.lock_changed.emit(summary)
+18 -3
View File
@@ -23,7 +23,22 @@ def is_file_locked(path: Path) -> bool:
"""Prüft eine vorhandene Datei durch exklusives Öffnen ohne Schreibzugriff."""
if not path.exists() or os.name != "nt":
return False
handle = ctypes.windll.kernel32.CreateFileW(
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
create_file = kernel32.CreateFileW
create_file.argtypes = [
wintypes.LPCWSTR,
wintypes.DWORD,
wintypes.DWORD,
ctypes.c_void_p,
wintypes.DWORD,
wintypes.DWORD,
wintypes.HANDLE,
]
create_file.restype = wintypes.HANDLE
close_handle = kernel32.CloseHandle
close_handle.argtypes = [wintypes.HANDLE]
close_handle.restype = wintypes.BOOL
handle = create_file(
str(path),
0x80000000,
0,
@@ -34,8 +49,8 @@ def is_file_locked(path: Path) -> bool:
)
invalid_handle = ctypes.c_void_p(-1).value
if handle == invalid_handle:
return ctypes.get_last_error() in {5, 32, 33}
ctypes.windll.kernel32.CloseHandle(handle)
return ctypes.get_last_error() in {32, 33}
close_handle(handle)
return False
+84 -16
View File
@@ -3,11 +3,11 @@
from __future__ import annotations
import sys
from datetime import datetime, timedelta
from datetime import datetime
from pathlib import Path
from PyQt6.QtCore import QEvent, Qt, QTimer, pyqtSignal
from PyQt6.QtGui import QAction, QCloseEvent, QFontDatabase
from PyQt6.QtGui import QAction, QCloseEvent, QFontDatabase, QResizeEvent
from PyQt6.QtWidgets import (
QAbstractItemView,
QApplication,
@@ -48,6 +48,38 @@ from syncvito.version import __version__
__all__ = ["MainWindow"]
UPDATE_CHECK_INTERVAL_MS = 30 * 60 * 1000
UPDATE_RETRY_INTERVAL_MS = 60 * 1000
PROGRESS_TEXT_WIDTH = 300
PROGRESS_BAR_WIDTH = 450
class _ElidedLabel(QLabel):
"""Hält dynamische Texte in einer festen Breite vollständig zugänglich."""
def __init__(self, text: str = "", parent: QWidget | None = None) -> None:
super().__init__(parent)
self._full_text = ""
self.setText(text)
def setText(self, text: str) -> None: # noqa: N802 - Qt API
self._full_text = text
self.setToolTip(text)
self.setAccessibleName(text)
self._refresh_elided_text()
def resizeEvent(self, event: QResizeEvent) -> None: # noqa: N802 - Qt API
super().resizeEvent(event)
self._refresh_elided_text()
def _refresh_elided_text(self) -> None:
visible_text = self.fontMetrics().elidedText(
self._full_text,
Qt.TextElideMode.ElideMiddle,
max(0, self.contentsRect().width()),
)
super().setText(visible_text)
class MainWindow(QMainWindow):
"""Zentrale SyncVito-Benutzeroberfläche."""
@@ -59,9 +91,10 @@ class MainWindow(QMainWindow):
self.config = config
self.config_store = config_store
self._allow_close = False
self._notified_update_version: str | None = None
self.setWindowTitle("SyncVito")
self.setWindowIcon(app_icon())
self.setMinimumSize(980, 680)
self.setMinimumSize(1040, 680)
self.resize(1220, 800)
self.coordinator = SyncCoordinator(config, config_store, self)
@@ -71,6 +104,15 @@ class MainWindow(QMainWindow):
self._connect_coordinator()
self._refresh_tasks()
self._set_global_status(*self.coordinator.current_overall_status())
self.update_check_timer = QTimer(self)
self.update_check_timer.setInterval(UPDATE_CHECK_INTERVAL_MS)
self.update_check_timer.timeout.connect(self._automatic_update_check)
self.update_check_timer.start()
self.update_retry_timer = QTimer(self)
self.update_retry_timer.setSingleShot(True)
self.update_retry_timer.setInterval(UPDATE_RETRY_INTERVAL_MS)
self.update_retry_timer.timeout.connect(self._automatic_update_check)
QTimer.singleShot(3000, self._automatic_update_check)
cleanup_attempts = 6 if "--updated" in sys.argv else 1
QTimer.singleShot(3000, lambda: self._cleanup_update_installers(cleanup_attempts))
@@ -159,9 +201,10 @@ class MainWindow(QMainWindow):
layout.addWidget(button)
self.nav_buttons.append(button)
self.nav_buttons[0].setChecked(True)
self.settings_nav_button = self.nav_buttons[2]
layout.addStretch(1)
version = QLabel("Version 1.0.0\nLokale Konfiguration")
version = QLabel(f"Version {__version__}\nby Davito")
version.setProperty("muted", True)
layout.addWidget(version)
return sidebar
@@ -340,7 +383,7 @@ class MainWindow(QMainWindow):
version_label = QLabel(f"Installierte Version: {__version__}")
version_label.setProperty("muted", True)
update_layout.addWidget(version_label)
self.update_check_box = QCheckBox("Beim Start automatisch nach Updates suchen")
self.update_check_box = QCheckBox("Beim Start und danach alle 30 Minuten automatisch nach Updates suchen")
self.update_check_box.setChecked(self.config.settings.check_updates_on_start)
update_layout.addWidget(self.update_check_box)
update_hint = QLabel(
@@ -366,15 +409,17 @@ class MainWindow(QMainWindow):
dock.setFixedHeight(66)
layout = QHBoxLayout(dock)
layout.setContentsMargins(28, 12, 28, 12)
self.progress_text = QLabel("Bereit")
self.progress_text.setMinimumWidth(250)
self.progress_text = _ElidedLabel("Bereit")
self.progress_text.setFixedWidth(PROGRESS_TEXT_WIDTH)
self.progress_text.setProperty("muted", True)
self.progress_bar = QProgressBar()
self.progress_bar.setFixedWidth(PROGRESS_BAR_WIDTH)
self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(0)
self.progress_bar.setFormat("%p %")
layout.addWidget(self.progress_text)
layout.addWidget(self.progress_bar, 1)
layout.addWidget(self.progress_bar)
layout.addStretch(1)
return dock
def _build_tray(self) -> None:
@@ -511,6 +556,7 @@ class MainWindow(QMainWindow):
def _save_settings(self) -> None:
settings = self.config.settings
old_autostart = settings.autostart
old_update_check = settings.check_updates_on_start
settings.autostart = self.autostart_check.isChecked()
settings.font_family = self.font_combo.currentText()
settings.font_size = self.font_size_spin.value()
@@ -528,18 +574,22 @@ class MainWindow(QMainWindow):
return
apply_theme(QApplication.instance(), settings)
self._append_log("Einstellungen", "Erfolg", "Darstellung und Systemstart wurden übernommen.")
if settings.check_updates_on_start and not old_update_check:
QTimer.singleShot(0, self._automatic_update_check)
elif not settings.check_updates_on_start:
self.update_retry_timer.stop()
def _automatic_update_check(self) -> None:
settings = self.config.settings
if not settings.check_updates_on_start or self.coordinator.is_running:
if not settings.check_updates_on_start:
self.update_retry_timer.stop()
return
if settings.last_update_check:
try:
last_check = datetime.fromisoformat(settings.last_update_check)
except ValueError:
last_check = datetime.min
if datetime.now() - last_check < timedelta(hours=24):
return
if self.coordinator.is_running or self.updater.busy:
if not self.update_retry_timer.isActive():
self.update_retry_timer.start()
return
self.update_retry_timer.stop()
self.update_check_timer.start(UPDATE_CHECK_INTERVAL_MS)
self._check_updates(manual=False)
def _check_updates(self, manual: bool) -> None:
@@ -558,12 +608,18 @@ class MainWindow(QMainWindow):
self._mark_update_checked()
self._update_progress(100, f"Version {info.version} ist verfügbar")
self._append_log("Update", "Hinweis", f"Version {info.version} ist verfügbar.")
self._set_update_indicator(info.version)
if not manual and self._notified_update_version == info.version:
return
self._notified_update_version = info.version
dialog = UpdateDialog(__version__, info, getattr(sys, "frozen", False), self)
if dialog.exec() == UpdateDialog.DialogCode.Accepted:
self.updater.download(info)
def _on_no_update(self, manual: bool) -> None:
self._mark_update_checked()
self._set_update_indicator(None)
self._notified_update_version = None
self._update_progress(100, "SyncVito ist aktuell")
self._append_log("Update", "Erfolg", "Die installierte Version ist aktuell.")
if manual:
@@ -579,6 +635,18 @@ class MainWindow(QMainWindow):
f"{message}\n\nBitte die Netzwerkverbindung prüfen und anschließend erneut versuchen.",
)
def _set_update_indicator(self, version: str | None) -> None:
available = bool(version)
self.settings_nav_button.setText("Einstellungen • UPDATE" if available else "Einstellungen")
self.settings_nav_button.setProperty("updateAvailable", available)
description = f"SyncVito {version} ist verfügbar" if version else ""
self.settings_nav_button.setToolTip(description)
self.settings_nav_button.setAccessibleDescription(description)
style = self.settings_nav_button.style()
style.unpolish(self.settings_nav_button)
style.polish(self.settings_nav_button)
self.settings_nav_button.update()
def _install_update(self, path: Path, info: UpdateInfo) -> None:
try:
self.updater.service.launch_installer(path, info.version)
+33 -6
View File
@@ -10,7 +10,7 @@ from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from syncvito.file_locks import LockingProcess, find_locking_processes
from syncvito.file_locks import LockingProcess, find_locking_processes, is_file_locked
from syncvito.models import ConflictMode, DeleteMode, SyncDirection, SyncTask
from syncvito.state_store import StateStore
@@ -117,12 +117,20 @@ class SyncEngine:
for index, operation in enumerate(operations, start=1):
try:
self._execute(operation, result)
except PermissionError as exc:
processes = tuple(find_locking_processes(operation.target))
result.blocked.append(BlockedFile(operation.target, processes))
result.errors.append(f"Zugriff verweigert: {operation.target} ({exc})")
except (OSError, shutil.Error) as exc:
result.errors.append(f"{operation.target}: {exc}")
failed_path = (
Path(exc.filename2 or exc.filename)
if isinstance(exc, OSError) and (exc.filename2 or exc.filename)
else operation.target
)
blocked_path = _blocked_path(operation, exc)
if blocked_path is not None:
processes = tuple(find_locking_processes(blocked_path))
result.blocked.append(BlockedFile(blocked_path, processes))
elif isinstance(exc, PermissionError):
result.errors.append(f"Zugriff verweigert: {failed_path} ({exc})")
else:
result.errors.append(f"{failed_path}: {exc}")
if progress:
detail = f"{operation.kind}: {operation.relative}"
progress(round(index / total * 100), detail)
@@ -295,6 +303,25 @@ class SyncEngine:
}
def _blocked_path(operation: _Operation, exc: OSError | shutil.Error) -> Path | None:
"""Ordnet auch Windows-Kopierfehler ohne Dateinamen der gesperrten Seite zu."""
winerror = getattr(exc, "winerror", None)
if winerror not in {5, 32, 33}:
return None
reported = (exc.filename2 or exc.filename) if isinstance(exc, OSError) else None
if reported and winerror in {32, 33}:
return Path(reported)
candidates = [Path(reported)] if reported else []
candidates.extend(path for path in (operation.source, operation.target) if path is not None)
for path in dict.fromkeys(candidates):
if is_file_locked(path):
return path
# Eine inzwischen freigegebene Datei beim nächsten Lauf erneut versuchen.
if winerror in {32, 33}:
return operation.source or operation.target
return None
def _atomic_copy(source: Path, target: Path) -> None:
temporary = target.with_name(f".{target.name}.syncvito.tmp")
try:
+6
View File
@@ -52,6 +52,12 @@ def _stylesheet(scale: float) -> str:
padding: {round(10 * scale)}px {round(12 * scale)}px; color: #9CACBF; }}
QPushButton[nav="true"]:hover {{ background: #17212E; color: #E8EDF4; }}
QPushButton[nav="true"]:checked {{ background: #1B2A3D; color: #FFFFFF; border-color: #2A568B; }}
QPushButton[nav="true"][updateAvailable="true"] {{ background: #3A2412; color: #FFF0DF;
border-color: #EA580C; }}
QPushButton[nav="true"][updateAvailable="true"]:hover {{ background: #4A2D14; color: #FFFFFF;
border-color: #FF7A2F; }}
QPushButton[nav="true"][updateAvailable="true"]:focus {{ border: 2px solid #FF9A5B;
padding: {round(9 * scale)}px {round(11 * scale)}px; }}
QLineEdit, QComboBox, QSpinBox, QListWidget, QTreeView, QTableWidget {{
background: #0E141D; border: 1px solid #304052; border-radius: {radius}px;
padding: {round(7 * scale)}px; color: #E8EDF4;
+1
View File
@@ -151,6 +151,7 @@ try {{
$bootstrapLog,
((Get-Date).ToString('o') + ' Starte verifizierten Installer.' + [Environment]::NewLine)
)
$env:PYINSTALLER_RESET_ENVIRONMENT = '1'
$setupArguments = @({setup_arguments})
& {_powershell_literal(installer_path)} @setupArguments
$setupExitCode = $LASTEXITCODE
+75
View File
@@ -0,0 +1,75 @@
from pathlib import Path
import pytest
from PyQt6.QtWidgets import QApplication
from syncvito.config_store import ConfigStore
from syncvito.coordinator import SyncCoordinator
from syncvito.models import AppConfig, SyncTask, TaskRuntimeState
from syncvito.sync_engine import BlockedFile, SyncResult
@pytest.fixture
def coordinator(tmp_path: Path):
application = QApplication.instance() or QApplication([])
task = SyncTask(name="AppData", source=str(tmp_path / "source"), target=str(tmp_path / "target"))
coordinator = SyncCoordinator(AppConfig(tasks=[task]), ConfigStore(tmp_path / "config.json"))
yield coordinator
coordinator._schedule_timer.stop()
coordinator._retry_timer.stop()
coordinator.stop()
assert application is not None
def test_many_locks_have_compact_banner_and_full_paths_in_log(coordinator: SyncCoordinator) -> None:
banners = []
logs = []
coordinator.lock_changed.connect(banners.append)
coordinator.log_created.connect(lambda *entry: logs.append(entry))
blocked = [BlockedFile(Path(f"AppData/Programm{index}/Cookies")) for index in range(300)]
task = coordinator.config.tasks[0]
coordinator._on_finished(task.id, SyncResult(copied=123, blocked=blocked))
assert len(banners[-1]) < 400
assert "123 kopiert" in banners[-1]
assert "300 Dateien gesperrt" in banners[-1]
assert "unvollständig" in banners[-1]
assert len(logs) == 301
assert str(blocked[-1].path) in logs[-1][2]
assert coordinator.task_state(task.id) is TaskRuntimeState.WAITING_FOR_RELEASE
coordinator._on_finished(task.id, SyncResult(copied=300))
assert banners[-1] == ""
assert task.id not in coordinator._retry_tasks
assert coordinator.task_state(task.id) is TaskRuntimeState.ACTIVE
def test_errors_are_visible_even_when_other_files_are_locked(coordinator: SyncCoordinator) -> None:
logs = []
coordinator.log_created.connect(lambda *entry: logs.append(entry))
task = coordinator.config.tasks[0]
coordinator._on_finished(task.id, SyncResult(blocked=[BlockedFile(Path("Cookies"))], errors=["Zugriff verweigert"]))
assert coordinator.task_state(task.id) is TaskRuntimeState.ERROR
assert any(level == "Fehler" and detail == "Zugriff verweigert" for _, level, detail in logs)
@pytest.mark.parametrize("all_locked", [False, True])
def test_retry_can_run_after_partial_release(
coordinator: SyncCoordinator, monkeypatch: pytest.MonkeyPatch, all_locked: bool
) -> None:
task = coordinator.config.tasks[0]
coordinator._retry_tasks[task.id] = [Path("dauerhaft"), Path("freigegeben")]
queued = []
monkeypatch.setattr(coordinator, "enqueue", queued.append)
monkeypatch.setattr("syncvito.coordinator.is_file_locked", lambda path: all_locked or path.name == "dauerhaft")
coordinator._check_retries()
assert queued == ([] if all_locked else [task.id])
def test_retry_queues_all_ready_tasks(coordinator: SyncCoordinator, monkeypatch: pytest.MonkeyPatch) -> None:
coordinator._retry_tasks = {"first": [], "second": []}
queued = []
monkeypatch.setattr(coordinator, "enqueue", queued.append)
coordinator._check_retries()
assert queued == ["first", "second"]
+74
View File
@@ -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"
+28
View File
@@ -0,0 +1,28 @@
from types import SimpleNamespace
from PyQt6.QtWidgets import QApplication
from syncvito.main_window import PROGRESS_BAR_WIDTH, PROGRESS_TEXT_WIDTH, MainWindow
def test_progress_widgets_keep_fixed_widths_for_long_paths() -> None:
application = QApplication.instance() or QApplication([])
owner = SimpleNamespace()
dock = MainWindow._build_progress_dock(owner)
dock.resize(900, 66)
dock.show()
application.processEvents()
bar_x = owner.progress_bar.x()
long_path = "Kopieren: " + "sehr-langer-ordner\\" * 20 + "wichtige-datei.ini"
owner.progress_text.setText(long_path)
application.processEvents()
assert owner.progress_text.width() == PROGRESS_TEXT_WIDTH
assert owner.progress_bar.width() == PROGRESS_BAR_WIDTH
assert owner.progress_bar.x() == bar_x
assert owner.progress_text.text() != long_path
assert "…" in owner.progress_text.text()
assert owner.progress_text.toolTip() == long_path
assert owner.progress_text.accessibleName() == long_path
dock.close()
+8 -1
View File
@@ -14,7 +14,14 @@ def test_healthy_active_tasks_are_green() -> None:
def test_waiting_lock_has_priority_over_active_task() -> None:
states = [TaskRuntimeState.ACTIVE, TaskRuntimeState.WAITING_FOR_RELEASE]
assert derive_overall_status(states, globally_paused=False) == ("error", "Wartet auf Dateifreigabe")
assert derive_overall_status(states, globally_paused=False) == ("error", "Unvollständig – Dateien gesperrt")
def test_waiting_task_does_not_hide_running_sync() -> None:
assert derive_overall_status([TaskRuntimeState.WAITING_FOR_RELEASE], False, running=True) == (
"error",
"Synchronisiert",
)
def test_error_has_highest_priority() -> None:
+33
View File
@@ -1,6 +1,8 @@
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
@@ -10,6 +12,37 @@ 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"
+112
View File
@@ -0,0 +1,112 @@
from types import SimpleNamespace
from syncvito.main_window import UPDATE_CHECK_INTERVAL_MS, UPDATE_RETRY_INTERVAL_MS, MainWindow
class FakeTimer:
def __init__(self, *, active: bool = False) -> None:
self.active = active
self.start_calls: list[int | None] = []
self.stop_calls = 0
def isActive(self) -> bool: # noqa: N802 - Qt API
return self.active
def start(self, interval: int | None = None) -> None:
self.active = True
self.start_calls.append(interval)
def stop(self) -> None:
self.active = False
self.stop_calls += 1
class FakeStyle:
def __init__(self) -> None:
self.unpolished = False
self.polished = False
def unpolish(self, _button: object) -> None:
self.unpolished = True
def polish(self, _button: object) -> None:
self.polished = True
class FakeButton:
def __init__(self) -> None:
self.text = ""
self.properties: dict[str, object] = {}
self.tooltip = ""
self.accessible_description = ""
self.button_style = FakeStyle()
self.updated = False
def setText(self, text: str) -> None: # noqa: N802 - Qt API
self.text = text
def setProperty(self, name: str, value: object) -> None: # noqa: N802 - Qt API
self.properties[name] = value
def setToolTip(self, text: str) -> None: # noqa: N802 - Qt API
self.tooltip = text
def setAccessibleDescription(self, text: str) -> None: # noqa: N802 - Qt API
self.accessible_description = text
def style(self) -> FakeStyle:
return self.button_style
def update(self) -> None:
self.updated = True
def _window(*, sync_running: bool = False, updater_busy: bool = False) -> SimpleNamespace:
checks: list[bool] = []
return SimpleNamespace(
config=SimpleNamespace(settings=SimpleNamespace(check_updates_on_start=True)),
coordinator=SimpleNamespace(is_running=sync_running),
updater=SimpleNamespace(busy=updater_busy),
update_check_timer=FakeTimer(active=True),
update_retry_timer=FakeTimer(),
_check_updates=lambda manual: checks.append(manual),
checks=checks,
)
def test_automatic_update_check_runs_without_daily_throttle() -> None:
window = _window()
MainWindow._automatic_update_check(window)
assert window.checks == [False]
assert window.update_check_timer.start_calls == [UPDATE_CHECK_INTERVAL_MS]
assert window.update_retry_timer.stop_calls == 1
def test_automatic_update_check_retries_while_sync_is_running() -> None:
window = _window(sync_running=True)
MainWindow._automatic_update_check(window)
assert window.checks == []
assert window.update_retry_timer.start_calls == [None]
assert UPDATE_RETRY_INTERVAL_MS == 60_000
def test_periodic_update_interval_is_thirty_minutes() -> None:
assert UPDATE_CHECK_INTERVAL_MS == 30 * 60 * 1000
def test_update_indicator_uses_text_and_accessible_description() -> None:
button = FakeButton()
window = SimpleNamespace(settings_nav_button=button)
MainWindow._set_update_indicator(window, "1.0.5")
assert button.text == "Einstellungen • UPDATE"
assert button.properties["updateAvailable"] is True
assert button.tooltip == "SyncVito 1.0.5 ist verfügbar"
assert button.accessible_description == button.tooltip
assert button.button_style.unpolished and button.button_style.polished
assert button.updated
+3
View File
@@ -162,4 +162,7 @@ def test_update_watcher_waits_before_installer(monkeypatch: pytest.MonkeyPatch,
script = base64.b64decode(encoded).decode("utf-16-le")
assert "foreach ($processId in @(111, 222))" in script
assert script.index("while (Get-Process") < script.index(str(installer))
reset_environment = "$env:PYINSTALLER_RESET_ENVIRONMENT = '1'"
assert reset_environment in script
assert script.index(reset_environment) < script.index(f"& '{installer}'")
assert "'/VERYSILENT'" in script
+1 -1
View File
@@ -1 +1 @@
1.0.1
1.0.7