diff --git a/CHANGELOG.md b/CHANGELOG.md index d5575ac..0546384 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,21 @@ All notable changes to SearchMob Desktop are documented here. The version scheme is Ubuntu-style `YY.MM.VV` and releases are tagged `vYY.MM.VV`. +## Unreleased + +### Added +- **You now actually get told when an update is out.** When SearchMob is open and a newer release is + available, it posts a system notification and shows a banner across the top of the window; the + served search pages show the same banner (only to you, on this device, never to other people on + your network). The check still runs about once a day through the privacy proxy and stays opt-out. +- **One-click update.** Clicking **Update** (in the banner or the notification) downloads the right + installer for your OS, verifies it against the release's published SHA-256 checksums, and hands it + to your system installer. On Linux, which ships several package formats, it opens the release page + so you can pick the one you use. The web banner links straight to the release. + +### Changed +- The **Settings** window is now constrained to a 4:3 aspect ratio. + ## 26.06.01 — 2026-06-02 ### Fixed diff --git a/src/searchmob_desktop/cli.py b/src/searchmob_desktop/cli.py index 44ed1a9..530d598 100644 --- a/src/searchmob_desktop/cli.py +++ b/src/searchmob_desktop/cli.py @@ -57,6 +57,7 @@ VersionTag, check_if_due, fetch_latest, + reconcile_pending_update, ) from searchmob_desktop.version import __version__ @@ -252,7 +253,8 @@ def _run_update_check_in_background(prefs_store: JsonPreferencesStore) -> None: """Fire-and-forget the throttled GitHub update check on a thread. Never blocks server startup, never crashes the CLI: a bare `except Exception` swallows - anything bubbling out (network, serialization, prefs IO). Prints an update banner via + anything bubbling out (network, serialization, prefs IO). Persists the throttle stamp and the + pending-update fields (which the served-page banner reads) and prints an update banner via `console` if a newer release is found; otherwise stays silent. """ @@ -268,8 +270,9 @@ def _do() -> None: client_factory=lambda: make_privacy_client(4.0), ) ) - if stamped != prefs.last_update_check_ms: - prefs_store.save(prefs.with_update_check_stamped(stamped)) + reconciled = reconcile_pending_update(prefs, info, stamped=stamped) + if reconciled != prefs: + prefs_store.save(reconciled) if info is not None: _print_update_available(info) except Exception: diff --git a/src/searchmob_desktop/gui/main_window.py b/src/searchmob_desktop/gui/main_window.py index 4732b37..b110d33 100644 --- a/src/searchmob_desktop/gui/main_window.py +++ b/src/searchmob_desktop/gui/main_window.py @@ -10,9 +10,11 @@ from __future__ import annotations import asyncio +import sys import time from html import escape from importlib.resources import as_file, files +from pathlib import Path from PySide6.QtCore import Qt, QThreadPool, QTimer, QUrl, Signal from PySide6.QtGui import QAction, QDesktopServices, QIcon, QKeySequence @@ -42,7 +44,12 @@ save_personalization, ) from searchmob_desktop.data.ranking_store import load_ranking_rules, save_ranking_rules -from searchmob_desktop.engines import EngineContext, SearchResult, aggregate +from searchmob_desktop.engines import ( + EngineContext, + SearchResult, + aggregate, + make_privacy_client, +) from searchmob_desktop.engines.correct import start_background_corrector from searchmob_desktop.engines.local_llm import LlmConfig, stream_answer from searchmob_desktop.engines.rank import ( @@ -71,6 +78,14 @@ from searchmob_desktop.gui.workers import AsyncWorker from searchmob_desktop.prefs import JsonPreferencesStore from searchmob_desktop.server import LOOPBACK_HOST +from searchmob_desktop.update import ( + VersionTag, + check_if_due, + fetch_latest, + reconcile_pending_update, +) +from searchmob_desktop.update_download import default_download_dir, download_and_verify +from searchmob_desktop.version import __version__ def app_icon() -> QIcon: @@ -160,6 +175,13 @@ def __init__( outer.setContentsMargins(20, 16, 20, 12) outer.setSpacing(12) + # "Update available" banner, pinned at the very top (hidden until a check finds a newer + # release). Surfaced from persisted prefs on launch and from the background check below. + self._pending_update: tuple[str, str] | None = None + self._update_notified = False + self._update_banner = self._build_update_banner() + outer.addWidget(self._update_banner) + # Search bar row: a roomy field with a prominent accent-filled action button. search_row = QHBoxLayout() search_row.setSpacing(10) @@ -312,6 +334,11 @@ def __init__( if not prefs.onboarding_completed or prefs.onboarding_version < ONBOARDING_VERSION: QTimer.singleShot(0, self._show_onboarding) + # Surface a pending update found by an earlier check (persisted in prefs). A fresh throttled + # check is kicked off on first show (see showEvent), so it only runs for a real launch. + self._update_check_started = False + self._surface_pending_update_from_prefs(prefs) + def _show_onboarding(self) -> None: from searchmob_desktop.gui.onboarding_dialog import OnboardingDialog @@ -484,6 +511,9 @@ def _setup_tray(self) -> None: tray.setContextMenu(menu) tray.activated.connect(self._on_tray_activated) + # Clicking the "update available" notification starts the update. messageClicked does not + # say which fired, so _on_notification_clicked no-ops unless an update is actually pending. + tray.messageClicked.connect(self._on_notification_clicked) tray.show() self._tray = tray @@ -507,6 +537,179 @@ def _quit_from_tray(self) -> None: self._really_quit = True self.close() + # --- Update notifier --------------------------------------------------------------------- + + def _build_update_banner(self) -> QFrame: + """A dismissible "update available" bar with an Update button and a close affordance.""" + bar = QFrame() + bar.setObjectName("updateBanner") + row = QHBoxLayout(bar) + row.setContentsMargins(14, 8, 8, 8) + row.setSpacing(10) + self._update_label = QLabel("An update is available.") + self._update_btn = QPushButton("Update") + self._update_btn.clicked.connect(self._on_update_clicked) + dismiss = QPushButton("✕") # MULTIPLICATION X, a compact close glyph + dismiss.setProperty("role", "dismiss") + dismiss.setToolTip("Dismiss until the next check") + dismiss.setFixedWidth(34) + dismiss.clicked.connect(self._update_banner_dismiss) + row.addWidget(self._update_label, stretch=1) + row.addWidget(self._update_btn) + row.addWidget(dismiss) + bar.hide() + return bar + + @staticmethod + def _current_version_code() -> int: + parsed = VersionTag.parse(__version__) + return parsed.to_version_code() if parsed is not None else 0 + + def _surface_pending_update_from_prefs(self, prefs: object) -> None: + """Show the banner for a prefs-recorded pending update, if still newer than this build.""" + version = getattr(prefs, "pending_update_version", "") + url = getattr(prefs, "pending_update_url", "") + if not version or not url: + return + parsed = VersionTag.parse(version) + if parsed is None or parsed.to_version_code() <= self._current_version_code(): + return + # Notify on launch too: the app being open with an update available is exactly the case the + # system notification is for. _surface_update guards against repeating it within a session. + self._surface_update(version, url, notify=True) + + def _surface_update(self, version: str, url: str, *, notify: bool) -> None: + """Show/refresh the banner and optionally post a one-per-session system notification.""" + self._pending_update = (version, url) + self._update_label.setText(f"SearchMob {version} is available.") + self._update_btn.setEnabled(True) + self._update_btn.setText("Update") + self._update_banner.show() + if notify and not self._update_notified: + self._update_notified = True + self._post_update_notification(version) + + def _post_update_notification(self, version: str) -> None: + """Post a native system notification through the tray icon, when one is available.""" + if self._tray is not None and QSystemTrayIcon.supportsMessages(): + self._tray.showMessage( + "Update available", + f"SearchMob {version} is available. Click here to update.", + app_icon(), + 8000, + ) + + def _on_notification_clicked(self) -> None: + """The tray notification was clicked: bring the window forward and start the update.""" + if self._pending_update is None: + return + self._show_from_tray() + self._on_update_clicked() + + def _update_banner_dismiss(self) -> None: + # Hide for this session only. The pending record stays in prefs, so the banner returns on + # the next launch or check until the user actually updates. + self._update_banner.hide() + + def showEvent(self, event): # type: ignore[no-untyped-def] + # Kick off the throttled update check the first time the window is shown (a real launch), + # not on construction, so headless/widget tests that never show it stay offline. + super().showEvent(event) + if not self._update_check_started: + self._update_check_started = True + QTimer.singleShot(0, self._start_update_check) + + def _start_update_check(self) -> None: + """Run the throttled GitHub check off-thread; surface a newer release if found.""" + + async def _probe() -> object: + prefs = self._prefs_store.load() + now_ms = int(time.time() * 1000) + info, stamped = await check_if_due( + prefs, + self._current_version_code(), + now_ms=now_ms, + client_factory=lambda: make_privacy_client(4.0), + ) + return prefs, info, stamped + + worker: AsyncWorker[object] = AsyncWorker(_probe) + worker.signals.finished.connect(self._on_update_check_done) + worker.signals.failed.connect(lambda _msg: None) # fail-soft: a check error stays silent + worker.start(self._pool) + + def _on_update_check_done(self, payload: object) -> None: + """Persist the check result (stamp + pending fields) and surface a newer release.""" + if not (isinstance(payload, tuple) and len(payload) == 3): + return + prefs, info, stamped = payload + reconciled = reconcile_pending_update(prefs, info, stamped=stamped) + if reconciled != prefs: + try: + self._prefs_store.save(reconciled) + except OSError: + pass + if info is not None: + self._surface_update(info.latest_version.formatted(), info.release_url, notify=True) + + def _on_update_clicked(self) -> None: + """Fetch-and-hand-off: download + verify the installer, or open the release page.""" + if self._pending_update is None: + return + version, url = self._pending_update + self._update_btn.setEnabled(False) + self._update_label.setText(f"Downloading SearchMob {version} …") + system = sys.platform + + async def _run() -> object: + # A generous read timeout: the installer is large, and httpx applies the timeout per + # network operation (max gap between bytes), not to the whole transfer. + async with make_privacy_client(60.0) as client: + info = await fetch_latest(client) + if info is None: + return ("page", url) + asset = info.asset_for_system(system) + sums = info.checksums_asset() + if asset is None or sums is None: + # Linux (multiple package formats) or a release without a usable asset/checksum: + # hand off to the release page so the user picks the right download. + return ("page", info.release_url) + path = await download_and_verify(client, asset, sums, default_download_dir()) + return ("file", str(path)) + + worker: AsyncWorker[object] = AsyncWorker(_run) + worker.signals.finished.connect( + lambda payload, v=version, u=url: self._on_update_download_done(payload, v, u) + ) + worker.signals.failed.connect( + lambda msg, v=version, u=url: self._on_update_download_failed(msg, v, u) + ) + worker.start(self._pool) + + def _on_update_download_done(self, payload: object, version: str, url: str) -> None: + self._update_btn.setEnabled(True) + self._update_label.setText(f"SearchMob {version} is available.") + kind, value = payload if isinstance(payload, tuple) and len(payload) == 2 else ("page", url) + if kind == "file": + # Hand off to the OS: opening the installer mounts the .dmg / launches the .msi. While + # builds are unsigned, Gatekeeper / SmartScreen will still prompt; that is unavoidable. + if not QDesktopServices.openUrl(QUrl.fromLocalFile(value)): + # Could not auto-open it: reveal the folder so the user can run it themselves. + QDesktopServices.openUrl(QUrl.fromLocalFile(str(Path(value).parent))) + self.statusBar().showMessage(f"Downloaded installer: {value}") + else: + QDesktopServices.openUrl(QUrl(value)) + + def _on_update_download_failed(self, message: str, version: str, url: str) -> None: + self._update_btn.setEnabled(True) + self._update_label.setText(f"SearchMob {version} is available.") + QMessageBox.warning( + self, + "Update download failed", + f"{message}\n\nOpening the release page so you can download it manually.", + ) + QDesktopServices.openUrl(QUrl(url)) + # --- Search ------------------------------------------------------------------------------ def _on_submit(self) -> None: @@ -820,6 +1023,13 @@ def _on_rules_changed() -> None: # The personalization toggle / export-import / reset may have changed in Settings; pick up # the new enabled state and reload the model (so a live toggle takes effect immediately). prefs_after = self._prefs_store.load() + # A "Check now" in Settings may have found (or cleared) a pending update; reflect it without + # re-notifying (the user is already here). Hide the banner if the pending record is gone. + if prefs_after.pending_update_version and prefs_after.pending_update_url: + self._surface_pending_update_from_prefs(prefs_after) + else: + self._pending_update = None + self._update_banner.hide() self._personalization_enabled = prefs_after.personalization_enabled self._personalization = ( load_personalization() diff --git a/src/searchmob_desktop/gui/settings_dialog.py b/src/searchmob_desktop/gui/settings_dialog.py index 9485d81..1ec15ae 100644 --- a/src/searchmob_desktop/gui/settings_dialog.py +++ b/src/searchmob_desktop/gui/settings_dialog.py @@ -67,7 +67,13 @@ from searchmob_desktop.gui.theme import DARK, LIGHT, SYSTEM from searchmob_desktop.gui.workers import AsyncWorker from searchmob_desktop.prefs import JsonPreferencesStore, UserPreferences -from searchmob_desktop.update import RELEASES_PAGE_URL, UpdateInfo, fetch_latest +from searchmob_desktop.update import ( + RELEASES_PAGE_URL, + UpdateInfo, + VersionTag, + fetch_latest, + reconcile_pending_update, +) from searchmob_desktop.version import __version__ if TYPE_CHECKING: @@ -145,7 +151,12 @@ def __init__( super().__init__(parent) self.setWindowTitle("Settings") self.setModal(True) - self.resize(780, 640) + # Constrain the window to a 4:3 aspect ratio: open at 4:3 and keep height locked to 3/4 of + # the width as the user resizes (see resizeEvent). The guard flag prevents the resize we + # make inside resizeEvent from recursing. + self._adjusting_aspect = False + self.setMinimumSize(640, 480) # 4:3 + self.resize(800, 600) # 4:3 self._prefs_store = prefs_store self._server_controller = server_controller @@ -204,6 +215,21 @@ def __init__( # --- Persistence helper ------------------------------------------------------------------ + def resizeEvent(self, event): # type: ignore[no-untyped-def] + # Lock the window to 4:3: after any resize, snap the height to 3/4 of the current width. The + # reentrancy guard stops the resize() below from triggering this handler again in a loop. + super().resizeEvent(event) + if self._adjusting_aspect: + return + width = self.width() + target_height = round(width * 3 / 4) + if abs(target_height - self.height()) > 1: + self._adjusting_aspect = True + try: + self.resize(width, target_height) + finally: + self._adjusting_aspect = False + def _save(self, new_prefs: UserPreferences) -> None: self._prefs = new_prefs try: @@ -1044,8 +1070,6 @@ def _on_toggled(checked: bool) -> None: return tab def _on_check_now(self) -> None: - from searchmob_desktop.update import VersionTag - # One check at a time: disabling the button stops rapid clicks from stacking concurrent # workers (and duplicate result dialogs). Re-enabled when the check finishes or fails. self._check_now_btn.setEnabled(False) @@ -1058,18 +1082,23 @@ async def _probe() -> UpdateInfo | None: def _on_finished(info_obj: object) -> None: self._check_now_btn.setEnabled(True) - stamped = replace(self._prefs, last_update_check_ms=int(time.time() * 1000)) - self._save(stamped) - if not isinstance(info_obj, UpdateInfo): + parsed = VersionTag.parse(__version__) + current = parsed.to_version_code() if parsed else 0 + info = info_obj if isinstance(info_obj, UpdateInfo) else None + # Persist the throttle stamp and the pending-update fields together, so the in-app and + # served-page banners pick up (or clear) the result of this manual check on next read. + newer = info if (info is not None and info.is_newer_than(current)) else None + self._prefs = reconcile_pending_update( + self._prefs, newer, stamped=int(time.time() * 1000) + ) + self._save(self._prefs) + if info is None: QMessageBox.warning( self, "Update check failed", f"Could not reach GitHub. Releases page: {RELEASES_PAGE_URL}", ) return - info: UpdateInfo = info_obj - parsed = VersionTag.parse(__version__) - current = parsed.to_version_code() if parsed else 0 if info.is_newer_than(current): v = info.latest_version QMessageBox.information( diff --git a/src/searchmob_desktop/gui/theme.py b/src/searchmob_desktop/gui/theme.py index 791dc56..37e072a 100644 --- a/src/searchmob_desktop/gui/theme.py +++ b/src/searchmob_desktop/gui/theme.py @@ -197,6 +197,31 @@ def build_qss(p: Palette) -> str: }} QLabel[role="caveat-text"] {{ color: {p.danger_text}; }} +/* "Update available" banner, pinned at the top of the window. Accent fill so it reads as a + notice; the Update button keeps its primary styling against it. */ +QFrame#updateBanner {{ + background-color: {p.accent}; + border: none; + border-radius: 10px; +}} +QFrame#updateBanner QLabel {{ background: transparent; color: {p.on_accent}; font-weight: 600; }} +QFrame#updateBanner QPushButton {{ + background-color: {p.on_accent}; + color: {p.accent}; + border: none; + border-radius: 9px; + padding: 7px 16px; + font-weight: 700; +}} +QFrame#updateBanner QPushButton:hover {{ background-color: {p.card_hover}; }} +QFrame#updateBanner QPushButton[role="dismiss"] {{ + background: transparent; + color: {p.on_accent}; + padding: 6px 10px; + font-weight: 700; +}} +QFrame#updateBanner QPushButton[role="dismiss"]:hover {{ background-color: {p.accent_hover}; }} + /* Contextual Wikipedia summary card, above the results. */ QFrame#summaryCard {{ background-color: {p.card}; diff --git a/src/searchmob_desktop/prefs.py b/src/searchmob_desktop/prefs.py index 556fff9..9155116 100644 --- a/src/searchmob_desktop/prefs.py +++ b/src/searchmob_desktop/prefs.py @@ -88,6 +88,12 @@ class UserPreferences: llm_model: str = "" update_check_enabled: bool = True last_update_check_ms: int = 0 + # The newest published release the last update check found (empty when none / up to date). These + # drive the in-app and served-page "update available" banners and the tray notification so a + # found update survives a restart until a later check supersedes or clears it. Non-secret: just + # a version string and the release URL. Set/cleared via update.reconcile_pending_update. + pending_update_version: str = "" + pending_update_url: str = "" # Set once the first-run setup wizard has been completed or skipped, so it never reappears. onboarding_completed: bool = False # The onboarding revision the user last saw. The wizard re-appears once after an update that diff --git a/src/searchmob_desktop/server/app.py b/src/searchmob_desktop/server/app.py index ada352b..4593155 100644 --- a/src/searchmob_desktop/server/app.py +++ b/src/searchmob_desktop/server/app.py @@ -74,6 +74,8 @@ render_results_page, render_settings_page, ) +from searchmob_desktop.update import VersionTag +from searchmob_desktop.version import __version__ LOOPBACK_HOST = "127.0.0.1" DEFAULT_PORT = 8787 @@ -96,6 +98,12 @@ _SUGGESTIONS_CONTENT_TYPE = "application/x-suggestions+json" _OPENSEARCH_CONTENT_TYPE = "application/opensearchdescription+xml" +# The running app's version code, used to decide whether a persisted pending-update record is still +# newer than what is installed (so the owner-only banner self-clears after an in-place update even +# before the next background check rewrites prefs). 0 if the version string is unparseable. +_parsed_version = VersionTag.parse(__version__) +_CURRENT_VERSION_CODE = _parsed_version.to_version_code() if _parsed_version is not None else 0 + # A source of autocomplete suggestions for a partial query. Implementations MUST be fail-soft: # any error, timeout, or unavailable backing source returns an empty list rather than raising, so @@ -506,6 +514,7 @@ async def home(request: Request) -> Response: settings_link=_is_settings_owner(request), rules=rules_provider(), editable=_is_owner(request), + update_banner=_owner_update_banner(request), ) return Response(body, media_type="text/html; charset=utf-8") @@ -524,6 +533,23 @@ def _is_loopback_request(request: Request) -> bool: client_host = request.client.host if request.client is not None else "" return is_loopback_host(client_host) + def _owner_update_banner(request: Request) -> tuple[str, str] | None: + # The "update available" banner is owner-only: a network visitor cannot install anything and + # should not learn the owner's version. Read live from prefs (set by the background check), + # and only show it when the pending version still parses newer than what is installed. + if not _is_loopback_request(request): + return None + prefs = _load_prefs() + if prefs is None: + return None + version, url = prefs.pending_update_version, prefs.pending_update_url + if not version or not url: + return None + parsed = VersionTag.parse(version) + if parsed is None or parsed.to_version_code() <= _CURRENT_VERSION_CODE: + return None + return version, url + async def _maybe_summary(query: str) -> SummaryBox | None: if summary_provider is None or not query.strip(): return None @@ -581,6 +607,7 @@ async def search_html(request: Request) -> Response: vertical=vertical.value, settings_link=_is_settings_owner(request), link_builder=link_builder, + update_banner=_owner_update_banner(request), ) return Response(body, media_type="text/html; charset=utf-8") diff --git a/src/searchmob_desktop/server/templates.py b/src/searchmob_desktop/server/templates.py index 52a33f9..b4bfe7f 100644 --- a/src/searchmob_desktop/server/templates.py +++ b/src/searchmob_desktop/server/templates.py @@ -75,6 +75,14 @@ ".settings-link:hover{border-color:var(--accent);color:var(--accent)}" ".settings-link+.theme-toggle{margin-left:0}" ".topbar .spacer{margin-left:auto}" + # Owner-only "update available" banner, pinned above the top bar. Accent fill so it reads as a + # notice without an icon set; the action is a high-contrast pill linking to the release. + ".updatebar{display:flex;align-items:center;gap:12px;padding:9px 18px;background:var(--accent);" + "color:#fff;font-size:13px}" + ".updatebar .msg{font-weight:600}" + ".updatebar .btn{margin-left:auto;background:#fff;color:var(--accent);border-radius:16px;" + "padding:5px 14px;font-weight:700;text-decoration:none;white-space:nowrap}" + ".updatebar .btn:hover{text-decoration:none;opacity:.92}" ".settings{max-width:680px;margin:0 auto;padding:24px 18px 60px}" ".settings h1{font-size:24px;margin:8px 0 18px}" ".settings .saved{color:#fff;background:var(--accent);display:inline-block;border-radius:6px;" @@ -252,6 +260,25 @@ def _theme_toggle_button() -> str: ) +def _update_banner(banner: tuple[str, str] | None) -> str: + """An "update available" notice bar linking to the new release. Empty when `banner` is None. + + `banner` is `(version, url)`. The server passes it only for the loopback owner (a network + visitor cannot install anything and should not see the owner's version), so this renderer just + formats it. The link opens the release page; the GUI offers the verified one-click install. + """ + if banner is None: + return "" + version, url = banner + return ( + '
' + f'SearchMob {escape(version)} is available.' + f'' + "Get the update" + "
" + ) + + def _settings_link(show: bool) -> str: """A Settings-page link, shown only to the loopback owner (the route itself is owner-only).""" if not show: @@ -283,19 +310,22 @@ def render_home_page( settings_link: bool = False, rules: RankingRules | None = None, editable: bool = False, + update_banner: tuple[str, str] | None = None, ) -> str: """The home page: a centered search box plus the OpenSearch link. `settings_link` adds a Settings link to the top bar; the server passes True only for the loopback owner, since the Settings route is owner-only. `rules` + `editable` add a scope (lens) selector below the search box for the loopback owner, so a scope can be chosen before searching - (the selector renders only when at least one lens exists). + (the selector renders only when at least one lens exists). `update_banner` is `(version, url)` + for the owner-only "update available" notice (None to omit). """ active_rules = rules if rules is not None else RankingRules() scope = _scope_bar(active_rules) if editable else "" head = _page_head("SearchMob") body = ( '' + f"{_update_banner(update_banner)}" '
' '' f"{_settings_link(settings_link)}" @@ -448,6 +478,7 @@ def render_results_page( vertical: str = "web", settings_link: bool = False, link_builder: Callable[[int, str], str] | None = None, + update_banner: tuple[str, str] | None = None, ) -> str: """The results page. Empty/blank query -> a placeholder; otherwise -> the merged results. @@ -473,6 +504,7 @@ def render_results_page( parts: list[str] = [] parts.append('') + parts.append(_update_banner(update_banner)) parts.append('
') parts.append('') parts.append('
') diff --git a/src/searchmob_desktop/update.py b/src/searchmob_desktop/update.py index 0988562..674b9cf 100644 --- a/src/searchmob_desktop/update.py +++ b/src/searchmob_desktop/update.py @@ -19,7 +19,7 @@ import json from collections.abc import Awaitable, Callable -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any, Final import httpx @@ -31,11 +31,15 @@ "LATEST_RELEASE_API_URL", "MAX_RESPONSE_BYTES", "RELEASES_PAGE_URL", + "SHA256SUMS_ASSET_NAME", + "ReleaseAsset", "UpdateInfo", "VersionTag", + "asset_for_system", "check_if_due", "fetch_latest", "is_update_check_due", + "reconcile_pending_update", ] LATEST_RELEASE_API_URL: Final = ( @@ -46,10 +50,16 @@ # Throttle window: about one calendar day. Mirrors the Android default. DEFAULT_INTERVAL_MS: Final = 24 * 3600 * 1000 -# Bounded body read: a GitHub release JSON is a few KiB; anything past 16 KiB -# is treated as suspect rather than parsed. Matches the bounded-read pattern +# Bounded body read: a GitHub release JSON carries the tag plus the per-platform asset list +# (installer download URLs + the SHA256SUMS entry), so it runs larger than the tag-only payload +# the first cut parsed. 64 KiB comfortably covers a release with a dozen assets while still +# treating anything past it as suspect rather than parsing it. Matches the bounded-read pattern # the security audit codified on the Android side. -MAX_RESPONSE_BYTES: Final = 16 * 1024 +MAX_RESPONSE_BYTES: Final = 64 * 1024 + +# The integrity-anchor asset published alongside the installers (see the release workflow). The +# updater verifies a downloaded installer's SHA-256 against the matching line in this file. +SHA256SUMS_ASSET_NAME: Final = "SHA256SUMS" @dataclass(frozen=True, slots=True) @@ -68,6 +78,10 @@ class VersionTag: def to_version_code(self) -> int: return self.year * 10000 + self.month * 100 + self.build + def formatted(self) -> str: + """The canonical `YY.MM.VV` string (zero-padded), matching the version-file format.""" + return f"{self.year:02d}.{self.month:02d}.{self.build:02d}" + @classmethod def parse(cls, tag: str) -> VersionTag | None: if not isinstance(tag, str): @@ -92,16 +106,62 @@ def parse(cls, tag: str) -> VersionTag | None: return cls(year=year, month=month, build=build) +@dataclass(frozen=True, slots=True) +class ReleaseAsset: + """One published installer (or the SHA256SUMS file) attached to a GitHub release.""" + + name: str + download_url: str + size: int = 0 + + @dataclass(frozen=True, slots=True) class UpdateInfo: - """A newer release the user could download.""" + """A newer release the user could download, plus its published assets.""" latest_version: VersionTag release_url: str + assets: tuple[ReleaseAsset, ...] = () def is_newer_than(self, current_code: int) -> bool: return self.latest_version.to_version_code() > current_code + def asset_for_system(self, system: str) -> ReleaseAsset | None: + """The single installer asset for the running platform, or None when it is ambiguous. + + macOS picks the `.dmg`, Windows the `.msi`. Linux is deliberately left to fall back to the + release page: this build ships `.deb`, `.rpm`, and `.flatpak`, and nothing in a running + instance reliably says which one the user installed, so guessing risks the wrong package. + """ + return asset_for_system(self.assets, system) + + def checksums_asset(self) -> ReleaseAsset | None: + """The SHA256SUMS asset used to verify a downloaded installer's integrity, if published.""" + for asset in self.assets: + if asset.name == SHA256SUMS_ASSET_NAME: + return asset + return None + + +def asset_for_system(assets: tuple[ReleaseAsset, ...], system: str) -> ReleaseAsset | None: + """Pick the installer asset matching `system` (a `sys.platform` string), or None. + + Returns the first asset whose name ends with the platform's installer suffix: `.dmg` on macOS + (`darwin`), `.msi` on Windows (`win32`/`win...`). Any other platform (Linux included) returns + None so the caller opens the release page instead of downloading a possibly-wrong package. + """ + sys_lower = system.lower() + if sys_lower.startswith("darwin"): + suffix = ".dmg" + elif sys_lower.startswith("win"): + suffix = ".msi" + else: + return None + for asset in assets: + if asset.name.lower().endswith(suffix): + return asset + return None + async def fetch_latest( client: httpx.AsyncClient, @@ -139,7 +199,36 @@ async def fetch_latest( if version is None: return None release_url = html_url if isinstance(html_url, str) and html_url else RELEASES_PAGE_URL - return UpdateInfo(latest_version=version, release_url=release_url) + return UpdateInfo( + latest_version=version, + release_url=release_url, + assets=_parse_assets(payload.get("assets")), + ) + + +def _parse_assets(raw: Any) -> tuple[ReleaseAsset, ...]: + """Parse the release `assets` array into `ReleaseAsset`s, skipping malformed entries. + + Fail-soft like the rest of the parser: a non-list, or an entry missing a usable name or + `browser_download_url`, is simply dropped rather than raising. + """ + if not isinstance(raw, list): + return () + assets: list[ReleaseAsset] = [] + for entry in raw: + if not isinstance(entry, dict): + continue + name = entry.get("name") + url = entry.get("browser_download_url") + if not isinstance(name, str) or not name: + continue + if not isinstance(url, str) or not url: + continue + size = entry.get("size") + assets.append( + ReleaseAsset(name=name, download_url=url, size=size if isinstance(size, int) else 0) + ) + return tuple(assets) def is_update_check_due( @@ -192,3 +281,27 @@ async def check_if_due( if info is None or not info.is_newer_than(current_code): return None, now_ms return info, now_ms + + +def reconcile_pending_update( + prefs: UserPreferences, info: UpdateInfo | None, *, stamped: int +) -> UserPreferences: + """Fold a check result into prefs: stamp the check time and set/clear the pending-update fields. + + `info` is the newer-than-current release (as returned by `check_if_due`), or None. `stamped` is + that call's returned timestamp. The pending fields drive the GUI/web banners, so they must both + appear when an update is found AND disappear once a check confirms the user is current (e.g. + after they updated). They are only cleared when a check actually ran (`stamped` advanced), so a + throttled no-op never wipes a still-valid banner. + """ + checked = stamped != prefs.last_update_check_ms + base = replace(prefs, last_update_check_ms=stamped) if checked else prefs + if info is not None: + return replace( + base, + pending_update_version=info.latest_version.formatted(), + pending_update_url=info.release_url, + ) + if checked: + return replace(base, pending_update_version="", pending_update_url="") + return base diff --git a/src/searchmob_desktop/update_download.py b/src/searchmob_desktop/update_download.py new file mode 100644 index 0000000..a6bd72d --- /dev/null +++ b/src/searchmob_desktop/update_download.py @@ -0,0 +1,160 @@ +"""Download a release installer and verify it against the published SHA256SUMS. + +This backs the GUI's one-click "Update" action on macOS and Windows, where the platform installer +is unambiguous (`.dmg` / `.msi`). The flow is deliberately a *fetch-and-hand-off*, not a silent +auto-install: download the right asset, prove its SHA-256 matches the release's signed-by-checksum +`SHA256SUMS`, then let the OS open the installer. (Linux ships several package formats, so the GUI +opens the release page there instead of guessing.) + +Every download goes through the same privacy-proxy `httpx.AsyncClient` the engines use (no cookies, +rotated User-Agent, no env proxies) and is bounded: the body is streamed with a hard size cap so a +hostile or mis-published asset cannot exhaust memory or disk, and the SHA-256 is computed while +streaming so a tampered byte fails the check before the file is ever opened. +""" + +from __future__ import annotations + +import hashlib +import os +import tempfile +from pathlib import Path + +import httpx + +from searchmob_desktop.update import ReleaseAsset + +__all__ = [ + "MAX_INSTALLER_BYTES", + "MAX_SUMS_BYTES", + "UpdateDownloadError", + "default_download_dir", + "download_and_verify", + "expected_digest", + "parse_sha256sums", +] + +# Hard ceiling on a downloaded installer. SearchMob's installers are well under 300 MiB; this bounds +# disk/memory use so a mis-published or hostile asset cannot stream an unbounded body. +MAX_INSTALLER_BYTES = 512 * 1024 * 1024 + +# The SHA256SUMS file is one short line per asset; a few KiB at most. +MAX_SUMS_BYTES = 64 * 1024 + + +class UpdateDownloadError(Exception): + """A download or integrity check failed. Carries a user-facing message.""" + + +def parse_sha256sums(text: str) -> dict[str, str]: + """Parse `SHA256SUMS` content into `{asset_name: lowercase_hex_digest}`. + + Accepts the standard `sha256sum` format: `<64-hex>`. Lines that + do not match (blank lines, comments, malformed entries) are skipped. A leading `*` (binary-mode + marker) on the name is stripped so the names match the published asset names. + """ + digests: dict[str, str] = {} + for line in text.splitlines(): + stripped = line.strip() + if not stripped: + continue + parts = stripped.split(None, 1) + if len(parts) != 2: + continue + digest, name = parts[0].lower(), parts[1].strip() + if name.startswith("*"): + name = name[1:] + if len(digest) == 64 and all(c in "0123456789abcdef" for c in digest): + digests[name] = digest + return digests + + +def expected_digest(sums_text: str, asset_name: str) -> str | None: + """The expected SHA-256 for `asset_name` from `SHA256SUMS` content, or None when absent.""" + return parse_sha256sums(sums_text).get(asset_name) + + +def default_download_dir() -> Path: + """A sensible place to drop the installer: the user's Downloads folder, else the temp dir. + + Downloads is discoverable for the user after the hand-off; the temp dir is the fail-safe when + no Downloads folder exists (or is not writable). + """ + downloads = Path.home() / "Downloads" + if downloads.is_dir() and os.access(downloads, os.W_OK): + return downloads + return Path(tempfile.gettempdir()) + + +async def _fetch_sums(client: httpx.AsyncClient, sums_asset: ReleaseAsset) -> str: + """Download the bounded SHA256SUMS body and return it as text, or raise UpdateDownloadError.""" + try: + async with client.stream("GET", sums_asset.download_url) as response: + response.raise_for_status() + chunks: list[bytes] = [] + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > MAX_SUMS_BYTES: + raise UpdateDownloadError("Checksum file was unexpectedly large; aborting.") + chunks.append(chunk) + except httpx.HTTPError as exc: + raise UpdateDownloadError(f"Could not fetch the checksum file: {exc}") from exc + return b"".join(chunks).decode("utf-8", errors="replace") + + +async def download_and_verify( + client: httpx.AsyncClient, + asset: ReleaseAsset, + sums_asset: ReleaseAsset, + dest_dir: Path, + *, + max_bytes: int = MAX_INSTALLER_BYTES, +) -> Path: + """Stream `asset` into `dest_dir`, verifying its SHA-256 against `sums_asset`. Return the path. + + Raises `UpdateDownloadError` on any transport failure, an oversized body, a missing checksum + entry, or a digest mismatch. The file is written to a temporary name and only moved into place + once the checksum verifies, so a partial or tampered download never lands at the final path. + """ + expected = expected_digest(await _fetch_sums(client, sums_asset), asset.name) + if expected is None: + raise UpdateDownloadError( + f"No published checksum for {asset.name}; refusing to open an unverified download." + ) + + dest_dir.mkdir(parents=True, exist_ok=True) + final_path = dest_dir / asset.name + # Write next to the final path so the move is an atomic rename on the same filesystem. + fd, tmp_name = tempfile.mkstemp(prefix=f".{asset.name}.", dir=str(dest_dir)) + tmp_path = Path(tmp_name) + hasher = hashlib.sha256() + try: + with os.fdopen(fd, "wb") as out: + try: + async with client.stream("GET", asset.download_url) as response: + response.raise_for_status() + total = 0 + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > max_bytes: + raise UpdateDownloadError( + "The download exceeded the expected size; aborting." + ) + hasher.update(chunk) + out.write(chunk) + except httpx.HTTPError as exc: + raise UpdateDownloadError(f"Download failed: {exc}") from exc + actual = hasher.hexdigest() + if actual != expected: + raise UpdateDownloadError( + "The downloaded file's checksum did not match the published value; " + "discarding it for your safety." + ) + os.replace(tmp_path, final_path) + finally: + # Remove the temp file on any failure (a successful os.replace already consumed it). + try: + tmp_path.unlink() + except OSError: + pass + return final_path diff --git a/tests/gui/test_update_banner_ui.py b/tests/gui/test_update_banner_ui.py new file mode 100644 index 0000000..9614476 --- /dev/null +++ b/tests/gui/test_update_banner_ui.py @@ -0,0 +1,86 @@ +"""GUI: the in-app update banner surfaces from prefs, and the Settings window is 4:3-locked.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +import pytest + +pytest.importorskip("PySide6") + +from searchmob_desktop.data.history import InMemoryHistoryStore +from searchmob_desktop.gui.main_window import MainWindow +from searchmob_desktop.gui.onboarding_dialog import ONBOARDING_VERSION +from searchmob_desktop.gui.settings_dialog import SettingsDialog +from searchmob_desktop.prefs import JsonPreferencesStore +from searchmob_desktop.update import VersionTag + + +def _newer_than_current() -> str: + code = MainWindow._current_version_code() + return VersionTag(code // 10000, (code // 100) % 100, (code % 100) + 1).formatted() + + +def _store_with_pending(tmp_path: Path, version: str) -> JsonPreferencesStore: + store = JsonPreferencesStore(path=tmp_path / "prefs.json") + store.save( + replace( + store.load(), + # Mark onboarding done so constructing the window does not schedule a blocking wizard + # (these tests assert on the update banner, not first-run). + onboarding_completed=True, + onboarding_version=ONBOARDING_VERSION, + pending_update_version=version, + pending_update_url="https://example.test/r/newer", + ) + ) + return store + + +def test_banner_shown_when_prefs_has_newer_pending(qapp: object, tmp_path: Path) -> None: + version = _newer_than_current() + window = MainWindow( + prefs_store=_store_with_pending(tmp_path, version), + history_store=InMemoryHistoryStore(), + ) + assert window._pending_update == (version, "https://example.test/r/newer") + assert not window._update_banner.isHidden() + assert f"SearchMob {version} is available." == window._update_label.text() + + +def test_banner_hidden_when_pending_not_newer(qapp: object, tmp_path: Path) -> None: + code = MainWindow._current_version_code() + same = VersionTag(code // 10000, (code // 100) % 100, code % 100).formatted() + window = MainWindow( + prefs_store=_store_with_pending(tmp_path, same), + history_store=InMemoryHistoryStore(), + ) + assert window._pending_update is None + assert window._update_banner.isHidden() + + +def test_dismiss_hides_banner_for_session(qapp: object, tmp_path: Path) -> None: + window = MainWindow( + prefs_store=_store_with_pending(tmp_path, _newer_than_current()), + history_store=InMemoryHistoryStore(), + ) + assert not window._update_banner.isHidden() + window._update_banner_dismiss() + assert window._update_banner.isHidden() + + +def test_settings_window_locked_to_4_3(qapp: object, tmp_path: Path) -> None: + from PySide6.QtCore import QSize + from PySide6.QtGui import QResizeEvent + + store = JsonPreferencesStore(path=tmp_path / "prefs.json") + dialog = SettingsDialog(prefs_store=store, history_store=InMemoryHistoryStore()) + # Opens at 4:3. + assert dialog.width() == 800 + assert dialog.height() == 600 + # A resize that breaks the ratio is snapped back: height tracks 3/4 of the width. Drive the + # handler directly (no processEvents) so this stays hermetic in the shared-QApplication session. + dialog.resize(1200, 1000) + dialog.resizeEvent(QResizeEvent(QSize(1200, 1000), QSize(800, 600))) + assert dialog.height() == round(dialog.width() * 3 / 4) == 900 diff --git a/tests/server/test_update_banner_route.py b/tests/server/test_update_banner_route.py new file mode 100644 index 0000000..7a053cc --- /dev/null +++ b/tests/server/test_update_banner_route.py @@ -0,0 +1,83 @@ +"""Served pages show the "update available" banner only to the loopback owner. + +A network visitor must never see it (they cannot install anything and it would leak the owner's +version), and a pending record that is not actually newer than the running build must not render. +""" + +from __future__ import annotations + +from dataclasses import replace + +from starlette.testclient import TestClient + +from searchmob_desktop.engines import EngineContext, SearchResult +from searchmob_desktop.prefs import UserPreferences +from searchmob_desktop.server.app import _CURRENT_VERSION_CODE, build_app +from searchmob_desktop.update import VersionTag + +# A version strictly newer than the running build, formatted the way prefs stores it. +_NEWER = VersionTag( + _CURRENT_VERSION_CODE // 10000, + (_CURRENT_VERSION_CODE // 100) % 100, + (_CURRENT_VERSION_CODE % 100) + 1, +).formatted() +_RELEASE_URL = "https://example.test/r/newer" + + +async def _metasearch(_ctx: EngineContext, _engines: object) -> list[SearchResult]: + return [SearchResult(title="One", url="https://a.example/1", snippet="s", engine="e")] + + +def _app(prefs: UserPreferences) -> object: + return build_app( + [lambda _c, _ctx: []], + bound_port_getter=lambda: 8787, + bound_host_getter=lambda: "0.0.0.0", + prefs_provider=lambda: prefs, + metasearch=_metasearch, # type: ignore[arg-type] + host_allowlist_enabled=False, + ) + + +def _pending_prefs() -> UserPreferences: + return UserPreferences( + pending_update_version=_NEWER, + pending_update_url=_RELEASE_URL, + ) + + +def test_owner_sees_update_banner_on_home_and_results() -> None: + app = _app(_pending_prefs()) + owner = TestClient(app, client=("127.0.0.1", 9)) # type: ignore[arg-type] + home = owner.get("/").text + assert 'class="updatebar"' in home + assert _RELEASE_URL in home + assert f"SearchMob {_NEWER} is available." in home + + results = owner.get("/search", params={"q": "hi"}).text + assert 'class="updatebar"' in results + assert _RELEASE_URL in results + + +def test_network_visitor_never_sees_update_banner() -> None: + app = _app(_pending_prefs()) + remote = TestClient(app, client=("192.168.1.20", 9)) # type: ignore[arg-type] + assert 'class="updatebar"' not in remote.get("/").text + assert 'class="updatebar"' not in remote.get("/search", params={"q": "hi"}).text + + +def test_no_banner_when_pending_not_newer_than_current() -> None: + # Pending equals the running version -> stale, must not render (self-clears after an update). + current = VersionTag( + _CURRENT_VERSION_CODE // 10000, + (_CURRENT_VERSION_CODE // 100) % 100, + _CURRENT_VERSION_CODE % 100, + ).formatted() + prefs = replace(_pending_prefs(), pending_update_version=current) + owner = TestClient(_app(prefs), client=("127.0.0.1", 9)) # type: ignore[arg-type] + assert 'class="updatebar"' not in owner.get("/").text + + +def test_no_banner_when_no_pending_record() -> None: + owner = TestClient(_app(UserPreferences()), client=("127.0.0.1", 9)) # type: ignore[arg-type] + assert 'class="updatebar"' not in owner.get("/").text diff --git a/tests/test_update.py b/tests/test_update.py index f3dd341..fcfc578 100644 --- a/tests/test_update.py +++ b/tests/test_update.py @@ -12,11 +12,15 @@ from searchmob_desktop.update import ( LATEST_RELEASE_API_URL, RELEASES_PAGE_URL, + SHA256SUMS_ASSET_NAME, + ReleaseAsset, UpdateInfo, VersionTag, + asset_for_system, check_if_due, fetch_latest, is_update_check_due, + reconcile_pending_update, ) # --- VersionTag.parse ------------------------------------------------------------------------- @@ -56,6 +60,11 @@ def test_version_code_formula_matches_android() -> None: assert VersionTag(26, 5, 1).to_version_code() == 260501 +def test_version_tag_formatted_is_zero_padded() -> None: + assert VersionTag(26, 6, 1).formatted() == "26.06.01" + assert VersionTag(26, 12, 10).formatted() == "26.12.10" + + def test_is_newer_than_compares_codes() -> None: info = UpdateInfo(latest_version=VersionTag(26, 5, 2), release_url="x") assert info.is_newer_than(260501) is True @@ -132,6 +141,141 @@ async def test_fetch_latest_returns_none_on_oversized_body() -> None: assert await fetch_latest(client) is None +# --- assets ----------------------------------------------------------------------------------- + + +def _release_json_with_assets() -> dict[str, object]: + return { + "tag_name": "v26.07.00", + "html_url": "https://example.test/r/v26.07.00", + "assets": [ + { + "name": "SearchMob-26.07.00.dmg", + "browser_download_url": "https://example.test/dl/app.dmg", + "size": 12345, + }, + { + "name": "SearchMob-26.07.00.msi", + "browser_download_url": "https://example.test/dl/app.msi", + }, + { + "name": "SearchMob_26.07.00_amd64.deb", + "browser_download_url": "https://example.test/dl/app.deb", + }, + { + "name": "SHA256SUMS", + "browser_download_url": "https://example.test/dl/SHA256SUMS", + }, + # Malformed entries (missing url / not a dict) are skipped, not fatal. + {"name": "broken-no-url"}, + "not-a-dict", + ], + } + + +@respx.mock +@pytest.mark.asyncio +async def test_fetch_latest_parses_assets() -> None: + respx.get(LATEST_RELEASE_API_URL).respond(200, json=_release_json_with_assets()) + async with httpx.AsyncClient() as client: + info = await fetch_latest(client) + assert info is not None + names = [a.name for a in info.assets] + assert names == [ + "SearchMob-26.07.00.dmg", + "SearchMob-26.07.00.msi", + "SearchMob_26.07.00_amd64.deb", + "SHA256SUMS", + ] + assert info.assets[0].size == 12345 + assert info.assets[1].size == 0 # missing size defaults to 0 + + +@respx.mock +@pytest.mark.asyncio +async def test_fetch_latest_tolerates_missing_assets_field() -> None: + respx.get(LATEST_RELEASE_API_URL).respond(200, json={"tag_name": "v26.07.00"}) + async with httpx.AsyncClient() as client: + info = await fetch_latest(client) + assert info is not None + assert info.assets == () + + +def _info_with_assets() -> UpdateInfo: + return UpdateInfo( + latest_version=VersionTag(26, 7, 0), + release_url="https://example.test/r", + assets=( + ReleaseAsset("SearchMob.dmg", "https://example.test/app.dmg"), + ReleaseAsset("SearchMob.msi", "https://example.test/app.msi"), + ReleaseAsset("SearchMob.deb", "https://example.test/app.deb"), + ReleaseAsset(SHA256SUMS_ASSET_NAME, "https://example.test/SHA256SUMS"), + ), + ) + + +def test_asset_for_system_picks_dmg_on_macos() -> None: + asset = _info_with_assets().asset_for_system("darwin") + assert asset is not None and asset.name == "SearchMob.dmg" + + +def test_asset_for_system_picks_msi_on_windows() -> None: + asset = _info_with_assets().asset_for_system("win32") + assert asset is not None and asset.name == "SearchMob.msi" + + +def test_asset_for_system_returns_none_on_linux() -> None: + # Linux ships several package formats; the GUI falls back to the release page rather than guess. + assert _info_with_assets().asset_for_system("linux") is None + + +def test_asset_for_system_returns_none_when_no_match() -> None: + assets = (ReleaseAsset("only.deb", "https://example.test/app.deb"),) + assert asset_for_system(assets, "darwin") is None + + +def test_checksums_asset_found_and_absent() -> None: + assert _info_with_assets().checksums_asset() is not None + no_sums = UpdateInfo(VersionTag(26, 7, 0), "u", assets=(ReleaseAsset("a.dmg", "u"),)) + assert no_sums.checksums_asset() is None + + +# --- reconcile_pending_update ----------------------------------------------------------------- + + +def test_reconcile_sets_pending_when_update_found() -> None: + prefs = UserPreferences(last_update_check_ms=0) + info = UpdateInfo(VersionTag(26, 7, 0), "https://example.test/r/v26.07.00") + out = reconcile_pending_update(prefs, info, stamped=10**12) + assert out.pending_update_version == "26.07.00" + assert out.pending_update_url == "https://example.test/r/v26.07.00" + assert out.last_update_check_ms == 10**12 + + +def test_reconcile_clears_pending_when_check_ran_and_up_to_date() -> None: + prefs = UserPreferences( + last_update_check_ms=0, + pending_update_version="26.07.00", + pending_update_url="https://example.test/old", + ) + out = reconcile_pending_update(prefs, None, stamped=10**12) + assert out.pending_update_version == "" + assert out.pending_update_url == "" + assert out.last_update_check_ms == 10**12 + + +def test_reconcile_leaves_pending_untouched_when_throttled() -> None: + # A throttled no-op (stamped == existing last check) must not wipe a still-valid banner. + prefs = UserPreferences( + last_update_check_ms=555, + pending_update_version="26.07.00", + pending_update_url="https://example.test/r", + ) + out = reconcile_pending_update(prefs, None, stamped=555) + assert out.pending_update_version == "26.07.00" + assert out == prefs + + # --- throttle --------------------------------------------------------------------------------- diff --git a/tests/test_update_download.py b/tests/test_update_download.py new file mode 100644 index 0000000..153e758 --- /dev/null +++ b/tests/test_update_download.py @@ -0,0 +1,124 @@ +"""Unit tests for the installer download + SHA-256 verification helper.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +import httpx +import pytest +import respx + +from searchmob_desktop.update import ReleaseAsset +from searchmob_desktop.update_download import ( + UpdateDownloadError, + download_and_verify, + expected_digest, + parse_sha256sums, +) + +_INSTALLER_URL = "https://example.test/dl/SearchMob.dmg" +_SUMS_URL = "https://example.test/dl/SHA256SUMS" + + +def _asset(name: str = "SearchMob.dmg") -> ReleaseAsset: + return ReleaseAsset(name=name, download_url=_INSTALLER_URL) + + +def _sums_asset() -> ReleaseAsset: + return ReleaseAsset(name="SHA256SUMS", download_url=_SUMS_URL) + + +def _sums_text(payload: bytes, name: str = "SearchMob.dmg", extra: str = "") -> str: + digest = hashlib.sha256(payload).hexdigest() + line = f"{digest} {name}\n" + return extra + line + + +# --- parse_sha256sums ------------------------------------------------------------------------- + + +def test_parse_sha256sums_reads_two_space_and_binary_marker() -> None: + digest = "a" * 64 + text = f"{digest} plain.dmg\n{digest} *binary.msi\n\n# a comment\nbad line\n" + parsed = parse_sha256sums(text) + assert parsed == {"plain.dmg": digest, "binary.msi": digest} + + +def test_parse_sha256sums_skips_non_hex_and_short_digests() -> None: + text = "zzzz bad.dmg\nABCDEF short.dmg\n" + assert parse_sha256sums(text) == {} + + +def test_parse_sha256sums_lowercases_digest() -> None: + digest_upper = "A" * 64 + assert parse_sha256sums(f"{digest_upper} x.dmg") == {"x.dmg": "a" * 64} + + +def test_expected_digest_returns_none_when_absent() -> None: + assert expected_digest("", "missing.dmg") is None + + +# --- download_and_verify ---------------------------------------------------------------------- + + +@respx.mock +@pytest.mark.asyncio +async def test_download_and_verify_writes_file_when_checksum_matches(tmp_path: Path) -> None: + payload = b"installer-bytes" * 100 + respx.get(_SUMS_URL).respond(200, text=_sums_text(payload)) + respx.get(_INSTALLER_URL).respond(200, content=payload) + async with httpx.AsyncClient() as client: + path = await download_and_verify(client, _asset(), _sums_asset(), tmp_path) + assert path == tmp_path / "SearchMob.dmg" + assert path.read_bytes() == payload + # No leftover temp files beside the final installer. + assert [p.name for p in tmp_path.iterdir()] == ["SearchMob.dmg"] + + +@respx.mock +@pytest.mark.asyncio +async def test_download_and_verify_rejects_checksum_mismatch(tmp_path: Path) -> None: + payload = b"the-real-bytes" + # The sums file describes a DIFFERENT payload, so the computed digest will not match. + respx.get(_SUMS_URL).respond(200, text=_sums_text(b"something-else")) + respx.get(_INSTALLER_URL).respond(200, content=payload) + async with httpx.AsyncClient() as client: + with pytest.raises(UpdateDownloadError, match="checksum did not match"): + await download_and_verify(client, _asset(), _sums_asset(), tmp_path) + # The mismatched download must not be left on disk under the final name. + assert not (tmp_path / "SearchMob.dmg").exists() + assert list(tmp_path.iterdir()) == [] + + +@respx.mock +@pytest.mark.asyncio +async def test_download_and_verify_errors_when_no_checksum_entry(tmp_path: Path) -> None: + # Sums file exists but has no line for our asset name -> refuse to open an unverified download. + respx.get(_SUMS_URL).respond(200, text=_sums_text(b"x", name="other.dmg")) + respx.get(_INSTALLER_URL).respond(200, content=b"x") + async with httpx.AsyncClient() as client: + with pytest.raises(UpdateDownloadError, match="No published checksum"): + await download_and_verify(client, _asset(), _sums_asset(), tmp_path) + + +@respx.mock +@pytest.mark.asyncio +async def test_download_and_verify_enforces_size_cap(tmp_path: Path) -> None: + payload = b"x" * 5000 + respx.get(_SUMS_URL).respond(200, text=_sums_text(payload)) + respx.get(_INSTALLER_URL).respond(200, content=payload) + async with httpx.AsyncClient() as client: + with pytest.raises(UpdateDownloadError, match="exceeded the expected size"): + await download_and_verify(client, _asset(), _sums_asset(), tmp_path, max_bytes=1000) + assert list(tmp_path.iterdir()) == [] + + +@respx.mock +@pytest.mark.asyncio +async def test_download_and_verify_raises_on_transport_error(tmp_path: Path) -> None: + respx.get(_SUMS_URL).respond(200, text=_sums_text(b"x")) + respx.get(_INSTALLER_URL).mock(side_effect=httpx.ConnectError("down")) + async with httpx.AsyncClient() as client: + with pytest.raises(UpdateDownloadError, match="Download failed"): + await download_and_verify(client, _asset(), _sums_asset(), tmp_path)