diff --git a/CHANGELOG.md b/CHANGELOG.md index acce328..cdf1ea9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,16 @@ All notable changes to this project will be documented in this file. ## [Unreleased] - 2026-07-28 +- Ported the launcher-catalog reconciliation routine into the Plan-D repository + as `scripts/softwarecenter_sync.py`. Catalog, registry, software root, and + local development root are now explicit CLI inputs instead of being derived + from the script location. +- The catalog runtime remains dry-run by default, retains both QSettings + profiles and `SUPPRESSED` semantics, starts no shell, and gates `--apply` + behind process readback plus a local profile/registry backup. +- Added regression coverage and a pinned-runtime scheduler migration/rollback + contract in `RUNTIME_DAILY_CARE.md`. No live scheduler or Windows Task is + changed by this source update. - The Windows build entrypoint now derives the release artifact directory and filename from `project.version` in `pyproject.toml`. Current builds therefore land under `releases/v1.2.0/SoftwareCenter-1.2.0-win64.exe` instead of diff --git a/README.md b/README.md index b940563..691dc74 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # SoftwareCenter [![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/) -[![Pytest 126 Passed](https://img.shields.io/badge/pytest-126%20passed-brightgreen.svg)](https://docs.pytest.org/) +[![Pytest 161 Passed](https://img.shields.io/badge/pytest-161%20passed-brightgreen.svg)](https://docs.pytest.org/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) [![PySide6](https://img.shields.io/badge/GUI-PySide6-green.svg)](https://doc.qt.io/qtforpython-6/) [![LLM Indexing Ready](https://img.shields.io/badge/LLM-Ready-blueviolet.svg)](llms.txt) @@ -132,6 +132,14 @@ GitHub Actions runs these smoke checks. The macOS smoke validates `.app` import, For the Windows Store path, `python scripts/run_windows_wack.py --dry-run` checks the local MSIX/AppCert paths and prints the reproducible WACK command. The real certification run should be executed from an elevated PowerShell against a fresh signed MSIX before submission. `python generate_store_screenshots.py` creates the reproducible Store screenshot set under `README/screenshots/store/`. +## Headless Launcher Catalog Care + +The optional catalog reconciler is provided as Plan-D runtime code in +`scripts/softwarecenter_sync.py`. Synchronized catalog and registry files are +explicit CLI inputs; the command is read-only unless `--apply` is supplied. +The pinned-runtime scheduler payload, apply gates, native readback, and rollback +procedure are documented in [RUNTIME_DAILY_CARE.md](RUNTIME_DAILY_CARE.md). + ## Exchange Format Profiles can be exported as `softwarecenter-profile-v1.json` and imported again later. The format carries tabs, view modes, and entries with `label`, `path`, `kind`, and optional `notes`, but does not copy local files or credentials. See [EXPORTFORMAT.md](EXPORTFORMAT.md) for details. diff --git a/README_de.md b/README_de.md index 98f0d7a..69abc87 100644 --- a/README_de.md +++ b/README_de.md @@ -1,7 +1,7 @@ # SoftwareCenter [![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](https://www.python.org/) -[![Pytest 126 Passed](https://img.shields.io/badge/pytest-126%20passed-brightgreen.svg)](https://docs.pytest.org/) +[![Pytest 161 Passed](https://img.shields.io/badge/pytest-161%20passed-brightgreen.svg)](https://docs.pytest.org/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) [![PySide6](https://img.shields.io/badge/GUI-PySide6-green.svg)](https://doc.qt.io/qtforpython-6/) [![LLM Indexing Ready](https://img.shields.io/badge/LLM-Ready-blueviolet.svg)](llms.txt) @@ -133,6 +133,14 @@ Die GitHub Actions führen diese Smoke-Checks ebenfalls aus; der macOS-Job prüf Für den Windows-Store-Pfad prüft `python scripts/run_windows_wack.py --dry-run` die lokalen MSIX/AppCert-Pfade und gibt den reproduzierbaren WACK-Befehl aus. Der eigentliche Zertifizierungslauf sollte aus einer erhöhten PowerShell gegen ein frisch signiertes MSIX vor der Einreichung ausgeführt werden. +## Headless-Katalogpflege + +Die optionale Katalogroutine liegt als Plan-D-Runtime-Code unter +`scripts/softwarecenter_sync.py`. Synchronisierte Katalog- und Registry-Dateien +werden als explizite CLI-Eingaben übergeben; ohne `--apply` bleibt der Lauf +read-only. Der gepinnte Scheduler-Payload, Apply-Gates, native Readbacks und das +Rollback stehen in [RUNTIME_DAILY_CARE.md](RUNTIME_DAILY_CARE.md). + ## Austauschformat Profile lassen sich als `softwarecenter-profile-v1.json` exportieren und wieder importieren. Das Format enthält Tabs, Ansichtsmodus und Einträge mit `label`, `path`, `kind` und optionalen `notes`, aber keine kopierten Dateien und keine Credentials. Details stehen in [EXPORTFORMAT.md](EXPORTFORMAT.md). diff --git a/RUNTIME_DAILY_CARE.md b/RUNTIME_DAILY_CARE.md new file mode 100644 index 0000000..f1fb38f --- /dev/null +++ b/RUNTIME_DAILY_CARE.md @@ -0,0 +1,203 @@ +# Launcher catalog daily-care runtime + +This document defines the code-side contract for migrating +`softwarecenter.launchboards.daily-care` away from executable code stored in +OneDrive. It does not authorize or perform a scheduler, Windows Task, profile, +or registry mutation. + +## Boundary + +- Runtime code: a reviewed SoftwareCenter commit materialized below + `C:\_Local_DEV\runtime\SoftwareCenter\`. +- Synchronized inputs: `software_apps.json` and `DESKTOP-REGISTRY.txt`, passed + as exact command-line arguments. +- Local profile state: the two QSettings namespaces `SoftwareCenter` and + `LaunchBoards`. +- Backups: local Plan-D storage only, never the synchronized software root. +- Default: dry-run. The scheduler payload below intentionally omits `--apply`. + +The runtime never derives synchronized data paths from `__file__`, never +invokes a shell, and never copies catalog or registry data into the repository. + +## CLI contract + +```powershell +$python = Join-Path $env:LOCALAPPDATA 'Programs\Python\Python312\python.exe' +$oneDrive = Join-Path $env:USERPROFILE 'OneDrive' +$runtimeRoot = 'C:\_Local_DEV\runtime\SoftwareCenter\' +$script = Join-Path $runtimeRoot 'scripts\softwarecenter_sync.py' + +& $python $script ` + --catalog (Join-Path $oneDrive '.TOPICS\.SOFTWARE\_tools\software_apps.json') ` + --registry (Join-Path $oneDrive 'Desktop\DESKTOP-REGISTRY.txt') ` + --software-root (Join-Path $oneDrive '.TOPICS\.SOFTWARE') ` + --local-dev-root 'C:\_Local_DEV' +``` + +Exit codes: + +- `0`: valid dry-run or completed apply. +- `2`: missing/invalid input or a failed apply gate. +- `3`: write operation failed after a backup was created. + +An apply run additionally requires `--apply`. It aborts while either managed +GUI is running and creates a local `launcher-catalog-profile-backup-v2` before +the first profile or registry write. The backup contains both original +QSettings projections and the exact original registry text plus SHA-256. If a +write fails after the backup, the runtime prints its exact path and immediately +attempts to restore both profiles and the registry. An incomplete rollback is +reported explicitly and still exits `3`. The registry bytes are captured once +immediately before the write phase and reused unchanged for both backup and +rollback; if that final read is unavailable, the apply exits before any write. + +## Pinned runtime materialization + +Only use a reviewed commit that is reachable from the intended release branch. +The target directory must not already contain unrelated files. + +```powershell +$repo = 'C:\_Local_DEV\repos\SoftwareCenter' +$approvedCommit = '' +$releaseRef = 'origin/master' +$runtimeRoot = "C:\_Local_DEV\runtime\SoftwareCenter\$approvedCommit" +$archive = Join-Path $env:TEMP "softwarecenter-$approvedCommit.tar" + +git -C $repo cat-file -e "$approvedCommit^{commit}" +git -C $repo merge-base --is-ancestor $approvedCommit $releaseRef +if ($LASTEXITCODE -ne 0) { + throw "$approvedCommit is not reachable from $releaseRef" +} +if (Test-Path -LiteralPath $runtimeRoot) { + throw "Runtime target already exists: $runtimeRoot" +} +if (Test-Path -LiteralPath $archive) { + throw "Temporary archive already exists: $archive" +} +New-Item -ItemType Directory -Path $runtimeRoot +git -C $repo archive --format=tar --output=$archive $approvedCommit +tar -xf $archive -C $runtimeRoot +Remove-Item -LiteralPath $archive +git -C $repo show -s --format='%H' $approvedCommit +``` + +Record the commit, runtime path, interpreter path, input hashes, and dry-run +output in the native migration receipt. Do not use a mutable OneDrive worktree +or an unpinned conflict copy. + +## Proposed native scheduler migration + +`ellmos-scheduler` 0.1.0 has no in-place payload-update command. Preserve the +old job as a disabled rollback target and introduce a separately identifiable +candidate. Run these commands only after: + +1. the pinned runtime dry-run exits `0`; +2. current `jobs --json`, `status --json`, and recent runs are captured; +3. the existing job has a recent native success receipt or its degraded + baseline is explicitly recorded; +4. the SoftwareCenter and LaunchBoards processes are absent; +5. a byte-for-byte scheduler database backup exists outside OneDrive. + +```powershell +$db = Join-Path $env:LOCALAPPDATA 'ellmos\scheduler.db' +$python = Join-Path $env:LOCALAPPDATA 'Programs\Python\Python312\python.exe' +$oneDrive = Join-Path $env:USERPROFILE 'OneDrive' +$approvedCommit = '' +$runtimeRoot = "C:\_Local_DEV\runtime\SoftwareCenter\$approvedCommit" +$script = Join-Path $runtimeRoot 'scripts\softwarecenter_sync.py' + +$argv = @( + $python, + $script, + '--catalog', + (Join-Path $oneDrive '.TOPICS\.SOFTWARE\_tools\software_apps.json'), + '--registry', + (Join-Path $oneDrive 'Desktop\DESKTOP-REGISTRY.txt'), + '--software-root', + (Join-Path $oneDrive '.TOPICS\.SOFTWARE'), + '--local-dev-root', + 'C:\_Local_DEV' +) +$payload = @{ + argv = $argv + cwd = $runtimeRoot +} | ConvertTo-Json -Compress +$schedule = @{ + kind = 'daily' + time = '06:20' + timezone = 'Europe/Berlin' +} | ConvertTo-Json -Compress + +ellmos-scheduler --db $db jobs --json +ellmos-scheduler --db $db status --json +ellmos-scheduler --db $db add ` + --id softwarecenter.launchboards.daily-care.v4 ` + --schedule $schedule ` + --executor command ` + --payload $payload ` + --lease-seconds 300 ` + --timeout-seconds 120 +ellmos-scheduler --db $db jobs --json +ellmos-scheduler --db $db disable softwarecenter.launchboards.daily-care +ellmos-scheduler --db $db jobs --json +``` + +Adding the candidate before disabling the old job keeps the rollback target +available if candidate creation fails. If the final disable or its readback +fails, immediately disable the `.v4` candidate before leaving the system. + +The existing Windows Task continues to run one native scheduler tick; its +action does not need to point at this script. Do not alter that task unless its +own native readback proves a separate defect. + +The first candidate execution must be observed through native scheduler run +history. The receipt must prove: + +- `executor=command`; +- argv contains the pinned runtime and all four explicit paths; +- `cwd` is the pinned runtime, not OneDrive; +- exit code `0`; +- output says `DRY-RUN`; +- input files, QSettings profiles, and registry remain unchanged. + +## Rollback + +If the candidate must be rolled back before its first run: + +```powershell +$db = Join-Path $env:LOCALAPPDATA 'ellmos\scheduler.db' +ellmos-scheduler --db $db disable softwarecenter.launchboards.daily-care.v4 +ellmos-scheduler --db $db enable softwarecenter.launchboards.daily-care +ellmos-scheduler --db $db jobs --json +``` + +After any apply run, also restore the registry text and both profile snapshots +from the reported `launcher-catalog-profile-backup-v2` file. The explicit +restore gate uses the same exact runtime/input bindings: + +```powershell +$python = Join-Path $env:LOCALAPPDATA 'Programs\Python\Python312\python.exe' +$oneDrive = Join-Path $env:USERPROFILE 'OneDrive' +$runtimeRoot = 'C:\_Local_DEV\runtime\SoftwareCenter\' +$script = Join-Path $runtimeRoot 'scripts\softwarecenter_sync.py' +$backup = 'C:\_Local_DEV\launcher_catalog_backups\.json' + +& $python $script ` + --catalog (Join-Path $oneDrive '.TOPICS\.SOFTWARE\_tools\software_apps.json') ` + --registry (Join-Path $oneDrive 'Desktop\DESKTOP-REGISTRY.txt') ` + --software-root (Join-Path $oneDrive '.TOPICS\.SOFTWARE') ` + --local-dev-root 'C:\_Local_DEV' ` + --apply ` + --restore-backup $backup +``` + +The restore command validates the backup format, registry binding and SHA-256, +then snapshots the current state before restoring both QSettings profiles and +the registry. If the restore itself fails after its first write, it compensates +back to that pre-restore snapshot and reports whether the compensation was +complete. Verify the registry hash and QSettings readback afterwards. Never +overwrite a newer user edit without first comparing it to the backup and +obtaining the applicable decision. + +The candidate job, pinned runtime, and database backup remain in place until a +successful run and rollback rehearsal have both been recorded. Cleanup is a +separate, explicit operation. diff --git a/scripts/softwarecenter_sync.py b/scripts/softwarecenter_sync.py new file mode 100644 index 0000000..bbf89f5 --- /dev/null +++ b/scripts/softwarecenter_sync.py @@ -0,0 +1,861 @@ +# -*- coding: utf-8 -*- +"""Synchronize the portable launcher catalog into SoftwareCenter profiles. + +The executable code lives in the Plan-D repository. Synchronized catalog and +registry files remain data inputs and must be supplied explicitly on every +invocation. The default mode is read-only; ``--apply`` is the only write gate. +""" +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import os +import re +import subprocess +import tempfile +from collections.abc import Callable, Iterable, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path + +from PySide6.QtCore import QSettings + + +ORG = "LukasGeiger" +PROFILES = ("SoftwareCenter", "LaunchBoards") +VALID_STATES = {"PENDING", "ACTIVE", "SUPPRESSED"} +META_KEY = "launcher_catalog/managed_paths_json" +PROFILE_IMAGE_PATTERNS = tuple( + re.compile( + rf"^{re.escape(profile)}" + r"(?:-[0-9][A-Za-z0-9._-]*-win64)?\.exe$", + re.IGNORECASE, + ) + for profile in PROFILES +) + + +@dataclass(frozen=True) +class RuntimeConfig: + catalog: Path + registry: Path + software_root: Path + local_dev_root: Path + backup_dir: Path + apply: bool + restore_backup: Path | None = None + + +def _absolute(path: Path) -> Path: + return Path(os.path.abspath(os.path.expandvars(str(path.expanduser())))) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Synchronizes the portable launcher catalog (dry-run by default)." + ) + parser.add_argument( + "--catalog", + type=Path, + required=True, + help="Exact path to the software-launcher-catalog-v2 JSON file.", + ) + parser.add_argument( + "--registry", + type=Path, + required=True, + help="Exact path to DESKTOP-REGISTRY.txt.", + ) + parser.add_argument( + "--software-root", + type=Path, + required=True, + help="Root used to resolve $SOFTWARE_ROOT launcher templates.", + ) + parser.add_argument( + "--local-dev-root", + type=Path, + required=True, + help="Root used to resolve $LOCAL_DEV_ROOT launcher templates.", + ) + parser.add_argument( + "--backup-dir", + type=Path, + help="Local backup directory (default: /launcher_catalog_backups).", + ) + parser.add_argument( + "--apply", + action="store_true", + help="Write both QSettings profiles and registry state after all gates pass.", + ) + parser.add_argument( + "--restore-backup", + type=Path, + help=( + "Restore both profiles and the registry from a " + "launcher-catalog-profile-backup-v2 file; requires --apply." + ), + ) + return parser + + +def parse_config(argv: Sequence[str] | None = None) -> RuntimeConfig: + args = build_parser().parse_args(argv) + local_dev_root = _absolute(args.local_dev_root) + return RuntimeConfig( + catalog=_absolute(args.catalog), + registry=_absolute(args.registry), + software_root=_absolute(args.software_root), + local_dev_root=local_dev_root, + backup_dir=_absolute( + args.backup_dir + if args.backup_dir is not None + else local_dev_root / "launcher_catalog_backups" + ), + apply=bool(args.apply), + restore_backup=( + _absolute(args.restore_backup) + if args.restore_backup is not None + else None + ), + ) + + +def _is_within(path: Path, parent: Path) -> bool: + try: + path.resolve(strict=False).relative_to(parent.resolve(strict=False)) + return True + except ValueError: + return False + + +def validate_config(config: RuntimeConfig) -> list[str]: + errors: list[str] = [] + if not config.catalog.is_file(): + errors.append(f"Katalog fehlt: {config.catalog}") + if not config.registry.is_file(): + errors.append(f"Registry fehlt: {config.registry}") + if not config.software_root.is_dir(): + errors.append(f"Software-Root fehlt: {config.software_root}") + if not config.local_dev_root.is_dir(): + errors.append(f"Local-Dev-Root fehlt: {config.local_dev_root}") + if config.catalog == config.registry: + errors.append("Katalog und Registry müssen verschiedene Dateien sein.") + if config.restore_backup is not None: + if not config.apply: + errors.append("--restore-backup erfordert zusätzlich --apply.") + if not config.restore_backup.is_file(): + errors.append(f"Restore-Backup fehlt: {config.restore_backup}") + if config.apply: + synchronized_root = config.software_root + if config.software_root.parent.name.casefold() == ".topics": + candidate = config.software_root.parent.parent + if _is_within(config.registry, candidate): + synchronized_root = candidate + if _is_within(config.backup_dir, synchronized_root): + errors.append( + "Backup-Verzeichnis darf nicht innerhalb des synchronisierten " + "Daten-Roots liegen." + ) + return errors + + +def resolve_template( + value: str, *, software_root: Path, local_dev_root: Path +) -> str: + return ( + value.replace("$SOFTWARE_ROOT", str(software_root)) + .replace("$LOCAL_DEV_ROOT", str(local_dev_root)) + ) + + +def norm(path: str) -> str: + return ( + os.path.normcase(os.path.normpath(os.path.abspath(path))) if path else "" + ) + + +def detect_kind(path: str) -> str: + lower = path.lower() + if lower.endswith((".bat", ".cmd", ".ps1")): + return "script" + if lower.endswith(".lnk"): + return "windows_shortcut" + if os.path.isdir(path): + return "directory" + return "file" + + +def is_profile_running() -> bool | None: + """Return profile state, or ``None`` when the native probe is unavailable.""" + try: + completed = subprocess.run( + ["tasklist", "/NH", "/FO", "CSV"], + capture_output=True, + text=True, + timeout=10, + check=False, + shell=False, + ) + if completed.returncode != 0: + return None + image_names = [ + row[0].strip() + for row in csv.reader(completed.stdout.splitlines()) + if row and row[0].strip() + ] + return any( + pattern.fullmatch(image_name) + for image_name in image_names + for pattern in PROFILE_IMAGE_PATTERNS + ) + except Exception: + return None + + +def load_profile(settings: QSettings) -> tuple[list[dict], int]: + tabs: list[dict] = [] + size = settings.beginReadArray("tabs") + for index in range(size): + settings.setArrayIndex(index) + name = str(settings.value("name", "Tab") or "Tab") + view_mode = str(settings.value("view_mode", "tiles") or "tiles") + entries_json = settings.value("entries_json", "") + entries: list[dict] = [] + if isinstance(entries_json, str) and entries_json.strip(): + try: + parsed = json.loads(entries_json) + if isinstance(parsed, list): + entries = [ + dict(entry) for entry in parsed if isinstance(entry, dict) + ] + except json.JSONDecodeError: + entries = [] + if not entries: + paths = settings.value("paths", []) + if isinstance(paths, str): + paths = [paths] + if isinstance(paths, Iterable): + entries = [ + { + "path": str(path), + "label": Path(str(path)).stem, + "kind": detect_kind(str(path)), + "notes": None, + } + for path in paths + if isinstance(path, str) and path.strip() + ] + tabs.append({"name": name, "view_mode": view_mode, "entries": entries}) + settings.endArray() + current = settings.value("current_tab", 0) + return tabs, current if isinstance(current, int) else 0 + + +def save_profile(settings: QSettings, tabs: list[dict], current_tab: int) -> None: + settings.setValue("current_tab", current_tab) + settings.remove("tabs") + settings.beginWriteArray("tabs") + for index, tab in enumerate(tabs): + settings.setArrayIndex(index) + entries = tab["entries"] + settings.setValue("name", tab["name"]) + settings.setValue("view_mode", tab.get("view_mode", "tiles")) + settings.setValue("entries_json", json.dumps(entries, ensure_ascii=False)) + settings.setValue( + "paths", [entry.get("path", "") for entry in entries] + ) + settings.endArray() + settings.sync() + if settings.status() != QSettings.Status.NoError: + raise OSError( + "QSettings-Profil konnte nicht geschrieben werden: " + f"{settings.fileName()}" + ) + + +def load_meta(settings: QSettings) -> dict[str, dict[str, str]]: + raw = settings.value(META_KEY, "") + if not isinstance(raw, str) or not raw.strip(): + return {} + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + return {} + if not isinstance(parsed, dict): + return {} + return { + str(key): dict(value) + for key, value in parsed.items() + if isinstance(value, dict) + } + + +def save_meta(settings: QSettings, meta: dict[str, dict[str, str]]) -> None: + settings.setValue( + META_KEY, json.dumps(meta, ensure_ascii=False, sort_keys=True) + ) + settings.sync() + if settings.status() != QSettings.Status.NoError: + raise OSError( + "QSettings-Metadaten konnten nicht geschrieben werden: " + f"{settings.fileName()}" + ) + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def backup_profiles( + backup_dir: Path, + staged: list[tuple[str, list[dict], int, dict[str, dict[str, str]]]], + registry_path: Path, + registry_bytes: bytes, +) -> Path: + """Back up both profiles and the registry source before an apply run.""" + backup_dir.mkdir(parents=True, exist_ok=True) + target = backup_dir / ( + f"profiles_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}.json" + ) + payload = { + "format": "launcher-catalog-profile-backup-v2", + "created_at": datetime.now().astimezone().isoformat(), + "registry": { + "source_path": str(registry_path), + "sha256": _sha256(registry_bytes), + "text": registry_bytes.decode("utf-8"), + }, + "profiles": { + profile: { + "tabs": tabs, + "current_tab": current_tab, + "managed_paths": meta, + } + for profile, tabs, current_tab, meta in staged + }, + } + target.write_text( + json.dumps(payload, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + return target + + +def ensure_board(tabs: list[dict], board_name: str) -> int: + for index, tab in enumerate(tabs): + if tab["name"] == board_name: + return index + tabs.append({"name": board_name, "view_mode": "tiles", "entries": []}) + return len(tabs) - 1 + + +def find_entry( + tabs: list[dict], candidates: Iterable[str] +) -> tuple[int, int] | None: + wanted = {norm(candidate) for candidate in candidates if candidate} + for tab_index, tab in enumerate(tabs): + for entry_index, entry in enumerate(tab["entries"]): + if norm(str(entry.get("path", ""))) in wanted: + return tab_index, entry_index + return None + + +def read_catalog(path: Path) -> list[dict]: + payload = json.loads(path.read_text(encoding="utf-8")) + if payload.get("format") != "software-launcher-catalog-v2": + raise ValueError(f"Unerwartetes Katalogformat in {path}") + apps = payload.get("apps") + if not isinstance(apps, list): + raise ValueError("Katalog enthält keine App-Liste") + normalized: list[dict] = [] + for index, app in enumerate(apps): + if not isinstance(app, dict): + raise ValueError(f"Katalogeintrag {index} ist kein Objekt") + app_id = app.get("id") + if not isinstance(app_id, str) or not app_id.strip(): + raise ValueError(f"Katalogeintrag {index} besitzt keine gültige ID") + launcher_template = app.get("launcher_template") + if not isinstance(launcher_template, str): + raise ValueError( + f"Katalogeintrag {app_id} besitzt kein String-Pfadtemplate" + ) + profiles = app.get("profiles", {}) + if not isinstance(profiles, dict): + raise ValueError( + f"Katalogeintrag {app_id} besitzt kein Profilobjekt" + ) + normalized.append(dict(app)) + return normalized + + +def registry_states_for_spec(spec: str) -> dict[str, str]: + result = {profile: "PENDING" for profile in PROFILES} + for profile, value in re.findall( + r"(?i)(SOFTWARECENTER|LAUNCHBOARDS)\s*=\s*([A-Z_]+)", spec + ): + profile_name = ( + "SoftwareCenter" + if profile.casefold() == "softwarecenter" + else "LaunchBoards" + ) + result[profile_name] = value.upper() + return result + + +def registry_states(path: Path) -> dict[str, dict[str, str]]: + states: dict[str, dict[str, str]] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + match = re.match(r"^\s*[A-Z_]+\s*\|\s*(.+?)\s*:\s*(.+?)\s*$", line) + if not match: + continue + states[match.group(1).strip()] = registry_states_for_spec( + match.group(2).strip() + ) + return states + + +def write_registry_updates( + path: Path, updates: dict[str, dict[str, str]] +) -> None: + if not updates: + return + original = path.read_text(encoding="utf-8") + lines = original.splitlines(keepends=True) + changed = False + remaining = set(updates) + rewritten: list[str] = [] + for line in lines: + match = re.match( + r"^(\s*[A-Z_]+\s*\|\s*)(.+?)(\s*:\s*)(.*?)(\r?\n)?$", line + ) + if not match or match.group(2).strip() not in updates: + rewritten.append(line) + continue + app_id = match.group(2).strip() + remaining.discard(app_id) + spec = match.group(4).strip() + desktop = re.search( + r"(?i)(?:^|;)\s*DESKTOP\s*=\s*(ON|OFF)\s*(?:;|$)", spec + ) + if desktop: + desktop_state = desktop.group(1).upper() + elif spec.upper() in {"ON", "OFF"}: + desktop_state = spec.upper() + else: + desktop_state = "OFF" + next_states = {**registry_states_for_spec(spec), **updates[app_id]} + rendered = "; ".join( + [f"DESKTOP={desktop_state}"] + + [ + f"{profile.upper()}={next_states[profile]}" + for profile in PROFILES + ] + ) + newline = match.group(5) or "\n" + rewritten.append( + f"{match.group(1)}{app_id}{match.group(3)}{rendered}{newline}" + ) + changed = True + if remaining: + raise ValueError( + "Registry enthält keine Zeile für: " + ", ".join(sorted(remaining)) + ) + if not changed: + return + with tempfile.NamedTemporaryFile( + "w", encoding="utf-8", delete=False, dir=path.parent, newline="" + ) as handle: + handle.writelines(rewritten) + temp_name = handle.name + os.replace(temp_name, path) + + +def replace_file_bytes(path: Path, content: bytes) -> None: + with tempfile.NamedTemporaryFile( + "wb", delete=False, dir=path.parent + ) as handle: + handle.write(content) + temp_name = handle.name + os.replace(temp_name, path) + + +@dataclass +class ProfileReport: + added: list[str] = field(default_factory=list) + updated: list[str] = field(default_factory=list) + suppressed: list[str] = field(default_factory=list) + blocked: list[str] = field(default_factory=list) + unchanged: list[str] = field(default_factory=list) + + +def reconcile_profile( + profile: str, + tabs: list[dict], + meta: dict[str, dict[str, str]], + apps: list[dict], + *, + software_root: Path, + local_dev_root: Path, +) -> tuple[ProfileReport, dict[str, str]]: + report = ProfileReport() + registry_update: dict[str, str] = {} + for app in apps: + app_id = str(app["id"]) + desired = str(app.get("profiles", {}).get(profile, "PENDING")).upper() + if desired not in VALID_STATES: + report.blocked.append( + f"{app_id} (ungültiger Registry-Zustand {desired})" + ) + continue + if desired == "SUPPRESSED": + report.unchanged.append(f"{app_id} (unterdrückt)") + continue + target = resolve_template( + str(app.get("launcher_template", "")), + software_root=software_root, + local_dev_root=local_dev_root, + ) + if not target or not os.path.isfile(target): + report.blocked.append(f"{app_id} (Startziel fehlt)") + continue + remembered = meta.get(app_id, {}) + hit = find_entry(tabs, [target, str(remembered.get("path", ""))]) + seen_before = bool(remembered.get("path")) + if hit is None: + if desired == "ACTIVE" and seen_before: + registry_update[app_id] = "SUPPRESSED" + report.suppressed.append(app_id) + continue + board = str(app.get("category", "Allgemein") or "Allgemein") + tab_index = ensure_board(tabs, board) + tabs[tab_index]["entries"].append( + { + "path": target, + "label": app.get("name", app_id), + "kind": detect_kind(target), + "notes": None, + } + ) + meta[app_id] = {"path": target} + registry_update[app_id] = "ACTIVE" + report.added.append(f'{app_id} (Board "{board}")') + continue + tab_index, entry_index = hit + entry = tabs[tab_index]["entries"][entry_index] + if norm(str(entry.get("path", ""))) != norm(target): + entry["path"] = target + entry["kind"] = detect_kind(target) + report.updated.append( + f'{app_id} (Board "{tabs[tab_index]["name"]}")' + ) + else: + report.unchanged.append( + f'{app_id} (Board "{tabs[tab_index]["name"]}")' + ) + meta[app_id] = {"path": target} + if desired == "PENDING": + registry_update[app_id] = "ACTIVE" + return report, registry_update + + +def print_report(profile: str, report: ProfileReport) -> None: + print(f"\n{profile}") + for title, entries, marker in ( + ("Neu", report.added, "+"), + ("Pfade aktualisiert", report.updated, "~"), + ("Vom Nutzer entfernt", report.suppressed, "-"), + ("Blockiert", report.blocked, "!"), + ): + print(f" {title}: {len(entries)}") + for entry in entries: + print(f" {marker} {entry}") + print(f" Unverändert: {len(report.unchanged)}") + + +SettingsFactory = Callable[[str, str], QSettings] + + +def _restore_original_state( + staged: list[ + tuple[QSettings, list[dict], int, dict[str, dict[str, str]]] + ], + backup_data: list[ + tuple[str, list[dict], int, dict[str, dict[str, str]]] + ], + registry_path: Path, + registry_bytes: bytes, +) -> list[str]: + errors: list[str] = [] + for (settings, _tabs, _current, _meta), ( + profile, + original_tabs, + original_current, + original_meta, + ) in zip(staged, backup_data, strict=True): + try: + save_profile(settings, original_tabs, original_current) + save_meta(settings, original_meta) + except (OSError, ValueError) as exc: + errors.append(f"{profile}: {exc}") + try: + replace_file_bytes(registry_path, registry_bytes) + except OSError as exc: + errors.append(f"Registry: {exc}") + return errors + + +def restore_backup_file( + config: RuntimeConfig, + *, + settings_factory: SettingsFactory, +) -> int: + assert config.restore_backup is not None + try: + payload = json.loads( + config.restore_backup.read_text(encoding="utf-8") + ) + if payload.get("format") != "launcher-catalog-profile-backup-v2": + raise ValueError("unerwartetes Backupformat") + registry = payload.get("registry") + profiles = payload.get("profiles") + if not isinstance(registry, dict) or not isinstance(profiles, dict): + raise ValueError("Backup enthält keine Registry-/Profildaten") + source_path = registry.get("source_path") + registry_text = registry.get("text") + registry_hash = registry.get("sha256") + if not all( + isinstance(value, str) + for value in (source_path, registry_text, registry_hash) + ): + raise ValueError("Backup-Registrydaten sind unvollständig") + if norm(source_path) != norm(str(config.registry)): + raise ValueError( + "Backup gehört zu einer anderen Registry-Datei: " + f"{source_path}" + ) + registry_bytes = registry_text.encode("utf-8") + if _sha256(registry_bytes) != registry_hash: + raise ValueError("Backup-Registryhash stimmt nicht") + validated: list[ + tuple[QSettings, list[dict], int, dict[str, dict[str, str]]] + ] = [] + for profile in PROFILES: + profile_data = profiles.get(profile) + if not isinstance(profile_data, dict): + raise ValueError(f"Backup-Profil fehlt: {profile}") + tabs = profile_data.get("tabs") + current_tab = profile_data.get("current_tab") + managed_paths = profile_data.get("managed_paths") + if not isinstance(tabs, list) or not isinstance(current_tab, int): + raise ValueError(f"Backup-Profil ungültig: {profile}") + if not isinstance(managed_paths, dict): + raise ValueError(f"Backup-Metadaten ungültig: {profile}") + validated.append( + ( + settings_factory(ORG, profile), + tabs, + current_tab, + managed_paths, + ) + ) + current_registry_bytes = config.registry.read_bytes() + current_data: list[ + tuple[str, list[dict], int, dict[str, dict[str, str]]] + ] = [] + for profile, ( + settings, + _tabs, + _current_tab, + _managed_paths, + ) in zip(PROFILES, validated, strict=True): + current_tabs, current_tab = load_profile(settings) + current_data.append( + (profile, current_tabs, current_tab, load_meta(settings)) + ) + except ( + OSError, + UnicodeError, + ValueError, + json.JSONDecodeError, + ) as exc: + print(f"ABBRUCH: Restore-Validierung fehlgeschlagen: {exc}") + print(f"Backup: {config.restore_backup}") + return 3 + mutation_started = False + try: + for settings, tabs, current_tab, managed_paths in validated: + mutation_started = True + save_profile(settings, tabs, current_tab) + save_meta(settings, managed_paths) + replace_file_bytes(config.registry, registry_bytes) + except (OSError, UnicodeError, ValueError) as exc: + print(f"ABBRUCH: Restore fehlgeschlagen: {exc}") + print(f"Backup: {config.restore_backup}") + if mutation_started: + rollback_errors = _restore_original_state( + validated, + current_data, + config.registry, + current_registry_bytes, + ) + if rollback_errors: + print("RESTORE-ROLLBACK UNVOLLSTÄNDIG:") + for error in rollback_errors: + print(f" - {error}") + else: + print( + "RESTORE-ROLLBACK ERFOLGREICH: " + "vorheriger Zustand wiederhergestellt." + ) + return 3 + print(f"Restore abgeschlossen: {config.restore_backup}") + return 0 + + +def execute( + config: RuntimeConfig, + *, + settings_factory: SettingsFactory = QSettings, + running_probe: Callable[[], bool | None] = is_profile_running, +) -> int: + errors = validate_config(config) + if errors: + for error in errors: + print(f"ABBRUCH: {error}") + return 2 + if config.apply: + running_state = running_probe() + if running_state is None: + print( + "ABBRUCH: Prozessstatus von SoftwareCenter und LaunchBoards " + "konnte nicht verlässlich gelesen werden." + ) + return 2 + if running_state: + print( + "ABBRUCH: SoftwareCenter oder LaunchBoards läuft. " + "Bitte beide schließen und erneut ausführen." + ) + return 2 + if config.restore_backup is not None: + return restore_backup_file( + config, + settings_factory=settings_factory, + ) + try: + apps = read_catalog(config.catalog) + state_overrides = registry_states(config.registry) + except (OSError, UnicodeError, ValueError, json.JSONDecodeError) as exc: + print(f"ABBRUCH: Eingabedaten ungültig: {exc}") + return 2 + for app in apps: + app.setdefault("profiles", {}).update( + state_overrides.get(str(app["id"]), {}) + ) + print( + f"Launcher-Katalog: {len(apps)} Apps | " + f"{'APPLY' if config.apply else 'DRY-RUN'}" + ) + all_updates: dict[str, dict[str, str]] = {} + staged: list[ + tuple[QSettings, list[dict], int, dict[str, dict[str, str]]] + ] = [] + backup_data: list[ + tuple[str, list[dict], int, dict[str, dict[str, str]]] + ] = [] + for profile in PROFILES: + settings = settings_factory(ORG, profile) + original_tabs, original_current_tab = load_profile(settings) + original_meta = load_meta(settings) + tabs = json.loads(json.dumps(original_tabs)) + meta = json.loads(json.dumps(original_meta)) + report, updates = reconcile_profile( + profile, + tabs, + meta, + apps, + software_root=config.software_root, + local_dev_root=config.local_dev_root, + ) + print_report(profile, report) + for app_id, state in updates.items(): + all_updates.setdefault(app_id, {})[profile] = state + staged.append((settings, tabs, original_current_tab, meta)) + backup_data.append( + (profile, original_tabs, original_current_tab, original_meta) + ) + if not config.apply: + print("\nDRY-RUN: keine Profile und keine Registry geschrieben.") + return 0 + missing_registry_ids = sorted(set(all_updates) - set(state_overrides)) + if missing_registry_ids: + print( + "ABBRUCH: Registry enthält keine Zeile für: " + + ", ".join(missing_registry_ids) + ) + return 2 + backup: Path | None = None + mutation_started = False + try: + original_registry_bytes = config.registry.read_bytes() + except OSError as exc: + print( + "ABBRUCH: Registry ist unmittelbar vor dem Apply nicht mehr " + f"stabil lesbar: {exc}" + ) + return 2 + try: + backup = backup_profiles( + config.backup_dir, + backup_data, + config.registry, + original_registry_bytes, + ) + for settings, tabs, current_tab, meta in staged: + mutation_started = True + save_profile(settings, tabs, current_tab) + save_meta(settings, meta) + write_registry_updates(config.registry, all_updates) + except (OSError, UnicodeError, ValueError) as exc: + print(f"ABBRUCH: Schreibvorgang fehlgeschlagen: {exc}") + if backup is not None: + print(f"Backup: {backup}") + if backup is not None and mutation_started: + rollback_errors = _restore_original_state( + staged, + backup_data, + config.registry, + original_registry_bytes, + ) + if rollback_errors: + print("ROLLBACK UNVOLLSTÄNDIG:") + for error in rollback_errors: + print(f" - {error}") + else: + print( + "ROLLBACK ERFOLGREICH: " + "Profile und Registry wiederhergestellt." + ) + elif backup is None: + print( + "KEIN ROLLBACK-WRITE: Backup-Erstellung scheiterte " + "vor Beginn der Mutation." + ) + return 3 + print(f"\nBackup: {backup}") + print( + f"Geschrieben: {len(all_updates)} Registry-Einträge, " + "beide lokale Profile aktualisiert." + ) + return 0 + + +def main(argv: Sequence[str] | None = None) -> int: + return execute(parse_config(argv)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_softwarecenter_sync.py b/tests/test_softwarecenter_sync.py new file mode 100644 index 0000000..8a73e35 --- /dev/null +++ b/tests/test_softwarecenter_sync.py @@ -0,0 +1,773 @@ +# -*- coding: utf-8 -*- +"""Regression coverage for the Plan-D launcher-catalog runtime.""" +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest +from PySide6.QtCore import QSettings + +from scripts import softwarecenter_sync as sync + + +def app(app_id: str, target: str, state: str = "PENDING") -> dict: + return { + "id": app_id, + "name": app_id, + "category": "DATA", + "launcher_template": target, + "profiles": { + "SoftwareCenter": state, + "LaunchBoards": state, + }, + } + + +def write_inputs( + tmp_path: Path, *, state: str = "PENDING", target: Path | None = None +) -> tuple[Path, Path, Path, Path]: + software_root = tmp_path / "software" + local_dev_root = tmp_path / "local-dev" + tools = software_root / "_tools" + tools.mkdir(parents=True) + local_dev_root.mkdir() + if target is None: + target = software_root / "DATA" / "Werkzeug.exe" + target.parent.mkdir() + target.write_bytes(b"exe") + catalog = tools / "software_apps.json" + catalog.write_text( + json.dumps( + { + "format": "software-launcher-catalog-v2", + "apps": [app("Werkzeug", str(target), state)], + } + ), + encoding="utf-8", + ) + registry = tmp_path / "DESKTOP-REGISTRY.txt" + registry.write_text( + f"DATA | Werkzeug : DESKTOP=OFF; " + f"SOFTWARECENTER={state}; LAUNCHBOARDS={state}\n", + encoding="utf-8", + ) + return catalog, registry, software_root, local_dev_root + + +def config_for( + tmp_path: Path, *, apply: bool = False, state: str = "PENDING" +) -> sync.RuntimeConfig: + catalog, registry, software_root, local_dev_root = write_inputs( + tmp_path, state=state + ) + return sync.RuntimeConfig( + catalog=catalog, + registry=registry, + software_root=software_root, + local_dev_root=local_dev_root, + backup_dir=local_dev_root / "backups", + apply=apply, + ) + + +def settings_factory(tmp_path: Path): + settings_root = tmp_path / "settings" + + def factory(_organization: str, profile: str) -> QSettings: + settings_root.mkdir(exist_ok=True) + return QSettings(str(settings_root / f"{profile}.ini"), QSettings.IniFormat) + + return factory + + +def test_cli_requires_exact_catalog_registry_and_roots(): + parser = sync.build_parser() + with pytest.raises(SystemExit): + parser.parse_args([]) + with pytest.raises(SystemExit): + parser.parse_args( + [ + "--catalog", + "catalog.json", + "--registry", + "registry.txt", + ] + ) + + +def test_parse_config_preserves_explicit_inputs_as_absolute_paths(tmp_path): + values = [ + "--catalog", + str(tmp_path / "catalog.json"), + "--registry", + str(tmp_path / "registry.txt"), + "--software-root", + str(tmp_path / "software"), + "--local-dev-root", + str(tmp_path / "dev"), + ] + result = sync.parse_config(values) + assert result.catalog == tmp_path / "catalog.json" + assert result.registry == tmp_path / "registry.txt" + assert result.software_root == tmp_path / "software" + assert result.local_dev_root == tmp_path / "dev" + assert result.backup_dir == tmp_path / "dev" / "launcher_catalog_backups" + assert result.apply is False + + +def test_missing_inputs_fail_closed_without_settings_access(tmp_path, capsys): + touched = False + + def factory(_organization: str, _profile: str) -> QSettings: + nonlocal touched + touched = True + raise AssertionError("settings must not be opened") + + config = sync.RuntimeConfig( + catalog=tmp_path / "missing.json", + registry=tmp_path / "missing.txt", + software_root=tmp_path / "missing-software", + local_dev_root=tmp_path / "missing-dev", + backup_dir=tmp_path / "backups", + apply=False, + ) + assert sync.execute(config, settings_factory=factory) == 2 + assert touched is False + assert "ABBRUCH" in capsys.readouterr().out + + +def test_apply_rejects_backup_inside_synchronized_software_root(tmp_path): + config = config_for(tmp_path, apply=True) + unsafe = sync.RuntimeConfig( + **{ + **config.__dict__, + "backup_dir": config.software_root / "backups", + } + ) + errors = sync.validate_config(unsafe) + assert any("Backup-Verzeichnis" in error for error in errors) + + +def test_apply_rejects_backup_anywhere_inside_detected_onedrive_root(tmp_path): + one_drive = tmp_path / "OneDrive" + software_root = one_drive / ".TOPICS" / ".SOFTWARE" + tools = software_root / "_tools" + desktop = one_drive / "Desktop" + local_dev = tmp_path / "local-dev" + tools.mkdir(parents=True) + desktop.mkdir() + local_dev.mkdir() + catalog = tools / "software_apps.json" + catalog.write_text( + '{"format":"software-launcher-catalog-v2","apps":[]}', + encoding="utf-8", + ) + registry = desktop / "DESKTOP-REGISTRY.txt" + registry.write_text("", encoding="utf-8") + config = sync.RuntimeConfig( + catalog=catalog, + registry=registry, + software_root=software_root, + local_dev_root=local_dev, + backup_dir=one_drive / ".backups", + apply=True, + ) + assert any( + "synchronisierten Daten-Roots" in error + for error in sync.validate_config(config) + ) + + +def test_dry_run_writes_neither_registry_profiles_nor_backup(tmp_path): + config = config_for(tmp_path) + original_registry = config.registry.read_bytes() + factory = settings_factory(tmp_path) + + assert sync.execute(config, settings_factory=factory) == 0 + + assert config.registry.read_bytes() == original_registry + assert not config.backup_dir.exists() + assert not list((tmp_path / "settings").glob("*.ini")) + + +def test_apply_is_blocked_while_a_profile_is_running(tmp_path, capsys): + config = config_for(tmp_path, apply=True) + original_registry = config.registry.read_bytes() + + assert ( + sync.execute( + config, + settings_factory=settings_factory(tmp_path), + running_probe=lambda: True, + ) + == 2 + ) + + assert config.registry.read_bytes() == original_registry + assert not config.backup_dir.exists() + assert "läuft" in capsys.readouterr().out + + +def test_apply_fails_closed_when_process_probe_is_unavailable(tmp_path, capsys): + config = config_for(tmp_path, apply=True) + + assert ( + sync.execute( + config, + settings_factory=settings_factory(tmp_path), + running_probe=lambda: None, + ) + == 2 + ) + + assert not config.backup_dir.exists() + assert "nicht verlässlich" in capsys.readouterr().out + + +def test_apply_backs_up_then_writes_both_profiles_and_registry(tmp_path): + config = config_for(tmp_path, apply=True) + factory = settings_factory(tmp_path) + + assert ( + sync.execute( + config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 0 + ) + + registry = config.registry.read_text(encoding="utf-8") + assert "SOFTWARECENTER=ACTIVE" in registry + assert "LAUNCHBOARDS=ACTIVE" in registry + backups = list(config.backup_dir.glob("profiles_*.json")) + assert len(backups) == 1 + backup = json.loads(backups[0].read_text(encoding="utf-8")) + assert backup["format"] == "launcher-catalog-profile-backup-v2" + assert backup["registry"]["text"].splitlines()[-1].endswith( + "LAUNCHBOARDS=PENDING" + ) + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + tabs, _current = sync.load_profile(settings) + assert tabs[0]["entries"][0]["label"] == "Werkzeug" + assert sync.load_meta(settings)["Werkzeug"]["path"].endswith( + "Werkzeug.exe" + ) + + +def test_mid_apply_failure_reports_backup_and_rolls_back_all_state( + tmp_path, monkeypatch, capsys +): + config = config_for(tmp_path, apply=True) + factory = settings_factory(tmp_path) + original_registry = config.registry.read_bytes() + real_save_profile = sync.save_profile + calls = 0 + + def fail_second_profile(settings, tabs, current_tab): + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("synthetic second-profile failure") + real_save_profile(settings, tabs, current_tab) + + monkeypatch.setattr(sync, "save_profile", fail_second_profile) + + assert ( + sync.execute( + config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 3 + ) + + output = capsys.readouterr().out + backups = list(config.backup_dir.glob("profiles_*.json")) + assert len(backups) == 1 + assert f"Backup: {backups[0]}" in output + assert "ROLLBACK ERFOLGREICH" in output + assert config.registry.read_bytes() == original_registry + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + assert sync.load_profile(settings)[0] == [] + assert sync.load_meta(settings) == {} + + +def test_backup_creation_failure_performs_no_restore_or_other_write( + tmp_path, monkeypatch, capsys +): + config = config_for(tmp_path, apply=True) + writes: list[str] = [] + + def fail_backup(*_args, **_kwargs): + raise OSError("synthetic backup failure") + + monkeypatch.setattr(sync, "backup_profiles", fail_backup) + monkeypatch.setattr( + sync, + "save_profile", + lambda *_args, **_kwargs: writes.append("profile"), + ) + monkeypatch.setattr( + sync, + "save_meta", + lambda *_args, **_kwargs: writes.append("meta"), + ) + monkeypatch.setattr( + sync, + "replace_file_bytes", + lambda *_args, **_kwargs: writes.append("registry"), + ) + + assert ( + sync.execute( + config, + settings_factory=settings_factory(tmp_path), + running_probe=lambda: False, + ) + == 3 + ) + + assert writes == [] + assert "KEIN ROLLBACK-WRITE" in capsys.readouterr().out + + +def test_registry_disappearing_before_apply_fails_cleanly_without_writes( + tmp_path, monkeypatch, capsys +): + config = config_for(tmp_path, apply=True) + writes: list[str] = [] + base_factory = settings_factory(tmp_path) + removed = False + + def deleting_factory(organization: str, profile: str) -> QSettings: + nonlocal removed + if not removed: + config.registry.unlink() + removed = True + return base_factory(organization, profile) + + monkeypatch.setattr( + sync, + "save_profile", + lambda *_args, **_kwargs: writes.append("profile"), + ) + monkeypatch.setattr( + sync, + "save_meta", + lambda *_args, **_kwargs: writes.append("meta"), + ) + monkeypatch.setattr( + sync, + "replace_file_bytes", + lambda *_args, **_kwargs: writes.append("registry"), + ) + + assert ( + sync.execute( + config, + settings_factory=deleting_factory, + running_probe=lambda: False, + ) + == 2 + ) + + assert writes == [] + assert not config.backup_dir.exists() + assert "unmittelbar vor dem Apply" in capsys.readouterr().out + + +def test_restore_backup_cli_restores_profiles_and_registry(tmp_path): + config = config_for(tmp_path, apply=True) + factory = settings_factory(tmp_path) + assert ( + sync.execute( + config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 0 + ) + backup = next(config.backup_dir.glob("profiles_*.json")) + config.registry.write_text("DATA | Werkzeug : BROKEN\n", encoding="utf-8") + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + sync.save_profile( + settings, + [ + { + "name": "Changed", + "view_mode": "tiles", + "entries": [], + } + ], + 0, + ) + restore_config = sync.RuntimeConfig( + **{ + **config.__dict__, + "restore_backup": backup, + } + ) + + assert ( + sync.execute( + restore_config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 0 + ) + + assert "SOFTWARECENTER=PENDING" in config.registry.read_text( + encoding="utf-8" + ) + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + assert sync.load_profile(settings)[0] == [] + assert sync.load_meta(settings) == {} + + +def test_failed_restore_compensates_to_the_pre_restore_state( + tmp_path, monkeypatch, capsys +): + config = config_for(tmp_path, apply=True) + factory = settings_factory(tmp_path) + assert ( + sync.execute( + config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 0 + ) + backup = next(config.backup_dir.glob("profiles_*.json")) + changed_registry = b"DATA | Werkzeug : BROKEN\r\n" + config.registry.write_bytes(changed_registry) + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + sync.save_profile( + settings, + [ + { + "name": "Changed", + "view_mode": "tiles", + "entries": [], + } + ], + 0, + ) + restore_config = sync.RuntimeConfig( + **{ + **config.__dict__, + "restore_backup": backup, + } + ) + real_save_profile = sync.save_profile + calls = 0 + + def fail_second_restore_profile(settings, tabs, current_tab): + nonlocal calls + calls += 1 + if calls == 2: + raise OSError("synthetic restore failure") + real_save_profile(settings, tabs, current_tab) + + monkeypatch.setattr( + sync, "save_profile", fail_second_restore_profile + ) + + assert ( + sync.execute( + restore_config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 3 + ) + + assert config.registry.read_bytes() == changed_registry + for profile in sync.PROFILES: + settings = factory(sync.ORG, profile) + assert sync.load_profile(settings)[0][0]["name"] == "Changed" + assert "RESTORE-ROLLBACK ERFOLGREICH" in capsys.readouterr().out + + +def test_restore_backup_requires_apply(tmp_path): + config = config_for(tmp_path) + backup = tmp_path / "backup.json" + backup.write_text("{}", encoding="utf-8") + restore_config = sync.RuntimeConfig( + **{ + **config.__dict__, + "restore_backup": backup, + } + ) + assert "--restore-backup erfordert" in "\n".join( + sync.validate_config(restore_config) + ) + + +def test_pending_app_is_added_once_and_activated(tmp_path): + target = tmp_path / "Werkzeug.exe" + target.write_bytes(b"exe") + tabs: list[dict] = [] + meta: dict[str, dict[str, str]] = {} + + report, updates = sync.reconcile_profile( + "SoftwareCenter", + tabs, + meta, + [app("Werkzeug", str(target))], + software_root=tmp_path, + local_dev_root=tmp_path, + ) + + assert report.added == ['Werkzeug (Board "DATA")'] + assert updates == {"Werkzeug": "ACTIVE"} + assert tabs[0]["entries"][0]["path"] == str(target) + assert meta["Werkzeug"]["path"] == str(target) + + +def test_deleted_active_entry_is_suppressed_not_readded(tmp_path): + target = tmp_path / "Werkzeug.exe" + target.write_bytes(b"exe") + + report, updates = sync.reconcile_profile( + "SoftwareCenter", + [], + {"Werkzeug": {"path": str(target)}}, + [app("Werkzeug", str(target), "ACTIVE")], + software_root=tmp_path, + local_dev_root=tmp_path, + ) + + assert report.suppressed == ["Werkzeug"] + assert updates == {"Werkzeug": "SUPPRESSED"} + + +def test_explicit_suppressed_state_never_readds_entry(tmp_path): + target = tmp_path / "Werkzeug.exe" + target.write_bytes(b"exe") + + report, updates = sync.reconcile_profile( + "SoftwareCenter", + [], + {}, + [app("Werkzeug", str(target), "SUPPRESSED")], + software_root=tmp_path, + local_dev_root=tmp_path, + ) + + assert report.added == [] + assert report.unchanged == ["Werkzeug (unterdrückt)"] + assert updates == {} + + +def test_active_entry_moved_by_user_is_not_moved_back(tmp_path): + target = tmp_path / "Werkzeug.exe" + target.write_bytes(b"exe") + tabs = [ + { + "name": "Eigene Auswahl", + "view_mode": "tiles", + "entries": [ + { + "path": str(target), + "label": "Werkzeug", + "kind": "file", + "notes": "persönlich", + } + ], + } + ] + meta = {"Werkzeug": {"path": str(target)}} + + report, updates = sync.reconcile_profile( + "SoftwareCenter", + tabs, + meta, + [app("Werkzeug", str(target), "ACTIVE")], + software_root=tmp_path, + local_dev_root=tmp_path, + ) + + assert not report.added and not report.updated and not report.suppressed + assert updates == {} + assert tabs[0]["name"] == "Eigene Auswahl" + assert tabs[0]["entries"][0]["notes"] == "persönlich" + + +def test_missing_target_stays_pending_for_a_later_build(tmp_path): + target = tmp_path / "does-not-exist.exe" + + report, updates = sync.reconcile_profile( + "SoftwareCenter", + [], + {}, + [app("Werkzeug", str(target))], + software_root=tmp_path, + local_dev_root=tmp_path, + ) + + assert report.blocked == ["Werkzeug (Startziel fehlt)"] + assert updates == {} + + +def test_registry_update_keeps_desktop_state_and_writes_profile_fields(tmp_path): + registry = tmp_path / "DESKTOP-REGISTRY.txt" + registry.write_text("DATA | Werkzeug : OFF\n", encoding="utf-8") + + sync.write_registry_updates( + registry, + { + "Werkzeug": { + "SoftwareCenter": "ACTIVE", + "LaunchBoards": "SUPPRESSED", + } + }, + ) + + text = registry.read_text(encoding="utf-8") + assert "DESKTOP=OFF" in text + assert "SOFTWARECENTER=ACTIVE" in text + assert "LAUNCHBOARDS=SUPPRESSED" in text + + +def test_registry_update_fails_closed_for_unknown_app_id(tmp_path): + registry = tmp_path / "DESKTOP-REGISTRY.txt" + registry.write_text("DATA | Bekannt : OFF\n", encoding="utf-8") + original = registry.read_bytes() + + with pytest.raises(ValueError, match="Unbekannt"): + sync.write_registry_updates( + registry, + {"Unbekannt": {"SoftwareCenter": "ACTIVE"}}, + ) + + assert registry.read_bytes() == original + + +def test_apply_detects_unknown_registry_id_before_any_profile_write(tmp_path): + config = config_for(tmp_path, apply=True) + config.registry.write_text("DATA | AndereApp : OFF\n", encoding="utf-8") + factory = settings_factory(tmp_path) + + assert ( + sync.execute( + config, + settings_factory=factory, + running_probe=lambda: False, + ) + == 2 + ) + + assert not config.backup_dir.exists() + assert not list((tmp_path / "settings").glob("*.ini")) + + +def test_portable_templates_use_only_explicit_roots(tmp_path): + software_root = tmp_path / "software" + local_dev_root = tmp_path / "local-dev" + value = ( + r"$SOFTWARE_ROOT\DATA\App\App.exe;" + r"$LOCAL_DEV_ROOT\repos\App\App.exe" + ) + + resolved = sync.resolve_template( + value, + software_root=software_root, + local_dev_root=local_dev_root, + ) + + assert str(software_root) in resolved + assert str(local_dev_root) in resolved + assert "$SOFTWARE_ROOT" not in resolved + assert "$LOCAL_DEV_ROOT" not in resolved + assert str(Path(sync.__file__).parent) not in resolved + + +def test_process_probe_uses_argv_without_shell(monkeypatch): + captured: dict = {} + + def fake_run(argv, **kwargs): + captured["argv"] = argv + captured.update(kwargs) + return subprocess.CompletedProcess(argv, 0, stdout="", stderr="") + + monkeypatch.setattr(sync.subprocess, "run", fake_run) + + assert sync.is_profile_running() is False + assert captured["argv"] == ["tasklist", "/NH", "/FO", "CSV"] + assert captured["shell"] is False + + +@pytest.mark.parametrize( + "image_name", + [ + "SoftwareCenter.exe", + "SoftwareCenter-1.2.0-win64.exe", + "LaunchBoards.exe", + "LaunchBoards-1.0.0-win64.exe", + ], +) +def test_process_probe_blocks_all_supported_profile_image_names( + monkeypatch, image_name +): + output = f'"{image_name}","1234","Console","1","10,000 K"\n' + + def fake_run(argv, **_kwargs): + return subprocess.CompletedProcess(argv, 0, stdout=output, stderr="") + + monkeypatch.setattr(sync.subprocess, "run", fake_run) + assert sync.is_profile_running() is True + + +@pytest.mark.parametrize( + "image_name", + [ + "SoftwareCenterHelper.exe", + "SoftwareCenter-copy.exe", + "MyLaunchBoards.exe", + "LaunchBoards-evil-win64.exe", + ], +) +def test_process_probe_does_not_block_similarly_named_foreign_processes( + monkeypatch, image_name +): + output = f'"{image_name}","1234","Console","1","10,000 K"\n' + + def fake_run(argv, **_kwargs): + return subprocess.CompletedProcess(argv, 0, stdout=output, stderr="") + + monkeypatch.setattr(sync.subprocess, "run", fake_run) + assert sync.is_profile_running() is False + + +def test_invalid_catalog_fails_closed_before_profile_writes(tmp_path): + config = config_for(tmp_path) + config.catalog.write_text('{"format":"wrong","apps":[]}', encoding="utf-8") + + assert ( + sync.execute( + config, + settings_factory=settings_factory(tmp_path), + ) + == 2 + ) + + assert not config.backup_dir.exists() + assert not list((tmp_path / "settings").glob("*.ini")) + + +def test_runtime_migration_doc_uses_host_variables_not_a_personal_home_path(): + document = ( + Path(__file__).resolve().parents[1] / "RUNTIME_DAILY_CARE.md" + ).read_text(encoding="utf-8") + assert r"C:\Users\lukas" not in document + assert "$env:USERPROFILE" in document + assert "$env:LOCALAPPDATA" in document