diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 31bc1ca..3a25463 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,7 +44,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -r requirements.txt pytest pytest-xdist ruff Pillow + pip install -r requirements.txt pytest pytest-qt pytest-xdist ruff Pillow - name: Compile sources run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 651afcf..2aafd4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ Format basiert auf [Keep a Changelog](https://keepachangelog.com/de/1.1.0/). ## [Unreleased] +### Sicheres Löschen +- Die Löschbestätigung lässt sich über das Menü **Bearbeiten** oder die Einstellungen ein- und ausschalten. +- Wenn die Bestätigung ausgeschaltet ist, verschiebt ExplorerPro ausgewählte Elemente in den System-Papierkorb. Schlägt das Verschieben fehl, bleibt das Element erhalten und ExplorerPro zeigt den Fehler an. +- Die Windows-Fehler-5-Behandlung entfernt bei Bedarf das Read-only-Attribut regulärer Einzel-Link-Dateien vor einem einzelnen Löschversuch und stellt es bei Fehlschlag wieder her. +- Die rekursive Löschbehandlung verwendet ab Python 3.12 den `onexc`-Callback und bei älteren unterstützten Python-Versionen `onerror`. + ### Datei-/Ordner-Eigenschaften, Pfad-Kopier-Suite & Terminal-Integration (2026-10-03, TW-EP-11) - **Datei- & Ordner-Eigenschaften Dialog (`src/gui/properties_dialog.py`)**: - Neuer standardkonformer Eigenschafts-Dialog mit Tabs für "Allgemein" und "Prüfsummen" (bzw. "Inhalt" für Textdateien). @@ -41,6 +47,21 @@ Format basiert auf [Keep a Changelog](https://keepachangelog.com/de/1.1.0/). - Dialog-Lebenszyklus: Verhindert das automatische Schließen des modalen Dialogs bei `_do_rename`, sodass der Rollback-Button für den Benutzer erreichbar und nutzbar bleibt; `Accepted`-Status wird beim Schließen nach erfolgreicher Umbenennung an den Browser übermittelt. - Eingabepfad-Deduplizierung unter Erhalt der Reihenfolge im Konstruktor. - `tests/test_bugsweep_batch_rename_resilience_20261001.py`: 8 neue hermetische Regressionstests (100% grün). +### Prüfsummen-Dialog (2026-10-01) + +- Das Schließen fordert den Abbruch an und hält den Dialog bis zum bestätigten Thread-Ende offen. Die Oberfläche bleibt dabei bedienbar; die Freigabe eines noch laufenden nativen Workers wird verhindert. +- Automatische zyklische Python-Bereinigung wird während des GUI-Betriebs im Anwendungsthread ausgeführt, damit Hintergrundthreads keine Qt-Objekte zerstören. + +### Laufwerksbelegung / Drive capacity (2026-09-30) + +- Die Ordner-Seitenleiste zeigt für jedes Laufwerk einen Belegungsbalken, den belegten Anteil in Prozent und freien, belegten sowie gesamten Speicher in binären Größen (GiB/TiB). +- Kapazitäten werden unabhängig im Hintergrund abgefragt, beim Öffnen und über „Laufwerksbelegung aktualisieren“ erneuert. Nicht verfügbare Laufwerke erhalten einen erklärenden Text statt falscher Nullwerte. +- Text und Balken sind auch in schmalen Seitenleisten lesbar; Pfadnavigation bleibt erhalten. Alle neuen Texte sind in DE/EN/ES/ZH/JA/RU vorhanden. +- Hintergrundabfragen liefern reine Python-Daten; ein Timer des Panels übernimmt fertige Ergebnisse im GUI-Thread. Dateisystem- und Sortiermodell gehören jetzt ihrem Dateibrowser und werden mit ihm gelöscht. +- Die Löschprüfungen bereinigen ausschließlich ihre eigenen Qt-Objekte; die Navigation wartet mit einem begrenzten Qt-Ereignislauf auf tatsächlich geladene Dateizeilen. +- Zyklische Python-Objekte werden während des GUI-Betriebs im Anwendungsthread bereinigt. Hintergrundabfragen werden vor dem Beenden abgewartet; Prüfsummen-Threads lösen keine automatische Löschung von Qt-GUI-Objekten mehr aus. +- Laufwerksabfragen laufen in eigenen Hilfsprozessen mit zehn Sekunden Zeitlimit. Hängende Datenträger werden als nicht verfügbar angezeigt; beim Beenden werden laufende Hilfsprozesse beendet und abgeholt. +- Source feature only: no EXE/MSIX release or Store submission performed. ### Repository-Lebenszyklus-Härtung & CI/CD-Parität (2026-09-30, Pfad A) - **CI/CD Lifecycle Workflows & Label-Governance**: diff --git a/README.md b/README.md index 339f5bb..97ce8aa 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,8 @@ ## 1. Features & Core Capabilities +The source version includes per-drive capacity bars in the folder sidebar, with used percentage and free/used/total space in GiB/TiB. Capacity requests run in the background; use **Refresh drive usage** to update them. Unavailable devices show an explanatory message. This addition is not yet part of a newly packaged release. + Standard operating system file managers are built for casual browsing and lack the heavy-lifting tools developers, researchers, and power users require daily. ExplorerPro addresses this gap by packaging pro-grade productivity utilities into a cohesive, responsive desktop interface with zero telemetry and 100% Local-First data isolation: - **Unified Multi-Tab Experience:** Browse multiple directories concurrently with tab pinning, breadcrumb navigation, drag-and-drop, and intelligent context menus. @@ -568,7 +570,7 @@ ExplorerPro provides comprehensive keyboard control designed for high-efficiency | Ctrl + Tab | Browser | Cycle through open directory tabs | | Ctrl + F | Global | Focus search bar and trigger FTS5 search | | F2 | Browser | Rename selected file or folder | -| Delete | Browser | Delete selected items (with confirmation dialog) | +| Delete | Browser | Delete selected items (with confirmation; without confirmation they go to the system trash) | | Ctrl + C | Browser | Copy selected files/folders to clipboard | | Ctrl + V | Browser | Paste files from clipboard (with collision-free auto-suffix) | | Ctrl + Shift + N | Browser | Create a new folder in current directory | diff --git a/README_de.md b/README_de.md index dc9d3c6..ce9d3e0 100644 --- a/README_de.md +++ b/README_de.md @@ -2,6 +2,8 @@ # ExplorerPro Suite +Der aktuelle Quellcode zeigt in der Ordner-Seitenleiste die Speicherbelegung je Laufwerk: Balken, Prozentanteil und freier/belegter/gesamter Speicher in GiB/TiB. Die Abfrage läuft im Hintergrund; **Laufwerksbelegung aktualisieren** erneuert die Werte. Nicht verfügbare Geräte werden verständlich gekennzeichnet. Diese Ergänzung ist noch nicht Bestandteil eines neu gebauten Releases. + [English](README.md) | **[Deutsch](README_de.md)** | [Maschinenlesbarer Kontext (llms.txt)](llms.txt) [![CI](https://github.com/file-bricks/ExplorerPro/actions/workflows/ci.yml/badge.svg)](https://github.com/file-bricks/ExplorerPro/actions/workflows/ci.yml) diff --git a/locales/translations.json b/locales/translations.json index ff39cfe..5f3a860 100644 --- a/locales/translations.json +++ b/locales/translations.json @@ -2846,5 +2846,69 @@ "zh": "打开选定的文件", "ja": "選択したファイルを開く", "ru": "Открывает выбранный файл" + }, + "Speicherbelegung": { + "de": "Speicherbelegung", + "en": "Storage usage", + "es": "Uso del almacenamiento", + "zh": "存储使用情况", + "ja": "ストレージ使用量", + "ru": "Использование хранилища" + }, + "Speicherbelegung wird ermittelt …": { + "de": "Speicherbelegung wird ermittelt …", + "en": "Checking storage usage …", + "es": "Calculando el uso del almacenamiento …", + "zh": "正在查询存储使用情况 …", + "ja": "ストレージ使用量を確認中 …", + "ru": "Проверка использования хранилища …" + }, + "Speicherbelegung nicht verfügbar": { + "de": "Speicherbelegung nicht verfügbar", + "en": "Storage usage unavailable", + "es": "Uso del almacenamiento no disponible", + "zh": "存储使用情况不可用", + "ja": "ストレージ使用量を取得できません", + "ru": "Данные об использовании хранилища недоступны" + }, + "{free} frei · {used} belegt · {total} gesamt": { + "de": "{free} frei · {used} belegt · {total} gesamt", + "en": "{free} free · {used} used · {total} total", + "es": "{free} libre · {used} usado · {total} total", + "zh": "{free} 可用 · {used} 已用 · {total} 总计", + "ja": "{free} 空き · {used} 使用中 · {total} 合計", + "ru": "{free} свободно · {used} занято · {total} всего" + }, + "{percent}% belegt": { + "de": "{percent}% belegt", + "en": "{percent}% used", + "es": "{percent}% usado", + "zh": "已用 {percent}%", + "ja": "{percent}% 使用中", + "ru": "{percent}% занято" + }, + "Laufwerksbelegung aktualisieren": { + "de": "Laufwerksbelegung aktualisieren", + "en": "Refresh drive usage", + "es": "Actualizar el uso de las unidades", + "zh": "刷新驱动器使用情况", + "ja": "ドライブ使用量を更新", + "ru": "Обновить использование дисков" + }, + "Wenn deaktiviert, werden Elemente ohne Rückfrage in den Papierkorb verschoben.": { + "de": "Wenn deaktiviert, werden Elemente ohne Rückfrage in den Papierkorb verschoben.", + "en": "When disabled, items are moved to the Recycle Bin without asking.", + "es": "Si se desactiva, los elementos se mueven a la papelera sin pedir confirmación.", + "zh": "禁用后，项目会直接移至回收站，不再询问。", + "ja": "無効にすると、確認なしで項目をごみ箱に移動します。", + "ru": "Если отключить, элементы будут перемещаться в корзину без запроса подтверждения." + }, + "Das Element konnte nicht in den Papierkorb verschoben werden.": { + "de": "Das Element konnte nicht in den Papierkorb verschoben werden.", + "en": "The item could not be moved to the Recycle Bin.", + "es": "No se pudo mover el elemento a la papelera.", + "zh": "无法将项目移至回收站。", + "ja": "項目をごみ箱に移動できませんでした。", + "ru": "Не удалось переместить элемент в корзину." } -} \ No newline at end of file +} diff --git a/pyproject.toml b/pyproject.toml index 99bb74d..9bdf87a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,11 +81,13 @@ Notice = "https://github.com/file-bricks/ExplorerPro/blob/master/NOTICE" [project.optional-dependencies] dev = [ "pytest>=9.1.1", + "pytest-qt>=4.5.0", "ruff>=0.9.0", "altgraph>=0.17.4", ] test = [ "pytest>=9.1.1", + "pytest-qt>=4.5.0", "Pillow>=10.0", ] build = [ diff --git a/src/core/delete_service.py b/src/core/delete_service.py new file mode 100644 index 0000000..3bc7720 --- /dev/null +++ b/src/core/delete_service.py @@ -0,0 +1,87 @@ +"""Permanent deletion with a Windows read-only retry for single-link files.""" + +import os +import shutil +import stat +import sys + +from translator import t + + +def _retry_readonly_file(operation, path, error): + """Retry a failed file deletion once; never relax directory permissions.""" + if ( + sys.platform != "win32" + or getattr(error, "winerror", None) != 5 + or operation not in (os.remove, os.unlink) + ): + raise error + try: + original = os.lstat(path) + except OSError: + raise error + if not ( + stat.S_ISREG(original.st_mode) + and original.st_file_attributes & stat.FILE_ATTRIBUTE_READONLY + and original.st_nlink == 1 + ): + # Attributes belong to the file, not its name. Clearing read-only on + # a hardlink would also modify aliases outside the selected tree. + raise error + + os.chmod(path, original.st_mode | stat.S_IWRITE) + try: + operation(path) + except OSError: + # An ACL or open handle can still prevent deletion. Restore the user's + # read-only bit on the same file, and propagate the retry error. + try: + current = os.lstat(path) + if ( + stat.S_ISREG(current.st_mode) + and (current.st_dev, current.st_ino) == (original.st_dev, original.st_ino) + ): + os.chmod(path, original.st_mode) + except OSError: + pass + raise + + +def _rmtree_error(operation, path, exc_info): + _retry_readonly_file(operation, path, exc_info[1]) + + +def _rmtree_exception(operation, path, error): + _retry_readonly_file(operation, path, error) + + +def _remove_tree(path): + if sys.version_info >= (3, 12): + shutil.rmtree(path, onexc=_rmtree_exception) + else: + shutil.rmtree(path, onerror=_rmtree_error) + + +def delete_path(path): + """Delete one selected entry, without traversing symlink/junction targets.""" + entry = os.lstat(path) + if stat.S_ISLNK(entry.st_mode): + os.unlink(path) + elif getattr(entry, "st_reparse_tag", None) == getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", 0xA0000003): + os.rmdir(path) + elif stat.S_ISDIR(entry.st_mode): + # Python 3.8+ also avoids descending into nested Windows junctions. + _remove_tree(path) + else: + try: + os.remove(path) + except OSError as error: + _retry_readonly_file(os.remove, path, error) + + +def move_to_trash(path): + """Move one selected item to the operating system trash/recycle bin.""" + from PySide6.QtCore import QFile + + if not QFile.moveToTrash(path): + raise OSError(t("Das Element konnte nicht in den Papierkorb verschoben werden.")) diff --git a/src/core/drive_usage.py b/src/core/drive_usage.py new file mode 100644 index 0000000..69d8dd0 --- /dev/null +++ b/src/core/drive_usage.py @@ -0,0 +1,138 @@ +"""Read capacity without scanning files or changing the filesystem.""" + +from dataclasses import dataclass +import os +import shutil +import json +from pathlib import Path +import subprocess +import sys +import threading +import tempfile + +QUERY_TIMEOUT_SECONDS = 10 +_query_lock = threading.Lock() +_query_processes = set() +_queries_stopped = False + + +def start_drive_queries(): + global _queries_stopped + with _query_lock: + _queries_stopped = False + + +def stop_drive_queries(): + """Stop only our capacity helpers, including reads stuck inside the OS.""" + global _queries_stopped + with _query_lock: + _queries_stopped = True + for process in tuple(_query_processes): + if process.poll() is None: + process.kill() + + +def _query_command(path, result_path): + if getattr(sys, 'frozen', False): + return [sys.executable, '--drive-capacity-query', path, result_path] + return [sys.executable, str(Path(__file__).resolve()), path, result_path] + + +def read_drive_usage_bounded(path): + """Isolate disk_usage so stalled devices cannot retain pool slots forever.""" + with tempfile.TemporaryDirectory(prefix='explorerpro-capacity-') as directory: + return _read_drive_usage_process(path, Path(directory) / 'result.json') + + +def _read_drive_usage_process(path, result_path): + with _query_lock: + if _queries_stopped: + raise OSError('Capacity queries stopped') + process = subprocess.Popen( + _query_command(path, str(result_path)), + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0, + ) + _query_processes.add(process) + try: + try: + process.wait(timeout=QUERY_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + raise OSError('Capacity query timed out') from None + if process.returncode: + raise OSError('Capacity query unavailable') + try: + values = json.loads(result_path.read_text(encoding='utf-8')) + return DriveUsage(**values) + except (OSError, ValueError, TypeError): + raise OSError('Invalid capacity query response') from None + finally: + with _query_lock: + _query_processes.discard(process) + + +def capacity_query_main(path, result_path): + """Read-only measurement; write its result to the parent's private temp file.""" + try: + usage = read_drive_usage(path) + except (OSError, ValueError): + return 1 + # Windowed frozen executables may have no stdout, even with redirected pipes. + with open(result_path, 'x', encoding='utf-8') as result: + json.dump({'total': usage.total, 'used': usage.used, 'free': usage.free}, result) + return 0 + + +@dataclass(frozen=True) +class DriveUsage: + total: int + used: int + free: int + + def __post_init__(self): + if self.total <= 0 or not 0 <= self.used <= self.total or not 0 <= self.free <= self.total: + raise ValueError("Invalid drive capacity") + + @property + def used_percent(self) -> float: + return self.used / self.total * 100 + + +def read_drive_usage(path: str) -> DriveUsage: + if os.name == "nt": + # Empty removable drives must return an error rather than opening an + # operating-system critical-error dialog in this background thread. + import ctypes + from ctypes import wintypes + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + set_mode = kernel.SetThreadErrorMode + set_mode.argtypes = (wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)) + set_mode.restype = wintypes.BOOL + get_mode = kernel.GetThreadErrorMode + get_mode.argtypes = () + get_mode.restype = wintypes.DWORD + previous = wintypes.DWORD() + if not set_mode(get_mode() | 0x0001, ctypes.byref(previous)): # SEM_FAILCRITICALERRORS + raise ctypes.WinError(ctypes.get_last_error()) + try: + usage = shutil.disk_usage(path) + finally: + set_mode(previous.value, None) + else: + usage = shutil.disk_usage(path) + return DriveUsage(usage.total, usage.used, usage.free) + + +def format_capacity(size: int) -> str: + """Use explicit binary units, retaining useful precision for small drives.""" + value = float(size) + for unit in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"): + if value < 1024 or unit == "PiB": + return f"{int(value)} B" if unit == "B" else f"{value:.1f} {unit}" + value /= 1024 + + +if __name__ == '__main__': + sys.exit(capacity_query_main(*sys.argv[1:]) if len(sys.argv) == 3 else 2) diff --git a/src/core/gui_gc.py b/src/core/gui_gc.py new file mode 100644 index 0000000..fd0eebf --- /dev/null +++ b/src/core/gui_gc.py @@ -0,0 +1,77 @@ +"""Keep cyclic Python garbage collection on the Qt application thread. + +PySide wrappers can be part of Python reference cycles. Automatic collection +may run in whichever thread happens to allocate, including a hashing worker; +destroying a GUI QObject there is unsafe. Reference counting is unaffected. +""" + +import gc + +from PySide6.QtCore import QObject, QThread, QTimer, Slot + + +class GuiGarbageCollector(QObject): + def __init__(self, application): + if QThread.currentThread() != application.thread(): + raise RuntimeError("Install the collector on the application thread") + super().__init__(application) + self._previously_enabled = gc.isenabled() + self._closed = False + self._ticks = 0 + self.timer = QTimer(self) + self.timer.setInterval(1000) + self.timer.timeout.connect(self.collect_if_needed) + if self._previously_enabled: + self.timer.start() + gc.disable() + + def _assert_gui_thread(self): + if QThread.currentThread() != self.thread(): + raise RuntimeError("Collect Qt reference cycles on the application thread") + + @Slot() + def collect_if_needed(self): + self._assert_gui_thread() + if self._closed: + return + self._ticks += 1 + counts = gc.get_count() + thresholds = gc.get_threshold() + # Periodically revisit older cycles even if allocation has subsided. + if self._ticks >= 30 or counts[2] >= thresholds[2]: + self.collect() + elif counts[1] >= thresholds[1]: + gc.collect(1) + elif counts[0] >= thresholds[0]: + gc.collect(0) + + def collect(self): + self._assert_gui_thread() + if not self._closed: + self._ticks = 0 + return gc.collect() + return 0 + + def close(self): + """Restore GC policy after application workers have stopped.""" + self._assert_gui_thread() + if self._closed: + return + try: + self.timer.stop() + self.collect() + finally: + self._closed = True + if self._previously_enabled: + gc.enable() + else: + gc.disable() + + +def install_gui_gc(application): + """Install once; the application retains the Python wrapper and Qt owner.""" + existing = getattr(application, "_gui_garbage_collector", None) + if existing is None or existing._closed: + existing = GuiGarbageCollector(application) + application._gui_garbage_collector = existing + return existing diff --git a/src/core/platform_utils.py b/src/core/platform_utils.py index a19cee8..54ade03 100644 --- a/src/core/platform_utils.py +++ b/src/core/platform_utils.py @@ -31,10 +31,12 @@ def get_terminal_command(directory: str) -> list[str]: """Return the platform-native terminal launcher command for a directory.""" if sys.platform.startswith("win"): if shutil.which("wt"): - return ["wt", "-d", directory] + # Resolve the target through Popen(cwd=...), not command text. + # Windows Terminal treats semicolons as command separators. + return ["wt", "-d", "."] if shutil.which("powershell"): - return ["powershell", "-NoExit", "-Command", f"Set-Location -LiteralPath '{directory}'"] - return ["cmd", "/K", f"cd /d {directory}"] + return ["powershell", "-NoExit"] + return ["cmd", "/K"] if sys.platform == "darwin": return ["open", "-a", "Terminal", directory] # Linux / BSD diff --git a/src/gui/browser/file_browser.py b/src/gui/browser/file_browser.py index 6a79c87..0ce0004 100644 --- a/src/gui/browser/file_browser.py +++ b/src/gui/browser/file_browser.py @@ -122,7 +122,7 @@ def _setup_ui(self): layout.setContentsMargins(0, 0, 0, 0) # Datei-System-Model - self.model = QFileSystemModel() + self.model = QFileSystemModel(self) self.model.setFilter( QDir.Filter.AllEntries | QDir.Filter.NoDotAndDotDot @@ -130,7 +130,7 @@ def _setup_ui(self): self.model.directoryLoaded.connect(self._on_directory_loaded) # Sortier-Proxy - self.proxy = QSortFilterProxyModel() + self.proxy = QSortFilterProxyModel(self) self.proxy.setSourceModel(self.model) self.proxy.setSortCaseSensitivity(Qt.CaseSensitivity.CaseInsensitive) @@ -775,7 +775,10 @@ def _show_diff(self, file1: str = "", file2: str = ""): dialog.exec() def delete_selection(self, target_paths: list = None) -> bool: - """Löscht ausgewählte Dateien oder Ordner nach Bestätigung.""" + """Delete selected entries, using the trash when confirmation is disabled.""" + from core.delete_service import delete_path, move_to_trash + from core.settings_manager import SettingsManager + if not target_paths: target_paths = self.get_selected_files() if not target_paths: @@ -790,26 +793,25 @@ def delete_selection(self, target_paths: list = None) -> bool: preview += f"\n... und {count - 5} weitere" msg = f"Möchten Sie diese {count} Elemente wirklich unwiderruflich löschen?\n\n{preview}" - reply = QMessageBox.question( - self, - "Löschen bestätigen", - msg, - QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, - QMessageBox.StandardButton.No - ) - - if reply != QMessageBox.StandardButton.Yes: - return False + confirm_delete = SettingsManager.instance().get("general", "confirm_delete", True) is not False + if confirm_delete: + reply = QMessageBox.question( + self, + "Löschen bestätigen", + msg, + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No + ) + if reply != QMessageBox.StandardButton.Yes: + return False errors = [] + delete_item = delete_path if confirm_delete else move_to_trash for path in target_paths: - if not os.path.exists(path): + if not os.path.lexists(path): continue try: - if os.path.isdir(path): - shutil.rmtree(path) - else: - os.remove(path) + delete_item(path) except OSError as exc: errors.append(f"{os.path.basename(path)}: {exc}") diff --git a/src/gui/checksum_dialog.py b/src/gui/checksum_dialog.py index 3b9afb6..7674b40 100644 --- a/src/gui/checksum_dialog.py +++ b/src/gui/checksum_dialog.py @@ -10,7 +10,7 @@ QLabel, QLineEdit, QPushButton, QProgressBar, QApplication, QMessageBox ) -from PySide6.QtCore import Slot +from PySide6.QtCore import Slot, QTimer from PySide6.QtGui import QFont from core.checksum_service import ChecksumWorker, verify_hash @@ -24,6 +24,10 @@ def __init__(self, filepath: str, parent=None): self.filepath = filepath self.worker = None self._calculated_hashes = {} + self._pending_close_result = None + self._close_timer = QTimer(self) + self._close_timer.setInterval(20) + self._close_timer.timeout.connect(self._finish_pending_close) filename = os.path.basename(filepath) self.setWindowTitle(f"Prüfsummen — {filename}") @@ -243,13 +247,34 @@ def _on_verify_text_changed(self, text: str): self.verify_result_label.setStyleSheet("color: #c00000; font-weight: bold;") def done(self, result: int): - if self.worker and self.worker.isRunning(): + if self.worker and not self.worker.wait(0): + # A timed wait does not make destroying a still-running QThread safe. + # Keep the dialog alive and its event loop responsive until it exits. + if self._pending_close_result is None: + self._pending_close_result = result self.worker.cancel() - self.worker.wait(1000) + self.close_btn.setEnabled(False) + self._close_timer.start() + return + self._close_timer.stop() + if self._pending_close_result is not None: + result = self._pending_close_result + self._pending_close_result = None super().done(result) + @Slot() + def _finish_pending_close(self): + if self.worker and not self.worker.wait(0): + return + self._close_timer.stop() + result = self._pending_close_result + self._pending_close_result = None + if result is not None: + super().done(result) + def closeEvent(self, event): - if self.worker and self.worker.isRunning(): - self.worker.cancel() - self.worker.wait(1000) + if self.worker and not self.worker.wait(0): + event.ignore() + self.done(QDialog.DialogCode.Rejected) + return super().closeEvent(event) diff --git a/src/gui/main_window.py b/src/gui/main_window.py index 199cce5..3bf6db8 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -22,6 +22,7 @@ from .preview.preview_panel import PreviewPanel from .status_bar import StatusBarWidget from version import __version__ +from translator import t class SearchToolBar(QToolBar): @@ -341,6 +342,16 @@ def _setup_menu(self): copy_path_action.triggered.connect(self._copy_path) edit_menu.addAction(copy_path_action) + edit_menu.addSeparator() + self.confirm_delete_action = QAction(t("Vor dem Löschen nachfragen"), self) + self.confirm_delete_action.setCheckable(True) + self.confirm_delete_action.setToolTip( + t("Wenn deaktiviert, werden Elemente ohne Rückfrage in den Papierkorb verschoben.") + ) + self._sync_delete_confirmation() + self.confirm_delete_action.triggered.connect(self._set_delete_confirmation) + edit_menu.addAction(self.confirm_delete_action) + edit_menu.aboutToShow.connect(self._sync_delete_confirmation) # ===== Ansicht-Menü ===== view_menu = menubar.addMenu("&Ansicht") @@ -792,11 +803,26 @@ def _show_settings(self): self._apply_settings() self.statusBar().showMessage("Einstellungen gespeichert", 3000) + def _sync_delete_confirmation(self): + from core.settings_manager import SettingsManager + + self.confirm_delete_action.setChecked( + SettingsManager.instance().get("general", "confirm_delete", True) is not False + ) + + def _set_delete_confirmation(self, checked): + from core.settings_manager import SettingsManager + + settings = SettingsManager.instance() + settings.set("general", "confirm_delete", checked) + settings.save() + def _apply_settings(self): """Wendet die gespeicherten Einstellungen auf das laufende Fenster an.""" from core.settings_manager import SettingsManager settings = SettingsManager.instance() + self._sync_delete_confirmation() show_hidden = bool(settings.get("general", "show_hidden_files", False)) if hasattr(self.file_browser, "set_show_hidden_files"): diff --git a/src/gui/settings_dialog.py b/src/gui/settings_dialog.py index 3a1a64b..4e4f4d7 100644 --- a/src/gui/settings_dialog.py +++ b/src/gui/settings_dialog.py @@ -15,7 +15,7 @@ ) from core.settings_manager import SettingsManager -from translator import TranslationSystem +from translator import TranslationSystem, t class SettingsDialog(QDialog): @@ -102,9 +102,11 @@ def _build_general_tab(self) -> QWidget: self.show_hidden_cb.setToolTip("Versteckte Dateien und Systemordner im Dateibrowser anzeigen") form.addRow(self.show_hidden_cb) - self.confirm_delete_cb = QCheckBox("Vor dem Löschen nachfragen") - self.confirm_delete_cb.setAccessibleName("Vor dem Löschen nachfragen") - self.confirm_delete_cb.setToolTip("Sicherheitsabfrage vor dem unwiderruflichen Löschen einblenden") + self.confirm_delete_cb = QCheckBox(t("Vor dem Löschen nachfragen")) + self.confirm_delete_cb.setAccessibleName(t("Vor dem Löschen nachfragen")) + self.confirm_delete_cb.setToolTip( + t("Wenn deaktiviert, werden Elemente ohne Rückfrage in den Papierkorb verschoben.") + ) form.addRow(self.confirm_delete_cb) self.remember_size_cb = QCheckBox("Fenstergröße merken") @@ -239,7 +241,7 @@ def _load_settings(self): self.start_folder_edit.setText(get("general", "start_folder", "") or "") self.show_hidden_cb.setChecked(bool(get("general", "show_hidden_files", False))) - self.confirm_delete_cb.setChecked(bool(get("general", "confirm_delete", True))) + self.confirm_delete_cb.setChecked(get("general", "confirm_delete", True) is not False) self.remember_size_cb.setChecked(bool(get("general", "remember_window_size", True))) self.auto_index_cb.setChecked(bool(get("index", "auto_index", True))) diff --git a/src/gui/sidebar/drive_capacity.py b/src/gui/sidebar/drive_capacity.py new file mode 100644 index 0000000..4fb8242 --- /dev/null +++ b/src/gui/sidebar/drive_capacity.py @@ -0,0 +1,79 @@ +"""Capacity widgets and independent background requests for drive rows.""" + +from concurrent.futures import ThreadPoolExecutor + +from PySide6.QtCore import Qt +from PySide6.QtWidgets import QLabel, QProgressBar, QVBoxLayout, QWidget + +from core.drive_usage import format_capacity, start_drive_queries, stop_drive_queries +from translator import t + +_executor = None + + +def capacity_executor(): + """Workers only return Python data; no Qt objects cross thread boundaries. + + OS reads run in bounded helpers, so failed drives release worker slots. + """ + global _executor + if _executor is None: + start_drive_queries() + _executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="drive-capacity") + return _executor + + +def shutdown_capacity_executor(): + """Stop capacity helpers, then drain Python workers before Qt teardown.""" + global _executor + if _executor is not None: + stop_drive_queries() + _executor.shutdown(wait=True, cancel_futures=True) + _executor = None + + +class DriveCapacityWidget(QWidget): + def __init__(self, path, parent=None): + super().__init__(parent) + self.path = path + # Tree selection and double-click navigation still receive mouse events. + self.setAttribute(Qt.WidgetAttribute.WA_TransparentForMouseEvents) + layout = QVBoxLayout(self) + layout.setContentsMargins(2, 3, 2, 3) + layout.setSpacing(2) + self.title = QLabel(path) + self.details = QLabel() + self.details.setWordWrap(True) + self.bar = QProgressBar() + self.bar.setRange(0, 1000) + self.bar.setFixedHeight(18) + self.bar.setAccessibleName(t("Speicherbelegung")) + layout.addWidget(self.title) + layout.addWidget(self.details) + layout.addWidget(self.bar) + self.set_loading() + + def set_loading(self): + self.details.setText(t("Speicherbelegung wird ermittelt …")) + self.bar.hide() + self.setAccessibleName(self.path) + self.setAccessibleDescription(self.details.text()) + + def set_usage(self, usage): + if usage is None: + self.details.setText(t("Speicherbelegung nicht verfügbar")) + self.bar.hide() + self.bar.setValue(0) + else: + self.details.setText(t("{free} frei · {used} belegt · {total} gesamt").format( + free=format_capacity(usage.free), used=format_capacity(usage.used), + total=format_capacity(usage.total))) + self.bar.setValue(round(usage.used_percent * 10)) + self.bar.setFormat(t("{percent}% belegt").format(percent=f"{usage.used_percent:.1f}")) + self.bar.show() + description = self.details.text() + if usage is not None: + description += " · " + self.bar.format() + self.setToolTip(self.path + "\n" + description) + self.setAccessibleDescription(description) + self.bar.setAccessibleDescription(description) diff --git a/src/gui/sidebar/sidebar_main.py b/src/gui/sidebar/sidebar_main.py index daf285d..a914721 100644 --- a/src/gui/sidebar/sidebar_main.py +++ b/src/gui/sidebar/sidebar_main.py @@ -5,12 +5,14 @@ Phase 5: Vollständige Integration """ +from concurrent.futures import CancelledError + from PySide6.QtWidgets import ( QWidget, QVBoxLayout, QHBoxLayout, QStackedWidget, QTreeWidget, QTreeWidgetItem, QListWidget, QListWidgetItem, QPushButton, QLabel, QFrame, QToolButton, QButtonGroup ) -from PySide6.QtCore import Qt, Signal, QDir, QStandardPaths +from PySide6.QtCore import Qt, Signal, Slot, QDir, QStandardPaths, QSize, QTimer import os # Module importieren - absolute Imports @@ -19,6 +21,9 @@ from modules.launcher import AppsPanel from modules.prompts import PromptsPanel from modules.sync import SyncPanel +from core.drive_usage import read_drive_usage_bounded as read_drive_usage +from gui.sidebar.drive_capacity import DriveCapacityWidget, capacity_executor +from translator import t class TreePanel(QWidget): @@ -28,8 +33,17 @@ class TreePanel(QWidget): def __init__(self, parent=None): super().__init__(parent) + self._drive_rows = {} + self._usage_requests = {} + self._resize_timer = QTimer(self) + self._resize_timer.setSingleShot(True) + self._resize_timer.timeout.connect(self._resize_drive_rows) + self._usage_timer = QTimer(self) + self._usage_timer.setInterval(50) + self._usage_timer.timeout.connect(self._collect_drive_usage) self._setup_ui() self._populate() + self.refresh_drive_usage() def _setup_ui(self): layout = QVBoxLayout(self) @@ -48,6 +62,9 @@ def _setup_ui(self): self.tree.itemExpanded.connect(self._on_item_expanded) layout.addWidget(self.tree) + self.refresh_drives_button = QPushButton(t("Laufwerksbelegung aktualisieren")) + self.refresh_drives_button.clicked.connect(self.refresh_drive_usage) + layout.addWidget(self.refresh_drives_button) def _populate(self): """Füllt den Baum mit Laufwerken und Schnellzugriff""" @@ -77,18 +94,77 @@ def _populate(self): # Laufwerke drives_item = QTreeWidgetItem(["💾 Laufwerke"]) drives_item.setFlags(drives_item.flags() & ~Qt.ItemFlag.ItemIsSelectable) + self.tree.addTopLevelItem(drives_item) for drive in QDir.drives(): path = drive.absolutePath() - item = QTreeWidgetItem([path]) + # The embedded widget paints the title; duplicate tree text would + # otherwise show through between its labels and progress bar. + item = QTreeWidgetItem([""]) item.setData(0, Qt.ItemDataRole.UserRole, path) + item.setData(0, Qt.ItemDataRole.AccessibleTextRole, path) item.setIcon(0, get_file_icon(path)) item.setChildIndicatorPolicy(QTreeWidgetItem.ChildIndicatorPolicy.ShowIndicator) drives_item.addChild(item) + capacity = DriveCapacityWidget(path) + self.tree.setItemWidget(item, 0, capacity) + item.setSizeHint(0, capacity.sizeHint()) + self._drive_rows[path] = (item, capacity) - self.tree.addTopLevelItem(drives_item) drives_item.setExpanded(True) + def refresh_drive_usage(self): + """One request per drive; repeat clicks cannot queue duplicate queries.""" + for path, (item, capacity) in self._drive_rows.items(): + if path in self._usage_requests: + continue + capacity.set_loading() + item.setSizeHint(0, capacity.sizeHint()) + self._usage_requests[path] = capacity_executor().submit(read_drive_usage, path) + if self._usage_requests: + self._usage_timer.start() + + @Slot() + def _collect_drive_usage(self): + for path, future in list(self._usage_requests.items()): + if not future.done(): + continue + try: + usage = future.result() + except (CancelledError, OSError, ValueError): + usage = None + self._on_drive_usage(path, usage) + if not self._usage_requests: + self._usage_timer.stop() + + @Slot(str, object) + def _on_drive_usage(self, path, usage): + self._usage_requests.pop(path, None) + row = self._drive_rows.get(path) + if row is not None: + item, capacity = row + capacity.set_usage(usage) + item.setToolTip(0, capacity.toolTip()) + item.setData(0, Qt.ItemDataRole.AccessibleDescriptionRole, capacity.accessibleDescription()) + item.setSizeHint(0, capacity.sizeHint()) + self._resize_drive_rows() + + def showEvent(self, event): + super().showEvent(event) + self.refresh_drive_usage() + + def resizeEvent(self, event): + super().resizeEvent(event) + self._resize_timer.start(0) + + @Slot() + def _resize_drive_rows(self): + """Wrapped labels need taller rows when the sidebar becomes narrow.""" + for item, capacity in self._drive_rows.values(): + height = capacity.layout().totalHeightForWidth(capacity.width()) + item.setSizeHint(0, QSize(0, max(height, capacity.sizeHint().height()))) + self.tree.doItemsLayout() + def _on_item_clicked(self, item: QTreeWidgetItem, column: int): path = item.data(0, Qt.ItemDataRole.UserRole) if path: diff --git a/src/main.py b/src/main.py index 63ea7ab..660f258 100644 --- a/src/main.py +++ b/src/main.py @@ -21,12 +21,19 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +# Frozen helpers must finish before importing Qt or constructing the application. +if __name__ == '__main__' and len(sys.argv) == 4 and sys.argv[1] == '--drive-capacity-query': + from core.drive_usage import capacity_query_main + sys.exit(capacity_query_main(sys.argv[2], sys.argv[3])) + from PySide6.QtWidgets import QApplication from PySide6.QtCore import Qt, QTranslator, QLibraryInfo from PySide6.QtGui import QIcon from app import ExplorerProApp from version import __version__ +from core.gui_gc import install_gui_gc +from gui.sidebar.drive_capacity import shutdown_capacity_executor def load_app_icon() -> QIcon: @@ -73,29 +80,36 @@ def main(): ) app = QApplication(sys.argv) - app.setApplicationName("ExplorerPro") - app.setOrganizationName("ExplorerPro") - from translator import get_translator - install_qt_translations(app, get_translator().get_language()) - set_application_version(app) - icon = load_app_icon() - if not icon.isNull(): - app.setWindowIcon(icon) - - # Style - app.setStyle("Fusion") - - # Dark Theme (optional) - # from gui.themes import apply_dark_theme - # apply_dark_theme(app) - - # Hauptfenster starten - explorer = ExplorerProApp() - if not icon.isNull(): - explorer.setWindowIcon(icon) - explorer.show() - - sys.exit(app.exec()) + collector = install_gui_gc(app) + try: + app.setApplicationName("ExplorerPro") + app.setOrganizationName("ExplorerPro") + from translator import get_translator + install_qt_translations(app, get_translator().get_language()) + set_application_version(app) + icon = load_app_icon() + if not icon.isNull(): + app.setWindowIcon(icon) + + # Style + app.setStyle("Fusion") + + # Dark Theme (optional) + # from gui.themes import apply_dark_theme + # apply_dark_theme(app) + + # Hauptfenster starten + explorer = ExplorerProApp() + if not icon.isNull(): + explorer.setWindowIcon(icon) + explorer.show() + exit_code = app.exec() + finally: + try: + shutdown_capacity_executor() + finally: + collector.close() + sys.exit(exit_code) if __name__ == "__main__": diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..9f2243b --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,29 @@ +"""Exercise the application's GUI-thread cyclic collection policy.""" + +import os + +import pytest + +os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen') + +from PySide6.QtWidgets import QApplication +from core.gui_gc import install_gui_gc +from gui.sidebar.drive_capacity import shutdown_capacity_executor + +_application = QApplication.instance() or QApplication([]) +_collector = install_gui_gc(_application) + + +@pytest.fixture(autouse=True) +def collect_gui_cycles(): + yield + _collector.collect() + + +@pytest.fixture(scope='session', autouse=True) +def gui_runtime(): + yield + try: + shutdown_capacity_executor() + finally: + _collector.close() diff --git a/tests/test_bugsweep_checksum_and_diff_20260924.py b/tests/test_bugsweep_checksum_and_diff_20260924.py index efe31e9..160d859 100644 --- a/tests/test_bugsweep_checksum_and_diff_20260924.py +++ b/tests/test_bugsweep_checksum_and_diff_20260924.py @@ -145,29 +145,32 @@ def test_is_binary_file_heuristic_control_chars(tmp_path: Path): assert is_binary_file(str(bin_file)) is True -def test_checksum_dialog_done_cancels_worker(tmp_path: Path): +def test_checksum_dialog_done_cancels_worker(tmp_path: Path, qtbot): """Prüft, dass ChecksumDialog.done() (via accept/reject/Schließen) den Worker abbricht.""" _ensure_app() sample = tmp_path / "big_sample.bin" sample.write_bytes(b"Z" * (1024 * 1024)) dlg = ChecksumDialog(str(sample)) + qtbot.addWidget(dlg) worker = dlg.worker assert worker is not None # Simuliere Schließen via accept (Button "Schließen") dlg.accept() assert worker.is_cancelled() is True - assert not worker.isRunning() + qtbot.waitUntil(lambda: not worker.isRunning(), timeout=3000) + qtbot.waitUntil(lambda: not dlg._close_timer.isActive(), timeout=3000) -def test_checksum_dialog_directory_handling(tmp_path: Path): +def test_checksum_dialog_directory_handling(tmp_path: Path, qtbot): """Prüft, dass ChecksumDialog bei einem Verzeichnis nicht abstürzt und klare Meldung liefert.""" _ensure_app() sub_dir = tmp_path / "some_directory" sub_dir.mkdir() dlg = ChecksumDialog(str(sub_dir)) + qtbot.addWidget(dlg) assert dlg.worker is None assert "Verzeichnis" in dlg.verify_result_label.text() dlg.close() diff --git a/tests/test_checksum_close_lifecycle.py b/tests/test_checksum_close_lifecycle.py new file mode 100644 index 0000000..3b79e94 --- /dev/null +++ b/tests/test_checksum_close_lifecycle.py @@ -0,0 +1,98 @@ +"""A dialog must retain its native QThread until the worker actually exits.""" + +import threading +import time +import subprocess +import sys +from pathlib import Path + +import pytest +from PySide6.QtWidgets import QDialog + +from core.checksum_service import ChecksumWorker +from gui.checksum_dialog import ChecksumDialog + + +@pytest.mark.parametrize('action,result', [ + ('accept', QDialog.DialogCode.Accepted), + ('reject', QDialog.DialogCode.Rejected), + ('close', QDialog.DialogCode.Rejected), +]) +def test_close_remains_responsive_and_keeps_running_worker(qtbot, monkeypatch, tmp_path, action, result): + entered, release = threading.Event(), threading.Event() + + def blocked_read(worker): + entered.set() + release.wait(5) + + monkeypatch.setattr(ChecksumWorker, 'run', blocked_read) + sample = tmp_path / 'sample.txt' + sample.write_text('slow storage', encoding='utf-8') + dialog = ChecksumDialog(str(sample)) + qtbot.addWidget(dialog) + dialog.show() + completed = [] + dialog.finished.connect(completed.append) + try: + assert entered.wait(2) + started = time.monotonic() + getattr(dialog, action)() + assert time.monotonic() - started < .3 + qtbot.wait(40) + assert dialog.isVisible() + assert dialog.worker.isRunning() + assert dialog.worker.is_cancelled() + assert completed == [] + # Repeated close requests must not emit duplicate completion signals. + getattr(dialog, action)() + release.set() + assert dialog.worker.wait(2000) + # The worker can exit just before the polling timer handles its end. + # A concurrent close request must preserve the first result and emit once. + dialog.done(1 - int(result)) + qtbot.waitUntil(lambda: completed == [int(result)], timeout=3000) + qtbot.wait(40) + assert completed == [int(result)] + assert not dialog.isVisible() + assert not dialog.worker.isRunning() + finally: + release.set() + dialog.worker.wait(2000) + dialog.close() + + +def test_finished_cleanup_does_not_destroy_running_native_thread(tmp_path): + script = ''' +import sys, threading +from pathlib import Path +sys.path.insert(0, 'src') +from PySide6.QtCore import QCoreApplication, QEvent, QTimer +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication +from core.checksum_service import ChecksumWorker +from gui.checksum_dialog import ChecksumDialog +app = QApplication([]) +app.setQuitOnLastWindowClosed(False) +entered, release = threading.Event(), threading.Event() +def read(worker): + entered.set() + release.wait(5) +ChecksumWorker.run = read +dialog = ChecksumDialog(sys.argv[1]) +dialog.show() +dialog.finished.connect(dialog.deleteLater) +assert entered.wait(2) +QTimer.singleShot(1500, release.set) +dialog.accept() +for _ in range(200): + QTest.qWait(10) +QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete) +print('native-close-ok') +''' + sample = tmp_path / 'slow.txt' + sample.write_text('sample', encoding='utf-8') + result = subprocess.run([sys.executable, '-X', 'faulthandler', '-c', script, str(sample)], + cwd=Path(__file__).resolve().parents[1], capture_output=True, + text=True, timeout=15) + assert result.returncode == 0, (result.returncode, result.stderr) + assert 'native-close-ok' in result.stdout diff --git a/tests/test_checksum_dialog.py b/tests/test_checksum_dialog.py index d9fdb8d..6ce1e6a 100644 --- a/tests/test_checksum_dialog.py +++ b/tests/test_checksum_dialog.py @@ -20,13 +20,14 @@ def _ensure_app(): return app -def test_checksum_dialog_initialization(tmp_path: Path): +def test_checksum_dialog_initialization(tmp_path: Path, qtbot): """Prüft Dialog-Aufbau und Hash-Berechnung.""" _ensure_app() sample = tmp_path / "test_dialog.txt" sample.write_text("ExplorerPro Dialog Test Content", encoding="utf-8") dlg = ChecksumDialog(str(sample)) + qtbot.addWidget(dlg) assert dlg.path_label.text() == str(sample) assert dlg.windowTitle().startswith("Prüfsummen — test_dialog.txt") assert "sha256" in dlg.hash_edits @@ -61,20 +62,24 @@ def test_checksum_dialog_initialization(tmp_path: Path): dlg.close() -def test_checksum_dialog_non_existent_file(tmp_path: Path): +def test_checksum_dialog_non_existent_file(tmp_path: Path, qtbot): """Prüft Dialog-Verhalten wenn Datei nicht existiert.""" _ensure_app() dlg = ChecksumDialog(str(tmp_path / "nicht_da.txt")) + qtbot.addWidget(dlg) assert "Fehler" in dlg.verify_result_label.text() dlg.close() -def test_checksum_dialog_close_cancels_worker(tmp_path: Path): +def test_checksum_dialog_close_cancels_worker(tmp_path: Path, qtbot): """Prüft, dass closeEvent den Worker sauber abbricht.""" _ensure_app() sample = tmp_path / "cancel_test.txt" sample.write_bytes(b"X" * (500 * 1024)) dlg = ChecksumDialog(str(sample)) + qtbot.addWidget(dlg) dlg.closeEvent(QCloseEvent()) assert dlg.worker is not None + qtbot.waitUntil(lambda: not dlg.worker.isRunning(), timeout=3000) + qtbot.waitUntil(lambda: not dlg._close_timer.isActive(), timeout=3000) diff --git a/tests/test_delete_confirmation.py b/tests/test_delete_confirmation.py new file mode 100644 index 0000000..0672ca6 --- /dev/null +++ b/tests/test_delete_confirmation.py @@ -0,0 +1,203 @@ +"""Menu, settings persistence and all browser deletion entry points.""" + +import json +import os +from pathlib import Path +from unittest.mock import Mock + +import pytest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtCore import QEvent, Qt +from PySide6.QtGui import QKeyEvent +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication, QMenu, QMessageBox + +from core import delete_service +from core.settings_manager import SettingsManager +import gui.browser.file_browser as browser_module +from gui.main_window import MainWindow +from gui.settings_dialog import SettingsDialog + +_app = QApplication.instance() or QApplication([]) + + +@pytest.fixture +def settings(tmp_path, monkeypatch): + monkeypatch.setattr(SettingsManager, "_instance", None) + monkeypatch.setattr(SettingsManager, "_get_config_path", lambda self: tmp_path / "settings.json") + return SettingsManager.instance() + + +@pytest.fixture +def window(settings): + win = MainWindow() + yield win + win.close() + win.deleteLater() + _app.sendPostedEvents(None, QEvent.Type.DeferredDelete) + + +def test_menu_checkbox_default_and_json_roundtrip(window, settings): + action = window.confirm_delete_action + assert action.isCheckable() and action.isChecked() + action.trigger() + assert settings.get("general", "confirm_delete") is False + assert json.loads(settings._config_path.read_text())["general"]["confirm_delete"] is False + settings._settings = {} + settings._load_settings() + assert settings.get("general", "confirm_delete") is False + second = MainWindow() + try: + assert not second.confirm_delete_action.isChecked() + action.trigger() + second.menuBar().actions()[1].menu().aboutToShow.emit() + assert second.confirm_delete_action.isChecked() + assert json.loads(settings._config_path.read_text())["general"]["confirm_delete"] is True + finally: + second.close() + second.deleteLater() + + +def test_settings_dialog_and_menu_share_confirmation(window, settings, monkeypatch): + window.confirm_delete_action.trigger() + + def accept(dialog): + assert not dialog.confirm_delete_cb.isChecked() + dialog.confirm_delete_cb.setChecked(True) + dialog.apply_to_settings() + return SettingsDialog.DialogCode.Accepted + + monkeypatch.setattr(SettingsDialog, "exec", accept) + window._show_settings() + assert window.confirm_delete_action.isChecked() + assert settings.get("general", "confirm_delete") is True + assert json.loads(settings._config_path.read_text())["general"]["confirm_delete"] is True + + +@pytest.mark.parametrize("invalid", [None, 0, "", "false", [], {}]) +def test_only_explicit_json_false_disables_confirmation(window, settings, tmp_path, monkeypatch, invalid): + settings._config_path.write_text(json.dumps({"general": {"confirm_delete": invalid}})) + settings._load_settings() + window._apply_settings() + assert window.confirm_delete_action.isChecked() + dialog = SettingsDialog(window) + assert dialog.confirm_delete_cb.isChecked() + target = tmp_path / "keep.txt" + target.write_text("keep") + question = Mock(return_value=QMessageBox.StandardButton.No) + monkeypatch.setattr(QMessageBox, "question", question) + assert not window.file_browser.delete_selection([str(target)]) + question.assert_called_once() + assert target.exists() + dialog.deleteLater() + + +def _context_delete(browser, target, monkeypatch): + for _ in range(100): + _app.processEvents() + index = browser.proxy.mapFromSource(browser.model.index(str(target))) + if index.isValid(): + break + QTest.qWait(10) + assert index.isValid() + browser.table.selectRow(index.row()) + + class CapturingMenu(QMenu): + def exec(self, *args): + next(a for a in self.actions() if a.text() == "Löschen").trigger() + + monkeypatch.setattr(browser_module, "QMenu", CapturingMenu) + browser._show_context_menu(browser.table.visualRect(index).center()) + + +@pytest.mark.parametrize("entry", ["direct", "keyboard", "context"]) +@pytest.mark.parametrize("confirm", [True, False]) +def test_all_entry_points_use_current_setting(window, settings, tmp_path, monkeypatch, entry, confirm): + target = tmp_path / "selected.txt" + target.write_text("delete") + browser = window.file_browser + browser.navigate_to(str(tmp_path)) + if not confirm: + window.confirm_delete_action.trigger() + question = Mock(return_value=QMessageBox.StandardButton.Yes) + monkeypatch.setattr(QMessageBox, "question", question) + trash = Mock(side_effect=lambda path: Path(path).unlink()) + monkeypatch.setattr(delete_service, "move_to_trash", trash) + if entry == "direct": + assert browser.delete_selection([str(target)]) + elif entry == "keyboard": + monkeypatch.setattr(browser, "get_selected_files", lambda: [str(target)]) + browser.table.keyPressEvent( + QKeyEvent(QEvent.Type.KeyPress, Qt.Key.Key_Delete, Qt.KeyboardModifier.NoModifier) + ) + else: + _context_delete(browser, target, monkeypatch) + assert not target.exists() + assert question.call_count == int(confirm) + assert trash.call_count == int(not confirm) + if confirm: + assert question.call_args.args[-1] == QMessageBox.StandardButton.No + + +def test_reenabled_confirmation_cancellation_preserves_file(window, tmp_path, monkeypatch): + target = tmp_path / "keep.txt" + target.write_text("keep") + window.confirm_delete_action.trigger() + window.confirm_delete_action.trigger() + monkeypatch.setattr(QMessageBox, "question", lambda *a: QMessageBox.StandardButton.No) + assert not window.file_browser.delete_selection([str(target)]) + assert target.read_text() == "keep" + + +def test_partial_failure_warns_and_continues(window, tmp_path, monkeypatch): + blocked = tmp_path / "blocked.txt" + blocked.write_text("keep") + good = tmp_path / "good.txt" + good.write_text("delete") + window.confirm_delete_action.trigger() + def trash(path): + if path == str(blocked): + raise PermissionError("blocked fixture") + Path(path).unlink() + + monkeypatch.setattr(delete_service, "move_to_trash", trash) + warning = Mock() + monkeypatch.setattr(QMessageBox, "warning", warning) + assert not window.file_browser.delete_selection([str(blocked), str(good)]) + assert blocked.read_text() == "keep" + assert not good.exists() + assert "blocked.txt" in warning.call_args.args[2] + warning.assert_called_once() + + +def test_disabled_confirmation_uses_trash_and_never_permanent_delete(window, tmp_path, monkeypatch): + target = tmp_path / "recoverable.txt" + target.write_text("keep recoverable") + window.confirm_delete_action.trigger() + trash = Mock() + monkeypatch.setattr(delete_service, "move_to_trash", trash) + permanent_delete = Mock(side_effect=AssertionError("must not permanently delete")) + monkeypatch.setattr(delete_service, "delete_path", permanent_delete) + monkeypatch.setattr(QMessageBox, "question", Mock(side_effect=AssertionError("must not ask"))) + + assert window.file_browser.delete_selection([str(target)]) + trash.assert_called_once_with(str(target)) + permanent_delete.assert_not_called() + assert target.exists() + + +def test_disabled_confirmation_preserves_item_when_trash_fails(window, tmp_path, monkeypatch): + target = tmp_path / "keep.txt" + target.write_text("keep") + window.confirm_delete_action.trigger() + monkeypatch.setattr( + delete_service, "move_to_trash", Mock(side_effect=OSError("Papierkorb nicht verfügbar")) + ) + warning = Mock() + monkeypatch.setattr(QMessageBox, "warning", warning) + + assert not window.file_browser.delete_selection([str(target)]) + assert target.read_text() == "keep" + warning.assert_called_once() diff --git a/tests/test_delete_service.py b/tests/test_delete_service.py new file mode 100644 index 0000000..b750847 --- /dev/null +++ b/tests/test_delete_service.py @@ -0,0 +1,250 @@ +"""Deletion must handle Windows read-only files without following links.""" + +import os +import stat +import subprocess +import sys +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from core import delete_service + + +def test_delete_regular_file_and_nested_folder(tmp_path): + folder = tmp_path / "selected" + (folder / "nested").mkdir(parents=True) + (folder / "nested" / "child.txt").write_text("child") + direct = tmp_path / "direct.txt" + direct.write_text("direct") + delete_service.delete_path(str(direct)) + delete_service.delete_path(str(folder)) + assert not direct.exists() + assert not folder.exists() + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows file attributes") +@pytest.mark.parametrize("nested", [False, True]) +def test_native_readonly_file_deletion(tmp_path, nested): + folder = tmp_path / "selected" + folder.mkdir() + child = folder / "readonly.txt" + child.write_text("read-only fixture") + os.chmod(child, stat.S_IREAD) + try: + delete_service.delete_path(str(folder if nested else child)) + assert not child.exists() + if nested: + assert not folder.exists() + finally: + if child.exists(): + os.chmod(child, stat.S_IWRITE) + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows read-only hardlink attributes") +@pytest.mark.parametrize("nested", [False, True]) +def test_readonly_hardlink_does_not_change_surviving_alias(tmp_path, nested): + outside = tmp_path / "outside.txt" + outside.write_text("unchanged bytes") + selected = tmp_path / "selected" + selected.mkdir() + alias = selected / "alias.txt" + os.link(outside, alias) + os.chmod(alias, stat.S_IREAD) + original_attributes = outside.stat().st_file_attributes + try: + with pytest.raises(PermissionError) as caught: + delete_service.delete_path(str(selected if nested else alias)) + assert caught.value.winerror == 5 + assert alias.exists() and outside.exists() + assert outside.read_text() == "unchanged bytes" + assert alias.read_text() == "unchanged bytes" + assert outside.stat().st_file_attributes == original_attributes + assert alias.stat().st_file_attributes == original_attributes + finally: + os.chmod(outside, stat.S_IWRITE) + + +def test_writable_hardlink_deletes_selected_name_only(tmp_path): + outside = tmp_path / "outside.txt" + outside.write_text("unchanged bytes") + alias = tmp_path / "selected.txt" + os.link(outside, alias) + delete_service.delete_path(str(alias)) + assert not alias.exists() + assert outside.read_text() == "unchanged bytes" + + +def _windows_error(code): + error = PermissionError(13, "access denied") + error.winerror = code + return error + + +@pytest.mark.parametrize("code", [5, 32]) +def test_permissions_and_sharing_failures_do_not_chmod_normal_files(tmp_path, monkeypatch, code): + path = tmp_path / "normal.txt" + path.write_text("keep") + error = _windows_error(code) + remove = Mock(side_effect=error) + chmod = Mock() + with monkeypatch.context() as patch: + patch.setattr(delete_service.os, "remove", remove) + patch.setattr(delete_service.os, "chmod", chmod) + with pytest.raises(PermissionError) as caught: + delete_service.delete_path(str(path)) + assert caught.value is error + remove.assert_called_once_with(str(path)) + chmod.assert_not_called() + assert path.read_text() == "keep" + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows file attributes") +def test_failed_readonly_retry_restores_attribute_and_propagates_error(tmp_path, monkeypatch): + path = tmp_path / "readonly.txt" + path.write_text("keep") + os.chmod(path, stat.S_IREAD) + native_remove = os.remove + calls = [] + retry_error = _windows_error(32) + + def remove(name): + calls.append(name) + if len(calls) == 1: + native_remove(name) # Real WinError 5 from the read-only attribute. + raise retry_error + + try: + with monkeypatch.context() as patch: + patch.setattr(delete_service.os, "remove", remove) + with pytest.raises(PermissionError) as caught: + delete_service.delete_path(str(path)) + assert caught.value is retry_error + assert len(calls) == 2 + assert path.stat().st_file_attributes & stat.FILE_ATTRIBUTE_READONLY + assert path.read_text() == "keep" + finally: + os.chmod(path, stat.S_IWRITE) + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows file attributes") +def test_sharing_failure_does_not_clear_readonly_attribute(tmp_path, monkeypatch): + path = tmp_path / "readonly.txt" + path.write_text("keep") + os.chmod(path, stat.S_IREAD) + try: + error = _windows_error(32) + with monkeypatch.context() as patch: + patch.setattr(delete_service.os, "remove", Mock(side_effect=error)) + chmod = Mock() + patch.setattr(delete_service.os, "chmod", chmod) + with pytest.raises(PermissionError) as caught: + delete_service.delete_path(str(path)) + assert caught.value is error + chmod.assert_not_called() + assert path.stat().st_file_attributes & stat.FILE_ATTRIBUTE_READONLY + finally: + os.chmod(path, stat.S_IWRITE) + + +def test_directory_error_does_not_retry_or_chmod(tmp_path, monkeypatch): + error = _windows_error(5) + chmod = Mock() + monkeypatch.setattr(delete_service.os, "chmod", chmod) + with pytest.raises(PermissionError) as caught: + delete_service._rmtree_error(os.rmdir, str(tmp_path), (PermissionError, error, None)) + assert caught.value is error + chmod.assert_not_called() + + +@pytest.mark.parametrize( + ("version", "callback_name"), + [((3, 10), "onerror"), ((3, 11), "onerror"), ((3, 12), "onexc")], +) +def test_remove_tree_selects_python_compatible_callback(monkeypatch, version, callback_name): + rmtree = Mock() + monkeypatch.setattr(delete_service.shutil, "rmtree", rmtree) + monkeypatch.setattr(delete_service, "sys", SimpleNamespace(version_info=version)) + + delete_service._remove_tree("folder") + + assert rmtree.call_count == 1 + assert rmtree.call_args.args == ("folder",) + assert callback_name in rmtree.call_args.kwargs + + +def test_move_to_trash_uses_qt_trash_api(monkeypatch, tmp_path): + from PySide6.QtCore import QFile + + target = tmp_path / "recoverable.txt" + target.write_text("keep recoverable") + move = Mock(return_value=True) + monkeypatch.setattr(QFile, "moveToTrash", move) + + delete_service.move_to_trash(str(target)) + + move.assert_called_once_with(str(target)) + assert target.read_text() == "keep recoverable" + + +def test_move_to_trash_failure_does_not_fall_back_to_permanent_delete(monkeypatch, tmp_path): + from PySide6.QtCore import QFile + + target = tmp_path / "keep.txt" + target.write_text("keep") + monkeypatch.setattr(QFile, "moveToTrash", Mock(return_value=False)) + permanent_delete = Mock(side_effect=AssertionError("must not permanently delete")) + monkeypatch.setattr(delete_service.os, "remove", permanent_delete) + + with pytest.raises(OSError, match="Papierkorb"): + delete_service.move_to_trash(str(target)) + + permanent_delete.assert_not_called() + assert target.read_text() == "keep" + + +@pytest.mark.parametrize("kind", ["file", "directory", "broken"]) +@pytest.mark.parametrize("nested", [False, True]) +def test_symlink_deletes_only_link(tmp_path, kind, nested): + outside = tmp_path / "outside" + outside.mkdir() + sentinel = outside / "keep.txt" + sentinel.write_text("keep") + target = outside if kind == "directory" else sentinel + if kind == "broken": + target = outside / "missing" + selected = tmp_path / "selected" + selected.mkdir() + link = selected / "link" + try: + link.symlink_to(target, target_is_directory=kind == "directory") + except OSError as exc: + pytest.skip(f"Symlinks unavailable: {exc}") + delete_service.delete_path(str(selected if nested else link)) + assert not os.path.lexists(link) + assert sentinel.read_text() == "keep" + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows junction") +@pytest.mark.parametrize("nested", [False, True]) +def test_junction_deletes_only_link(tmp_path, nested): + outside = tmp_path / "outside" + outside.mkdir() + sentinel = outside / "keep.txt" + sentinel.write_text("keep") + selected = tmp_path / "selected" + selected.mkdir() + junction = selected / "junction" + result = subprocess.run( + ["cmd", "/c", "mklink", "/J", str(junction), str(outside)], + capture_output=True, text=True, + ) + assert result.returncode == 0, result.stderr + try: + delete_service.delete_path(str(selected if nested else junction)) + assert not os.path.lexists(junction) + assert sentinel.read_text() == "keep" + finally: + if os.path.lexists(junction): + os.rmdir(junction) diff --git a/tests/test_drive_capacity.py b/tests/test_drive_capacity.py new file mode 100644 index 0000000..ac9dbc7 --- /dev/null +++ b/tests/test_drive_capacity.py @@ -0,0 +1,249 @@ +import os +import threading +import time +from concurrent.futures import Future +from types import SimpleNamespace + +os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen') + +import pytest +from PySide6.QtCore import QDir, Qt +from PySide6.QtWidgets import QApplication +from PySide6.QtTest import QTest + +from core.drive_usage import DriveUsage, format_capacity, read_drive_usage +from gui.sidebar.drive_capacity import DriveCapacityWidget +from gui.sidebar.sidebar_main import TreePanel +from translator import get_translator + +app = QApplication.instance() or QApplication([]) + + +def wait_until(predicate): + deadline = time.monotonic() + 4 + while not predicate(): + assert time.monotonic() < deadline, 'Background result did not arrive' + app.processEvents() + QTest.qWait(5) + + +@pytest.fixture(autouse=True) +def german(): + tr = get_translator() + old = tr.get_language() + tr.set_language('de') + yield + tr.set_language(old) + + +@pytest.mark.parametrize('used,free,percent', [(0, 1000, 0), (1, 999, .1), (500, 500, 50), (1000, 0, 100)]) +def test_usage_values_and_bar(used, free, percent): + widget = DriveCapacityWidget('X:/') + widget.set_usage(DriveUsage(1000, used, free)) + assert widget.bar.value() == round(percent * 10) + assert widget.bar.format() == f'{percent:.1f}% belegt' + assert f'{free} B frei' in widget.details.text() + assert f'{used} B belegt' in widget.details.text() + assert '1000 B gesamt' in widget.details.text() + assert not widget.bar.isHidden() + assert widget.bar.format() in widget.accessibleDescription() + + +@pytest.mark.parametrize('values', [(0, 0, 0), (100, -1, 20), (100, 101, 0), (100, 30, 101)]) +def test_invalid_capacity_not_shown_as_real_data(values): + with pytest.raises(ValueError): + DriveUsage(*values) + + +def test_unavailable_discards_previous_measurement(): + widget = DriveCapacityWidget('X:/') + widget.set_usage(DriveUsage(100, 80, 20)) + widget.set_usage(None) + assert widget.details.text() == 'Speicherbelegung nicht verfügbar' + assert widget.bar.isHidden() + assert '80' not in widget.accessibleDescription() + + +def test_real_operating_system_capacity(tmp_path): + import shutil + expected = shutil.disk_usage(tmp_path) + actual = read_drive_usage(str(tmp_path)) + assert actual.total == expected.total + # Free space can change between OS calls; compare accounting and valid range. + assert actual.used + actual.free == actual.total if os.name == 'nt' else actual.used + actual.free <= actual.total + assert 0 <= actual.used_percent <= 100 + + +def test_binary_units(): + assert format_capacity(1024) == '1.0 KiB' + assert format_capacity(1024 ** 3) == '1.0 GiB' + + +def make_panel(monkeypatch, read): + monkeypatch.setattr(QDir, 'drives', lambda: [SimpleNamespace(absolutePath=lambda: 'X:/')]) + monkeypatch.setattr('gui.sidebar.sidebar_main.read_drive_usage', read) + return TreePanel() + + +def test_cancelled_drive_query_is_reported_as_unavailable(monkeypatch): + panel = make_panel(monkeypatch, lambda _: DriveUsage(100, 20, 80)) + try: + wait_until(lambda: not panel._usage_requests) + cancelled = Future() + cancelled.cancel() + panel._usage_requests['X:/'] = cancelled + + panel._collect_drive_usage() + + assert not panel._usage_requests + assert panel._drive_rows['X:/'][1].details.text() == 'Speicherbelegung nicht verfügbar' + finally: + panel.close() + + +def test_slow_query_keeps_gui_responsive_and_deduplicates(monkeypatch): + entered, release = threading.Event(), threading.Event() + threads = [] + main_thread = threading.get_ident() + + def read(path): + threads.append(threading.get_ident()) + entered.set() + assert release.wait(4) + return DriveUsage(1000, 250, 750) + + panel = make_panel(monkeypatch, read) + try: + wait_until(entered.is_set) + for _ in range(5): + panel.refresh_drive_usage() + app.processEvents() + assert len(threads) == 1 + assert threads[0] != main_thread + assert panel._drive_rows['X:/'][1].bar.isHidden() + release.set() + wait_until(lambda: not panel._usage_requests) + assert panel._drive_rows['X:/'][1].bar.value() == 250 + panel.refresh_drives_button.click() + wait_until(lambda: not panel._usage_requests) + assert len(threads) == 2 + finally: + release.set() + for future in panel._usage_requests.values(): + future.result(timeout=4) + panel.close() + + +def test_failed_query_and_recovery(monkeypatch): + def unavailable(path): + raise OSError('Device not ready') + + panel = make_panel(monkeypatch, unavailable) + wait_until(lambda: not panel._usage_requests) + item, widget = panel._drive_rows['X:/'] + assert item.text(0) == '' # embedded title must not overlap tree text + assert item.data(0, Qt.ItemDataRole.AccessibleTextRole) == 'X:/' + assert 'nicht verfügbar' in widget.details.text() + monkeypatch.setattr('gui.sidebar.sidebar_main.read_drive_usage', lambda path: DriveUsage(100, 100, 0)) + panel.refresh_drive_usage() + wait_until(lambda: not panel._usage_requests) + assert widget.bar.value() == 1000 + assert item.data(0, Qt.ItemDataRole.UserRole) == 'X:/' + selected = [] + panel.folder_selected.connect(selected.append) + panel._on_item_clicked(item, 0) + assert selected == ['X:/'] + panel.close() + + +def test_panel_destroyed_while_query_is_running(monkeypatch): + from PySide6.QtCore import QCoreApplication, QEvent, QObject + from shiboken6 import isValid + entered, release = threading.Event(), threading.Event() + + def read(path): + entered.set() + assert release.wait(4) + return DriveUsage(100, 20, 80) + + panel = make_panel(monkeypatch, read) + futures = list(panel._usage_requests.values()) + unrelated = None + try: + wait_until(entered.is_set) + unrelated = QObject() + unrelated.deleteLater() + panel.deleteLater() + QCoreApplication.sendPostedEvents(panel, QEvent.Type.DeferredDelete) + assert isValid(unrelated), 'Panel cleanup must not delete unrelated Qt objects' + finally: + release.set() + for future in futures: + future.result(timeout=4) + if unrelated is not None and isValid(unrelated): + QCoreApplication.sendPostedEvents(unrelated, QEvent.Type.DeferredDelete) + app.processEvents() + + +def test_capacity_text_placeholders_in_every_language(): + tr = get_translator() + for lang in tr.get_supported_languages(): + tr.set_language(lang) + widget = DriveCapacityWidget('X:/') + widget.set_usage(DriveUsage(1000, 500, 500)) + assert '500 B' in widget.details.text() and '1000 B' in widget.details.text() + assert '{' not in widget.details.text() + assert '50.0' in widget.bar.format() + + +def test_narrow_sidebar_preserves_wrapped_capacity_text(monkeypatch): + panel = make_panel(monkeypatch, lambda path: DriveUsage(1024 ** 4, 1024 ** 3, 1024 ** 4 - 1024 ** 3)) + panel.resize(200, 600) + panel.show() + wait_until(lambda: not panel._usage_requests) + app.processEvents() + item, widget = panel._drive_rows['X:/'] + required = widget.layout().totalHeightForWidth(widget.width()) + assert item.sizeHint(0).height() >= required + assert widget.height() >= required + panel.close() + + +def test_pending_capacity_job_finishes_before_interpreter_exit(): + import subprocess + import sys + from pathlib import Path + root = Path(__file__).resolve().parents[1] + script = ''' +import sys, time +sys.path.insert(0, 'src') +from PySide6.QtWidgets import QApplication +from core.drive_usage import DriveUsage +from gui.sidebar import drive_capacity +app = QApplication([]) +def slow_read(path): + time.sleep(.2) + print('capacity-job-finished', flush=True) + return DriveUsage(100, 20, 80) +drive_capacity.capacity_executor().submit(slow_read, 'X:/') +''' + result = subprocess.run([sys.executable, '-c', script], cwd=root, capture_output=True, text=True, timeout=20) + assert result.returncode == 0, result.stderr + assert 'capacity-job-finished' in result.stdout + + +def test_destroyed_panel_cancels_queued_layout_refresh(monkeypatch, capsys): + from PySide6.QtCore import QCoreApplication, QEvent + from shiboken6 import isValid + panel = make_panel(monkeypatch, lambda path: DriveUsage(100, 20, 80)) + wait_until(lambda: not panel._usage_requests) + panel.resize(200, 600) + panel.show() + timer = panel._resize_timer + timer.start(0) + assert timer.isActive() + panel.deleteLater() + QCoreApplication.sendPostedEvents(panel, QEvent.Type.DeferredDelete) + assert not isValid(timer) + app.processEvents() + assert 'RuntimeError' not in capsys.readouterr().err diff --git a/tests/test_drive_query_process.py b/tests/test_drive_query_process.py new file mode 100644 index 0000000..86b8d2a --- /dev/null +++ b/tests/test_drive_query_process.py @@ -0,0 +1,100 @@ +"""Exercise real helper termination rather than substituting completed futures.""" + +import subprocess +import sys +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import pytest + +from core import drive_usage +from gui.sidebar import drive_capacity + + +@pytest.fixture(autouse=True) +def query_lifecycle(): + drive_usage.start_drive_queries() + yield + drive_capacity.shutdown_capacity_executor() + drive_usage.stop_drive_queries() + + +def sleeper_command(path, result_path): + return [sys.executable, '-c', 'import time; time.sleep(60)'] + + +def test_real_helper_matches_os_capacity(tmp_path): + expected = drive_usage.read_drive_usage(str(tmp_path)) + actual = drive_usage.read_drive_usage_bounded(str(tmp_path)) + assert actual.total == expected.total + assert 0 <= actual.used_percent <= 100 + assert not drive_usage._query_processes + + +def test_timed_out_helper_is_reaped(monkeypatch): + monkeypatch.setattr(drive_usage, '_query_command', sleeper_command) + monkeypatch.setattr(drive_usage, 'QUERY_TIMEOUT_SECONDS', .2) + processes = [] + real_popen = subprocess.Popen + + def record(*args, **kwargs): + process = real_popen(*args, **kwargs) + processes.append(process) + return process + + monkeypatch.setattr(drive_usage.subprocess, 'Popen', record) + with pytest.raises(OSError, match='timed out'): + drive_usage.read_drive_usage_bounded('blocked') + assert processes[0].poll() is not None + assert not drive_usage._query_processes + + +def test_four_stalled_reads_release_slot_for_healthy_fifth(monkeypatch, tmp_path): + real_command = drive_usage._query_command + monkeypatch.setattr(drive_usage, '_query_command', + lambda path, result: sleeper_command(path, result) if path == 'blocked' else real_command(path, result)) + monkeypatch.setattr(drive_usage, 'QUERY_TIMEOUT_SECONDS', 1) + with ThreadPoolExecutor(max_workers=4) as executor: + stalled = [executor.submit(drive_usage.read_drive_usage_bounded, 'blocked') for _ in range(4)] + healthy = executor.submit(drive_usage.read_drive_usage_bounded, str(tmp_path)) + assert healthy.result(timeout=6).total > 0 + for future in stalled: + with pytest.raises(OSError, match='timed out'): + future.result() + assert not drive_usage._query_processes + + +def test_shutdown_kills_running_helpers_and_cancels_queued(monkeypatch): + monkeypatch.setattr(drive_usage, '_query_command', sleeper_command) + monkeypatch.setattr(drive_usage, 'QUERY_TIMEOUT_SECONDS', 60) + executor = drive_capacity.capacity_executor() + futures = [executor.submit(drive_usage.read_drive_usage_bounded, 'blocked') for _ in range(8)] + deadline = time.monotonic() + 5 + while len(drive_usage._query_processes) != 4: + assert time.monotonic() < deadline + time.sleep(.01) + processes = tuple(drive_usage._query_processes) + started = time.monotonic() + drive_capacity.shutdown_capacity_executor() + assert time.monotonic() - started < 3 + assert all(process.poll() is not None for process in processes) + assert all(future.done() for future in futures) + assert not drive_usage._query_processes + + +def test_packaged_entrypoint_dispatches_before_qt(tmp_path): + root = Path(__file__).resolve().parents[1] + output = tmp_path / 'result.json' + result = subprocess.run([sys.executable, str(root / 'src/main.py'), + '--drive-capacity-query', str(tmp_path), str(output)], + capture_output=True, text=True, timeout=10) + assert result.returncode == 0, result.stderr + assert '"total"' in output.read_text(encoding='utf-8') + + +def test_helper_works_without_stdout(monkeypatch, tmp_path): + monkeypatch.setattr(sys, 'stdout', None) + output = tmp_path / 'result.json' + assert drive_usage.capacity_query_main(str(tmp_path), str(output)) == 0 + assert '"total"' in output.read_text(encoding='utf-8') diff --git a/tests/test_file_browser.py b/tests/test_file_browser.py index 2be2b0c..3da0658 100644 --- a/tests/test_file_browser.py +++ b/tests/test_file_browser.py @@ -32,6 +32,19 @@ def test_dnd_flags_enabled(): assert browser.table.dragEnabled(), "table.dragEnabled() muss True sein" +def test_browser_destruction_disposes_its_models(): + from PySide6.QtCore import QCoreApplication, QEvent + from shiboken6 import isValid + _ensure_app() + browser = FileBrowser() + model, proxy = browser.model, browser.proxy + browser.deleteLater() + QCoreApplication.sendPostedEvents(browser, QEvent.Type.DeferredDelete) + assert not isValid(browser) + assert not isValid(model), 'Filesystem gatherer must not outlive its browser' + assert not isValid(proxy), 'Sort model must not outlive its browser' + + def test_do_file_drop_copies_file(tmp_path): """_do_file_drop kopiert eine Quelldatei ohne das Original zu löschen.""" _ensure_app() diff --git a/tests/test_gui_gc.py b/tests/test_gui_gc.py new file mode 100644 index 0000000..eb0e946 --- /dev/null +++ b/tests/test_gui_gc.py @@ -0,0 +1,241 @@ +"""Native process regressions for collection and worker shutdown.""" + +import os +import subprocess +import sys +from pathlib import Path + +import pytest + + +@pytest.mark.parametrize("automatic_before", [True, False]) +@pytest.mark.parametrize("use_timer", [True, False]) +def test_qt_cycles_are_destroyed_on_gui_thread(automatic_before, use_timer): + script = f''' +import gc, sys, threading +sys.path.insert(0, 'src') +from PySide6.QtCore import QObject +from PySide6.QtTest import QTest +from PySide6.QtWidgets import QApplication +from core.gui_gc import install_gui_gc +app = QApplication([]) +gc.enable() if {automatic_before!r} else gc.disable() +collector = install_gui_gc(app) +assert collector is install_gui_gc(app) +assert not gc.isenabled() +gui_thread = threading.get_ident() +deleted = [] +qt_deleted = [] +errors = [] +class Probe(QObject): + def __init__(self): + super().__init__() + self.cycle = self + self.destroyed.connect(lambda: qt_deleted.append(threading.get_ident())) + def __del__(self): + deleted.append(threading.get_ident()) +probe = Probe() +del probe +def allocate(): + for _ in range(10000): + cycle = [] + cycle.append(cycle) + for operation in (collector.collect, lambda: install_gui_gc(QApplication.instance()).close()): + try: + operation() + except RuntimeError: + errors.append('wrong-thread-rejected') +thread = threading.Thread(target=allocate) +thread.start() +thread.join(timeout=10) +assert not thread.is_alive() +assert errors == ['wrong-thread-rejected', 'wrong-thread-rejected'] +assert deleted == [] and qt_deleted == [], (deleted, qt_deleted) +if {use_timer!r}: + collector.timer.setInterval(1) + collector.timer.start() + QTest.qWait(100) +else: + collector.collect() +assert deleted == [gui_thread], deleted +assert qt_deleted == [gui_thread], qt_deleted +collector.close() +collector.close() +assert gc.isenabled() is {automatic_before!r} +assert not collector.timer.isActive() +print('gui-collection-ok') +''' + result = subprocess.run( + [sys.executable, "-c", script], cwd=Path(__file__).resolve().parents[1], + capture_output=True, text=True, timeout=25, + ) + assert result.returncode == 0, result.stderr + assert "gui-collection-ok" in result.stdout + + +def _isolated_qt_env(tmp_path): + profile = tmp_path / 'profile' + directories = { + 'USERPROFILE': profile, + 'HOME': profile, + 'APPDATA': profile / 'AppData' / 'Roaming', + 'LOCALAPPDATA': profile / 'AppData' / 'Local', + 'XDG_CONFIG_HOME': profile / '.config', + 'XDG_CACHE_HOME': profile / '.cache', + } + for directory in directories.values(): + directory.mkdir(parents=True, exist_ok=True) + env = os.environ.copy() + env.update({key: str(value) for key, value in directories.items()}) + env['QT_QPA_PLATFORM'] = 'offscreen' + return env + + +@pytest.mark.parametrize('automatic_before', [True, False]) +def test_close_restores_gc_policy_when_collection_fails(tmp_path, automatic_before): + script = f''' +import gc, sys +sys.path.insert(0, 'src') +from PySide6.QtWidgets import QApplication +from core.gui_gc import install_gui_gc +app = QApplication([]) +gc.enable() if {automatic_before!r} else gc.disable() +collector = install_gui_gc(app) +def fail_collect(): + raise RuntimeError('collect-failed') +collector.collect = fail_collect +try: + collector.close() +except RuntimeError as error: + assert str(error) == 'collect-failed' +else: + raise AssertionError('collection failure was swallowed') +assert collector._closed +assert gc.isenabled() is {automatic_before!r} +collector.close() +assert gc.isenabled() is {automatic_before!r} +print('gc-close-error-restored') +''' + result = subprocess.run( + [sys.executable, '-c', script], + cwd=Path(__file__).resolve().parents[1], + env=_isolated_qt_env(tmp_path), + capture_output=True, + text=True, + timeout=25, + ) + assert result.returncode == 0, result.stderr + assert 'gc-close-error-restored' in result.stdout + + +def test_install_start_failure_keeps_existing_gc_policy(tmp_path): + script = ''' +import gc, sys +sys.path.insert(0, 'src') +from PySide6.QtWidgets import QApplication +import core.gui_gc as gui_gc +app = QApplication([]) +class Signal: + def connect(self, callback): + self.callback = callback +class FailingTimer: + def __init__(self, parent): + self.timeout = Signal() + def setInterval(self, interval): + self.interval = interval + def start(self): + raise RuntimeError('timer-start-failed') +gui_gc.QTimer = FailingTimer +gc.enable() +try: + gui_gc.install_gui_gc(app) +except RuntimeError as error: + assert str(error) == 'timer-start-failed' +else: + raise AssertionError('timer start failure was swallowed') +assert gc.isenabled(), 'GC policy changed before timer start succeeded' +print('gc-install-start-error-restored') +''' + result = subprocess.run( + [sys.executable, '-c', script], + cwd=Path(__file__).resolve().parents[1], + env=_isolated_qt_env(tmp_path), + capture_output=True, + text=True, + timeout=25, + ) + assert result.returncode == 0, result.stderr + assert 'gc-install-start-error-restored' in result.stdout + + +def test_main_closes_gc_collector_when_application_setup_fails(tmp_path): + script = ''' +import sys, types +sys.path.insert(0, 'src') +import main +sys.argv = [] +closed = [] +class Collector: + def close(self): + closed.append(True) +collector = Collector() +main.install_gui_gc = lambda app: collector +class Translator: + def get_language(self): + return 'en' +module = types.ModuleType('translator') +module.get_translator = lambda: Translator() +sys.modules['translator'] = module +def fail_setup(app, language): + raise RuntimeError('application-setup-failed') +main.install_qt_translations = fail_setup +try: + main.main() +except RuntimeError as error: + assert str(error) == 'application-setup-failed' +else: + raise AssertionError('setup failure was swallowed') +assert closed == [True], closed +print('main-setup-close-restored') +''' + result = subprocess.run( + [sys.executable, '-c', script], + cwd=Path(__file__).resolve().parents[1], + env=_isolated_qt_env(tmp_path), + capture_output=True, + text=True, + timeout=25, + ) + assert result.returncode == 0, result.stderr + assert 'main-setup-close-restored' in result.stdout + + +def test_capacity_shutdown_drains_workers_before_restoring_gc(): + script = ''' +import gc, sys, threading, time +sys.path.insert(0, 'src') +from PySide6.QtWidgets import QApplication +from core.gui_gc import install_gui_gc +from gui.sidebar.drive_capacity import capacity_executor, shutdown_capacity_executor +app = QApplication([]) +collector = install_gui_gc(app) +entered = threading.Event() +def worker(): + entered.set() + time.sleep(.1) + assert not gc.isenabled() + return 'finished' +future = capacity_executor().submit(worker) +assert entered.wait(5) +shutdown_capacity_executor() +assert future.done() and future.result() == 'finished' +collector.close() +assert gc.isenabled() +print('shutdown-ok') +''' + result = subprocess.run( + [sys.executable, "-c", script], cwd=Path(__file__).resolve().parents[1], + capture_output=True, text=True, timeout=25, + ) + assert result.returncode == 0, result.stderr + assert "shutdown-ok" in result.stdout diff --git a/tests/test_properties_and_terminal.py b/tests/test_properties_and_terminal.py index 38a4054..b04695c 100644 --- a/tests/test_properties_and_terminal.py +++ b/tests/test_properties_and_terminal.py @@ -77,6 +77,27 @@ def test_get_terminal_command_windows(self): assert isinstance(cmd, list) assert any("cmd" in c or "powershell" in c or "wt" in c for c in cmd) + def test_windows_terminal_does_not_receive_directory_as_command_text(self): + directory = r"C:\Users\test\a; new-tab -p PowerShell" + with patch("sys.platform", "win32"), patch( + "shutil.which", + side_effect=lambda executable: "wt.exe" if executable == "wt" else None, + ): + assert get_terminal_command(directory) == ["wt", "-d", "."] + + def test_windows_powershell_fallback_does_not_interpolate_directory(self): + directory = r"C:\Users\test\x'; Start-Process calc; '" + with patch("sys.platform", "win32"), patch( + "shutil.which", + side_effect=lambda executable: "powershell.exe" if executable == "powershell" else None, + ): + assert get_terminal_command(directory) == ["powershell", "-NoExit"] + + def test_windows_cmd_fallback_does_not_interpolate_directory(self): + directory = r"C:\Users\test\a&calc" + with patch("sys.platform", "win32"), patch("shutil.which", return_value=None): + assert get_terminal_command(directory) == ["cmd", "/K"] + def test_get_terminal_command_darwin(self): with patch("sys.platform", "darwin"): cmd = get_terminal_command("/Users/test") @@ -96,6 +117,34 @@ def test_open_terminal_in_directory_launches_process(self, tmp_path): args, kwargs = mock_popen.call_args assert kwargs.get("cwd") == str(target) + def test_open_terminal_uses_cwd_for_windows_metacharacter_path(self, tmp_path): + target = tmp_path / "x'&calc" + target.mkdir() + with ( + patch("sys.platform", "win32"), + patch("shutil.which", return_value=None), + patch("subprocess.Popen") as mock_popen, + ): + open_terminal_in_directory(str(target)) + + args, kwargs = mock_popen.call_args + assert args[0] == ["cmd", "/K"] + assert kwargs["cwd"] == str(target) + + def test_open_terminal_uses_cwd_with_windows_terminal(self, tmp_path): + target = tmp_path / "a; new-tab -p PowerShell" + target.mkdir() + with ( + patch("sys.platform", "win32"), + patch("shutil.which", side_effect=lambda executable: "wt.exe" if executable == "wt" else None), + patch("subprocess.Popen") as mock_popen, + ): + open_terminal_in_directory(str(target)) + + args, kwargs = mock_popen.call_args + assert args[0] == ["wt", "-d", "."] + assert kwargs["cwd"] == str(target) + class TestCopyPathOperations: """Pfad-Kopier-Funktionen in FileBrowser.""" diff --git a/tests/test_store_regressions_20260928.py b/tests/test_store_regressions_20260928.py index a7fe322..39b834e 100644 --- a/tests/test_store_regressions_20260928.py +++ b/tests/test_store_regressions_20260928.py @@ -11,6 +11,7 @@ from PySide6.QtCore import QItemSelectionModel from PySide6.QtGui import QContextMenuEvent from PySide6.QtWidgets import QApplication, QInputDialog, QMenu +from PySide6.QtTest import QTest import gui.browser.file_browser as browser_module from gui.browser.file_browser import FileBrowser @@ -102,11 +103,11 @@ def test_navigation_loads_files_and_selection_reaches_preview(tmp_path): browser.resize(800, 500) browser.show() browser.navigate_to(str(tmp_path)) - for _ in range(100): - _app().processEvents() - if browser.proxy.rowCount(browser.table.rootIndex()) > 0: - break - time.sleep(.01) + # QFileSystemModel gathers rows asynchronously. Pump the actual Qt event + # loop with a bounded deadline instead of assuming a one-second OS scan. + deadline = time.monotonic() + 5 + while browser.proxy.rowCount(browser.table.rootIndex()) == 0 and time.monotonic() < deadline: + QTest.qWait(10) root = browser.proxy.mapToSource(browser.table.rootIndex()) assert Path(browser.model.filePath(root)) == tmp_path assert browser.proxy.rowCount(browser.table.rootIndex()) > 0