5 Commits
Author SHA1 Message Date
Davito bb5bfda4fe Release SyncVito 1.0.8 2026-09-24 16:13:36 +02:00
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
17 changed files with 797 additions and 54 deletions

No files matched your search

+43
View File
@@ -1,5 +1,48 @@
# Änderungsprotokoll # Änderungsprotokoll
## 1.0.8 – 2026-09-24
### Behoben
- Der Updatewächter wartet zusätzlich zu den auslösenden PyInstaller-Prozessen auf weitere Instanzen derselben installierten Anwendung. Solange diese noch laufen, wird der Installer nicht gestartet.
- Das tatsächliche Ende des Installers und sein Exitcode werden zuverlässig erfasst. Der Wächter wartet dabei nicht auf die anschließend neu gestartete Anwendung.
- Abgebrochene Installationen und weiterhin laufende Instanzen werden protokolliert und beim nächsten Programmstart mit der tatsächlich gestarteten Version angezeigt.
- Installations- und Protokollpfade mit Leerzeichen oder Apostrophen werden beim Aufruf des Installers korrekt übergeben.
## 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 ## 1.0.2 – 2026-09-24
### Behoben ### Behoben
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "syncvito" name = "syncvito"
version = "1.0.2" version = "1.0.8"
description = "Windows-Desktopwerkzeug für lokale und NAS-Synchronisation" description = "Windows-Desktopwerkzeug für lokale und NAS-Synchronisation"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
+25 -16
View File
@@ -25,7 +25,7 @@ def derive_overall_status(
if TaskRuntimeState.ERROR in states: if TaskRuntimeState.ERROR in states:
return "error", "Fehler" return "error", "Fehler"
if TaskRuntimeState.WAITING_FOR_RELEASE in states: 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: if globally_paused:
return "paused", "Pausiert" return "paused", "Pausiert"
if TaskRuntimeState.ACTIVE not in states: if TaskRuntimeState.ACTIVE not in states:
@@ -205,25 +205,31 @@ class SyncCoordinator(QObject):
if result.blocked: if result.blocked:
paths = [blocked.path for blocked in result.blocked] paths = [blocked.path for blocked in result.blocked]
self._retry_tasks[task_id] = paths self._retry_tasks[task_id] = paths
descriptions: list[str] = [] message = (
for blocked in result.blocked: f"{result.copied} kopiert, {len(result.blocked)} Dateien gesperrt – Sicherung unvollständig. "
names = ", ".join(f"{process.name} (PID {process.pid})" for process in blocked.processes) "Andere Dateien wurden weiterverarbeitet. Automatischer Wiederholungsversuch nach Freigabe; "
descriptions.append(f"{blocked.path.name}: {names or 'sperrender Prozess nicht ermittelbar'}") "ggf. betroffene Programme schließen. Details im Protokoll."
message = "Zurückgestellt – " + "; ".join(descriptions) )
self._lock_messages[task_id] = message self._lock_messages[task_id] = message
self._emit_lock_summary() self._emit_lock_summary()
self.log_created.emit(task.name, "Blockiert", message) 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) 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: else:
self._retry_tasks.pop(task_id, None)
self._lock_messages.pop(task_id, None) self._lock_messages.pop(task_id, None)
self._emit_lock_summary() 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" summary = f"{result.copied} kopiert, {result.deleted} gelöscht, {result.conflicts} Konfliktkopien"
self.log_created.emit(task.name, "Erfolg", summary) self.log_created.emit(task.name, "Erfolg", summary)
self._set_task_state(task_id, TaskRuntimeState.ACTIVE) self._set_task_state(task_id, TaskRuntimeState.ACTIVE)
@@ -273,13 +279,12 @@ class SyncCoordinator(QObject):
if self.config.settings.paused or self.is_running: if self.config.settings.paused or self.is_running:
return return
for task_id, paths in list(self._retry_tasks.items()): 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 continue
self.log_created.emit( self.log_created.emit(
"Automatischer Versuch", "Hinweis", "Blockierte oder nicht erreichbare Datei wird erneut geprüft." "Automatischer Versuch", "Hinweis", "Blockierte oder nicht erreichbare Datei wird erneut geprüft."
) )
self.enqueue(task_id) self.enqueue(task_id)
break
def _task(self, task_id: str) -> SyncTask | None: def _task(self, task_id: str) -> SyncTask | None:
return next((task for task in self.config.tasks if task.id == task_id), 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) self.overall_status_changed.emit(status, text)
def _emit_lock_summary(self) -> None: 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.""" """Prüft eine vorhandene Datei durch exklusives Öffnen ohne Schreibzugriff."""
if not path.exists() or os.name != "nt": if not path.exists() or os.name != "nt":
return False 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), str(path),
0x80000000, 0x80000000,
0, 0,
@@ -34,8 +49,8 @@ def is_file_locked(path: Path) -> bool:
) )
invalid_handle = ctypes.c_void_p(-1).value invalid_handle = ctypes.c_void_p(-1).value
if handle == invalid_handle: if handle == invalid_handle:
return ctypes.get_last_error() in {5, 32, 33} return ctypes.get_last_error() in {32, 33}
ctypes.windll.kernel32.CloseHandle(handle) close_handle(handle)
return False return False
+93 -14
View File
@@ -3,11 +3,11 @@
from __future__ import annotations from __future__ import annotations
import sys import sys
from datetime import datetime, timedelta from datetime import datetime
from pathlib import Path from pathlib import Path
from PyQt6.QtCore import QEvent, Qt, QTimer, pyqtSignal 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 ( from PyQt6.QtWidgets import (
QAbstractItemView, QAbstractItemView,
QApplication, QApplication,
@@ -48,6 +48,38 @@ from syncvito.version import __version__
__all__ = ["MainWindow"] __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): class MainWindow(QMainWindow):
"""Zentrale SyncVito-Benutzeroberfläche.""" """Zentrale SyncVito-Benutzeroberfläche."""
@@ -59,9 +91,10 @@ class MainWindow(QMainWindow):
self.config = config self.config = config
self.config_store = config_store self.config_store = config_store
self._allow_close = False self._allow_close = False
self._notified_update_version: str | None = None
self.setWindowTitle("SyncVito") self.setWindowTitle("SyncVito")
self.setWindowIcon(app_icon()) self.setWindowIcon(app_icon())
self.setMinimumSize(980, 680) self.setMinimumSize(1040, 680)
self.resize(1220, 800) self.resize(1220, 800)
self.coordinator = SyncCoordinator(config, config_store, self) self.coordinator = SyncCoordinator(config, config_store, self)
@@ -71,7 +104,17 @@ class MainWindow(QMainWindow):
self._connect_coordinator() self._connect_coordinator()
self._refresh_tasks() self._refresh_tasks()
self._set_global_status(*self.coordinator.current_overall_status()) 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) QTimer.singleShot(3000, self._automatic_update_check)
QTimer.singleShot(1000, self._show_previous_update_error)
cleanup_attempts = 6 if "--updated" in sys.argv else 1 cleanup_attempts = 6 if "--updated" in sys.argv else 1
QTimer.singleShot(3000, lambda: self._cleanup_update_installers(cleanup_attempts)) QTimer.singleShot(3000, lambda: self._cleanup_update_installers(cleanup_attempts))
if "--updated" in sys.argv: if "--updated" in sys.argv:
@@ -159,6 +202,7 @@ class MainWindow(QMainWindow):
layout.addWidget(button) layout.addWidget(button)
self.nav_buttons.append(button) self.nav_buttons.append(button)
self.nav_buttons[0].setChecked(True) self.nav_buttons[0].setChecked(True)
self.settings_nav_button = self.nav_buttons[2]
layout.addStretch(1) layout.addStretch(1)
version = QLabel(f"Version {__version__}\nby Davito") version = QLabel(f"Version {__version__}\nby Davito")
@@ -340,7 +384,7 @@ class MainWindow(QMainWindow):
version_label = QLabel(f"Installierte Version: {__version__}") version_label = QLabel(f"Installierte Version: {__version__}")
version_label.setProperty("muted", True) version_label.setProperty("muted", True)
update_layout.addWidget(version_label) 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) self.update_check_box.setChecked(self.config.settings.check_updates_on_start)
update_layout.addWidget(self.update_check_box) update_layout.addWidget(self.update_check_box)
update_hint = QLabel( update_hint = QLabel(
@@ -366,15 +410,17 @@ class MainWindow(QMainWindow):
dock.setFixedHeight(66) dock.setFixedHeight(66)
layout = QHBoxLayout(dock) layout = QHBoxLayout(dock)
layout.setContentsMargins(28, 12, 28, 12) layout.setContentsMargins(28, 12, 28, 12)
self.progress_text = QLabel("Bereit") self.progress_text = _ElidedLabel("Bereit")
self.progress_text.setMinimumWidth(250) self.progress_text.setFixedWidth(PROGRESS_TEXT_WIDTH)
self.progress_text.setProperty("muted", True) self.progress_text.setProperty("muted", True)
self.progress_bar = QProgressBar() self.progress_bar = QProgressBar()
self.progress_bar.setFixedWidth(PROGRESS_BAR_WIDTH)
self.progress_bar.setRange(0, 100) self.progress_bar.setRange(0, 100)
self.progress_bar.setValue(0) self.progress_bar.setValue(0)
self.progress_bar.setFormat("%p %") self.progress_bar.setFormat("%p %")
layout.addWidget(self.progress_text) layout.addWidget(self.progress_text)
layout.addWidget(self.progress_bar, 1) layout.addWidget(self.progress_bar)
layout.addStretch(1)
return dock return dock
def _build_tray(self) -> None: def _build_tray(self) -> None:
@@ -511,6 +557,7 @@ class MainWindow(QMainWindow):
def _save_settings(self) -> None: def _save_settings(self) -> None:
settings = self.config.settings settings = self.config.settings
old_autostart = settings.autostart old_autostart = settings.autostart
old_update_check = settings.check_updates_on_start
settings.autostart = self.autostart_check.isChecked() settings.autostart = self.autostart_check.isChecked()
settings.font_family = self.font_combo.currentText() settings.font_family = self.font_combo.currentText()
settings.font_size = self.font_size_spin.value() settings.font_size = self.font_size_spin.value()
@@ -528,18 +575,22 @@ class MainWindow(QMainWindow):
return return
apply_theme(QApplication.instance(), settings) apply_theme(QApplication.instance(), settings)
self._append_log("Einstellungen", "Erfolg", "Darstellung und Systemstart wurden übernommen.") 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: def _automatic_update_check(self) -> None:
settings = self.config.settings 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 return
if settings.last_update_check: if self.coordinator.is_running or self.updater.busy:
try: if not self.update_retry_timer.isActive():
last_check = datetime.fromisoformat(settings.last_update_check) self.update_retry_timer.start()
except ValueError:
last_check = datetime.min
if datetime.now() - last_check < timedelta(hours=24):
return return
self.update_retry_timer.stop()
self.update_check_timer.start(UPDATE_CHECK_INTERVAL_MS)
self._check_updates(manual=False) self._check_updates(manual=False)
def _check_updates(self, manual: bool) -> None: def _check_updates(self, manual: bool) -> None:
@@ -558,12 +609,18 @@ class MainWindow(QMainWindow):
self._mark_update_checked() self._mark_update_checked()
self._update_progress(100, f"Version {info.version} ist verfügbar") self._update_progress(100, f"Version {info.version} ist verfügbar")
self._append_log("Update", "Hinweis", 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) dialog = UpdateDialog(__version__, info, getattr(sys, "frozen", False), self)
if dialog.exec() == UpdateDialog.DialogCode.Accepted: if dialog.exec() == UpdateDialog.DialogCode.Accepted:
self.updater.download(info) self.updater.download(info)
def _on_no_update(self, manual: bool) -> None: def _on_no_update(self, manual: bool) -> None:
self._mark_update_checked() self._mark_update_checked()
self._set_update_indicator(None)
self._notified_update_version = None
self._update_progress(100, "SyncVito ist aktuell") self._update_progress(100, "SyncVito ist aktuell")
self._append_log("Update", "Erfolg", "Die installierte Version ist aktuell.") self._append_log("Update", "Erfolg", "Die installierte Version ist aktuell.")
if manual: if manual:
@@ -579,6 +636,18 @@ class MainWindow(QMainWindow):
f"{message}\n\nBitte die Netzwerkverbindung prüfen und anschließend erneut versuchen.", 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: def _install_update(self, path: Path, info: UpdateInfo) -> None:
try: try:
self.updater.service.launch_installer(path, info.version) self.updater.service.launch_installer(path, info.version)
@@ -598,6 +667,16 @@ class MainWindow(QMainWindow):
self._append_log("Update", "Erfolg", message) self._append_log("Update", "Erfolg", message)
self.tray.showMessage("SyncVito aktualisiert", message, QSystemTrayIcon.MessageIcon.Information, 5000) self.tray.showMessage("SyncVito aktualisiert", message, QSystemTrayIcon.MessageIcon.Information, 5000)
def _show_previous_update_error(self) -> None:
message = self.updater.service.take_install_error()
if message:
self._append_log("Update", "Fehler", message)
QMessageBox.warning(
self,
"Update nicht abgeschlossen",
f"{message}\n\nGestartete Version: {__version__}\nProtokolle: {self.updater.service.update_directory}",
)
def _cleanup_update_installers(self, attempts_remaining: int = 1) -> None: def _cleanup_update_installers(self, attempts_remaining: int = 1) -> None:
_removed, pending = self.updater.service.cleanup_installers() _removed, pending = self.updater.service.cleanup_installers()
if pending and attempts_remaining > 1: if pending and attempts_remaining > 1:
+33 -6
View File
@@ -10,7 +10,7 @@ from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime
from pathlib import Path 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.models import ConflictMode, DeleteMode, SyncDirection, SyncTask
from syncvito.state_store import StateStore from syncvito.state_store import StateStore
@@ -117,12 +117,20 @@ class SyncEngine:
for index, operation in enumerate(operations, start=1): for index, operation in enumerate(operations, start=1):
try: try:
self._execute(operation, result) 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: 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: if progress:
detail = f"{operation.kind}: {operation.relative}" detail = f"{operation.kind}: {operation.relative}"
progress(round(index / total * 100), detail) 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: def _atomic_copy(source: Path, target: Path) -> None:
temporary = target.with_name(f".{target.name}.syncvito.tmp") temporary = target.with_name(f".{target.name}.syncvito.tmp")
try: try:
+6
View File
@@ -52,6 +52,12 @@ def _stylesheet(scale: float) -> str:
padding: {round(10 * scale)}px {round(12 * scale)}px; color: #9CACBF; }} padding: {round(10 * scale)}px {round(12 * scale)}px; color: #9CACBF; }}
QPushButton[nav="true"]:hover {{ background: #17212E; color: #E8EDF4; }} QPushButton[nav="true"]:hover {{ background: #17212E; color: #E8EDF4; }}
QPushButton[nav="true"]:checked {{ background: #1B2A3D; color: #FFFFFF; border-color: #2A568B; }} 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 {{ QLineEdit, QComboBox, QSpinBox, QListWidget, QTreeView, QTableWidget {{
background: #0E141D; border: 1px solid #304052; border-radius: {radius}px; background: #0E141D; border: 1px solid #304052; border-radius: {radius}px;
padding: {round(7 * scale)}px; color: #E8EDF4; padding: {round(7 * scale)}px; color: #E8EDF4;
+58 -10
View File
@@ -116,6 +116,7 @@ def _start_installer_after_exit(
arguments: list[str], arguments: list[str],
wait_pids: Iterable[int], wait_pids: Iterable[int],
bootstrap_log_path: Path, bootstrap_log_path: Path,
application_path: Path,
) -> None: ) -> None:
"""Startet Inno Setup erst nach dem Ende aller PyInstaller-Prozesse.""" """Startet Inno Setup erst nach dem Ende aller PyInstaller-Prozesse."""
if os.name != "nt": if os.name != "nt":
@@ -129,42 +130,71 @@ def _start_installer_after_exit(
normalized_pids = sorted({int(pid) for pid in wait_pids if int(pid) > 0}) normalized_pids = sorted({int(pid) for pid in wait_pids if int(pid) > 0})
pid_list = ", ".join(str(pid) for pid in normalized_pids) pid_list = ", ".join(str(pid) for pid in normalized_pids)
setup_arguments = ", ".join(_powershell_literal(argument) for argument in arguments) setup_arguments = _powershell_literal(subprocess.list2cmdline(arguments))
result_path = bootstrap_log_path.parent / "update-result.json"
script = f""" script = f"""
$ErrorActionPreference = 'Stop' $ErrorActionPreference = 'Stop'
$bootstrapLog = {_powershell_literal(bootstrap_log_path)} $bootstrapLog = {_powershell_literal(bootstrap_log_path)}
$applicationPath = {_powershell_literal(application_path)}
$resultPath = {_powershell_literal(result_path)}
$setupExitCode = $null
function Write-UpdateResult([bool]$success, [string]$message) {{
$result = @{{ success = $success; message = $message; exit_code = $setupExitCode }} | ConvertTo-Json
[System.IO.File]::WriteAllText($resultPath, $result, [System.Text.UTF8Encoding]::new($false))
}}
try {{ try {{
[System.IO.File]::WriteAllText( [System.IO.File]::WriteAllText(
$bootstrapLog, $bootstrapLog,
((Get-Date).ToString('o') + ' Warte auf SyncVito-Prozesse: {pid_list}' + [Environment]::NewLine) ((Get-Date).ToString('o') + ' Warte auf SyncVito-Prozesse: {pid_list}' + [Environment]::NewLine)
) )
foreach ($processId in @({pid_list})) {{
$deadline = [DateTime]::UtcNow.AddMinutes(2) $deadline = [DateTime]::UtcNow.AddMinutes(2)
while (Get-Process -Id $processId -ErrorAction SilentlyContinue) {{ $lastWaiting = ''
while ($true) {{
$waiting = @(
foreach ($processId in @({pid_list})) {{
Get-Process -Id $processId -ErrorAction SilentlyContinue
}}
Get-Process -Name ([System.IO.Path]::GetFileNameWithoutExtension($applicationPath)) -ErrorAction SilentlyContinue |
Where-Object {{ -not $_.Path -or $_.Path -eq $applicationPath }}
)
if ($waiting.Count -eq 0) {{ break }}
$waitingIds = ($waiting.Id | Sort-Object -Unique) -join ', '
if ($waitingIds -ne $lastWaiting) {{
[System.IO.File]::AppendAllText($bootstrapLog, 'Noch aktiv: ' + $waitingIds + [Environment]::NewLine)
$lastWaiting = $waitingIds
}}
if ([DateTime]::UtcNow -ge $deadline) {{ if ([DateTime]::UtcNow -ge $deadline) {{
throw "SyncVito-Prozess $processId wurde nicht innerhalb von zwei Minuten beendet." throw "SyncVito ist noch aktiv (Prozesse: $waitingIds). Bitte alle Instanzen im Infobereich ueber Beenden schliessen und das Update erneut starten."
}} }}
Start-Sleep -Milliseconds 200 Start-Sleep -Milliseconds 200
}} }}
}}
[System.IO.File]::AppendAllText( [System.IO.File]::AppendAllText(
$bootstrapLog, $bootstrapLog,
((Get-Date).ToString('o') + ' Starte verifizierten Installer.' + [Environment]::NewLine) ((Get-Date).ToString('o') + ' Starte verifizierten Installer.' + [Environment]::NewLine)
) )
$setupArguments = @({setup_arguments}) $env:PYINSTALLER_RESET_ENVIRONMENT = '1'
& {_powershell_literal(installer_path)} @setupArguments $setupArguments = {setup_arguments}
$setupExitCode = $LASTEXITCODE $setup = Start-Process -FilePath {_powershell_literal(installer_path)} -ArgumentList $setupArguments -PassThru -WindowStyle Hidden
# Nur auf Setup warten: -Wait wuerde auch den neu gestarteten Client abwarten.
$setup.WaitForExit()
$setupExitCode = $setup.ExitCode
[System.IO.File]::AppendAllText( [System.IO.File]::AppendAllText(
$bootstrapLog, $bootstrapLog,
((Get-Date).ToString('o') + ' Installer-Exitcode: ' + $setupExitCode + [Environment]::NewLine) ((Get-Date).ToString('o') + ' Installer-Exitcode: ' + $setupExitCode + [Environment]::NewLine)
) )
if ($null -eq $setupExitCode -or $setupExitCode -ne 0) {{
throw "Installation abgebrochen (Exitcode: $setupExitCode). Details stehen im Inno-Setup-Protokoll im Update-Verzeichnis."
}}
Write-UpdateResult $true ''
exit $setupExitCode exit $setupExitCode
}} }}
catch {{ catch {{
$failureMessage = $_.Exception.Message
[System.IO.File]::AppendAllText( [System.IO.File]::AppendAllText(
$bootstrapLog, $bootstrapLog,
((Get-Date).ToString('o') + ' Bootstrap-Fehler: ' + $_.Exception.Message + [Environment]::NewLine) ((Get-Date).ToString('o') + ' Bootstrap-Fehler: ' + $_.Exception.Message + [Environment]::NewLine)
) )
Write-UpdateResult $false $failureMessage
exit 1 exit 1
}} }}
""" """
@@ -296,7 +326,11 @@ class GiteaUpdateService:
raise UpdateInstallError("Updates können nur aus der installierten Windows-Version gestartet werden.") raise UpdateInstallError("Updates können nur aus der installierten Windows-Version gestartet werden.")
installer = installer_path.resolve() installer = installer_path.resolve()
expected_name = f"SyncVitoSetup-{version}.exe" expected_name = f"SyncVitoSetup-{version}.exe"
if installer.parent != self.update_directory.resolve() or installer.name != expected_name or not installer.is_file(): if (
installer.parent != self.update_directory.resolve()
or installer.name != expected_name
or not installer.is_file()
):
raise UpdateInstallError("Der geprüfte Installer wurde nicht gefunden.") raise UpdateInstallError("Der geprüfte Installer wurde nicht gefunden.")
safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", version).strip("._") or "unknown" safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", version).strip("._") or "unknown"
@@ -316,10 +350,24 @@ class GiteaUpdateService:
f"/LOG={setup_log}", f"/LOG={setup_log}",
] ]
try: try:
_start_installer_after_exit(installer, arguments, wait_pids, bootstrap_log) _start_installer_after_exit(installer, arguments, wait_pids, bootstrap_log, current_executable)
except (OSError, ValueError, UpdateError) as exc: except (OSError, ValueError, UpdateError) as exc:
raise UpdateInstallError(f"Die Installation konnte nicht gestartet werden: {exc}") from exc raise UpdateInstallError(f"Die Installation konnte nicht gestartet werden: {exc}") from exc
def take_install_error(self) -> str | None:
"""Liest das Ergebnis des externen Wächters einmalig; Protokolle bleiben erhalten."""
result_path = self.update_directory / "update-result.json"
try:
result = json.loads(result_path.read_text(encoding="utf-8"))
if not isinstance(result, dict) or not isinstance(result.get("success"), bool):
return None
result_path.unlink()
except (OSError, ValueError):
return None
if result["success"]:
return None
return str(result.get("message") or "Die Installation wurde nicht abgeschlossen.")
def cleanup_installers(self) -> tuple[int, int]: def cleanup_installers(self) -> tuple[int, int]:
"""Entfernt ausschließlich von SyncVito verwaltete Setup-Dateien.""" """Entfernt ausschließlich von SyncVito verwaltete Setup-Dateien."""
try: try:
+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: def test_waiting_lock_has_priority_over_active_task() -> None:
states = [TaskRuntimeState.ACTIVE, TaskRuntimeState.WAITING_FOR_RELEASE] 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: def test_error_has_highest_priority() -> None:
+33
View File
@@ -1,6 +1,8 @@
import os import os
from pathlib import Path from pathlib import Path
import pytest
from syncvito.models import ConflictMode, DeleteMode, SyncDirection, SyncTask from syncvito.models import ConflictMode, DeleteMode, SyncDirection, SyncTask
from syncvito.state_store import StateStore from syncvito.state_store import StateStore
from syncvito.sync_engine import SyncEngine from syncvito.sync_engine import SyncEngine
@@ -10,6 +12,37 @@ def _engine(tmp_path: Path) -> SyncEngine:
return SyncEngine(StateStore(tmp_path / "state.json")) 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: def test_one_way_copy_and_mirror_delete(tmp_path: Path) -> None:
source = tmp_path / "source" source = tmp_path / "source"
target = tmp_path / "target" 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
+161
View File
@@ -0,0 +1,161 @@
"""Führt den echten PowerShell-Wächter mit harmlosen Testprogrammen aus."""
import base64
import json
import os
import shutil
import subprocess
import time
from pathlib import Path
import pytest
from syncvito import updater
pytestmark = pytest.mark.skipif(os.name != "nt", reason="Windows-Prozesslebenszyklus")
def wait_for_file(path: Path, timeout: float = 15) -> None:
deadline = time.monotonic() + timeout
while not path.exists():
if time.monotonic() > deadline:
pytest.fail(f"Testprozess hat keine Datei geschrieben: {path}")
time.sleep(0.05)
@pytest.fixture(scope="module")
def helper_exe(tmp_path_factory: pytest.TempPathFactory) -> Path:
root = tmp_path_factory.mktemp("watcher-helper")
source = root / "helper.cs"
source.write_text(
"""
using System;
using System.IO;
using System.Threading;
class Helper {
static int Main(string[] args) {
File.WriteAllLines(args[0], args);
var deadline = DateTime.UtcNow.AddSeconds(30);
while (!File.Exists(args[1])) {
if (DateTime.UtcNow > deadline) return 99;
Thread.Sleep(50);
}
return Int32.Parse(args[2]);
}
}
""",
encoding="utf-8",
)
compiler = Path(os.environ.get("SYSTEMROOT", r"C:\Windows")) / "Microsoft.NET/Framework64/v4.0.30319/csc.exe"
if not compiler.exists():
pytest.skip("Windows-.NET-Compiler für Testprogramm nicht vorhanden")
executable = root / "SyncVitoWatcherTest.exe"
subprocess.run(
[str(compiler), "/nologo", "/target:winexe", f"/out:{executable}", str(source)],
check=True,
capture_output=True,
timeout=30,
)
return executable
@pytest.mark.parametrize("exit_code", [0, 5])
def test_watcher_waits_for_all_instances_and_real_setup_exit(
helper_exe: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, exit_code: int
) -> None:
root = tmp_path / "Ordner mit Leerzeichen und 'Apostroph'"
root.mkdir()
installer = root / "SetupTest.exe"
shutil.copy2(helper_exe, installer)
unrelated_dir = root / "anderer Installationspfad"
unrelated_dir.mkdir()
unrelated_exe = unrelated_dir / helper_exe.name
shutil.copy2(helper_exe, unrelated_exe)
children = []
releases = []
def start_client(exe: Path, name: str) -> subprocess.Popen:
ready = root / f"{name}.ready"
release = root / f"{name}.release"
releases.append(release)
process = subprocess.Popen([str(exe), str(ready), str(release), "0"])
children.append(process)
wait_for_file(ready)
return process
def launch_watcher(executable: str, arguments: list[str]) -> None:
children.append(subprocess.Popen([executable, *arguments], creationflags=subprocess.CREATE_NO_WINDOW))
try:
parent = start_client(helper_exe, "parent")
extra = start_client(helper_exe, "extra")
unrelated = start_client(unrelated_exe, "unrelated")
monkeypatch.setattr(updater, "_start_detached", launch_watcher)
output = root / "setup.arguments"
setup_release = root / "setup.release"
releases.append(setup_release)
log = root / "bootstrap.log"
arguments = [str(output), str(setup_release), str(exit_code), f"/DIR={root}", f"/LOG={root / 'setup.log'}"]
updater._start_installer_after_exit(installer, arguments, [parent.pid], log, helper_exe)
watcher = children[-1]
wait_for_file(log)
assert not output.exists()
releases[0].touch()
parent.wait(timeout=10)
time.sleep(0.5)
assert not output.exists(), "Weitere Instanz wurde nicht abgewartet"
releases[1].touch()
extra.wait(timeout=10)
wait_for_file(output)
assert unrelated.poll() is None, "Anderer Installationspfad darf weiterlaufen"
assert output.read_text(encoding="utf-8-sig").splitlines() == arguments
assert watcher.poll() is None, "Wächter muss auf Setup warten"
assert not (root / "update-result.json").exists()
setup_release.touch()
assert watcher.wait(timeout=15) == (0 if exit_code == 0 else 1)
receipt = json.loads((root / "update-result.json").read_text(encoding="utf-8"))
assert receipt["success"] is (exit_code == 0)
assert receipt["exit_code"] == exit_code
assert f"Installer-Exitcode: {exit_code}" in log.read_text(encoding="utf-8-sig")
finally:
for release in releases:
release.touch()
for child in children:
try:
child.wait(timeout=10)
except subprocess.TimeoutExpired:
child.kill()
child.wait(timeout=5)
def test_watcher_reports_timeout_without_starting_installer(
helper_exe: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
ready = tmp_path / "client.ready"
release = tmp_path / "client.release"
log = tmp_path / "bootstrap.log"
client = subprocess.Popen([str(helper_exe), str(ready), str(release), "0"])
watchers = []
def launch_with_short_deadline(executable: str, arguments: list[str]) -> None:
script = base64.b64decode(arguments[-1]).decode("utf-16-le")
script = script.replace("AddMinutes(2)", "AddMilliseconds(400)")
arguments[-1] = base64.b64encode(script.encode("utf-16-le")).decode("ascii")
watchers.append(subprocess.Popen([executable, *arguments], creationflags=subprocess.CREATE_NO_WINDOW))
try:
wait_for_file(ready)
monkeypatch.setattr(updater, "_start_detached", launch_with_short_deadline)
updater._start_installer_after_exit(tmp_path / "not-started.exe", ["/VERYSILENT"], [], log, helper_exe)
assert watchers[0].wait(timeout=15) == 1
receipt = json.loads((tmp_path / "update-result.json").read_text(encoding="utf-8"))
assert not receipt["success"]
assert str(client.pid) in receipt["message"]
assert "Beenden" in receipt["message"]
assert "Starte verifizierten Installer" not in log.read_text(encoding="utf-8-sig")
assert client.poll() is None
finally:
release.touch()
client.wait(timeout=10)
for watcher in watchers:
watcher.wait(timeout=10)
+28 -2
View File
@@ -1,5 +1,6 @@
import base64 import base64
import hashlib import hashlib
import json
import os import os
from pathlib import Path from pathlib import Path
@@ -154,12 +155,37 @@ def test_update_watcher_waits_before_installer(monkeypatch: pytest.MonkeyPatch,
powershell = Path(os.environ.get("SYSTEMROOT", r"C:\Windows")) / "System32/WindowsPowerShell/v1.0/powershell.exe" powershell = Path(os.environ.get("SYSTEMROOT", r"C:\Windows")) / "System32/WindowsPowerShell/v1.0/powershell.exe"
monkeypatch.setattr(Path, "is_file", lambda self: self == powershell) monkeypatch.setattr(Path, "is_file", lambda self: self == powershell)
updater._start_installer_after_exit(installer, ["/VERYSILENT"], [222, 111], bootstrap_log) updater._start_installer_after_exit(
installer, ["/VERYSILENT"], [222, 111], bootstrap_log, tmp_path / "SyncVito.exe"
)
arguments = captured["arguments"] arguments = captured["arguments"]
assert isinstance(arguments, list) assert isinstance(arguments, list)
encoded = arguments[arguments.index("-EncodedCommand") + 1] encoded = arguments[arguments.index("-EncodedCommand") + 1]
script = base64.b64decode(encoded).decode("utf-16-le") script = base64.b64decode(encoded).decode("utf-16-le")
assert "foreach ($processId in @(111, 222))" in script assert "foreach ($processId in @(111, 222))" in script
assert script.index("while (Get-Process") < script.index(str(installer)) assert script.index("while ($true)") < script.index(str(installer))
reset_environment = "$env:PYINSTALLER_RESET_ENVIRONMENT = '1'"
assert reset_environment in script
assert script.index(reset_environment) < script.index("$setup = Start-Process")
assert "'/VERYSILENT'" in script assert "'/VERYSILENT'" in script
@pytest.mark.parametrize("success", [False, True])
def test_installer_result_is_consumed_but_logs_are_kept(tmp_path: Path, success: bool) -> None:
service = GiteaUpdateService("1.0.7", update_directory=tmp_path)
receipt = tmp_path / "update-result.json"
receipt.write_text(json.dumps({"success": success, "message": "Installation abgebrochen"}), encoding="utf-8")
log = tmp_path / "SyncVitoSetup-1.0.8.log"
log.write_text("Diagnose")
assert service.take_install_error() == (None if success else "Installation abgebrochen")
assert service.take_install_error() is None
assert log.read_text() == "Diagnose"
def test_incomplete_installer_result_can_be_read_later(tmp_path: Path) -> None:
service = GiteaUpdateService("1.0.7", update_directory=tmp_path)
receipt = tmp_path / "update-result.json"
receipt.write_text('{"success":', encoding="utf-8")
assert service.take_install_error() is None
assert receipt.exists()
+1 -1
View File
@@ -1 +1 @@
1.0.2 1.0.8