From 2dedeadce05a1a4779e807f778371985b9beca5a Mon Sep 17 00:00:00 2001 From: hieshima <62245675+hieshima@users.noreply.github.com> Date: Wed, 13 May 2026 22:56:28 -0600 Subject: [PATCH] feat: add verified Windows update installation --- .github/workflows/release.yml | 31 +- core/update_installer.py | 1084 ++++++++++++++++++++++++++ core/updater.py | 32 +- main_qml.py | 4 + tests/test_backend.py | 499 ++++++++++++ tests/test_locale_manager.py | 43 ++ tests/test_update_installer.py | 1178 +++++++++++++++++++++++++++++ tests/test_updater.py | 45 ++ tools/generate_update_manifest.py | 86 +++ tools/verify_update_manifest.py | 29 + ui/backend.py | 344 ++++++++- ui/locale_manager.py | 99 +++ ui/qml/ScrollPage.qml | 147 +++- 13 files changed, 3595 insertions(+), 26 deletions(-) create mode 100644 core/update_installer.py create mode 100644 tests/test_update_installer.py create mode 100644 tools/generate_update_manifest.py create mode 100644 tools/verify_update_manifest.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 9801f03..19b53fa 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -224,8 +224,33 @@ jobs: needs: [build-windows, build-macos, build-linux] runs-on: ubuntu-latest steps: + - uses: actions/checkout@v6 + - uses: actions/download-artifact@v8 + - uses: actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Generate update metadata + run: | + mkdir -p update-assets + cp Mouser-Windows/Mouser-Windows.zip update-assets/ + cp Mouser-macOS/Mouser-macOS.zip update-assets/ + cp Mouser-macOS-intel/Mouser-macOS-intel.zip update-assets/ + cp Mouser-Linux/Mouser-Linux.zip update-assets/ + python tools/generate_update_manifest.py \ + --repo "${{ github.repository }}" \ + --tag "${{ github.ref_name }}" \ + --commit "${{ github.sha }}" \ + --asset-dir update-assets \ + --output "mouser-${{ github.ref_name }}-update.json" + + - uses: actions/upload-artifact@v7 + with: + name: Mouser-Update-Metadata + path: mouser-${{ github.ref_name }}-update.json + - name: Create or update GitHub Release env: GH_TOKEN: ${{ github.token }} @@ -238,7 +263,8 @@ jobs: Mouser-Windows/Mouser-Windows.zip \ Mouser-macOS/Mouser-macOS.zip \ Mouser-macOS-intel/Mouser-macOS-intel.zip \ - Mouser-Linux/Mouser-Linux.zip + Mouser-Linux/Mouser-Linux.zip \ + mouser-${{ github.ref_name }}-update.json else gh release create "${{ github.ref_name }}" \ --repo "${{ github.repository }}" \ @@ -247,5 +273,6 @@ jobs: Mouser-Windows/Mouser-Windows.zip \ Mouser-macOS/Mouser-macOS.zip \ Mouser-macOS-intel/Mouser-macOS-intel.zip \ - Mouser-Linux/Mouser-Linux.zip + Mouser-Linux/Mouser-Linux.zip \ + mouser-${{ github.ref_name }}-update.json fi diff --git a/core/update_installer.py b/core/update_installer.py new file mode 100644 index 0000000..39af05d --- /dev/null +++ b/core/update_installer.py @@ -0,0 +1,1084 @@ +"""Verified update metadata, archive staging, and install planning.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +import errno +import hashlib +import json +import os +from pathlib import Path, PurePosixPath +import platform +import re +import shutil +import stat +import subprocess +import sys +import tempfile +import time +import urllib.request +import zipfile + +from core.version import APP_VERSION + + +APP_ID = "io.github.tombadash.mouser" +STABLE_CHANNEL = "stable" +MANIFEST_SCHEMA_VERSION = 1 +DEFAULT_MANIFEST_NAME = "mouser-v{version}-update.json" +DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 120.0 +MAX_ARCHIVE_UNCOMPRESSED_BYTES = 750 * 1024 * 1024 +_UPDATE_MANIFEST_URL_ENV = "MOUSER_UPDATE_MANIFEST_URL" +_WINDOWS_SYNCHRONIZE = 0x00100000 +_WINDOWS_WAIT_OBJECT_0 = 0x00000000 +_WINDOWS_WAIT_TIMEOUT = 0x00000102 +_WINDOWS_WAIT_FAILED = 0xFFFFFFFF +_WINDOWS_ERROR_ACCESS_DENIED = 5 +_WINDOWS_ERROR_INVALID_PARAMETER = 87 + + +class UpdateInstallError(Exception): + """Raised when update metadata, staging, or installation is unsafe.""" + + def __init__(self, code: str, message: str): + super().__init__(message) + self.code = code + self.message = message + + +@dataclass(frozen=True) +class UpdateAsset: + platform: str + name: str + url: str + size: int + sha256: str + + +@dataclass(frozen=True) +class UpdateManifest: + schema: int + app_id: str + channel: str + version: str + tag: str + build_number: int + expires_at: str + commit: str + release_notes_url: str + assets: dict[str, UpdateAsset] + + +@dataclass(frozen=True) +class ArchiveRequirements: + require_windows_app: bool = False + + +@dataclass(frozen=True) +class StagedUpdate: + archive_path: Path + stage_dir: Path + app_root: Path + platform_key: str + asset_name: str + + +@dataclass(frozen=True) +class RuntimeLocation: + executable: Path + install_root: Path + app_data_dir: Path + frozen: bool + platform_key: str + update_supported: bool + reason: str = "" + + +@dataclass(frozen=True) +class InstallPlan: + platform_key: str + can_install: bool + status: str + message: str + asset: UpdateAsset | None = None + staged: StagedUpdate | None = None + + +@dataclass(frozen=True) +class WindowsUpdatePlan: + current_pid: int + install_root: str + staged_root: str + backup_root: str + result_marker: str + target_version: str = "" + target_build_number: int = 0 + executable_name: str = "Mouser.exe" + + def to_dict(self) -> dict[str, object]: + return { + "current_pid": self.current_pid, + "install_root": self.install_root, + "staged_root": self.staged_root, + "backup_root": self.backup_root, + "result_marker": self.result_marker, + "target_version": self.target_version, + "target_build_number": self.target_build_number, + "executable_name": self.executable_name, + } + + @classmethod + def from_dict(cls, data) -> "WindowsUpdatePlan": + if not isinstance(data, dict): + raise UpdateInstallError("invalid_plan", "Update plan is not valid.") + try: + return cls( + current_pid=int(data["current_pid"]), + install_root=str(data["install_root"]), + staged_root=str(data["staged_root"]), + backup_root=str(data["backup_root"]), + result_marker=str(data["result_marker"]), + target_version=str(data.get("target_version") or ""), + target_build_number=int(data.get("target_build_number") or 0), + executable_name=str(data.get("executable_name") or "Mouser.exe"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise UpdateInstallError("invalid_plan", "Update plan is incomplete.") from exc + + +@dataclass(frozen=True) +class ValidatedWindowsUpdatePlan: + plan: WindowsUpdatePlan + install_root: Path + staged_root: Path + backup_root: Path + result_marker: Path + + +def build_number_from_version(version: str) -> int: + value = (version or "").strip() + if value.startswith("v"): + value = value[1:] + match = re.fullmatch(r"(\d+)\.(\d+)\.(\d+)", value) + if not match: + raise UpdateInstallError("invalid_version", "Version must be major.minor.patch.") + major, minor, patch = (int(part) for part in match.groups()) + return major * 10000 + minor * 100 + patch + + +def current_build_number() -> int: + return build_number_from_version(APP_VERSION) + + +def manifest_name_for_version(version: str) -> str: + value = (version or "").strip() + if value.startswith("v"): + value = value[1:] + build_number_from_version(value) + return DEFAULT_MANIFEST_NAME.format(version=value) + + +def _parse_datetime(value: str) -> float: + text = str(value or "").strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError as exc: + raise UpdateInstallError("invalid_expiry", "Release metadata expiry is invalid.") from exc + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.timestamp() + + +def _asset_from_payload(platform_key: str, data) -> UpdateAsset: + if not isinstance(data, dict): + raise UpdateInstallError("missing_asset", "Selected update asset is missing.") + name = str(data.get("name") or "").strip() + url = str(data.get("url") or "").strip() + sha256 = str(data.get("sha256") or "").strip().lower() + try: + size = int(data.get("size")) + except (TypeError, ValueError) as exc: + raise UpdateInstallError("missing_asset_size", "Selected update asset size is missing.") from exc + if not name or not url or not sha256: + raise UpdateInstallError("missing_asset_fields", "Selected update asset is incomplete.") + if size <= 0: + raise UpdateInstallError("missing_asset_size", "Selected update asset size is invalid.") + if not re.fullmatch(r"[0-9a-f]{64}", sha256): + raise UpdateInstallError("invalid_sha256", "Selected update asset checksum is invalid.") + return UpdateAsset(platform_key, name, url, size, sha256) + + +def verify_update_manifest( + payload: dict, + *, + platform_key: str, + now: float | None = None, + highest_trusted_build: int | None = None, +) -> UpdateManifest: + if not isinstance(payload, dict): + raise UpdateInstallError("invalid_metadata", "Release metadata is not valid.") + payload = payload.get("payload", payload) + if not isinstance(payload, dict): + raise UpdateInstallError("invalid_metadata", "Release metadata is incomplete.") + if int(payload.get("schema", 0)) != MANIFEST_SCHEMA_VERSION: + raise UpdateInstallError("unsupported_schema", "Release metadata schema is not supported.") + if str(payload.get("app_id") or "") != APP_ID: + raise UpdateInstallError("wrong_app", "Release metadata is for a different app.") + if str(payload.get("channel") or "") != STABLE_CHANNEL: + raise UpdateInstallError("wrong_channel", "Release metadata is for a different channel.") + expires_at = str(payload.get("expires_at") or "") + if _parse_datetime(expires_at) <= (time.time() if now is None else float(now)): + raise UpdateInstallError("expired_metadata", "Release metadata has expired.") + version = str(payload.get("version") or "").strip() + tag = str(payload.get("tag") or "").strip() + commit = str(payload.get("commit") or "").strip() + release_notes_url = str(payload.get("release_notes_url") or "").strip() + if not version or not tag or not commit or not release_notes_url: + raise UpdateInstallError("missing_metadata_fields", "Release metadata is incomplete.") + try: + build_number = int(payload.get("build_number")) + except (TypeError, ValueError) as exc: + raise UpdateInstallError("missing_build_number", "Release metadata build number is missing.") from exc + if build_number < build_number_from_version(version): + raise UpdateInstallError("invalid_build_number", "Release metadata build number is invalid.") + if highest_trusted_build is not None and build_number < int(highest_trusted_build): + raise UpdateInstallError("older_build", "Release metadata is older than the trusted version.") + assets_payload = payload.get("assets") + if not isinstance(assets_payload, dict): + raise UpdateInstallError("missing_assets", "Release metadata has no assets.") + selected = _asset_from_payload(platform_key, assets_payload.get(platform_key)) + assets = { + str(key): _asset_from_payload(str(key), value) + for key, value in assets_payload.items() + } + assets[selected.platform] = selected + return UpdateManifest( + schema=MANIFEST_SCHEMA_VERSION, + app_id=APP_ID, + channel=STABLE_CHANNEL, + version=version, + tag=tag, + build_number=build_number, + expires_at=expires_at, + commit=commit, + release_notes_url=release_notes_url, + assets=assets, + ) + + +def platform_key(sys_platform: str | None = None, machine: str | None = None) -> str: + system = sys_platform or sys.platform + arch = (machine or platform.machine() or "").lower() + if system.startswith("win"): + if arch in {"arm64", "aarch64"}: + return "windows-arm64" + return "windows-x64" + if system == "darwin": + if arch in {"arm64", "aarch64"}: + return "macos-arm64" + return "macos-x86_64" + if system.startswith("linux"): + if arch in {"arm64", "aarch64"}: + return "linux-arm64" + return "linux-x64" + return f"{system}-{arch or 'unknown'}" + + +def sha256_file(path: str | os.PathLike) -> str: + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def verify_file(path: str | os.PathLike, *, expected_sha256: str, expected_size: int) -> None: + file_path = Path(path) + if file_path.stat().st_size != int(expected_size): + raise UpdateInstallError("size_mismatch", "Downloaded update size does not match.") + actual = sha256_file(file_path) + if actual.lower() != str(expected_sha256).lower(): + raise UpdateInstallError("sha256_mismatch", "Downloaded update checksum does not match.") + + +def _cancelled(cancel_event) -> bool: + return bool(cancel_event is not None and getattr(cancel_event, "is_set")()) + + +def download_to_file( + url: str, + target: str | os.PathLike, + *, + timeout: float = DEFAULT_DOWNLOAD_TIMEOUT_SECONDS, + expected_size: int | None = None, + cancel_event=None, + progress_callback=None, +) -> Path: + if _cancelled(cancel_event): + raise UpdateInstallError("cancelled", "Update cancelled.") + request = urllib.request.Request( + url, + headers={"User-Agent": f"Mouser/{APP_VERSION}"}, + ) + target_path = Path(target) + target_path.parent.mkdir(parents=True, exist_ok=True) + success = False + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + total = 0 + with open(target_path, "wb") as out: + while True: + if _cancelled(cancel_event): + raise UpdateInstallError("cancelled", "Update cancelled.") + chunk = response.read(1024 * 1024) + if not chunk: + break + total += len(chunk) + if expected_size is not None and total > int(expected_size): + raise UpdateInstallError( + "size_mismatch", + "Downloaded update size does not match.", + ) + out.write(chunk) + if progress_callback: + progress_callback(total) + success = True + finally: + if not success: + try: + target_path.unlink() + except OSError: + pass + return target_path + + +def fetch_json(url: str, *, timeout: float = 10.0): + request = urllib.request.Request( + url, + headers={"Accept": "application/json", "User-Agent": f"Mouser/{APP_VERSION}"}, + ) + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8-sig")) + + +def manifest_url_for_release(tag: str, repo: str = "TomBadash/Mouser") -> str: + override = os.environ.get(_UPDATE_MANIFEST_URL_ENV, "").strip() + if override: + return override + return f"https://github.com/{repo}/releases/download/{tag}/{manifest_name_for_version(tag)}" + + +def fetch_update_manifest_for_release( + tag: str, + *, + repo: str = "TomBadash/Mouser", + target_platform: str | None = None, + timeout: float = 10.0, + highest_trusted_build: int | None = None, +) -> UpdateManifest: + key = target_platform or platform_key() + payload = fetch_json(manifest_url_for_release(tag, repo), timeout=timeout) + return verify_update_manifest( + payload, + platform_key=key, + highest_trusted_build=highest_trusted_build, + ) + + +def _normalized_member_name(name: str) -> str: + raw = str(name or "") + if "\x00" in raw: + raise UpdateInstallError("unsafe_archive", "Update archive contains an unsafe path.") + if raw.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:", raw): + raise UpdateInstallError("unsafe_archive", "Update archive contains an absolute path.") + normalized = raw.replace("\\", "/") + if normalized.startswith("//"): + raise UpdateInstallError("unsafe_archive", "Update archive contains an unsafe path.") + parts = PurePosixPath(normalized).parts + if not parts or any(part in {"", ".", ".."} for part in parts): + raise UpdateInstallError("unsafe_archive", "Update archive contains path traversal.") + return "/".join(parts) + + +def _entry_mode(info: zipfile.ZipInfo) -> int: + return (info.external_attr >> 16) & 0o177777 + + +def _is_regular_or_dir(info: zipfile.ZipInfo) -> bool: + mode = _entry_mode(info) + if mode == 0: + return True + if stat.S_IFMT(mode) == 0: + return True + return stat.S_ISREG(mode) or stat.S_ISDIR(mode) + + +def validate_zip_archive( + zip_path: str | os.PathLike, + *, + requirements: ArchiveRequirements | None = None, + max_uncompressed_bytes: int = MAX_ARCHIVE_UNCOMPRESSED_BYTES, +) -> str: + requirements = requirements or ArchiveRequirements() + seen: set[str] = set() + roots: set[str] = set() + files: set[str] = set() + total_size = 0 + with zipfile.ZipFile(zip_path) as zf: + infos = zf.infolist() + if not infos: + raise UpdateInstallError("empty_archive", "Update archive is empty.") + bad_crc = zf.testzip() + if bad_crc: + raise UpdateInstallError("bad_archive", "Update archive failed integrity checks.") + for info in infos: + normalized = _normalized_member_name(info.filename) + key = normalized.casefold() + if key in seen: + raise UpdateInstallError("duplicate_archive_entry", "Update archive contains duplicate paths.") + seen.add(key) + if not _is_regular_or_dir(info): + raise UpdateInstallError("unsafe_archive", "Update archive contains unsupported file types.") + total_size += max(0, int(info.file_size)) + if total_size > max_uncompressed_bytes: + raise UpdateInstallError("archive_too_large", "Update archive is too large.") + roots.add(normalized.split("/", 1)[0]) + if not info.is_dir(): + files.add(normalized) + if len(roots) != 1: + raise UpdateInstallError("invalid_archive_root", "Update archive must contain one app folder.") + root = next(iter(roots)) + if requirements.require_windows_app: + if f"{root}/Mouser.exe" not in files: + raise UpdateInstallError("missing_entrypoint", "Update archive does not contain Mouser.exe.") + if not any(path.startswith(f"{root}/_internal/") for path in files): + raise UpdateInstallError("missing_runtime", "Update archive does not contain the runtime folder.") + return root + + +def _copy_zip_member(source, dest, *, declared_size: int, max_bytes: int) -> int: + written = 0 + declared = int(declared_size) + while True: + chunk = source.read(1024 * 1024) + if not chunk: + break + written += len(chunk) + if written > declared: + raise UpdateInstallError( + "bad_archive", + "Update archive entry size is invalid.", + ) + if written > int(max_bytes): + raise UpdateInstallError( + "archive_too_large", + "Update archive is too large.", + ) + dest.write(chunk) + if written != declared: + raise UpdateInstallError( + "bad_archive", + "Update archive entry size is invalid.", + ) + return written + + +def extract_validated_zip( + zip_path: str | os.PathLike, + stage_dir: str | os.PathLike, + *, + requirements: ArchiveRequirements | None = None, +) -> StagedUpdate: + root = validate_zip_archive(zip_path, requirements=requirements) + stage_path = Path(stage_dir) + if stage_path.exists(): + shutil.rmtree(stage_path) + stage_path.mkdir(parents=True) + success = False + try: + with zipfile.ZipFile(zip_path) as zf: + total_written = 0 + for info in zf.infolist(): + normalized = _normalized_member_name(info.filename) + target = stage_path.joinpath(*PurePosixPath(normalized).parts) + if info.is_dir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info) as source, open(target, "wb") as dest: + total_written += _copy_zip_member( + source, + dest, + declared_size=int(info.file_size), + max_bytes=MAX_ARCHIVE_UNCOMPRESSED_BYTES - total_written, + ) + success = True + finally: + if not success: + shutil.rmtree(stage_path, ignore_errors=True) + return StagedUpdate( + archive_path=Path(zip_path), + stage_dir=stage_path, + app_root=stage_path / root, + platform_key="", + asset_name=Path(zip_path).name, + ) + + +def same_volume_windows_stage_dir( + install_root: str | os.PathLike, + tag: str, + *, + pid: int | None = None, +) -> Path: + root = Path(install_root).resolve() + safe_tag = re.sub(r"[^A-Za-z0-9._-]+", "-", str(tag or "update")).strip(".-") + if not safe_tag: + safe_tag = "update" + return root.with_name(f".{root.name}.update-{safe_tag}-{pid or os.getpid()}") + + +def _probe_directory_writable(directory: Path) -> bool: + try: + handle, marker = tempfile.mkstemp( + prefix=".mouser-update-write-test-", + dir=str(directory), + ) + except OSError: + return False + try: + with os.fdopen(handle, "wb") as marker_file: + marker_file.write(b"mouser") + marker_file.flush() + os.fsync(marker_file.fileno()) + except OSError: + try: + os.close(handle) + except OSError: + pass + try: + Path(marker).unlink() + except OSError: + pass + return False + try: + Path(marker).unlink() + except OSError: + return False + return True + + +def locate_runtime( + *, + executable: str | os.PathLike | None = None, + sys_platform: str | None = None, + frozen: bool | None = None, + app_data_dir: str | os.PathLike | None = None, +) -> RuntimeLocation: + exe = Path(executable or sys.executable).resolve() + system = sys_platform or sys.platform + is_frozen = bool(getattr(sys, "frozen", False) if frozen is None else frozen) + if app_data_dir is None and (sys_platform or sys.platform).startswith("win"): + base = os.environ.get("LOCALAPPDATA") or str(Path.home() / "AppData" / "Local") + app_data_dir = Path(base) / "Mouser" / "updates" + data_dir = Path(app_data_dir or Path.home() / ".mouser" / "updates").resolve() + key = platform_key(system) + if not is_frozen: + return RuntimeLocation(exe, exe.parent, data_dir, False, key, False, "source run") + if system.startswith("win"): + root = exe.parent + if not (root / "Mouser.exe").exists() or not (root / "_internal").exists(): + return RuntimeLocation(exe, root, data_dir, True, key, False, "unsupported install layout") + if not _probe_directory_writable(root.parent): + return RuntimeLocation(exe, root, data_dir, True, key, False, "install path not writable") + return RuntimeLocation(exe, root, data_dir, True, key, True) + return RuntimeLocation(exe, exe.parent, data_dir, True, key, False, "manual install required") + + +def plan_install_for_platform( + manifest: UpdateManifest, + *, + runtime: RuntimeLocation | None = None, + staged: StagedUpdate | None = None, +) -> InstallPlan: + runtime = runtime or locate_runtime() + asset = manifest.assets.get(runtime.platform_key) + if asset is None: + return InstallPlan( + runtime.platform_key, + False, + "manual_fallback", + "No update is available for this platform.", + ) + if not runtime.platform_key.startswith("windows"): + system = "macOS" if runtime.platform_key.startswith("macos") else "Linux" + return InstallPlan( + runtime.platform_key, + False, + "manual_fallback", + f"A new Mouser release is available. Install manually on {system}.", + asset, + staged, + ) + if not runtime.update_supported: + return InstallPlan( + runtime.platform_key, + False, + "manual_fallback", + "Mouser cannot safely update this install automatically. Please install manually from the release page.", + asset, + staged, + ) + return InstallPlan( + runtime.platform_key, + True, + "ready_to_install", + "Mouser is ready to install the verified update.", + asset, + staged, + ) + + +class ProcessRunner: + def run(self, argv, *, cwd=None, env=None, timeout=None): + return subprocess.run( + list(argv), + cwd=cwd, + env=env, + timeout=timeout, + check=False, + capture_output=True, + text=True, + shell=False, + ) + + def popen(self, argv, *, cwd=None, env=None): + return subprocess.Popen( + list(argv), + cwd=cwd, + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + shell=False, + ) + + +def write_windows_update_plan(plan: WindowsUpdatePlan, path: str | os.PathLike) -> Path: + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(plan.to_dict(), indent=2, sort_keys=True), encoding="utf-8") + return target + + +def read_windows_update_plan(path: str | os.PathLike) -> WindowsUpdatePlan: + return WindowsUpdatePlan.from_dict(json.loads(Path(path).read_text(encoding="utf-8"))) + + +def _safe_result_marker_for_state(path: str | os.PathLike) -> Path: + return Path(path).resolve().parent / "last-update-result.txt" + + +def validate_windows_update_plan( + plan: WindowsUpdatePlan, + *, + state_path: str | os.PathLike | None = None, +) -> ValidatedWindowsUpdatePlan: + install_root = Path(plan.install_root).resolve() + staged_root = Path(plan.staged_root).resolve() + backup_root = Path(plan.backup_root).resolve() + result_marker = Path(plan.result_marker).resolve() + + def reject() -> None: + raise UpdateInstallError("invalid_plan", "Update plan is not valid.") + + if plan.current_pid <= 0: + reject() + if plan.executable_name != "Mouser.exe": + reject() + if not install_root.is_dir(): + reject() + if not (install_root / "Mouser.exe").is_file(): + reject() + if not (install_root / "_internal").is_dir(): + reject() + if not staged_root.is_dir(): + reject() + if staged_root.name != install_root.name: + reject() + if not (staged_root / "Mouser.exe").is_file(): + reject() + if not (staged_root / "_internal").is_dir(): + reject() + staged_parent = staged_root.parent + if staged_parent.parent != install_root.parent: + reject() + if not staged_parent.name.startswith(f".{install_root.name}.update-"): + reject() + if backup_root.parent != install_root.parent: + reject() + if not backup_root.name.startswith(f"{install_root.name}.backup-"): + reject() + if state_path is not None and result_marker != _safe_result_marker_for_state(state_path): + reject() + return ValidatedWindowsUpdatePlan( + plan=plan, + install_root=install_root, + staged_root=staged_root, + backup_root=backup_root, + result_marker=result_marker, + ) + + +def stage_windows_update_helper( + source_executable: str | os.PathLike, + helper_dir: str | os.PathLike, +) -> Path: + source = Path(source_executable).resolve() + if not source.is_file(): + raise UpdateInstallError("missing_helper_source", "Update helper could not be prepared.") + # The Windows release is a PyInstaller one-dir bundle; the helper exe needs + # its adjacent runtime directory to start after we move it out of install_root. + runtime_dir = source.parent / "_internal" + if not runtime_dir.is_dir(): + raise UpdateInstallError("missing_helper_runtime", "Update helper runtime is missing.") + target_dir = Path(helper_dir).resolve() + target_root = target_dir / "MouserUpdateHelper" + try: + target_root.relative_to(source.parent) + except ValueError: + pass + else: + raise UpdateInstallError("unsafe_helper_location", "Update helper must be outside the install folder.") + + target_dir.mkdir(parents=True, exist_ok=True) + temp_root = target_dir / f".{target_root.name}.{os.getpid()}.tmp" + if temp_root.exists(): + shutil.rmtree(temp_root) + try: + temp_root.mkdir(parents=True) + target = temp_root / source.name + shutil.copy2(source, target) + shutil.copytree(runtime_dir, temp_root / "_internal", symlinks=True) + if target_root.exists(): + shutil.rmtree(target_root) + temp_root.rename(target_root) + temp_root = None + return target_root / source.name + finally: + if temp_root is not None and temp_root.exists(): + shutil.rmtree(temp_root, ignore_errors=True) + + +def launch_windows_update_helper( + plan_path: str | os.PathLike, + *, + executable: str | os.PathLike | None = None, + helper_dir: str | os.PathLike | None = None, + runner: ProcessRunner | None = None, +) -> None: + source = Path(executable or sys.executable) + helper = ( + stage_windows_update_helper(source, helper_dir) + if helper_dir is not None + else source.resolve() + ) + exe = str(helper) + env = dict(os.environ) + env["PYINSTALLER_RESET_ENVIRONMENT"] = "1" + (runner or ProcessRunner()).popen( + [exe, "--mouser-apply-update", str(plan_path)], + cwd=str(Path(exe).resolve().parent), + env=env, + ) + + +def _windows_pid_exists(pid: int, *, kernel32=None, get_last_error=None) -> bool: + if kernel32 is None: + import ctypes + from ctypes import wintypes + + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD] + kernel32.OpenProcess.restype = wintypes.HANDLE + kernel32.WaitForSingleObject.argtypes = [wintypes.HANDLE, wintypes.DWORD] + kernel32.WaitForSingleObject.restype = wintypes.DWORD + kernel32.CloseHandle.argtypes = [wintypes.HANDLE] + kernel32.CloseHandle.restype = wintypes.BOOL + get_last_error = ctypes.get_last_error + + handle = kernel32.OpenProcess(_WINDOWS_SYNCHRONIZE, False, int(pid)) + if not handle: + error = int(get_last_error() if get_last_error is not None else 0) + if error == _WINDOWS_ERROR_INVALID_PARAMETER: + return False + if error == _WINDOWS_ERROR_ACCESS_DENIED: + return True + # Unknown OpenProcess failures are treated as alive so the installer + # times out instead of replacing files while the old process may run. + return True + try: + status = int(kernel32.WaitForSingleObject(handle, 0)) + if status == _WINDOWS_WAIT_TIMEOUT: + return True + if status == _WINDOWS_WAIT_OBJECT_0: + return False + if status == _WINDOWS_WAIT_FAILED: + return True + return True + finally: + kernel32.CloseHandle(handle) + + +def _posix_pid_exists(pid: int) -> bool: + try: + os.kill(pid, 0) + return True + except OSError as exc: + if exc.errno == errno.ESRCH: + return False + if exc.errno == errno.EPERM: + return True + return True + + +def _pid_exists( + pid: int, + *, + sys_platform: str | None = None, + windows_api=None, + get_last_error=None, +) -> bool: + if pid <= 0: + return False + if (sys_platform or sys.platform).startswith("win"): + return _windows_pid_exists( + pid, + kernel32=windows_api, + get_last_error=get_last_error, + ) + return _posix_pid_exists(pid) + + +def _apply_validated_windows_update_plan( + validated: ValidatedWindowsUpdatePlan, + *, + wait_timeout: float = 30.0, + runner: ProcessRunner | None = None, +) -> int: + plan = validated.plan + result_marker = validated.result_marker + start = time.time() + while _pid_exists(plan.current_pid): + if time.time() - start > wait_timeout: + _write_update_result( + result_marker, + "failed", + plan.target_version, + plan.target_build_number, + "Mouser did not exit", + ) + return 2 + time.sleep(0.2) + + install_root = validated.install_root + staged_root = validated.staged_root + backup_root = validated.backup_root + try: + if backup_root.exists(): + raise UpdateInstallError("backup_exists", "Update backup path already exists.") + install_root.rename(backup_root) + try: + staged_root.rename(install_root) + except Exception as install_exc: + restore_error = None + try: + if install_root.exists(): + shutil.rmtree(install_root) + except Exception as exc: + restore_error = exc + if restore_error is None: + for attempt in range(2): + try: + backup_root.rename(install_root) + _write_update_result( + result_marker, + "failed", + plan.target_version, + plan.target_build_number, + f"Update failed before replacement completed: {install_exc}", + ) + return 1 + except Exception as exc: + restore_error = exc + if attempt == 0: + time.sleep(0.5) + _write_update_result( + result_marker, + "failed", + plan.target_version, + plan.target_build_number, + ( + "Update failed and rollback could not be restored. " + f"Previous install remains at {backup_root}; " + f"expected install path is {install_root}. {restore_error}" + ), + ) + return 1 + _write_update_result( + result_marker, + "installed", + plan.target_version, + plan.target_build_number, + "", + ) + try: + staged_parent = staged_root.parent + if ( + staged_parent != install_root.parent + and staged_parent.name.startswith(f".{install_root.name}.update-") + ): + staged_parent.rmdir() + except OSError: + pass + executable = install_root / plan.executable_name + if executable.exists() and os.access(executable, os.X_OK): + env = dict(os.environ) + env["PYINSTALLER_RESET_ENVIRONMENT"] = "1" + (runner or ProcessRunner()).popen( + [str(executable)], cwd=str(install_root), env=env + ) + return 0 + except Exception as exc: + _write_update_result( + result_marker, + "failed", + plan.target_version, + plan.target_build_number, + str(exc), + ) + return 1 + + +def apply_windows_update_from_state( + path: str | os.PathLike, + *, + runner: ProcessRunner | None = None, +) -> int: + plan = None + safe_marker = _safe_result_marker_for_state(path) + try: + plan = read_windows_update_plan(path) + validated = validate_windows_update_plan(plan, state_path=path) + except Exception as exc: + _write_update_result( + safe_marker, + "failed", + plan.target_version if plan is not None else "", + plan.target_build_number if plan is not None else 0, + str(exc), + ) + return 1 + return _apply_validated_windows_update_plan(validated, runner=runner) + + +def _write_update_result( + path: str | os.PathLike, + status: str, + version: str, + build_number: int, + message: str, +) -> None: + target = Path(path) + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text( + json.dumps( + { + "status": status, + "version": version, + "build_number": int(build_number or 0), + "message": message, + }, + sort_keys=True, + ), + encoding="utf-8", + ) + + +def read_update_result(path: str | os.PathLike) -> dict[str, object] | None: + target = Path(path) + if not target.exists(): + return None + text = target.read_text(encoding="utf-8").strip() + if not text: + return None + if text == "installed": + return {"status": "installed", "version": "", "build_number": 0, "message": ""} + if text.startswith("failed:"): + return { + "status": "failed", + "version": "", + "build_number": 0, + "message": text.split(":", 1)[1].strip(), + } + data = json.loads(text) + if not isinstance(data, dict): + return None + return data + + +def cleanup_stale_update_state(app_data_dir: str | os.PathLike) -> None: + root = Path(app_data_dir) + pending = root / "pending-update.json" + try: + plan = read_windows_update_plan(pending) + except Exception: + plan = None + if plan is not None: + try: + install_root = Path(plan.install_root).resolve() + staged_root = Path(plan.staged_root).resolve() + staged_parent = staged_root.parent + if ( + staged_parent.name.startswith(f".{install_root.name}.update-") + and staged_parent.parent == install_root.parent + ): + shutil.rmtree(staged_parent, ignore_errors=True) + except Exception: + pass + for name in ("pending-update.json",): + try: + (root / name).unlink() + except FileNotFoundError: + pass + for name in ("downloads", "staged", "helper"): + path = root / name + if path.exists(): + shutil.rmtree(path, ignore_errors=True) + + +def prepare_downloaded_asset( + asset: UpdateAsset, + *, + download_dir: str | os.PathLike | None = None, + timeout: float = DEFAULT_DOWNLOAD_TIMEOUT_SECONDS, + cancel_event=None, + progress_callback=None, +) -> Path: + root = Path(download_dir or tempfile.mkdtemp(prefix="mouser-update-download-")) + target = root / asset.name + download_to_file( + asset.url, + target, + timeout=timeout, + expected_size=asset.size, + cancel_event=cancel_event, + progress_callback=progress_callback, + ) + if _cancelled(cancel_event): + raise UpdateInstallError("cancelled", "Update cancelled.") + verify_file(target, expected_sha256=asset.sha256, expected_size=asset.size) + return target diff --git a/core/updater.py b/core/updater.py index 23f7e89..cf6ad68 100644 --- a/core/updater.py +++ b/core/updater.py @@ -5,7 +5,9 @@ from dataclasses import dataclass from email.utils import parsedate_to_datetime import json +import os import re +import sys import time import urllib.error import urllib.request @@ -15,6 +17,7 @@ DEFAULT_RELEASE_REPO = "TomBadash/Mouser" _GITHUB_API = "https://api.github.com/repos/{repo}/releases/latest" +_LATEST_RELEASE_URL_ENV = "MOUSER_UPDATE_LATEST_RELEASE_URL" _USER_AGENT = f"Mouser/{APP_VERSION}" DEFAULT_AUTO_CHECK_INTERVAL_SECONDS = 24 * 60 * 60 @@ -35,6 +38,7 @@ class UpdateCheckState: backoff_until: float = 0.0 last_seen_latest_version: str = "" skipped_version: str = "" + highest_trusted_build: int = 0 @classmethod def from_dict(cls, data) -> "UpdateCheckState": @@ -54,6 +58,7 @@ def number(name): backoff_until=number("backoff_until"), last_seen_latest_version=str(data.get("last_seen_latest_version") or ""), skipped_version=str(data.get("skipped_version") or ""), + highest_trusted_build=int(number("highest_trusted_build")), ) def to_dict(self) -> dict[str, object]: @@ -64,6 +69,7 @@ def to_dict(self) -> dict[str, object]: "backoff_until": self.backoff_until, "last_seen_latest_version": self.last_seen_latest_version, "skipped_version": self.skipped_version, + "highest_trusted_build": self.highest_trusted_build, } @@ -130,6 +136,13 @@ def _retry_after_until(headers, now: float) -> float: return now +def _latest_release_url(repo: str) -> str: + override = os.environ.get(_LATEST_RELEASE_URL_ENV, "").strip() + if override: + return override + return _GITHUB_API.format(repo=repo) + + def _request(repo: str, state: UpdateCheckState) -> urllib.request.Request: headers = { "Accept": "application/vnd.github+json", @@ -139,7 +152,11 @@ def _request(repo: str, state: UpdateCheckState) -> urllib.request.Request: headers["If-None-Match"] = state.etag if state.last_modified: headers["If-Modified-Since"] = state.last_modified - return urllib.request.Request(_GITHUB_API.format(repo=repo), headers=headers) + return urllib.request.Request(_latest_release_url(repo), headers=headers) + + +def _read_json_response(response): + return json.loads(response.read().decode("utf-8-sig")) def _state_after_attempt(state: UpdateCheckState, now: float, **updates) -> UpdateCheckState: @@ -220,7 +237,18 @@ def check_latest_release( reachable=False, rate_limited=bool(backoff > now), ) - payload = json.loads(response.read().decode("utf-8")) + try: + payload = _read_json_response(response) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + print( + f"[update] release metadata could not be read: {exc}", + file=sys.stderr, + ) + return UpdateCheckResult( + None, + _state_after_attempt(state, now), + reachable=False, + ) release = _parse_release(payload) next_state = _state_after_attempt( state, diff --git a/main_qml.py b/main_qml.py index 7d8a969..a320f96 100644 --- a/main_qml.py +++ b/main_qml.py @@ -392,6 +392,10 @@ def _runtime_launch_path() -> str: def main(): _print_startup_times() _t5 = _time.perf_counter() + if len(sys.argv) >= 3 and sys.argv[1] == "--mouser-apply-update": + from core.update_installer import apply_windows_update_from_state + + raise SystemExit(apply_windows_update_from_state(sys.argv[2])) argv, hid_backend, start_hidden, force_show = _parse_cli_args(sys.argv) cfg = load_config() cfg_settings = cfg.get("settings", {}) diff --git a/tests/test_backend.py b/tests/test_backend.py index d5d9467..1310276 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -1,10 +1,14 @@ import copy +import json +from pathlib import Path import sys +import tempfile import unittest from types import SimpleNamespace from unittest.mock import patch from core.config import DEFAULT_CONFIG +from core.updater import UpdateCheckState try: from PySide6.QtCore import QCoreApplication, Qt, QUrl @@ -43,6 +47,9 @@ def __init__( self.gesture_callback = None self.status_callback = None self.debug_enabled = None + self.start_count = 0 + self.stop_count = 0 + self.start_error = None def set_profile_change_callback(self, cb): self.profile_callback = cb @@ -68,6 +75,14 @@ def set_status_callback(self, cb): def set_debug_enabled(self, enabled): self.debug_enabled = enabled + def start(self): + self.start_count += 1 + if self.start_error: + raise self.start_error + + def stop(self): + self.stop_count += 1 + @unittest.skipIf(Backend is None, "PySide6 not installed in test environment") class BackendDeviceLayoutTests(unittest.TestCase): @@ -157,6 +172,490 @@ def test_manual_update_check_triggers_immediate_fetch(self): start_check.assert_called_once_with(manual=True) + def test_prepare_latest_update_uses_manual_fallback_for_macos(self): + from pathlib import Path + + from core.update_installer import ( + APP_ID, + RuntimeLocation, + UpdateAsset, + UpdateManifest, + ) + + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + backend._update_state = UpdateCheckState(highest_trusted_build=30699) + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "macos-arm64": UpdateAsset( + "macos-arm64", + "Mouser-macOS.zip", + "https://example.test/Mouser-macOS.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=Path("/Applications/Mouser.app/Contents/MacOS/Mouser"), + install_root=Path("/Applications/Mouser.app"), + app_data_dir=Path("/tmp/mouser"), + frozen=True, + platform_key="macos-arm64", + update_supported=False, + ) + + with ( + patch("ui.backend.fetch_update_manifest_for_release", return_value=manifest) as fetch_manifest, + patch("ui.backend.locate_runtime", return_value=runtime), + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + + fetch_manifest.assert_called_once_with( + "v3.7.0", + repo="TomBadash/Mouser", + highest_trusted_build=30699, + ) + self.assertEqual(backend.updateInstallStatus, "manual_fallback") + self.assertFalse(backend.updateInstallCanInstall) + self.assertEqual(backend.updateInstallMessage, "macos") + + def test_prepare_latest_update_keeps_windows_install_hidden_by_default(self): + from pathlib import Path + + from core.update_installer import ( + APP_ID, + RuntimeLocation, + UpdateAsset, + UpdateManifest, + ) + + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "windows-x64": UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.test/Mouser-Windows.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=Path("C:/Mouser/Mouser.exe"), + install_root=Path("C:/Mouser"), + app_data_dir=Path("C:/Users/test/AppData/Local/Mouser"), + frozen=True, + platform_key="windows-x64", + update_supported=True, + ) + + with ( + patch.dict("os.environ", {}, clear=True), + patch("ui.backend.fetch_update_manifest_for_release", return_value=manifest), + patch("ui.backend.locate_runtime", return_value=runtime), + patch("ui.backend.prepare_downloaded_asset") as prepare_asset, + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + prepare_asset.assert_not_called() + self.assertEqual(backend.updateInstallStatus, "manual_fallback") + self.assertFalse(backend.updateInstallCanInstall) + self.assertEqual(backend.updateInstallMessage, "windows") + self.assertFalse(backend.updateInstallEnabled) + + def test_prepare_latest_update_uses_manual_fallback_for_unsupported_windows_runtime(self): + from pathlib import Path + + from core.update_installer import ( + APP_ID, + RuntimeLocation, + UpdateAsset, + UpdateManifest, + ) + + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "windows-x64": UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.test/Mouser-Windows.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=Path("C:/Program Files/Mouser/Mouser.exe"), + install_root=Path("C:/Program Files/Mouser"), + app_data_dir=Path("C:/Users/test/AppData/Local/Mouser"), + frozen=True, + platform_key="windows-x64", + update_supported=False, + reason="install path not writable", + ) + + with ( + patch.dict("os.environ", {"MOUSER_ENABLE_UPDATE_INSTALL": "1"}), + patch("ui.backend.fetch_update_manifest_for_release", return_value=manifest), + patch("ui.backend.locate_runtime", return_value=runtime), + patch("ui.backend.prepare_downloaded_asset") as prepare_asset, + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + + prepare_asset.assert_not_called() + self.assertEqual(backend.updateInstallStatus, "manual_fallback") + self.assertFalse(backend.updateInstallCanInstall) + self.assertEqual(backend.updateInstallMessage, "windows") + self.assertTrue(backend.updateInstallEnabled) + + def test_prepare_latest_update_stages_windows_bundle_next_to_install_root(self): + from core.update_installer import ( + APP_ID, + RuntimeLocation, + StagedUpdate, + UpdateAsset, + UpdateManifest, + ) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + install.mkdir() + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "windows-x64": UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.test/Mouser-Windows.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=install / "Mouser.exe", + install_root=install, + app_data_dir=root / "data", + frozen=True, + platform_key="windows-x64", + update_supported=True, + ) + archive = root / "Mouser-Windows.zip" + staged_app = root / ".Mouser.update-v3.7.0-1234" / "Mouser" + staged = StagedUpdate( + archive_path=archive, + stage_dir=staged_app.parent, + app_root=staged_app, + platform_key="windows-x64", + asset_name="Mouser-Windows.zip", + ) + + with ( + patch.dict("os.environ", {"MOUSER_ENABLE_UPDATE_INSTALL": "1"}), + patch("ui.backend.fetch_update_manifest_for_release", return_value=manifest), + patch("ui.backend.locate_runtime", return_value=runtime), + patch("ui.backend.prepare_downloaded_asset", return_value=archive), + patch("ui.backend.extract_validated_zip", return_value=staged) as extract_zip, + patch("ui.backend.os.getpid", return_value=1234), + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + + extract_zip.assert_called_once() + stage_arg = extract_zip.call_args.args[1] + self.assertEqual(stage_arg.parent, install.resolve().parent) + self.assertEqual(stage_arg.name, ".Mouser.update-v3.7.0-1234") + self.assertEqual(backend.updateInstallStatus, "ready_to_install") + self.assertTrue(backend.updateInstallCanInstall) + + def test_install_prepared_update_launches_helper_and_quits(self): + backend = self._make_backend() + backend._pending_update_plan_path = "/tmp/pending-update.json" + backend._update_install_can_install = True + + with ( + patch.dict("os.environ", {"MOUSER_ENABLE_UPDATE_INSTALL": "1"}), + patch("ui.backend.launch_windows_update_helper") as launch_helper, + patch("ui.backend.QCoreApplication.quit") as quit_app, + ): + backend.installPreparedUpdate() + + launch_helper.assert_called_once_with("/tmp/pending-update.json", helper_dir=None) + quit_app.assert_called_once() + self.assertEqual(backend.updateInstallStatus, "installing") + + def test_install_prepared_update_restarts_engine_when_helper_launch_fails(self): + engine = _FakeEngine() + backend = self._make_backend(engine=engine) + backend._pending_update_plan_path = "/tmp/pending-update.json" + backend._update_install_can_install = True + + with ( + patch.dict("os.environ", {"MOUSER_ENABLE_UPDATE_INSTALL": "1"}), + patch( + "ui.backend.launch_windows_update_helper", + side_effect=OSError("helper failed"), + ), + patch("ui.backend.QCoreApplication.quit") as quit_app, + ): + backend.installPreparedUpdate() + + self.assertEqual(engine.stop_count, 1) + self.assertEqual(engine.start_count, 1) + quit_app.assert_not_called() + self.assertEqual(backend.updateInstallStatus, "error") + + def test_cancel_update_preparation_sets_inline_cancelled_state(self): + backend = self._make_backend() + backend._update_install_status = "downloading" + + backend.cancelUpdatePreparation() + + self.assertTrue(backend._update_cancel.is_set()) + self.assertEqual(backend.updateInstallStatus, "cancelled") + + def test_cancel_before_prepare_worker_runs_does_not_fetch_metadata(self): + from core.update_installer import RuntimeLocation + + with tempfile.TemporaryDirectory() as tmp: + data = Path(tmp) + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + runtime = RuntimeLocation( + executable=data / "Mouser.exe", + install_root=data, + app_data_dir=data, + frozen=True, + platform_key="windows-x64", + update_supported=True, + ) + captured = {} + + def make_thread(*, target, name, daemon): + captured["target"] = target + captured["name"] = name + captured["daemon"] = daemon + return SimpleNamespace(start=lambda: None) + + with patch("ui.backend.threading.Thread", side_effect=make_thread): + backend.prepareLatestUpdate() + + self.assertEqual(captured["name"], "MouserPrepareUpdate") + self.assertTrue(captured["daemon"]) + + backend.cancelUpdatePreparation() + with ( + patch("ui.backend.fetch_update_manifest_for_release") as fetch_manifest, + patch("ui.backend.locate_runtime", return_value=runtime), + ): + captured["target"]() + _ensure_qapp().processEvents() + + fetch_manifest.assert_not_called() + self.assertEqual(backend.updateInstallStatus, "cancelled") + self.assertFalse(backend.updateInstallCanInstall) + + def test_cancel_during_metadata_fetch_keeps_cancelled_state(self): + from pathlib import Path + + from core.update_installer import APP_ID, RuntimeLocation, UpdateAsset, UpdateManifest + + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "macos-arm64": UpdateAsset( + "macos-arm64", + "Mouser-macOS.zip", + "https://example.test/Mouser-macOS.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=Path("/Applications/Mouser.app/Contents/MacOS/Mouser"), + install_root=Path("/Applications/Mouser.app"), + app_data_dir=Path("/tmp/mouser"), + frozen=True, + platform_key="macos-arm64", + update_supported=False, + ) + + def fetch_and_cancel(*args, **kwargs): + backend._update_cancel.set() + return manifest + + with ( + patch("ui.backend.fetch_update_manifest_for_release", side_effect=fetch_and_cancel), + patch("ui.backend.locate_runtime", return_value=runtime), + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + + self.assertEqual(backend.updateInstallStatus, "cancelled") + self.assertFalse(backend.updateInstallCanInstall) + + def test_prepare_latest_update_cleans_install_adjacent_stage_on_error(self): + from core.update_installer import ( + APP_ID, + RuntimeLocation, + UpdateAsset, + UpdateInstallError, + UpdateManifest, + ) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + install.mkdir() + backend = self._make_backend() + backend._latest_update_version = "3.7.0" + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "windows-x64": UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.test/Mouser-Windows.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=install / "Mouser.exe", + install_root=install, + app_data_dir=root / "data", + frozen=True, + platform_key="windows-x64", + update_supported=True, + ) + archive = root / "Mouser-Windows.zip" + stage_dir = root / ".Mouser.update-v3.7.0-1234" + + def fail_extract(_archive, stage_arg, **_kwargs): + self.assertEqual(stage_arg, stage_dir) + stage_arg.mkdir(parents=True) + (stage_arg / "partial.txt").write_text("partial", encoding="utf-8") + raise UpdateInstallError("bad_archive", "bad archive") + + with ( + patch.dict("os.environ", {"MOUSER_ENABLE_UPDATE_INSTALL": "1"}), + patch("ui.backend.fetch_update_manifest_for_release", return_value=manifest), + patch("ui.backend.locate_runtime", return_value=runtime), + patch("ui.backend.prepare_downloaded_asset", return_value=archive), + patch("ui.backend.same_volume_windows_stage_dir", return_value=stage_dir), + patch("ui.backend.extract_validated_zip", side_effect=fail_extract), + ): + backend._runPrepareLatestUpdate() + _ensure_qapp().processEvents() + + self.assertEqual(backend.updateInstallStatus, "error") + self.assertEqual(backend.updateInstallMessage, "bad_archive") + self.assertFalse(stage_dir.exists()) + + def test_startup_consumes_successful_update_marker_and_persists_build_number(self): + from core.update_installer import RuntimeLocation + + with tempfile.TemporaryDirectory() as tmp: + data = Path(tmp) + marker = data / "last-update-result.txt" + marker.write_text( + json.dumps( + {"status": "installed", "version": "3.7.0", "build_number": 30700} + ), + encoding="utf-8", + ) + runtime = RuntimeLocation( + executable=data / "Mouser.exe", + install_root=data, + app_data_dir=data, + frozen=True, + platform_key="windows-x64", + update_supported=True, + ) + cfg = copy.deepcopy(DEFAULT_CONFIG) + cfg.setdefault("settings", {})["update_check_state"] = { + "highest_trusted_build": 30600 + } + + with ( + patch("ui.backend.load_config", return_value=cfg), + patch("ui.backend.save_config") as save_config, + patch("ui.backend.supports_login_startup", return_value=False), + patch("ui.backend.locate_runtime", return_value=runtime), + ): + backend = Backend(engine=_FakeEngine()) + + self.assertFalse(marker.exists()) + self.assertEqual(backend.updateInstallStatus, "installed") + self.assertEqual(backend.updateInstallMessage, "3.7.0") + self.assertEqual(backend._update_state.highest_trusted_build, 30700) + save_config.assert_called_with(backend._cfg) + def test_update_check_result_persists_state_on_main_thread(self): backend = self._make_backend() state = { diff --git a/tests/test_locale_manager.py b/tests/test_locale_manager.py index b28d799..eb58810 100644 --- a/tests/test_locale_manager.py +++ b/tests/test_locale_manager.py @@ -21,6 +21,49 @@ def test_key_capture_error_messages_exist_in_all_locales(self): for key in required: self.assertTrue(strings[key].strip()) + def test_update_install_messages_exist_in_all_locales(self): + required = { + "scroll.update_idle", + "scroll.update_available", + "scroll.update_checking", + "scroll.update_downloading", + "scroll.update_verifying", + "scroll.update_ready", + "scroll.update_installing", + "scroll.update_installed", + "scroll.update_installed_version", + "scroll.update_cancelled", + "scroll.update_manual", + "scroll.update_manual_windows", + "scroll.update_manual_macos", + "scroll.update_manual_linux", + "scroll.update_no_asset", + "scroll.update_error", + "scroll.update_error_check_first", + "scroll.update_error_network_error", + "scroll.update_error_metadata_missing", + "scroll.update_error_metadata_invalid", + "scroll.update_error_permission_denied", + "scroll.update_error_file_error", + "scroll.update_error_install_failed", + "scroll.update_error_sha256_mismatch", + "scroll.update_error_size_mismatch", + "scroll.update_error_expired_metadata", + "scroll.update_error_older_build", + "scroll.update_check", + "scroll.update_download", + "scroll.update_cancel", + "scroll.update_verify", + "scroll.update_install", + "scroll.update_open_release", + } + + for locale, strings in _TRANSLATIONS.items(): + with self.subTest(locale=locale): + self.assertTrue(required.issubset(strings)) + for key in required: + self.assertTrue(strings[key].strip()) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_update_installer.py b/tests/test_update_installer.py new file mode 100644 index 0000000..ec5ccb0 --- /dev/null +++ b/tests/test_update_installer.py @@ -0,0 +1,1178 @@ +from datetime import datetime, timedelta, timezone +import errno +import io +import json +from pathlib import Path +import stat +import subprocess +import sys +import tempfile +import time +import unittest +import zipfile +from unittest.mock import patch + +from core.update_installer import ( + APP_ID, + ArchiveRequirements, + DEFAULT_DOWNLOAD_TIMEOUT_SECONDS, + UpdateInstallError, + WindowsUpdatePlan, + apply_windows_update_from_state, + build_number_from_version, + cleanup_stale_update_state, + download_to_file, + extract_validated_zip, + fetch_json, + launch_windows_update_helper, + locate_runtime, + manifest_name_for_version, + manifest_url_for_release, + platform_key, + plan_install_for_platform, + prepare_downloaded_asset, + read_update_result, + same_volume_windows_stage_dir, + sha256_file, + stage_windows_update_helper, + validate_zip_archive, + verify_file, + verify_update_manifest, + write_windows_update_plan, + _normalized_member_name, + _pid_exists, +) +from core.update_installer import UpdateManifest, UpdateAsset, RuntimeLocation + + +def _payload(**updates): + expires = (datetime.now(timezone.utc) + timedelta(days=1)).replace(microsecond=0) + payload = { + "schema": 1, + "app_id": APP_ID, + "channel": "stable", + "version": "3.7.0", + "tag": "v3.7.0", + "build_number": 30700, + "expires_at": expires.isoformat().replace("+00:00", "Z"), + "commit": "abc123", + "release_notes_url": "https://github.com/TomBadash/Mouser/releases/tag/v3.7.0", + "assets": { + "windows-x64": { + "name": "Mouser-Windows.zip", + "url": "https://github.com/TomBadash/Mouser/releases/download/v3.7.0/Mouser-Windows.zip", + "size": 123, + "sha256": "a" * 64, + }, + "macos-arm64": { + "name": "Mouser-macOS.zip", + "url": "https://github.com/TomBadash/Mouser/releases/download/v3.7.0/Mouser-macOS.zip", + "size": 456, + "sha256": "b" * 64, + }, + }, + } + payload.update(updates) + return payload + + +def _zip(path: Path, entries): + with zipfile.ZipFile(path, "w") as zf: + for name, data in entries: + zf.writestr(name, data) + + +class _CancelEvent: + def __init__(self, value=True): + self.value = value + + def is_set(self): + return self.value + + +class _FakeRunner: + def __init__(self): + self.popen_calls = [] + + def popen(self, argv, *, cwd=None, env=None): + self.popen_calls.append((list(argv), cwd, dict(env or {}))) + return object() + + +class _FakeKernel32: + def __init__(self, *, handle=100, wait_result=0x00000102): + self.handle = handle + self.wait_result = wait_result + self.open_calls = [] + self.wait_calls = [] + self.close_calls = [] + + def OpenProcess(self, access, inherit_handle, pid): + self.open_calls.append((access, inherit_handle, pid)) + return self.handle + + def WaitForSingleObject(self, handle, milliseconds): + self.wait_calls.append((handle, milliseconds)) + return self.wait_result + + def CloseHandle(self, handle): + self.close_calls.append(handle) + return True + + +class _FakeHTTPResponse: + def __init__(self, payload: bytes): + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return self._payload + + +class UpdateInstallerTests(unittest.TestCase): + def test_build_number_from_version_uses_semver_digits(self): + self.assertEqual(build_number_from_version("v3.7.0"), 30700) + self.assertEqual(build_number_from_version("3.7.12"), 30712) + with self.assertRaises(UpdateInstallError): + build_number_from_version("3.7") + + def test_manifest_name_matches_release_url_contract(self): + self.assertEqual(manifest_name_for_version("v3.7.0"), "mouser-v3.7.0-update.json") + self.assertTrue( + manifest_url_for_release("v3.7.0").endswith( + "/v3.7.0/mouser-v3.7.0-update.json" + ) + ) + + def test_manifest_url_can_use_test_endpoint_override(self): + with patch.dict( + "os.environ", + {"MOUSER_UPDATE_MANIFEST_URL": "http://127.0.0.1:8765/update.json"}, + ): + self.assertEqual( + manifest_url_for_release("v3.7.0"), + "http://127.0.0.1:8765/update.json", + ) + + def test_fetch_json_accepts_utf8_bom_response(self): + payload = b'\xef\xbb\xbf{"schema": 1, "app_id": "io.github.tombadash.mouser"}' + + with patch("urllib.request.urlopen", return_value=_FakeHTTPResponse(payload)): + parsed = fetch_json("https://example.test/update.json") + + self.assertEqual(parsed["schema"], 1) + self.assertEqual(parsed["app_id"], APP_ID) + + def test_update_manifest_verifies_selected_platform(self): + manifest = verify_update_manifest(_payload(), platform_key="windows-x64") + + self.assertEqual(manifest.version, "3.7.0") + self.assertEqual(manifest.build_number, 30700) + self.assertEqual(manifest.assets["windows-x64"].name, "Mouser-Windows.zip") + + def test_update_manifest_accepts_ci_payload_wrapper(self): + manifest = verify_update_manifest({"payload": _payload()}, platform_key="windows-x64") + + self.assertEqual(manifest.version, "3.7.0") + + def test_update_manifest_rejects_wrong_app(self): + payload = _payload(app_id="other.app") + + with self.assertRaises(UpdateInstallError) as ctx: + verify_update_manifest(payload, platform_key="windows-x64") + + self.assertEqual(ctx.exception.code, "wrong_app") + + def test_update_manifest_rejects_missing_selected_asset(self): + with self.assertRaises(UpdateInstallError) as ctx: + verify_update_manifest(_payload(), platform_key="linux-x64") + + self.assertEqual(ctx.exception.code, "missing_asset") + + def test_update_manifest_rejects_older_build_number(self): + with self.assertRaises(UpdateInstallError) as ctx: + verify_update_manifest( + _payload(), + platform_key="windows-x64", + highest_trusted_build=30701, + ) + + self.assertEqual(ctx.exception.code, "older_build") + + def test_platform_key_detects_common_architectures(self): + self.assertEqual(platform_key("win32", "AMD64"), "windows-x64") + self.assertEqual(platform_key("win32", "ARM64"), "windows-arm64") + self.assertEqual(platform_key("darwin", "arm64"), "macos-arm64") + self.assertEqual(platform_key("linux", "x86_64"), "linux-x64") + + def test_verify_file_checks_size_and_sha256(self): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "asset.zip" + path.write_bytes(b"abc") + + verify_file(path, expected_sha256=sha256_file(path), expected_size=3) + + with self.assertRaises(UpdateInstallError) as ctx: + verify_file(path, expected_sha256="0" * 64, expected_size=3) + self.assertEqual(ctx.exception.code, "sha256_mismatch") + + def test_archive_validator_accepts_windows_bundle_shape(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "Mouser-Windows.zip" + _zip( + archive, + [ + ("Mouser/Mouser.exe", b"exe"), + ("Mouser/_internal/runtime.dll", b"dll"), + ], + ) + + root = validate_zip_archive( + archive, + requirements=ArchiveRequirements(require_windows_app=True), + ) + + self.assertEqual(root, "Mouser") + + def test_archive_validator_rejects_traversal_and_duplicates(self): + with tempfile.TemporaryDirectory() as tmp: + traversal = Path(tmp) / "bad.zip" + _zip(traversal, [("../evil.txt", b"no")]) + with self.assertRaises(UpdateInstallError): + validate_zip_archive(traversal) + + duplicate = Path(tmp) / "dup.zip" + _zip(duplicate, [("Mouser/a.txt", b"1"), ("mouser/A.txt", b"2")]) + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(duplicate) + self.assertEqual(ctx.exception.code, "duplicate_archive_entry") + + def test_archive_validator_rejects_unsafe_path_forms(self): + cases = [ + ("/tmp/evil.txt", "unsafe_archive"), + ("C:/Temp/evil.txt", "unsafe_archive"), + ("//server/share/evil.txt", "unsafe_archive"), + ] + with tempfile.TemporaryDirectory() as tmp: + for name, code in cases: + with self.subTest(name=name): + archive = Path(tmp) / f"{abs(hash(name))}.zip" + _zip(archive, [(name, b"no")]) + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(archive) + self.assertEqual(ctx.exception.code, code) + + with self.assertRaises(UpdateInstallError) as ctx: + _normalized_member_name("Mouser/\x00evil.txt") + self.assertEqual(ctx.exception.code, "unsafe_archive") + + def test_archive_validator_rejects_symlink_entries(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "symlink.zip" + info = zipfile.ZipInfo("Mouser/link") + info.external_attr = (stat.S_IFLNK | 0o777) << 16 + with zipfile.ZipFile(archive, "w") as zf: + zf.writestr(info, "target") + + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(archive) + + self.assertEqual(ctx.exception.code, "unsafe_archive") + + def test_archive_validator_rejects_bad_crc(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "bad-crc.zip" + _zip(archive, [("Mouser/file.txt", b"CORRUPT_ME")]) + data = bytearray(archive.read_bytes()) + offset = data.index(b"CORRUPT_ME") + data[offset] = ord("X") + archive.write_bytes(data) + + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(archive) + + self.assertEqual(ctx.exception.code, "bad_archive") + + def test_archive_validator_rejects_empty_and_too_large_archives(self): + with tempfile.TemporaryDirectory() as tmp: + empty = Path(tmp) / "empty.zip" + with zipfile.ZipFile(empty, "w"): + pass + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(empty) + self.assertEqual(ctx.exception.code, "empty_archive") + + large = Path(tmp) / "large.zip" + _zip(large, [("Mouser/file.bin", b"123456")]) + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive(large, max_uncompressed_bytes=5) + self.assertEqual(ctx.exception.code, "archive_too_large") + + def test_archive_validator_rejects_incomplete_windows_bundle_shape(self): + with tempfile.TemporaryDirectory() as tmp: + no_exe = Path(tmp) / "no-exe.zip" + _zip(no_exe, [("Mouser/_internal/runtime.dll", b"dll")]) + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive( + no_exe, + requirements=ArchiveRequirements(require_windows_app=True), + ) + self.assertEqual(ctx.exception.code, "missing_entrypoint") + + no_runtime = Path(tmp) / "no-runtime.zip" + _zip(no_runtime, [("Mouser/Mouser.exe", b"exe")]) + with self.assertRaises(UpdateInstallError) as ctx: + validate_zip_archive( + no_runtime, + requirements=ArchiveRequirements(require_windows_app=True), + ) + self.assertEqual(ctx.exception.code, "missing_runtime") + + def test_extract_validated_zip_writes_only_inside_stage_dir(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "Mouser-Windows.zip" + _zip( + archive, + [ + ("Mouser/Mouser.exe", b"exe"), + ("Mouser/_internal/runtime.dll", b"dll"), + ], + ) + staged = extract_validated_zip( + archive, + Path(tmp) / "stage", + requirements=ArchiveRequirements(require_windows_app=True), + ) + + self.assertTrue((staged.app_root / "Mouser.exe").exists()) + self.assertTrue((staged.app_root / "_internal" / "runtime.dll").exists()) + + def test_extract_validated_zip_removes_partial_stage_on_failure(self): + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "Mouser-Windows.zip" + stage = Path(tmp) / "stage" + _zip( + archive, + [ + ("Mouser/Mouser.exe", b"exe"), + ("Mouser/_internal/runtime.dll", b"dll"), + ], + ) + + with ( + patch( + "core.update_installer._copy_zip_member", + side_effect=OSError("write failed"), + ), + self.assertRaises(OSError), + ): + extract_validated_zip( + archive, + stage, + requirements=ArchiveRequirements(require_windows_app=True), + ) + + self.assertFalse(stage.exists()) + + def test_extract_validated_zip_rejects_member_size_mismatch(self): + class _FakeInfo: + filename = "Mouser/Mouser.exe" + file_size = 1 + external_attr = 0 + + def is_dir(self): + return False + + class _FakeRuntimeInfo: + filename = "Mouser/_internal/runtime.dll" + file_size = 1 + external_attr = 0 + + def is_dir(self): + return False + + class _FakeZip: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def infolist(self): + return [_FakeInfo(), _FakeRuntimeInfo()] + + def testzip(self): + return None + + def open(self, info): + return io.BytesIO(b"too large") + + with tempfile.TemporaryDirectory() as tmp: + archive = Path(tmp) / "Mouser-Windows.zip" + archive.write_bytes(b"not used by fake zip") + stage = Path(tmp) / "stage" + + with ( + patch("core.update_installer.zipfile.ZipFile", _FakeZip), + self.assertRaises(UpdateInstallError) as ctx, + ): + extract_validated_zip( + archive, + stage, + requirements=ArchiveRequirements(require_windows_app=True), + ) + + self.assertEqual(ctx.exception.code, "bad_archive") + self.assertFalse(stage.exists()) + + def test_platform_install_strategy_returns_manual_fallback_for_macos(self): + manifest = UpdateManifest( + schema=1, + app_id=APP_ID, + channel="stable", + version="3.7.0", + tag="v3.7.0", + build_number=30700, + expires_at="2026-06-01T00:00:00Z", + commit="abc", + release_notes_url="https://example.test", + assets={ + "macos-arm64": UpdateAsset( + "macos-arm64", + "Mouser-macOS.zip", + "https://example.test/Mouser-macOS.zip", + 1, + "a" * 64, + ) + }, + ) + runtime = RuntimeLocation( + executable=Path("/Applications/Mouser.app/Contents/MacOS/Mouser"), + install_root=Path("/Applications/Mouser.app"), + app_data_dir=Path("/tmp/mouser"), + frozen=True, + platform_key="macos-arm64", + update_supported=False, + reason="manual", + ) + + plan = plan_install_for_platform(manifest, runtime=runtime) + + self.assertFalse(plan.can_install) + self.assertEqual(plan.status, "manual_fallback") + self.assertIn("manual", plan.message.lower()) + + def test_locate_runtime_classifies_source_windows_onedir_and_manual_platforms(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "repo" / "main.py" + source.parent.mkdir() + source.write_text("print('x')", encoding="utf-8") + + source_runtime = locate_runtime( + executable=source, + sys_platform="darwin", + frozen=False, + app_data_dir=root / "data", + ) + self.assertFalse(source_runtime.update_supported) + self.assertEqual(source_runtime.reason, "source run") + + install = root / "Mouser Install With Spaces" + install.mkdir() + exe = install / "Mouser.exe" + exe.write_text("exe", encoding="utf-8") + (install / "_internal").mkdir() + windows_runtime = locate_runtime( + executable=exe, + sys_platform="win32", + frozen=True, + app_data_dir=root / "data", + ) + self.assertTrue(windows_runtime.update_supported) + self.assertEqual(windows_runtime.install_root, install.resolve()) + + mac_runtime = locate_runtime( + executable=root / "Mouser.app" / "Contents" / "MacOS" / "Mouser", + sys_platform="darwin", + frozen=True, + app_data_dir=root / "data", + ) + self.assertFalse(mac_runtime.update_supported) + self.assertEqual(mac_runtime.reason, "manual install required") + + def test_locate_runtime_rejects_unsupported_windows_layout(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + exe = root / "Mouser.exe" + exe.write_text("exe", encoding="utf-8") + + runtime = locate_runtime( + executable=exe, + sys_platform="win32", + frozen=True, + app_data_dir=root / "data", + ) + + self.assertFalse(runtime.update_supported) + self.assertEqual(runtime.reason, "unsupported install layout") + + def test_locate_runtime_rejects_windows_install_parent_without_write_probe(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + install.mkdir() + exe = install / "Mouser.exe" + exe.write_text("exe", encoding="utf-8") + (install / "_internal").mkdir() + + with patch("core.update_installer._probe_directory_writable", return_value=False) as probe: + runtime = locate_runtime( + executable=exe, + sys_platform="win32", + frozen=True, + app_data_dir=root / "data", + ) + + probe.assert_called_once_with(install.resolve().parent) + self.assertFalse(runtime.update_supported) + self.assertEqual(runtime.reason, "install path not writable") + + def test_download_to_file_reports_progress_and_honors_pre_cancel(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "asset.zip" + source.write_bytes(b"downloaded") + target = root / "download" / "asset.zip" + progress = [] + + download_to_file(source.as_uri(), target, progress_callback=progress.append) + + self.assertEqual(target.read_bytes(), b"downloaded") + self.assertEqual(progress[-1], len(b"downloaded")) + + with self.assertRaises(UpdateInstallError) as ctx: + download_to_file(source.as_uri(), root / "cancel.zip", cancel_event=_CancelEvent()) + self.assertEqual(ctx.exception.code, "cancelled") + + def test_prepare_downloaded_asset_verifies_file_and_progress(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "asset.zip" + source.write_bytes(b"verified") + asset = UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + source.as_uri(), + source.stat().st_size, + sha256_file(source), + ) + progress = [] + + path = prepare_downloaded_asset( + asset, + download_dir=root / "downloads", + progress_callback=progress.append, + ) + + self.assertEqual(path.read_bytes(), b"verified") + self.assertEqual(progress[-1], source.stat().st_size) + + def test_prepare_downloaded_asset_uses_default_download_timeout(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + target_path = root / "downloads" / "Mouser-Windows.zip" + target_path.parent.mkdir() + target_path.write_bytes(b"verified") + asset = UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.invalid/Mouser-Windows.zip", + target_path.stat().st_size, + sha256_file(target_path), + ) + + with patch("core.update_installer.download_to_file") as download: + download.return_value = target_path + + prepare_downloaded_asset(asset, download_dir=target_path.parent) + + self.assertEqual( + download.call_args.kwargs["timeout"], + DEFAULT_DOWNLOAD_TIMEOUT_SECONDS, + ) + + def test_prepare_downloaded_asset_honors_explicit_download_timeout(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + target_path = root / "downloads" / "Mouser-Windows.zip" + target_path.parent.mkdir() + target_path.write_bytes(b"verified") + asset = UpdateAsset( + "windows-x64", + "Mouser-Windows.zip", + "https://example.invalid/Mouser-Windows.zip", + target_path.stat().st_size, + sha256_file(target_path), + ) + + with patch("core.update_installer.download_to_file") as download: + download.return_value = target_path + + prepare_downloaded_asset( + asset, + download_dir=target_path.parent, + timeout=7.5, + ) + + self.assertEqual(download.call_args.kwargs["timeout"], 7.5) + + def test_download_to_file_aborts_when_stream_exceeds_expected_size(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "asset.zip" + source.write_bytes(b"too large") + target = root / "download" / "asset.zip" + + with self.assertRaises(UpdateInstallError) as ctx: + download_to_file(source.as_uri(), target, expected_size=3) + + self.assertEqual(ctx.exception.code, "size_mismatch") + self.assertFalse(target.exists()) + + def test_same_volume_windows_stage_dir_is_adjacent_to_install_root(self): + install_root = Path("C:/Users/example/Mouser") + + stage_dir = same_volume_windows_stage_dir(install_root, "v3.7.0/test", pid=42) + + self.assertEqual(stage_dir.parent, install_root.resolve().parent) + self.assertEqual(stage_dir.name, ".Mouser.update-v3.7.0-test-42") + + def test_launch_windows_helper_copies_executable_outside_install_root(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" + install.mkdir() + exe = install / "Mouser.exe" + exe.write_text("exe", encoding="utf-8") + internal = install / "_internal" + internal.mkdir() + (internal / "python312.dll").write_text("runtime", encoding="utf-8") + plan_path = root / "pending-update.json" + plan_path.write_text("{}", encoding="utf-8") + helper_dir = root / "data" / "helper" + runner = _FakeRunner() + + helper = stage_windows_update_helper(exe, helper_dir) + self.assertTrue(helper.exists()) + self.assertEqual(helper, (helper_dir / "MouserUpdateHelper" / "Mouser.exe").resolve()) + self.assertNotEqual(helper.parent, install) + self.assertTrue((helper.parent / "_internal" / "python312.dll").exists()) + + launch_windows_update_helper( + plan_path, + executable=exe, + helper_dir=helper_dir, + runner=runner, + ) + + argv, cwd, env = runner.popen_calls[0] + self.assertEqual(Path(argv[0]).parent, (helper_dir / "MouserUpdateHelper").resolve()) + self.assertEqual(argv[1:], ["--mouser-apply-update", str(plan_path)]) + self.assertEqual(cwd, str((helper_dir / "MouserUpdateHelper").resolve())) + self.assertEqual(env["PYINSTALLER_RESET_ENVIRONMENT"], "1") + + def test_pid_exists_windows_uses_wait_handle_without_os_kill(self): + kernel32 = _FakeKernel32(handle=123, wait_result=0x00000102) + + with patch( + "core.update_installer.os.kill", + side_effect=AssertionError("Windows path must not signal the process"), + ): + self.assertTrue( + _pid_exists( + 4242, + sys_platform="win32", + windows_api=kernel32, + get_last_error=lambda: 0, + ) + ) + + self.assertEqual(kernel32.open_calls, [(0x00100000, False, 4242)]) + self.assertEqual(kernel32.wait_calls, [(123, 0)]) + self.assertEqual(kernel32.close_calls, [123]) + + def test_pid_exists_windows_reports_exited_and_invalid_processes(self): + exited = _FakeKernel32(handle=123, wait_result=0x00000000) + + self.assertFalse( + _pid_exists( + 4242, + sys_platform="win32", + windows_api=exited, + get_last_error=lambda: 0, + ) + ) + self.assertEqual(exited.close_calls, [123]) + + missing = _FakeKernel32(handle=0) + self.assertFalse( + _pid_exists( + 4242, + sys_platform="win32", + windows_api=missing, + get_last_error=lambda: 87, + ) + ) + + def test_pid_exists_posix_treats_permission_denied_as_alive(self): + with patch( + "core.update_installer.os.kill", + side_effect=PermissionError(errno.EPERM, "not permitted"), + ): + self.assertTrue(_pid_exists(4242, sys_platform="linux")) + + with patch( + "core.update_installer.os.kill", + side_effect=ProcessLookupError(errno.ESRCH, "not found"), + ): + self.assertFalse(_pid_exists(4242, sys_platform="linux")) + + with patch( + "core.update_installer.os.kill", + side_effect=OSError(errno.EIO, "unknown"), + ): + self.assertTrue(_pid_exists(4242, sys_platform="linux")) + + @unittest.skipUnless(sys.platform.startswith("win"), "Windows process check") + def test_pid_exists_windows_observes_live_process_without_terminating_it(self): + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(2)"], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + self.assertTrue(_pid_exists(proc.pid)) + time.sleep(0.1) + self.assertIsNone(proc.poll()) + finally: + proc.terminate() + proc.wait(timeout=5) + + def test_windows_helper_staging_requires_onedir_runtime_outside_install_root(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" + install.mkdir() + exe = install / "Mouser.exe" + exe.write_text("exe", encoding="utf-8") + + with self.assertRaises(UpdateInstallError) as ctx: + stage_windows_update_helper(exe, root / "helper") + self.assertEqual(ctx.exception.code, "missing_helper_runtime") + + (install / "_internal").mkdir() + with self.assertRaises(UpdateInstallError) as ctx: + stage_windows_update_helper(exe, install / "helper") + self.assertEqual(ctx.exception.code, "unsafe_helper_location") + self.assertFalse((install / "helper").exists()) + + def test_apply_windows_update_from_state_preserves_backup_and_installs_stage(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=77) + staged = stage_parent / "install" + backup = root / "install.backup-123" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "Mouser.exe").chmod(0o755) + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(backup), + result_marker=str(marker), + ) + write_windows_update_plan(plan, state) + + runner = _FakeRunner() + + with patch("core.update_installer._pid_exists", return_value=False): + result = apply_windows_update_from_state(state, runner=runner) + + self.assertEqual(result, 0) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "new") + self.assertEqual((backup / "Mouser.exe").read_text(encoding="utf-8"), "old") + self.assertEqual(read_update_result(marker)["status"], "installed") + self.assertEqual(len(runner.popen_calls), 1) + + def test_apply_windows_update_from_state_restores_backup_after_transient_rollback_failure(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=83) + staged = stage_parent / "Mouser" + backup = root / "Mouser.backup-123" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(backup), + result_marker=str(marker), + target_version="3.7.0", + target_build_number=30700, + ) + write_windows_update_plan(plan, state) + original_rename = Path.rename + rollback_attempts = {"count": 0} + + def rename_side_effect(path, target): + if Path(path) == staged.resolve() and Path(target) == install.resolve(): + raise OSError("replacement locked") + if Path(path) == backup.resolve() and Path(target) == install.resolve(): + rollback_attempts["count"] += 1 + if rollback_attempts["count"] == 1: + raise OSError("backup briefly locked") + return original_rename(path, target) + + with ( + patch("core.update_installer._pid_exists", return_value=False), + patch("core.update_installer.Path.rename", autospec=True, side_effect=rename_side_effect), + patch("core.update_installer.time.sleep"), + ): + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertEqual(rollback_attempts["count"], 2) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + update_result = read_update_result(marker) + self.assertEqual(update_result["status"], "failed") + self.assertIn("replacement locked", update_result["message"]) + + def test_apply_windows_update_from_state_reports_backup_path_when_rollback_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=84) + staged = stage_parent / "Mouser" + backup = root / "Mouser.backup-123" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(backup), + result_marker=str(marker), + target_version="3.7.0", + target_build_number=30700, + ) + write_windows_update_plan(plan, state) + original_rename = Path.rename + + def rename_side_effect(path, target): + if Path(path) == staged.resolve() and Path(target) == install.resolve(): + raise OSError("replacement locked") + if Path(path) == backup.resolve() and Path(target) == install.resolve(): + raise OSError("backup locked") + return original_rename(path, target) + + with ( + patch("core.update_installer._pid_exists", return_value=False), + patch("core.update_installer.Path.rename", autospec=True, side_effect=rename_side_effect), + patch("core.update_installer.time.sleep"), + ): + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertFalse(install.exists()) + self.assertTrue(backup.exists()) + update_result = read_update_result(marker) + self.assertEqual(update_result["status"], "failed") + self.assertIn(str(backup.resolve()), update_result["message"]) + self.assertIn(str(install.resolve()), update_result["message"]) + + def test_apply_windows_update_from_state_threads_runner_seam(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=78) + staged = stage_parent / "install" + backup = root / "install.backup-123" + marker = root / "result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "Mouser.exe").chmod(0o755) + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(backup), + result_marker=str(root / "last-update-result.txt"), + ) + state.write_text(json.dumps(plan.to_dict()), encoding="utf-8") + runner = _FakeRunner() + + with patch("core.update_installer._pid_exists", return_value=False): + result = apply_windows_update_from_state(state, runner=runner) + + self.assertEqual(result, 0) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "new") + self.assertEqual(len(runner.popen_calls), 1) + + def test_apply_windows_update_from_state_rejects_malformed_paths_before_rename(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" + staged = root / "staged" + outside = root / "outside" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir() + outside.mkdir() + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + (outside / "keep.txt").write_text("keep", encoding="utf-8") + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(outside), + result_marker=str(marker), + ) + write_windows_update_plan(plan, state) + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + self.assertEqual((outside / "keep.txt").read_text(encoding="utf-8"), "keep") + self.assertEqual(read_update_result(marker)["status"], "failed") + + def test_apply_windows_update_from_state_rejects_mismatched_staged_root_name(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=79) + staged = stage_parent / "Other" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(root / "Mouser.backup-123"), + result_marker=str(marker), + ) + write_windows_update_plan(plan, state) + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + self.assertEqual(read_update_result(marker)["status"], "failed") + + def test_apply_windows_update_from_state_rejects_existing_backup_without_deleting(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=80) + staged = stage_parent / "Mouser" + backup = root / "Mouser.backup-123" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + backup.mkdir() + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + (backup / "keep.txt").write_text("keep", encoding="utf-8") + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(backup), + result_marker=str(marker), + ) + write_windows_update_plan(plan, state) + + with patch("core.update_installer._pid_exists", return_value=False): + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + self.assertEqual((backup / "keep.txt").read_text(encoding="utf-8"), "keep") + self.assertEqual(read_update_result(marker)["status"], "failed") + + def test_apply_windows_update_from_state_rejects_external_result_marker_safely(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=81) + staged = stage_parent / "Mouser" + external = root / "external" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + external.mkdir() + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=123, + install_root=str(install), + staged_root=str(staged), + backup_root=str(root / "Mouser.backup-123"), + result_marker=str(external / "last-update-result.txt"), + ) + state.write_text(json.dumps(plan.to_dict()), encoding="utf-8") + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertFalse((external / "last-update-result.txt").exists()) + safe_marker = root / "last-update-result.txt" + self.assertEqual(read_update_result(safe_marker)["status"], "failed") + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + + def test_apply_windows_update_from_state_reports_unreadable_plan_to_safe_marker(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state = root / "pending-update.json" + state.write_text("{", encoding="utf-8") + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + marker = root / "last-update-result.txt" + update_result = read_update_result(marker) + self.assertEqual(update_result["status"], "failed") + self.assertEqual(update_result["version"], "") + self.assertEqual(update_result["build_number"], 0) + + def test_apply_windows_update_from_state_reports_incomplete_plan_to_safe_marker(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state = root / "pending-update.json" + state.write_text(json.dumps({"current_pid": 123}), encoding="utf-8") + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + marker = root / "last-update-result.txt" + update_result = read_update_result(marker) + self.assertEqual(update_result["status"], "failed") + self.assertEqual(update_result["version"], "") + self.assertEqual(update_result["build_number"], 0) + + def test_apply_windows_update_from_state_rejects_non_positive_pid(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "Mouser" + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=82) + staged = stage_parent / "Mouser" + marker = root / "last-update-result.txt" + state = root / "pending-update.json" + install.mkdir() + staged.mkdir(parents=True) + (install / "Mouser.exe").write_text("old", encoding="utf-8") + (install / "_internal").mkdir() + (staged / "Mouser.exe").write_text("new", encoding="utf-8") + (staged / "_internal").mkdir() + plan = WindowsUpdatePlan( + current_pid=0, + install_root=str(install), + staged_root=str(staged), + backup_root=str(root / "Mouser.backup-123"), + result_marker=str(marker), + ) + write_windows_update_plan(plan, state) + + result = apply_windows_update_from_state(state) + + self.assertEqual(result, 1) + self.assertEqual((install / "Mouser.exe").read_text(encoding="utf-8"), "old") + self.assertEqual(read_update_result(marker)["status"], "failed") + + def test_cleanup_stale_update_state_removes_pending_staging_and_helper(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + install = root / "install" / "Mouser" + install.mkdir(parents=True) + stage_parent = same_volume_windows_stage_dir(install, "v3.7.0", pid=99) + staged_root = stage_parent / "Mouser" + staged_root.mkdir(parents=True) + (staged_root / "Mouser.exe").write_text("new", encoding="utf-8") + plan = WindowsUpdatePlan( + current_pid=0, + install_root=str(install), + staged_root=str(staged_root), + backup_root=str(root / "install" / "Mouser.backup"), + result_marker=str(root / "last-update-result.txt"), + ) + (root / "pending-update.json").write_text( + json.dumps(plan.to_dict()), + encoding="utf-8", + ) + for name in ("downloads", "staged", "helper"): + path = root / name + path.mkdir() + (path / "file.txt").write_text("x", encoding="utf-8") + + cleanup_stale_update_state(root) + + self.assertFalse((root / "pending-update.json").exists()) + self.assertFalse(stage_parent.exists()) + for name in ("downloads", "staged", "helper"): + self.assertFalse((root / name).exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_updater.py b/tests/test_updater.py index e7b55dd..e99c058 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -1,4 +1,5 @@ import json +import io import unittest import urllib.error from unittest.mock import patch @@ -65,6 +66,36 @@ def test_fetch_latest_release_parses_github_response(self): self.assertIn("TomBadash/Mouser", request.full_url) self.assertEqual(request.get_header("User-agent"), f"Mouser/{APP_VERSION}") + def test_check_latest_release_accepts_utf8_bom_response(self): + payload = ( + b'\xef\xbb\xbf{"tag_name":"v3.7.1",' + b'"html_url":"https://github.com/TomBadash/Mouser/releases/tag/v3.7.1"}' + ) + + with patch("urllib.request.urlopen", return_value=_FakeResponse(payload)): + result = check_latest_release(now=10.0, manual=True) + + self.assertEqual(result.release.tag_name, "v3.7.1") + self.assertTrue(result.reachable) + + def test_fetch_latest_release_can_use_test_endpoint_override(self): + payload = { + "tag_name": "v3.7.1", + "html_url": "https://example.test/releases/v3.7.1", + } + with ( + patch.dict( + "os.environ", + {"MOUSER_UPDATE_LATEST_RELEASE_URL": "http://127.0.0.1:8765/release.json"}, + ), + patch("urllib.request.urlopen", return_value=_FakeResponse(payload)) as mocked, + ): + release = fetch_latest_release(timeout=1) + + self.assertEqual(release.tag_name, "v3.7.1") + request = mocked.call_args.args[0] + self.assertEqual(request.full_url, "http://127.0.0.1:8765/release.json") + def test_fetch_latest_release_returns_none_on_malformed_response(self): with patch( "urllib.request.urlopen", @@ -103,6 +134,20 @@ def test_check_latest_release_records_attempt_on_malformed_response(self): self.assertEqual(result.state.last_check, 50.0) self.assertEqual(result.state.etag, '"malformed"') + def test_check_latest_release_logs_unreadable_json_response(self): + stderr = io.StringIO() + + with ( + patch("urllib.request.urlopen", return_value=_FakeResponse(b"{")), + patch("sys.stderr", new=stderr), + ): + result = check_latest_release(now=60.0, manual=True) + + self.assertIsNone(result.release) + self.assertFalse(result.reachable) + self.assertEqual(result.state.last_check, 60.0) + self.assertIn("[update] release metadata could not be read:", stderr.getvalue()) + def test_check_latest_release_records_attempt_on_network_error(self): with patch("urllib.request.urlopen", side_effect=OSError("network down")): result = check_latest_release(now=75.0) diff --git a/tools/generate_update_manifest.py b/tools/generate_update_manifest.py new file mode 100644 index 0000000..538d2c4 --- /dev/null +++ b/tools/generate_update_manifest.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Generate Mouser update metadata from release assets.""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import json +from pathlib import Path + +from core.update_installer import ( + APP_ID, + MANIFEST_SCHEMA_VERSION, + STABLE_CHANNEL, + build_number_from_version, + sha256_file, +) + + +_ASSET_PLATFORM_KEYS = { + "Mouser-Windows.zip": "windows-x64", + "Mouser-macOS.zip": "macos-arm64", + "Mouser-macOS-intel.zip": "macos-x86_64", + "Mouser-Linux.zip": "linux-x64", + "Mouser-Windows-arm64.zip": "windows-arm64", +} + + +def _version_from_tag(tag: str) -> str: + return tag[1:] if tag.startswith("v") else tag + + +def build_payload(args) -> dict: + asset_dir = Path(args.asset_dir) + version = _version_from_tag(args.tag) + assets = {} + for name, platform_key in _ASSET_PLATFORM_KEYS.items(): + path = asset_dir / name + if not path.exists(): + continue + assets[platform_key] = { + "name": name, + "url": f"https://github.com/{args.repo}/releases/download/{args.tag}/{name}", + "size": path.stat().st_size, + "sha256": sha256_file(path), + } + if not assets: + raise SystemExit(f"No known Mouser assets found in {asset_dir}") + expires_at = ( + datetime.now(timezone.utc) + timedelta(days=int(args.expires_days)) + ).replace(microsecond=0) + return { + "schema": MANIFEST_SCHEMA_VERSION, + "app_id": APP_ID, + "channel": STABLE_CHANNEL, + "version": version, + "tag": args.tag, + "build_number": int(args.build_number or build_number_from_version(version)), + "expires_at": expires_at.isoformat().replace("+00:00", "Z"), + "commit": args.commit, + "release_notes_url": f"https://github.com/{args.repo}/releases/tag/{args.tag}", + "assets": assets, + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--repo", required=True) + parser.add_argument("--tag", required=True) + parser.add_argument("--commit", required=True) + parser.add_argument("--asset-dir", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--expires-days", default="30") + parser.add_argument("--build-number", default="") + args = parser.parse_args() + + payload = build_payload(args) + Path(args.output).write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/verify_update_manifest.py b/tools/verify_update_manifest.py new file mode 100644 index 0000000..2d9631e --- /dev/null +++ b/tools/verify_update_manifest.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +"""Verify Mouser update metadata for the selected platform.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from core.update_installer import platform_key, verify_update_manifest + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--manifest", required=True) + parser.add_argument("--platform-key", default=platform_key()) + args = parser.parse_args() + + data = json.loads(Path(args.manifest).read_text(encoding="utf-8")) + manifest = verify_update_manifest(data, platform_key=args.platform_key) + print( + f"verified {manifest.version} build {manifest.build_number} " + f"for {args.platform_key}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ui/backend.py b/ui/backend.py index d04595f..db48f74 100644 --- a/ui/backend.py +++ b/ui/backend.py @@ -4,13 +4,16 @@ """ import os +import json import re +import shutil import sys import threading import time +import urllib.error import webbrowser -from PySide6.QtCore import QMetaObject, QObject, Property, QTimer, Signal, Slot, Qt, QUrl +from PySide6.QtCore import QCoreApplication, QMetaObject, QObject, Property, QTimer, Signal, Slot, Qt, QUrl from core.accessibility import is_process_trusted from core.config import ( @@ -51,6 +54,21 @@ check_latest_release, is_newer, ) +from core.update_installer import ( + ArchiveRequirements, + UpdateInstallError, + WindowsUpdatePlan, + cleanup_stale_update_state, + extract_validated_zip, + fetch_update_manifest_for_release, + launch_windows_update_helper, + locate_runtime, + plan_install_for_platform, + prepare_downloaded_asset, + read_update_result, + same_volume_windows_stage_dir, + write_windows_update_plan, +) from core.version import APP_VERSION @@ -169,6 +187,11 @@ def _open_url(url: str) -> bool: return bool(webbrowser.open(qurl.toString())) +def _update_install_enabled() -> bool: + value = os.environ.get("MOUSER_ENABLE_UPDATE_INSTALL", "") + return value.strip().lower() in {"1", "true", "yes", "on"} + + class Backend(QObject): """QML-exposed backend that bridges the engine and configuration.""" @@ -191,6 +214,7 @@ class Backend(QObject): deviceLayoutChanged = Signal() knownAppsChanged = Signal() updateAvailable = Signal(str, str) + updateInstallChanged = Signal() # Internal cross-thread signals _profileSwitchRequest = Signal(str) @@ -203,6 +227,8 @@ class Backend(QObject): _statusMessageRequest = Signal(str) _updateAvailableRequest = Signal(str, str, bool, object) _updateCheckFinishedRequest = Signal(bool, bool, object) + _updateInstallStateRequest = Signal(str, str, bool) + _updateInstallProgressRequest = Signal(int) def __init__(self, engine=None, parent=None, root_dir=None): super().__init__(parent) @@ -240,6 +266,14 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._latest_update_url = "" self._latest_update_version = "" self._update_check_in_progress = False + self._update_install_status = "idle" + self._update_install_message = "" + self._update_install_can_install = False + self._update_install_progress = 0 + self._update_cancel = threading.Event() + self._pending_update_plan = None + self._pending_update_plan_path = None + self._pending_update_helper_dir = None self._update_state = UpdateCheckState.from_dict( self._cfg.get("settings", {}).get("update_check_state", {}) ) @@ -268,6 +302,10 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._handleUpdateAvailable, Qt.QueuedConnection) self._updateCheckFinishedRequest.connect( self._handleUpdateCheckFinished, Qt.QueuedConnection) + self._updateInstallStateRequest.connect( + self._handleUpdateInstallState, Qt.QueuedConnection) + self._updateInstallProgressRequest.connect( + self._handleUpdateInstallProgress, Qt.QueuedConnection) # Wire engine callbacks if engine: @@ -313,6 +351,8 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._cfg.setdefault("settings", {})["start_at_login"] = False self._sync_connected_device_info() self._configureUpdateChecks() + self._consumeUpdateResultMarker() + self._cleanupStaleUpdatePreparation() # ── Properties ───────────────────────────────────────────── @@ -492,6 +532,47 @@ def debugMode(self): def checkForUpdates(self): return bool(self._cfg.get("settings", {}).get("check_for_updates", True)) + @Property(bool, constant=True) + def isWindows(self): + return sys.platform.startswith("win") + + @Property(bool, constant=True) + def isLinux(self): + return sys.platform.startswith("linux") + + @Property(str, notify=updateInstallChanged) + def latestUpdateVersion(self): + return self._latest_update_version + + @Property(str, notify=updateInstallChanged) + def updateInstallStatus(self): + return self._update_install_status + + @Property(str, notify=updateInstallChanged) + def updateInstallMessage(self): + return self._update_install_message + + @Property(int, notify=updateInstallChanged) + def updateInstallProgress(self): + return int(self._update_install_progress) + + @Property(bool, notify=updateInstallChanged) + def updateInstallCanInstall(self): + return self._update_install_can_install + + @Property(bool, constant=True) + def updateInstallEnabled(self): + return _update_install_enabled() + + @Property(bool, notify=updateInstallChanged) + def updateInstallInProgress(self): + return self._update_install_status in { + "checking", + "downloading", + "verifying", + "installing", + } + @Property(bool, notify=debugEventsEnabledChanged) def debugEventsEnabled(self): return self._debug_events_enabled @@ -768,6 +849,12 @@ def _handleUpdateAvailable(self, version, url, manual, state_data): self._persistUpdateCheckState(state_data) self._latest_update_version = str(version or "") self._latest_update_url = str(url or "") + self._update_install_status = "available" + self._update_install_message = "" + self._update_install_can_install = False + self._pending_update_plan = None + self._pending_update_plan_path = None + self.updateInstallChanged.emit() self.updateAvailable.emit(self._latest_update_version, self._latest_update_url) self.statusMessage.emit(f"Mouser {self._latest_update_version} is available") @@ -781,6 +868,208 @@ def _handleUpdateCheckFinished(self, manual, reachable, state_data): else: self.statusMessage.emit("Could not check for updates") + def _setUpdateInstallState(self, status, message="", can_install=False): + self._update_install_status = str(status or "idle") + self._update_install_message = str(message or "") + self._update_install_can_install = bool(can_install) + if status not in {"downloading", "verifying", "ready_to_install"}: + self._update_install_progress = 0 + self.updateInstallChanged.emit() + + @Slot(str, str, bool) + def _handleUpdateInstallState(self, status, message, can_install): + self._setUpdateInstallState(status, message, can_install) + + @Slot(int) + def _handleUpdateInstallProgress(self, value): + self._update_install_progress = max(0, min(100, int(value))) + self.updateInstallChanged.emit() + + def _trustedBuildNumber(self): + try: + return int(self._update_state.highest_trusted_build or 0) + except (TypeError, ValueError): + return 0 + + def _updateErrorCode(self, exc): + if isinstance(exc, UpdateInstallError): + return exc.code + if isinstance(exc, urllib.error.HTTPError): + return "metadata_missing" if exc.code == 404 else "network_error" + if isinstance(exc, (urllib.error.URLError, TimeoutError)): + return "network_error" + if isinstance(exc, (json.JSONDecodeError, UnicodeDecodeError)): + return "metadata_invalid" + if isinstance(exc, PermissionError): + return "permission_denied" + if isinstance(exc, OSError): + return "file_error" + return "error" + + def _updateProgressCallback(self, expected_size): + def _progress(downloaded): + if expected_size: + self._updateInstallProgressRequest.emit( + int(min(100, max(0, downloaded * 100 / expected_size))) + ) + + return _progress + + def _raiseIfUpdateCancelled(self): + if self._update_cancel.is_set(): + raise UpdateInstallError("cancelled", "Update cancelled.") + + def _cleanupUpdatePreparation(self, stage_dir=None): + try: + cleanup_stale_update_state(locate_runtime().app_data_dir) + except Exception: + pass + if stage_dir is not None: + try: + shutil.rmtree(stage_dir, ignore_errors=True) + except Exception: + pass + + def _consumeUpdateResultMarker(self): + try: + runtime = locate_runtime() + marker = runtime.app_data_dir / "last-update-result.txt" + result = read_update_result(marker) + if not result: + return + try: + marker.unlink() + except OSError: + pass + status = str(result.get("status") or "") + version = str(result.get("version") or "") + build_number = int(result.get("build_number") or 0) + if status == "installed": + if build_number > self._trustedBuildNumber(): + next_state = UpdateCheckState( + **{ + **self._update_state.to_dict(), + "highest_trusted_build": build_number, + } + ) + self._persistUpdateCheckState(next_state.to_dict()) + self._setUpdateInstallState("installed", version, False) + self.statusMessage.emit( + f"Updated to {version}" if version else "Update installed" + ) + elif status == "failed": + self._setUpdateInstallState("error", "install_failed", False) + except Exception as exc: + print(f"[update] failed to consume update result marker: {exc}") + + def _cleanupStaleUpdatePreparation(self): + try: + cleanup_stale_update_state(locate_runtime().app_data_dir) + except Exception: + pass + + def _runPrepareLatestUpdate(self): + stage_dir = None + try: + version = self._latest_update_version + if not version: + self._updateInstallStateRequest.emit( + "error", "check_first", False + ) + return + tag = version if version.startswith("v") else f"v{version}" + self._raiseIfUpdateCancelled() + self._updateInstallStateRequest.emit("checking", "", False) + self._raiseIfUpdateCancelled() + manifest = fetch_update_manifest_for_release( + tag, + repo=DEFAULT_RELEASE_REPO, + highest_trusted_build=self._trustedBuildNumber(), + ) + self._raiseIfUpdateCancelled() + runtime = locate_runtime() + self._raiseIfUpdateCancelled() + asset = manifest.assets.get(runtime.platform_key) + if asset is None: + self._raiseIfUpdateCancelled() + self._updateInstallStateRequest.emit( + "manual_fallback", + "no_asset", + False, + ) + return + if not runtime.platform_key.startswith("windows"): + plan_install_for_platform(manifest, runtime=runtime) + self._raiseIfUpdateCancelled() + platform_message = ( + "macos" if runtime.platform_key.startswith("macos") else "linux" + ) + self._updateInstallStateRequest.emit( + "manual_fallback", platform_message, False + ) + return + if not self.updateInstallEnabled: + self._raiseIfUpdateCancelled() + self._updateInstallStateRequest.emit( + "manual_fallback", "windows", False + ) + return + if not runtime.update_supported: + self._raiseIfUpdateCancelled() + self._updateInstallStateRequest.emit( + "manual_fallback", "windows", False + ) + return + + self._updateInstallStateRequest.emit("downloading", "", False) + archive_path = prepare_downloaded_asset( + asset, + download_dir=runtime.app_data_dir / "downloads" / tag, + cancel_event=self._update_cancel, + progress_callback=self._updateProgressCallback(asset.size), + ) + self._raiseIfUpdateCancelled() + self._updateInstallStateRequest.emit("verifying", "", False) + stage_dir = same_volume_windows_stage_dir(runtime.install_root, tag) + staged = extract_validated_zip( + archive_path, + stage_dir, + requirements=ArchiveRequirements(require_windows_app=True), + ) + self._raiseIfUpdateCancelled() + plan = plan_install_for_platform(manifest, runtime=runtime, staged=staged) + if not plan.can_install or not plan.staged: + self._updateInstallStateRequest.emit( + plan.status, plan.message, plan.can_install + ) + return + backup_root = runtime.install_root.with_name( + f"{runtime.install_root.name}.backup-{int(time.time())}" + ) + result_marker = runtime.app_data_dir / "last-update-result.txt" + state_path = runtime.app_data_dir / "pending-update.json" + windows_plan = WindowsUpdatePlan( + current_pid=os.getpid(), + install_root=str(runtime.install_root), + staged_root=str(plan.staged.app_root), + backup_root=str(backup_root), + result_marker=str(result_marker), + target_version=manifest.version, + target_build_number=manifest.build_number, + ) + write_windows_update_plan(windows_plan, state_path) + self._pending_update_plan = windows_plan + self._pending_update_plan_path = state_path + self._pending_update_helper_dir = runtime.app_data_dir / "helper" / tag + self._updateInstallStateRequest.emit("ready_to_install", "", True) + except Exception as exc: + code = self._updateErrorCode(exc) + self._cleanupUpdatePreparation(stage_dir) + if code == "cancelled": + self._updateInstallStateRequest.emit("cancelled", "", False) + return + self._updateInstallStateRequest.emit("error", code, False) + # ── Slots ────────────────────────────────────────────────── @Slot(str, str) @@ -835,6 +1124,59 @@ def openLatestReleasePage(self): ) _open_url(url) + @Slot() + def prepareLatestUpdate(self): + if self.updateInstallInProgress: + self.statusMessage.emit("Update is already in progress") + return + self._update_cancel.clear() + self._setUpdateInstallState("checking") + thread = threading.Thread( + target=self._runPrepareLatestUpdate, + name="MouserPrepareUpdate", + daemon=True, + ) + thread.start() + + @Slot() + def cancelUpdatePreparation(self): + if self._update_install_status not in {"checking", "downloading", "verifying"}: + return + self._update_cancel.set() + self._setUpdateInstallState("cancelled") + + @Slot() + def installPreparedUpdate(self): + if not self.updateInstallEnabled: + self.statusMessage.emit("Open the release page to install manually") + return + if not self._pending_update_plan_path or not self._update_install_can_install: + self.statusMessage.emit("Update is not ready to install") + return + self._setUpdateInstallState("installing") + engine_stopped = False + try: + if self._engine: + # Release mouse hooks and HID grabs before replacing binaries. + self._engine.stop() + engine_stopped = True + launch_windows_update_helper( + self._pending_update_plan_path, + helper_dir=self._pending_update_helper_dir, + ) + except Exception as exc: + if engine_stopped and self._engine: + try: + self._engine.start() + except Exception as restart_exc: + print( + f"[update] failed to restart remapping after update error: {restart_exc}", + file=sys.stderr, + ) + self._setUpdateInstallState("error", self._updateErrorCode(exc), False) + return + QCoreApplication.quit() + @Slot(bool) def setStartAtLogin(self, value): enabled = bool(value) diff --git a/ui/locale_manager.py b/ui/locale_manager.py index 6b6bc36..d6d1f34 100644 --- a/ui/locale_manager.py +++ b/ui/locale_manager.py @@ -131,6 +131,39 @@ "scroll.start_minimized": "Start minimized", "scroll.check_for_updates": "Check for updates", "scroll.check_for_updates_desc": "Notify when a newer Mouser release is available. Downloads and installation stay manual.", + "scroll.update_idle": "Mouser can check for new releases.", + "scroll.update_available": "Mouser %1 is available.", + "scroll.update_checking": "Checking for updates...", + "scroll.update_downloading": "Downloading the update...", + "scroll.update_verifying": "Verifying the update...", + "scroll.update_ready": "Ready to install after Mouser quits.", + "scroll.update_installing": "Installing update...", + "scroll.update_installed": "Update installed.", + "scroll.update_installed_version": "Updated to %1.", + "scroll.update_cancelled": "Update cancelled.", + "scroll.update_manual": "A new Mouser release is available. Download it from the release page and install manually.", + "scroll.update_manual_windows": "A new Mouser release is available. Download it from the release page and install manually on Windows.", + "scroll.update_manual_macos": "A new Mouser release is available. Download it from the release page and install manually on macOS.", + "scroll.update_manual_linux": "A new Mouser release is available. Download it from the release page and install manually on Linux.", + "scroll.update_no_asset": "No update package is available for this computer.", + "scroll.update_error": "Update could not be prepared.", + "scroll.update_error_check_first": "Check for updates first.", + "scroll.update_error_network_error": "Mouser could not reach the update service. Try again later.", + "scroll.update_error_metadata_missing": "Update details are not ready yet. Open the release page to install manually.", + "scroll.update_error_metadata_invalid": "Update details could not be read. Open the release page to install manually.", + "scroll.update_error_permission_denied": "Mouser does not have permission to prepare the update.", + "scroll.update_error_file_error": "Mouser could not write the update files.", + "scroll.update_error_install_failed": "The update did not finish. Open the release page to install manually.", + "scroll.update_error_sha256_mismatch": "The download did not pass verification. Try again later.", + "scroll.update_error_size_mismatch": "The download did not pass verification. Try again later.", + "scroll.update_error_expired_metadata": "Update details are out of date. Try again later.", + "scroll.update_error_older_build": "Mouser rejected an older update.", + "scroll.update_check": "Check", + "scroll.update_download": "Download", + "scroll.update_verify": "Verify", + "scroll.update_install": "Install", + "scroll.update_cancel": "Cancel", + "scroll.update_open_release": "Open release", "scroll.scroll_speed": "Scroll Speed", "scroll.scroll_speed_desc": "Adjust how fast the page scrolls per wheel click. 1.0\u00d7 is the system default.", "scroll.scroll_speed_presets": "Presets:", @@ -312,6 +345,39 @@ "scroll.start_minimized": "\u542f\u52a8\u65f6\u6700\u5c0f\u5316", "scroll.check_for_updates": "\u68c0\u67e5\u66f4\u65b0", "scroll.check_for_updates_desc": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u65f6\u901a\u77e5\u3002\u4e0b\u8f7d\u548c\u5b89\u88c5\u4ecd\u9700\u624b\u52a8\u5b8c\u6210\u3002", + "scroll.update_idle": "Mouser \u53ef\u4ee5\u68c0\u67e5\u65b0\u7248\u672c\u3002", + "scroll.update_available": "Mouser %1 \u53ef\u7528\u3002", + "scroll.update_checking": "\u6b63\u5728\u68c0\u67e5\u66f4\u65b0...", + "scroll.update_downloading": "\u6b63\u5728\u4e0b\u8f7d\u66f4\u65b0...", + "scroll.update_verifying": "\u6b63\u5728\u9a8c\u8bc1\u66f4\u65b0...", + "scroll.update_ready": "Mouser \u9000\u51fa\u540e\u5373\u53ef\u5b89\u88c5\u3002", + "scroll.update_installing": "\u6b63\u5728\u5b89\u88c5\u66f4\u65b0...", + "scroll.update_installed": "\u66f4\u65b0\u5df2\u5b89\u88c5\u3002", + "scroll.update_installed_version": "\u5df2\u66f4\u65b0\u5230 %1\u3002", + "scroll.update_cancelled": "\u66f4\u65b0\u5df2\u53d6\u6d88\u3002", + "scroll.update_manual": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8bf7\u4ece\u53d1\u5e03\u9875\u4e0b\u8f7d\u5e76\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_manual_windows": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8bf7\u4ece\u53d1\u5e03\u9875\u4e0b\u8f7d\u5e76\u5728 Windows \u4e0a\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_manual_macos": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8bf7\u4ece\u53d1\u5e03\u9875\u4e0b\u8f7d\u5e76\u5728 macOS \u4e0a\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_manual_linux": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8bf7\u4ece\u53d1\u5e03\u9875\u4e0b\u8f7d\u5e76\u5728 Linux \u4e0a\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_no_asset": "\u6b64\u7535\u8111\u6ca1\u6709\u53ef\u7528\u7684\u66f4\u65b0\u5305\u3002", + "scroll.update_error": "\u65e0\u6cd5\u51c6\u5907\u66f4\u65b0\u3002", + "scroll.update_error_check_first": "\u8bf7\u5148\u68c0\u67e5\u66f4\u65b0\u3002", + "scroll.update_error_network_error": "Mouser \u65e0\u6cd5\u8fde\u63a5\u66f4\u65b0\u670d\u52a1\u3002\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002", + "scroll.update_error_metadata_missing": "\u66f4\u65b0\u4fe1\u606f\u5c1a\u672a\u51c6\u5907\u597d\u3002\u8bf7\u6253\u5f00\u53d1\u5e03\u9875\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_error_metadata_invalid": "\u65e0\u6cd5\u8bfb\u53d6\u66f4\u65b0\u4fe1\u606f\u3002\u8bf7\u6253\u5f00\u53d1\u5e03\u9875\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_error_permission_denied": "Mouser \u6ca1\u6709\u51c6\u5907\u66f4\u65b0\u7684\u6743\u9650\u3002", + "scroll.update_error_file_error": "Mouser \u65e0\u6cd5\u5199\u5165\u66f4\u65b0\u6587\u4ef6\u3002", + "scroll.update_error_install_failed": "\u66f4\u65b0\u672a\u5b8c\u6210\u3002\u8bf7\u6253\u5f00\u53d1\u5e03\u9875\u624b\u52a8\u5b89\u88c5\u3002", + "scroll.update_error_sha256_mismatch": "\u4e0b\u8f7d\u672a\u901a\u8fc7\u9a8c\u8bc1\u3002\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002", + "scroll.update_error_size_mismatch": "\u4e0b\u8f7d\u672a\u901a\u8fc7\u9a8c\u8bc1\u3002\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002", + "scroll.update_error_expired_metadata": "\u66f4\u65b0\u4fe1\u606f\u5df2\u8fc7\u671f\u3002\u8bf7\u7a0d\u540e\u91cd\u8bd5\u3002", + "scroll.update_error_older_build": "Mouser \u5df2\u62d2\u7edd\u8f83\u65e7\u7684\u66f4\u65b0\u3002", + "scroll.update_check": "\u68c0\u67e5", + "scroll.update_download": "\u4e0b\u8f7d", + "scroll.update_verify": "\u9a8c\u8bc1", + "scroll.update_install": "\u5b89\u88c5", + "scroll.update_cancel": "\u53d6\u6d88", + "scroll.update_open_release": "\u6253\u5f00\u53d1\u5e03\u9875", "scroll.scroll_speed": "\u6eda\u8f6e\u901f\u5ea6", "scroll.scroll_speed_desc": "\u8c03\u6574\u6bcf\u6b21\u6eda\u8f6e\u6eda\u52a8\u7684\u9875\u9762\u79fb\u52a8\u901f\u5ea6\u30021.0\u00d7 \u4e3a\u7cfb\u7edf\u9ed8\u8ba4\u3002", "scroll.scroll_speed_presets": "\u9884\u8bbe\uff1a", @@ -487,6 +553,39 @@ "scroll.start_minimized": "\u555f\u52d5\u6642\u6700\u5c0f\u5316", "scroll.check_for_updates": "\u6aa2\u67e5\u66f4\u65b0", "scroll.check_for_updates_desc": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u6642\u901a\u77e5\u3002\u4e0b\u8f09\u548c\u5b89\u88dd\u4ecd\u9700\u624b\u52d5\u5b8c\u6210\u3002", + "scroll.update_idle": "Mouser \u53ef\u4ee5\u6aa2\u67e5\u65b0\u7248\u672c\u3002", + "scroll.update_available": "Mouser %1 \u53ef\u7528\u3002", + "scroll.update_checking": "\u6b63\u5728\u6aa2\u67e5\u66f4\u65b0...", + "scroll.update_downloading": "\u6b63\u5728\u4e0b\u8f09\u66f4\u65b0...", + "scroll.update_verifying": "\u6b63\u5728\u9a57\u8b49\u66f4\u65b0...", + "scroll.update_ready": "Mouser \u9000\u51fa\u5f8c\u5373\u53ef\u5b89\u88dd\u3002", + "scroll.update_installing": "\u6b63\u5728\u5b89\u88dd\u66f4\u65b0...", + "scroll.update_installed": "\u66f4\u65b0\u5df2\u5b89\u88dd\u3002", + "scroll.update_installed_version": "\u5df2\u66f4\u65b0\u5230 %1\u3002", + "scroll.update_cancelled": "\u66f4\u65b0\u5df2\u53d6\u6d88\u3002", + "scroll.update_manual": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8acb\u5f9e\u767c\u5e03\u9801\u4e0b\u8f09\u4e26\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_manual_windows": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8acb\u5f9e\u767c\u5e03\u9801\u4e0b\u8f09\u4e26\u5728 Windows \u4e0a\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_manual_macos": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8acb\u5f9e\u767c\u5e03\u9801\u4e0b\u8f09\u4e26\u5728 macOS \u4e0a\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_manual_linux": "\u6709\u65b0\u7248 Mouser \u53ef\u7528\u3002\u8acb\u5f9e\u767c\u5e03\u9801\u4e0b\u8f09\u4e26\u5728 Linux \u4e0a\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_no_asset": "\u6b64\u96fb\u8166\u6c92\u6709\u53ef\u7528\u7684\u66f4\u65b0\u5957\u4ef6\u3002", + "scroll.update_error": "\u7121\u6cd5\u6e96\u5099\u66f4\u65b0\u3002", + "scroll.update_error_check_first": "\u8acb\u5148\u6aa2\u67e5\u66f4\u65b0\u3002", + "scroll.update_error_network_error": "Mouser \u7121\u6cd5\u9023\u63a5\u66f4\u65b0\u670d\u52d9\u3002\u8acb\u7a0d\u5f8c\u91cd\u8a66\u3002", + "scroll.update_error_metadata_missing": "\u66f4\u65b0\u8cc7\u8a0a\u5c1a\u672a\u6e96\u5099\u597d\u3002\u8acb\u958b\u555f\u767c\u5e03\u9801\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_error_metadata_invalid": "\u7121\u6cd5\u8b80\u53d6\u66f4\u65b0\u8cc7\u8a0a\u3002\u8acb\u958b\u555f\u767c\u5e03\u9801\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_error_permission_denied": "Mouser \u6c92\u6709\u6e96\u5099\u66f4\u65b0\u7684\u6b0a\u9650\u3002", + "scroll.update_error_file_error": "Mouser \u7121\u6cd5\u5beb\u5165\u66f4\u65b0\u6a94\u6848\u3002", + "scroll.update_error_install_failed": "\u66f4\u65b0\u672a\u5b8c\u6210\u3002\u8acb\u958b\u555f\u767c\u5e03\u9801\u624b\u52d5\u5b89\u88dd\u3002", + "scroll.update_error_sha256_mismatch": "\u4e0b\u8f09\u672a\u901a\u904e\u9a57\u8b49\u3002\u8acb\u7a0d\u5f8c\u91cd\u8a66\u3002", + "scroll.update_error_size_mismatch": "\u4e0b\u8f09\u672a\u901a\u904e\u9a57\u8b49\u3002\u8acb\u7a0d\u5f8c\u91cd\u8a66\u3002", + "scroll.update_error_expired_metadata": "\u66f4\u65b0\u8cc7\u8a0a\u5df2\u904e\u671f\u3002\u8acb\u7a0d\u5f8c\u91cd\u8a66\u3002", + "scroll.update_error_older_build": "Mouser \u5df2\u62d2\u7d55\u8f03\u820a\u7684\u66f4\u65b0\u3002", + "scroll.update_check": "\u6aa2\u67e5", + "scroll.update_download": "\u4e0b\u8f09", + "scroll.update_verify": "\u9a57\u8b49", + "scroll.update_install": "\u5b89\u88dd", + "scroll.update_cancel": "\u53d6\u6d88", + "scroll.update_open_release": "\u958b\u555f\u767c\u5e03\u9801", "scroll.scroll_speed": "\u6eda\u8f2a\u901f\u5ea6", "scroll.scroll_speed_desc": "\u8abf\u6574\u6bcf\u6b21\u6eda\u8f2a\u6eda\u52d5\u7684\u9801\u9762\u79fb\u52d5\u901f\u5ea6\u30021.0\u00d7 \u70ba\u7cfb\u7d71\u9810\u8a2d\u3002", "scroll.scroll_speed_presets": "\u9810\u8a2d\uff1a", diff --git a/ui/qml/ScrollPage.qml b/ui/qml/ScrollPage.qml index 729cdf3..160aa4d 100644 --- a/ui/qml/ScrollPage.qml +++ b/ui/qml/ScrollPage.qml @@ -11,6 +11,43 @@ Item { // Reactive shortcut — all s["key"] bindings update when lm.languageChanged fires property var s: lm.strings + function updateStatusText() { + if (backend.updateInstallStatus === "checking") + return s["scroll.update_checking"] + if (backend.updateInstallStatus === "downloading") + return s["scroll.update_downloading"] + if (backend.updateInstallStatus === "verifying") + return s["scroll.update_verifying"] + if (backend.updateInstallStatus === "ready_to_install") + return s["scroll.update_ready"] + if (backend.updateInstallStatus === "installing") + return s["scroll.update_installing"] + if (backend.updateInstallStatus === "installed") + return backend.updateInstallMessage ? s["scroll.update_installed_version"].replace("%1", backend.updateInstallMessage) : s["scroll.update_installed"] + if (backend.updateInstallStatus === "cancelled") + return s["scroll.update_cancelled"] + if (backend.updateInstallStatus === "manual_fallback") { + if (backend.updateInstallMessage === "macos") + return s["scroll.update_manual_macos"] + if (backend.updateInstallMessage === "linux") + return s["scroll.update_manual_linux"] + if (backend.updateInstallMessage === "windows") + return s["scroll.update_manual_windows"] + if (backend.updateInstallMessage === "no_asset") + return s["scroll.update_no_asset"] + return s["scroll.update_manual"] + } + if (backend.updateInstallStatus === "error") { + var key = "scroll.update_error_" + backend.updateInstallMessage + if (s[key]) + return s[key] + return s["scroll.update_error"] + } + if (backend.latestUpdateVersion) + return s["scroll.update_available"].replace("%1", backend.latestUpdateVersion) + return s["scroll.update_idle"] + } + readonly property var appearanceOptions: [ { label: s["scroll.system"], value: "system" }, { label: s["scroll.light"], value: "light" }, @@ -763,50 +800,118 @@ Item { Rectangle { width: parent.width - height: 62 + height: 118 radius: 10 color: scrollPage.theme.bgSubtle - RowLayout { + ColumnLayout { anchors { fill: parent leftMargin: 16 rightMargin: 16 + topMargin: 10 + bottomMargin: 10 } spacing: 12 - Column { + RowLayout { Layout.fillWidth: true - spacing: 3 + spacing: 12 - Text { - text: s["scroll.check_for_updates"] - font { - family: uiState.fontFamily - pixelSize: 13 + Column { + Layout.fillWidth: true + spacing: 3 + + Text { + text: s["scroll.check_for_updates"] + font { + family: uiState.fontFamily + pixelSize: 13 + } + color: scrollPage.theme.textPrimary } - color: scrollPage.theme.textPrimary + + Text { + width: parent.width + text: s["scroll.check_for_updates_desc"] + font { + family: uiState.fontFamily + pixelSize: 11 + } + color: scrollPage.theme.textSecondary + wrapMode: Text.WordWrap + } + } + + Switch { + id: checkUpdatesSwitch + checked: backend.checkForUpdates + focusPolicy: Qt.StrongFocus + Material.accent: scrollPage.theme.accent + Accessible.name: s["scroll.check_for_updates"] + onClicked: backend.setCheckForUpdates(checked) } + } + + RowLayout { + Layout.fillWidth: true + spacing: 8 Text { - width: parent.width - text: s["scroll.check_for_updates_desc"] + Layout.fillWidth: true + text: scrollPage.updateStatusText() font { family: uiState.fontFamily pixelSize: 11 } color: scrollPage.theme.textSecondary - wrapMode: Text.WordWrap + elide: Text.ElideRight } - } - Switch { - id: checkUpdatesSwitch - checked: backend.checkForUpdates - focusPolicy: Qt.StrongFocus - Material.accent: scrollPage.theme.accent - Accessible.name: s["scroll.check_for_updates"] - onClicked: backend.setCheckForUpdates(checked) + ProgressBar { + Layout.preferredWidth: 120 + visible: backend.updateInstallStatus === "downloading" + from: 0 + to: 100 + value: backend.updateInstallProgress + } + + Button { + text: s["scroll.update_check"] + enabled: !backend.updateInstallInProgress + onClicked: backend.manualCheckForUpdates() + } + + Button { + text: backend.isWindows ? s["scroll.update_download"] : s["scroll.update_verify"] + visible: backend.latestUpdateVersion !== "" + && !backend.updateInstallCanInstall + && (!backend.isWindows || backend.updateInstallEnabled) + enabled: !backend.updateInstallInProgress + onClicked: backend.prepareLatestUpdate() + } + + Button { + text: s["scroll.update_cancel"] + visible: backend.updateInstallStatus === "checking" + || backend.updateInstallStatus === "downloading" + || backend.updateInstallStatus === "verifying" + onClicked: backend.cancelUpdatePreparation() + } + + Button { + text: s["scroll.update_install"] + visible: backend.updateInstallCanInstall && backend.updateInstallEnabled + enabled: !backend.updateInstallInProgress + onClicked: backend.installPreparedUpdate() + } + + Button { + text: s["scroll.update_open_release"] + visible: backend.latestUpdateVersion !== "" + enabled: !backend.updateInstallInProgress + onClicked: backend.openLatestReleasePage() + } } } }