7 Commits
Author SHA1 Message Date
Davito 4efc1efa18 Release SyncVito 1.0.9 2026-09-24 16:46:22 +02:00
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
Davito 8eb8cc1d0d Release SyncVito 1.0.2 2026-09-24 12:50:54 +02:00
17 changed files with 940 additions and 63 deletions

No files matched your search

+59
View File
@@ -1,5 +1,64 @@
# Änderungsprotokoll
## 1.0.9 – 2026-09-24
### Behoben
- Beim Start einer Synchronisation wird die Fortschrittsanzeige sofort zurückgesetzt. Bereits beim Einlesen von Quelle und Ziel erscheinen der aktuelle Ordner und die Anzahl gefundener Dateien; eine vorherige Updateanzeige bleibt nicht mehr unverändert stehen.
- Der Gesamtstatus kennzeichnet einen laufenden Vorgang eindeutig mit „Synchronisierung läuft“.
- Windows-Verzeichnisverknüpfungen (Junctions) werden beim Einlesen übersprungen, damit insbesondere AppData nicht mehrfach oder zyklisch durchsucht wird.
- Bei einer Unterauswahl werden nicht ausgewählte Verzeichniszweige nicht mehr vollständig durchsucht.
## 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
### 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.9"
description = "Windows-Desktopwerkzeug für lokale und NAS-Synchronisation"
readme = "README.md"
requires-python = ">=3.11"
+27 -17
View File
@@ -25,12 +25,12 @@ 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", "Synchronisierung läuft" if running else "Unvollständig – Dateien gesperrt"
if globally_paused:
return "paused", "Pausiert"
if TaskRuntimeState.ACTIVE not in states:
return "paused", "Inaktiv"
return "active", "Synchronisiert" if running else "Bereit"
return "active", "Synchronisierung läuft" if running else "Bereit"
class _WorkerSignals(QObject):
@@ -186,6 +186,7 @@ class SyncCoordinator(QObject):
self._start_next()
return
self._running_task_id = task_id
self.progress_changed.emit(0, f"{task.name}: Synchronisation wird gestartet …")
self.idle_changed.emit(False)
self._emit_overall_status()
self.log_created.emit(task.name, "Start", "Synchronisation gestartet.")
@@ -205,25 +206,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 +280,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 +302,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
+95 -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,7 +104,17 @@ 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)
QTimer.singleShot(1000, self._show_previous_update_error)
cleanup_attempts = 6 if "--updated" in sys.argv else 1
QTimer.singleShot(3000, lambda: self._cleanup_update_installers(cleanup_attempts))
if "--updated" in sys.argv:
@@ -159,9 +202,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 +384,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 +410,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 +557,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 +575,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 +609,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 +636,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)
@@ -598,6 +667,16 @@ class MainWindow(QMainWindow):
self._append_log("Update", "Erfolg", message)
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:
_removed, pending = self.updater.service.cleanup_installers()
if pending and attempts_remaining > 1:
+68 -10
View File
@@ -4,13 +4,15 @@ from __future__ import annotations
import os
import shutil
import stat
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from time import monotonic
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
@@ -89,9 +91,13 @@ class SyncEngine:
task.validate()
source_root = Path(task.source)
target_root = Path(task.target)
if progress:
progress(0, f"{task.name}: Ordner werden geprüft …")
self._validate_roots(source_root, target_root)
source_files, source_dirs = self._inventory(source_root, task.selected_paths)
target_files, target_dirs = self._inventory(target_root, task.selected_paths)
source_files, source_dirs = self._inventory(source_root, task.selected_paths, progress, "Quelle")
target_files, target_dirs = self._inventory(target_root, task.selected_paths, progress, "Ziel")
if progress:
progress(0, f"{task.name}: Änderungen werden verglichen …")
previous = self.state_store.get(task.id)
if task.direction is SyncDirection.ONE_WAY:
@@ -117,12 +123,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)
@@ -149,15 +163,31 @@ class SyncEngine:
raise UnavailablePathError(f"Ziel ist kein Ordner: {target_root}")
@staticmethod
def _inventory(root: Path, selected_paths: list[str]) -> tuple[dict[str, FileRecord], set[str]]:
def _inventory(
root: Path,
selected_paths: list[str],
progress: ProgressCallback | None = None,
label: str = "Ordner",
) -> tuple[dict[str, FileRecord], set[str]]:
files: dict[str, FileRecord] = {}
directories: set[str] = set()
selections = tuple(Path(item).parts for item in selected_paths)
last_report = monotonic()
if progress:
progress(0, f"{label} wird eingelesen: {root}")
try:
for current, dir_names, file_names in os.walk(root):
current_path = Path(current)
relative_dir = current_path.relative_to(root)
dir_names[:] = [name for name in dir_names if not (current_path / name).is_symlink()]
dir_names[:] = [
name
for name in dir_names
if _is_selected((relative_dir / name).parts, selections)
and not _is_directory_link(current_path / name)
]
if progress and monotonic() - last_report >= 0.25:
progress(0, f"{label}: {len(files)} Dateien gefunden – {current_path}")
last_report = monotonic()
for directory_name in dir_names:
relative = relative_dir / directory_name
if _is_selected(relative.parts, selections):
@@ -174,6 +204,9 @@ class SyncEngine:
except OSError:
continue
files[str(relative)] = FileRecord(path, file_stat.st_size, file_stat.st_mtime_ns)
if progress and monotonic() - last_report >= 0.25:
progress(0, f"{label}: {len(files)} Dateien gefunden – {current_path}")
last_report = monotonic()
except OSError as exc:
raise UnavailablePathError(f"Ordner kann nicht gelesen werden: {root}") from exc
return files, directories
@@ -295,6 +328,31 @@ class SyncEngine:
}
def _is_directory_link(path: Path) -> bool:
"""Erkennt auch Windows-Junctions unter Python 3.11 ohne Path.is_junction."""
metadata = path.lstat()
return stat.S_ISLNK(metadata.st_mode) or getattr(metadata, "st_reparse_tag", 0) == stat.IO_REPARSE_TAG_MOUNT_POINT
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;
+60 -12
View File
@@ -116,6 +116,7 @@ def _start_installer_after_exit(
arguments: list[str],
wait_pids: Iterable[int],
bootstrap_log_path: Path,
application_path: Path,
) -> None:
"""Startet Inno Setup erst nach dem Ende aller PyInstaller-Prozesse."""
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})
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"""
$ErrorActionPreference = 'Stop'
$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 {{
[System.IO.File]::WriteAllText(
$bootstrapLog,
((Get-Date).ToString('o') + ' Warte auf SyncVito-Prozesse: {pid_list}' + [Environment]::NewLine)
)
foreach ($processId in @({pid_list})) {{
$deadline = [DateTime]::UtcNow.AddMinutes(2)
while (Get-Process -Id $processId -ErrorAction SilentlyContinue) {{
if ([DateTime]::UtcNow -ge $deadline) {{
throw "SyncVito-Prozess $processId wurde nicht innerhalb von zwei Minuten beendet."
$deadline = [DateTime]::UtcNow.AddMinutes(2)
$lastWaiting = ''
while ($true) {{
$waiting = @(
foreach ($processId in @({pid_list})) {{
Get-Process -Id $processId -ErrorAction SilentlyContinue
}}
Start-Sleep -Milliseconds 200
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) {{
throw "SyncVito ist noch aktiv (Prozesse: $waitingIds). Bitte alle Instanzen im Infobereich ueber Beenden schliessen und das Update erneut starten."
}}
Start-Sleep -Milliseconds 200
}}
[System.IO.File]::AppendAllText(
$bootstrapLog,
((Get-Date).ToString('o') + ' Starte verifizierten Installer.' + [Environment]::NewLine)
)
$setupArguments = @({setup_arguments})
& {_powershell_literal(installer_path)} @setupArguments
$setupExitCode = $LASTEXITCODE
$env:PYINSTALLER_RESET_ENVIRONMENT = '1'
$setupArguments = {setup_arguments}
$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(
$bootstrapLog,
((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
}}
catch {{
$failureMessage = $_.Exception.Message
[System.IO.File]::AppendAllText(
$bootstrapLog,
((Get-Date).ToString('o') + ' Bootstrap-Fehler: ' + $_.Exception.Message + [Environment]::NewLine)
)
Write-UpdateResult $false $failureMessage
exit 1
}}
"""
@@ -296,7 +326,11 @@ class GiteaUpdateService:
raise UpdateInstallError("Updates können nur aus der installierten Windows-Version gestartet werden.")
installer = installer_path.resolve()
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.")
safe_version = re.sub(r"[^0-9A-Za-z._-]+", "_", version).strip("._") or "unknown"
@@ -316,10 +350,24 @@ class GiteaUpdateService:
f"/LOG={setup_log}",
]
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:
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]:
"""Entfernt ausschließlich von SyncVito verwaltete Setup-Dateien."""
try:
+95
View File
@@ -0,0 +1,95 @@
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"]
def test_start_resets_stale_progress_before_worker_starts(
coordinator: SyncCoordinator, monkeypatch: pytest.MonkeyPatch
) -> None:
from types import SimpleNamespace
messages = []
workers = []
coordinator.progress_changed.connect(lambda *entry: messages.append(entry))
def start(worker) -> None:
assert messages and messages[-1][0] == 0
assert "AppData" in messages[-1][1]
workers.append(worker)
monkeypatch.setattr(coordinator, "pool", SimpleNamespace(start=start))
coordinator.enqueue(coordinator.config.tasks[0].id)
assert len(workers) == 1
assert coordinator.current_overall_status() == ("active", "Synchronisierung läuft")
+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",
"Synchronisierung läuft",
)
def test_error_has_highest_priority() -> None:
+99
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,103 @@ 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"
+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 hashlib
import json
import os
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"
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"]
assert isinstance(arguments, list)
encoded = arguments[arguments.index("-EncodedCommand") + 1]
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))
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
@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.1
1.0.9