From 5a1a004dafc37197595d38c40fcce8bc1c2f81b8 Mon Sep 17 00:00:00 2001 From: ila Date: Thu, 18 Jun 2026 11:58:06 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20non-blocking=20updates=20=E2=80=94=20do?= =?UTF-8?q?wnload=20in=20app,=20apply=20at=20launch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Startup no longer blocks on a download. New split: - services/installer.py: download_and_stage() (runs while the app is open → staging\app.new, verified) and apply_staged() (launcher swaps it into app\ at next launch, when the exe isn't locked). - launcher.py: slimmed to seed + apply_staged + launch; no network at startup. - main_window: a new version lights a dot on the ⚙ 配置 button instead of a blocking banner; the old open-folder banner is removed. - settings_dialog: "检查并更新" downloads + stages, then "下次启动生效". Verified end-to-end over a local HTTP server (download→stage→apply); 101 tests pass (test_installer +11, test_launcher rewritten). Co-Authored-By: Claude Opus 4.8 --- src/app/main_window.py | 131 ++++---------------- src/app/widgets/settings_dialog.py | 69 ++++++++++- src/launcher.py | 190 +++++------------------------ src/services/installer.py | 186 ++++++++++++++++++++++++++++ tests/test_installer.py | 130 ++++++++++++++++++++ tests/test_launcher.py | 135 ++++++-------------- 6 files changed, 468 insertions(+), 373 deletions(-) create mode 100644 src/services/installer.py create mode 100644 tests/test_installer.py diff --git a/src/app/main_window.py b/src/app/main_window.py index 79fa8ff..8f3c54e 100644 --- a/src/app/main_window.py +++ b/src/app/main_window.py @@ -71,8 +71,10 @@ class MainWindow(QMainWindow): layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(0) - # Update notification banner (hidden until a newer version is found). - layout.addWidget(self._create_update_banner()) + # Update state: when a newer version is found the gear button shows a + # dot; the user downloads it from the settings dialog (no startup block). + self._update_info = None + self._update_found.connect(self._on_update_found) # No in-app title bar: the OS window title (setWindowTitle) already # shows the app name and version, so an in-content header would just @@ -120,13 +122,14 @@ class MainWindow(QMainWindow): return container def _open_settings(self): - """Open the settings dialog; persist + re-check on save (docs/07 §4.4).""" + """Open the settings dialog (update config + manual update). Persist on save.""" dlg = SettingsDialog( self, update_source=self._config.get("update_source", ""), update_user=self._config.get("update_user", ""), update_pass=self._config.get("update_pass", ""), current_version=APP_VERSION, + update_info=self._update_info, ) if dlg.exec() == QDialog.Accepted: self._config.update(dlg.values()) @@ -134,54 +137,20 @@ class MainWindow(QMainWindow): self._refresh_update_check() def _refresh_update_check(self): - """Re-run the update check after the source/credentials changed.""" - self._update_banner.setVisible(False) + """Clear the indicator and re-check after the source/credentials changed.""" + self._update_info = None + self._set_update_indicator(False) self._start_update_check() # ------------------------------------------------------------------ - # Update notification (stage 2: notify-only, see docs/10-lan-update.md) + # Update availability (download happens in the settings dialog) # ------------------------------------------------------------------ - def _create_update_banner(self): - """A thin info bar shown when a newer version is found. Hidden by default.""" - self._update_info = None - self._update_found.connect(self._on_update_found) # connect once - - bar = QWidget() - bar.setObjectName("updateBanner") - bar.setVisible(False) - row = QHBoxLayout(bar) - row.setContentsMargins(12, 5, 8, 5) - row.setSpacing(8) - - self._update_label = QLabel() - self._update_label.setObjectName("updateBannerText") - - open_btn = QPushButton("打开更新目录") - open_btn.setObjectName("updateBannerOpen") - open_btn.setCursor(Qt.PointingHandCursor) - open_btn.clicked.connect(self._open_update_source) - - close_btn = QPushButton("✕") - close_btn.setObjectName("updateBannerClose") - close_btn.setFixedWidth(24) - close_btn.setCursor(Qt.PointingHandCursor) - close_btn.setToolTip("关闭") - close_btn.clicked.connect(lambda: self._update_banner.setVisible(False)) - - row.addWidget(self._update_label) - row.addStretch() - row.addWidget(open_btn) - row.addWidget(close_btn) - - self._update_banner = bar - return bar - def _start_update_check(self): """Check the configured update source for a newer version, off the UI thread. - Safe to call again after the settings change; the result signal is - connected once in _create_update_banner, not here. + Notify-only: a new version lights the gear button's dot; the user + downloads it from the settings dialog. Startup is never blocked. """ source = self._config.get("update_source", "") if not source: @@ -201,47 +170,17 @@ class MainWindow(QMainWindow): threading.Thread(target=worker, name="update-check", daemon=True).start() def _on_update_found(self, info): - """Show the update banner (runs on the UI thread via queued signal).""" + """A newer version is available — mark the gear button (UI thread).""" self._update_info = info - text = "发现新版本 v{}，当前 v{}。".format(info.version, APP_VERSION) - if info.notes: - text += " " + info.notes - self._update_label.setText(text) - self._update_banner.setVisible(True) + self._set_update_indicator(True, info.version) - def _open_update_source(self): - """Open the update folder in the file explorer. - - Prefer the version folder from the manifest; fall back to the configured - source root (which we know exists — the manifest was just read from it). - os.startfile is the reliable way to open a directory on Windows; - QDesktopServices.openUrl mishandles file:// URLs to folders (ShellExecute - error 2), so it is only a secondary fallback. - """ - if not self._update_info: - return - candidates = [ - self._update_info.source, - self._config.get("update_source", ""), - ] - for path in candidates: - if path and path.lower().startswith(("http://", "https://")): - if QDesktopServices.openUrl(QUrl(path)): - return - if not path or not Path(path).exists(): - continue - try: - os.startfile(path) # noqa: S606 — native Explorer open - return - except (OSError, AttributeError) as exc: - logger.warning("startfile failed for %s: %s", path, exc) - if QDesktopServices.openUrl(QUrl.fromLocalFile(path)): - return - shown = self._update_info.source or self._config.get("update_source", "") - QMessageBox.information( - self, "更新目录", - "无法自动打开更新目录，请手动前往：\n{}".format(shown), - ) + def _set_update_indicator(self, available, version=""): + self._settings_btn.setText("⚙ 配置 ●" if available else "⚙ 配置") + self._settings_btn.setProperty("hasUpdate", bool(available)) + self._settings_btn.setToolTip( + "发现新版本 v{}，点击进入设置更新".format(version) if available else "") + self._settings_btn.style().unpolish(self._settings_btn) + self._settings_btn.style().polish(self._settings_btn) def _create_work_area(self): """Horizontal splitter: left material panel | canvas | right params.""" @@ -616,33 +555,6 @@ class MainWindow(QMainWindow): background-color: #f0f0f0; } - /* Update notification banner */ - #updateBanner { - background-color: #eef6ff; - border-bottom: 1px solid #cfe3fa; - } - #updateBannerText { - font-family: "Microsoft YaHei", "Segoe UI", sans-serif; - font-size: 12px; - color: #1b4f86; - } - #updateBannerOpen { - font-size: 12px; - padding: 3px 12px; - border: 1px solid #0078d4; - border-radius: 3px; - color: #0078d4; - background: transparent; - } - #updateBannerOpen:hover { background: #d8e9fb; } - #updateBannerClose { - font-size: 12px; - border: none; - color: #6a8bab; - background: transparent; - } - #updateBannerClose:hover { color: #1b4f86; } - /* Flow tab bar */ #tabBarContainer { background-color: #f0f0f0; @@ -677,6 +589,7 @@ class MainWindow(QMainWindow): padding: 4px 14px; } #settingsBtn:hover { color: #0078d4; background: #e8f0fb; } + #settingsBtn[hasUpdate="true"] { color: #c42b1c; font-weight: bold; } /* Work splitter */ QSplitter#workSplitter::handle { diff --git a/src/app/widgets/settings_dialog.py b/src/app/widgets/settings_dialog.py index 9066138..fa7807b 100644 --- a/src/app/widgets/settings_dialog.py +++ b/src/app/widgets/settings_dialog.py @@ -20,24 +20,34 @@ from PySide6.QtWidgets import ( ) from version import APP_VERSION -from services.update_service import is_newer, load_manifest +from services.file_service import get_app_dir +from services import installer +from services.update_service import check_for_update, is_newer, load_manifest logger = logging.getLogger(__name__) class SettingsDialog(QDialog): - """Edit update_source / update_user / update_pass with a connection test.""" + """Edit update config; test the connection; download+stage an update. - _test_done = Signal(str) # result message, delivered to the UI thread + Downloading happens here (app is running, writes to staging\\app.new); the + actual app/app.old swap is applied by the launcher on the next start, since + Windows locks the running exe (docs/10-lan-update.md). + """ + + _test_done = Signal(str) # connection-test result message (UI thread) + _update_done = Signal(bool, str) # (ok, message) after download+stage def __init__(self, parent=None, *, update_source="", update_user="", - update_pass="", current_version=APP_VERSION): + update_pass="", current_version=APP_VERSION, update_info=None): super().__init__(parent) self.setWindowTitle("设置") self.setMinimumWidth(420) self._current_version = current_version + self._update_info = update_info self._setup_ui(update_source, update_user, update_pass) self._test_done.connect(self._on_test_done) + self._update_done.connect(self._on_update_done) # ── UI ─────────────────────────────────────────────────────────────────── @@ -83,7 +93,19 @@ class SettingsDialog(QDialog): self._result.setWordWrap(True) col.addWidget(self._result) - hint = QLabel("提示：建议使用只读账号；生产环境请走 HTTPS（凭据为明文存储）。") + # ── update section ─────────────────────────────────────────────────── + update_row = QHBoxLayout() + self._update_status = QLabel(self._update_status_text()) + self._update_status.setObjectName("settingsUpdate") + self._update_status.setWordWrap(True) + update_row.addWidget(self._update_status, 1) + self._update_btn = QPushButton("检查并更新") + self._update_btn.clicked.connect(self._on_update) + update_row.addWidget(self._update_btn) + col.addLayout(update_row) + + hint = QLabel("提示：建议使用只读账号；生产环境请走 HTTPS（凭据为明文存储）。" + "更新下载完成后，下次启动时自动生效。") hint.setObjectName("settingsHint") hint.setWordWrap(True) col.addWidget(hint) @@ -144,3 +166,40 @@ class SettingsDialog(QDialog): def _on_test_done(self, msg): self._result.setText(msg) self._test_btn.setEnabled(True) + + # ── manual update: download + stage (swap applied by launcher next start) ── + + def _update_status_text(self): + if self._update_info: + return "发现新版本 v{}（当前 v{}）。".format( + self._update_info.version, self._current_version) + return "当前 v{}。".format(self._current_version) + + def _on_update(self): + source = self._source_edit.text().strip() + if not source: + self._update_status.setText("请先填写更新地址。") + return + user, password = self._user_edit.text(), self._pass_edit.text() + install_root = installer.install_root_for(get_app_dir()) + self._update_btn.setEnabled(False) + self._update_status.setText("正在下载…") + + def worker(): + try: + info = check_for_update(source, self._current_version, user, password) + if info is None: + self._update_done.emit(True, "已是最新版本（或更新源不可用）。") + return + installer.download_and_stage(install_root, info, user, password) + self._update_done.emit( + True, "已下载 v{}，下次启动时自动更新。".format(info.version)) + except Exception as exc: + logger.info("Update download failed: %s", exc) + self._update_done.emit(False, "更新失败：{}".format(exc)) + + threading.Thread(target=worker, name="settings-update", daemon=True).start() + + def _on_update_done(self, ok, msg): + self._update_status.setText(msg) + self._update_btn.setEnabled(True) diff --git a/src/launcher.py b/src/launcher.py index f6dfe6e..ddd6405 100644 --- a/src/launcher.py +++ b/src/launcher.py @@ -1,54 +1,36 @@ -"""CMBot update launcher (docs/10-lan-update.md stage 3/4). +"""CMBot launcher (docs/10-lan-update.md). Compiled to Launcher.exe (PyInstaller onefile) and run instead of launching app/CMBot.exe directly. On each start it: 1. seeds default config/templates into ~/.cmbot on first run; - 2. reads update_source / credentials from ~/.cmbot/config/app_config.json; - 3. checks the manifest and, when a newer version is advertised, downloads the - release zip, verifies SHA-256, extracts it, and swaps it into app\\ while - keeping the previous version in app.old\\ for rollback; - 4. launches app\\CMBot.exe. + 2. applies a previously staged update if one is ready (fast local swap of + staging\\app.new into app\\, keeping app.old\\ for rollback); + 3. launches app\\CMBot.exe. -Degrades safely: an unreachable source, a bad download, or a non-writable -install root leaves the existing app\\ in place and launches it. The running -program's data lives in ~/.cmbot (get_data_dir), so updates never touch it. +It does NOT download anything — downloading happens inside the app (manual +"更新" button / background), so startup is never blocked on the network. The +running program's data lives in ~/.cmbot (get_data_dir), untouched by updates. """ import argparse -import hashlib import logging import shutil import subprocess import sys -import zipfile from pathlib import Path -from urllib import error # Make 'services' importable both frozen (PyInstaller --paths src) and from source. sys.path.insert(0, str(Path(__file__).resolve().parent)) -from services.update_service import check_for_update, download # noqa: E402 -from services.file_service import get_data_dir # noqa: E402 -from services.config_service import load_config # noqa: E402 +from services import installer # noqa: E402 +from services.file_service import get_data_dir # noqa: E402 APP_EXE = "CMBot.exe" -VERSION_FILE = "version.txt" CONFIG_FILES = ("app_config.json", "templates.json") logger = logging.getLogger("launcher") -# ── helpers ────────────────────────────────────────────────────────────────── - -def _read_version(app_dir): - """Return the version recorded in app/version.txt, or '' if absent.""" - try: - # utf-8-sig tolerates a BOM written by PowerShell. - return (app_dir / VERSION_FILE).read_text(encoding="utf-8-sig").strip() - except OSError: - return "" - - def seed_defaults(app_dir, data_dir): """First run: copy factory config/templates from app\\config into the data root, without overwriting any existing user file.""" @@ -64,128 +46,20 @@ def seed_defaults(app_dir, data_dir): logger.info("Seeded default %s into %s", name, dst) -def is_writable(path): - """True if a file can be created under *path* (so self-update can proceed).""" - try: - path.mkdir(parents=True, exist_ok=True) - probe = path / ".write_test" - probe.write_text("x", encoding="ascii") - probe.unlink() - return True - except OSError: - return False - - -def _sha256(path): - h = hashlib.sha256() - with open(str(path), "rb") as f: - for chunk in iter(lambda: f.read(65536), b""): - h.update(chunk) - return h.hexdigest() - - -def _rm(path): - if path.is_dir(): - shutil.rmtree(str(path), ignore_errors=True) - elif path.exists(): - path.unlink() - - -def _find_package_root(extract_dir): - """Locate the folder that holds CMBot.exe inside an extracted zip. - - Handles both a flat zip (exe at root) and one wrapped in a single folder. - """ - if (extract_dir / APP_EXE).exists(): - return extract_dir - subs = [p for p in extract_dir.iterdir() if p.is_dir()] - if len(subs) == 1 and (subs[0] / APP_EXE).exists(): - return subs[0] - return None - - -def _swap_in(app_dir, old_dir, package_root): - """Rename app->app.old, then package_root->app. Roll back on failure. - - Never overwrites a running app dir: if app\\CMBot.exe is locked (running), - the rename raises and the caller degrades to the existing version. - """ - if old_dir.exists(): - shutil.rmtree(str(old_dir)) - if app_dir.exists(): - app_dir.rename(old_dir) - try: - package_root.rename(app_dir) - except OSError: - if not app_dir.exists() and old_dir.exists(): - old_dir.rename(app_dir) # restore previous version - raise - - -# ── update flow ────────────────────────────────────────────────────────────── - -def _try_update(install_root, app_dir, source, user, password, local_version): - info = check_for_update(source, local_version, user, password) - if info is None: - return # no update / unreachable / not newer — already logged - - if not is_writable(install_root): - logger.warning("Install root not writable, skipping update: %s", install_root) - return - - staging = install_root / "staging" - staging.mkdir(parents=True, exist_ok=True) - zip_path = staging / "{}.zip".format(info.version) - extract_dir = staging / "app.new" - _rm(zip_path) - _rm(extract_dir) - - logger.info("Downloading v%s from %s", info.version, info.source) - download(info.source, zip_path, user, password) - - if info.size and zip_path.stat().st_size != info.size: - raise ValueError("download size mismatch: got {}, expected {}".format( - zip_path.stat().st_size, info.size)) - if info.sha256 and _sha256(zip_path).lower() != info.sha256.lower(): - raise ValueError("sha256 mismatch") - - with zipfile.ZipFile(str(zip_path)) as z: - z.extractall(str(extract_dir)) - package_root = _find_package_root(extract_dir) - if not package_root: - raise ValueError("downloaded package does not contain {}".format(APP_EXE)) - pkg_version = _read_version(package_root) - if pkg_version != info.version: - raise ValueError("package version '{}' != manifest '{}'".format( - pkg_version, info.version)) - - _swap_in(app_dir, install_root / "app.old", package_root) - _rm(zip_path) - _rm(extract_dir) - logger.info("Installed and switched to v%s", info.version) - - def run(install_root, no_launch=False): - """Seed config, attempt an update, then launch the app. Returns exit code.""" + """Seed config, apply any staged update, then launch the app. Returns exit code.""" install_root = Path(install_root) app_dir = install_root / "app" data_dir = get_data_dir() seed_defaults(app_dir, data_dir) - cfg = load_config() - source = cfg.get("update_source", "") - user = cfg.get("update_user", "") - password = cfg.get("update_pass", "") - - local_version = _read_version(app_dir) - if source: - try: - _try_update(install_root, app_dir, source, user, password, local_version) - except (OSError, ValueError, error.URLError) as exc: - logger.warning("Update skipped (using local version): %s", exc) - else: - logger.info("No update_source configured, skipping update check.") + try: + applied = installer.apply_staged(install_root) + if applied: + logger.info("Applied staged update v%s", applied) + except OSError as exc: + logger.warning("Apply staged update failed (using current version): %s", exc) exe = app_dir / APP_EXE if not exe.exists(): @@ -205,25 +79,12 @@ def _default_install_root(): return Path.cwd() -def main(argv=None): - parser = argparse.ArgumentParser(description="CMBot update launcher") - parser.add_argument("--install-root", default=None, - help="install root (default: Launcher.exe's folder)") - parser.add_argument("--no-launch", action="store_true", - help="update only, do not start the app") - args = parser.parse_args(argv) - - root = Path(args.install_root) if args.install_root else _default_install_root() - _setup_logging(root) - return run(root, no_launch=args.no_launch) - - def _setup_logging(root): """Configure logging defensively for both console and --windowed builds. - A PyInstaller --windowed exe has no stdout/stderr (StreamHandler(None) would - fail), and a read-only install root makes the file handler fail. Add each - handler only when it can be created so the launcher never crashes on logging. + A PyInstaller --windowed exe has no stdout/stderr; a read-only install root + makes the file handler fail. Add each handler only when it can be created so + the launcher never crashes on logging. """ handlers = [] if sys.stderr is not None: @@ -236,5 +97,18 @@ def _setup_logging(root): handlers=handlers) +def main(argv=None): + parser = argparse.ArgumentParser(description="CMBot launcher") + parser.add_argument("--install-root", default=None, + help="install root (default: Launcher.exe's folder)") + parser.add_argument("--no-launch", action="store_true", + help="apply staged update only, do not start the app") + args = parser.parse_args(argv) + + root = Path(args.install_root) if args.install_root else _default_install_root() + _setup_logging(root) + return run(root, no_launch=args.no_launch) + + if __name__ == "__main__": sys.exit(main()) diff --git a/src/services/installer.py b/src/services/installer.py new file mode 100644 index 0000000..13edba5 --- /dev/null +++ b/src/services/installer.py @@ -0,0 +1,186 @@ +"""Stage and apply portable-layout updates (docs/10-lan-update.md). + +Split into two phases because Windows locks a running .exe and its folder: + +- download_and_stage(): runs WHILE the app is open — downloads the release zip, + verifies SHA-256, extracts it to staging\\app.new. No swap (app\\ is in use). +- apply_staged(): runs at launch time from the launcher, when the app is NOT + running — swaps staging\\app.new into app\\, keeping the old one as app.old\\. + +All paths are relative to the install root (the folder that holds app\\ and +Launcher.exe). +""" +import hashlib +import logging +import shutil +import zipfile +from pathlib import Path + +from services.update_service import download, is_newer + +logger = logging.getLogger(__name__) + +APP_EXE = "CMBot.exe" +VERSION_FILE = "version.txt" + + +# ── layout helpers ─────────────────────────────────────────────────────────── + +def app_dir(install_root): + return Path(install_root) / "app" + + +def old_dir(install_root): + return Path(install_root) / "app.old" + + +def staging_dir(install_root): + return Path(install_root) / "staging" + + +def staged_app(install_root): + """staging\\app.new — a downloaded+verified package waiting to be applied.""" + return staging_dir(install_root) / "app.new" + + +def install_root_for(app_path): + """Install root given the running app\\ directory (its parent).""" + return Path(app_path).resolve().parent + + +def read_version(folder): + """version.txt inside *folder*, or '' if absent (utf-8-sig tolerates BOM).""" + try: + return (Path(folder) / VERSION_FILE).read_text(encoding="utf-8-sig").strip() + except OSError: + return "" + + +def is_writable(path): + """True if a file can be created under *path* (self-update needs this).""" + path = Path(path) + try: + path.mkdir(parents=True, exist_ok=True) + probe = path / ".write_test" + probe.write_text("x", encoding="ascii") + probe.unlink() + return True + except OSError: + return False + + +# ── internal ───────────────────────────────────────────────────────────────── + +def _sha256(path): + h = hashlib.sha256() + with open(str(path), "rb") as f: + for chunk in iter(lambda: f.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def _rm(path): + path = Path(path) + if path.is_dir(): + shutil.rmtree(str(path), ignore_errors=True) + elif path.exists(): + path.unlink() + + +def _find_package_root(extract_dir): + """Folder holding CMBot.exe — flat zip (root) or wrapped in one subfolder.""" + extract_dir = Path(extract_dir) + if (extract_dir / APP_EXE).exists(): + return extract_dir + subs = [p for p in extract_dir.iterdir() if p.is_dir()] + if len(subs) == 1 and (subs[0] / APP_EXE).exists(): + return subs[0] + return None + + +# ── phase 1: download + stage (runs while the app is open) ─────────────────── + +def download_and_stage(install_root, info, update_user="", update_pass=""): + """Download, verify and extract *info* into staging\\app.new. Does NOT swap. + + Returns the staged version string. Raises (OSError/ValueError) on any failure + (not writable, download error, size/hash mismatch, bad package). + """ + install_root = Path(install_root) + if not is_writable(install_root): + raise OSError("安装目录不可写：{}".format(install_root)) + + st = staging_dir(install_root) + st.mkdir(parents=True, exist_ok=True) + zip_path = st / "{}.zip".format(info.version) + extract_tmp = st / "extract.tmp" + target = staged_app(install_root) + _rm(zip_path) + _rm(extract_tmp) + _rm(target) + + logger.info("Downloading v%s from %s", info.version, info.source) + download(info.source, zip_path, update_user, update_pass) + + if info.size and zip_path.stat().st_size != info.size: + raise ValueError("下载大小不符：得到 {}，期望 {}".format( + zip_path.stat().st_size, info.size)) + if info.sha256 and _sha256(zip_path).lower() != info.sha256.lower(): + raise ValueError("SHA-256 校验不通过") + + with zipfile.ZipFile(str(zip_path)) as z: + z.extractall(str(extract_tmp)) + root = _find_package_root(extract_tmp) + if not root: + raise ValueError("更新包缺少 {}".format(APP_EXE)) + pkg_ver = read_version(root) + if pkg_ver != info.version: + raise ValueError("包内版本 {} 与清单 {} 不一致".format(pkg_ver, info.version)) + + root.rename(target) + _rm(extract_tmp) + _rm(zip_path) + logger.info("Staged update v%s at %s", info.version, target) + return info.version + + +def staged_version(install_root): + """Version ready in staging\\app.new, or '' if none/invalid.""" + src = staged_app(install_root) + return read_version(src) if (src / APP_EXE).exists() else "" + + +# ── phase 2: apply (runs at launch, app not running) ───────────────────────── + +def apply_staged(install_root): + """Swap a ready staging\\app.new into app\\ (keeping app.old\\). Returns the + applied version, or '' if nothing valid/newer was staged. + + Never overwrites a running app\\: if app\\CMBot.exe is locked the rename + raises and the caller degrades to the current version. + """ + install_root = Path(install_root) + src = staged_app(install_root) + if not (src / APP_EXE).exists(): + return "" + + ver = read_version(src) + cur = read_version(app_dir(install_root)) + if cur and ver and not is_newer(ver, cur): + _rm(src) # stale/not newer — discard + return "" + + a = app_dir(install_root) + o = old_dir(install_root) + if o.exists(): + shutil.rmtree(str(o)) + if a.exists(): + a.rename(o) + try: + src.rename(a) + except OSError: + if not a.exists() and o.exists(): + o.rename(a) # restore previous version + raise + logger.info("Applied staged update v%s", ver) + return ver diff --git a/tests/test_installer.py b/tests/test_installer.py new file mode 100644 index 0000000..57cf6b6 --- /dev/null +++ b/tests/test_installer.py @@ -0,0 +1,130 @@ +"""Tests for services.installer — no GUI/network (download is mocked).""" +import hashlib +import shutil +import sys +import tempfile +import unittest +import zipfile +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from services import installer +from services.update_service import UpdateInfo + + +def _make_release_zip(dst_zip, version, exe_body, wrap=False): + tmp = Path(tempfile.mkdtemp()) + try: + base = tmp / ("CMBot-" + version) if wrap else tmp + base.mkdir(parents=True, exist_ok=True) + (base / "CMBot.exe").write_text(exe_body, encoding="ascii") + (base / "version.txt").write_text(version, encoding="ascii") + with zipfile.ZipFile(str(dst_zip), "w", zipfile.ZIP_DEFLATED) as z: + for p in base.rglob("*"): + z.write(str(p), str(p.relative_to(tmp))) + finally: + shutil.rmtree(str(tmp), ignore_errors=True) + return hashlib.sha256(dst_zip.read_bytes()).hexdigest() + + +class _Base(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + self.root = self.tmp / "install" + self.app = self.root / "app" + self.app.mkdir(parents=True) + (self.app / "CMBot.exe").write_text("OLD-1.0.0", encoding="ascii") + (self.app / "version.txt").write_text("1.0.0", encoding="ascii") + + def tearDown(self): + shutil.rmtree(str(self.tmp), ignore_errors=True) + + def _info(self, version, zip_path): + return UpdateInfo(version=version, source="http://x/{}.zip".format(version), + sha256=hashlib.sha256(zip_path.read_bytes()).hexdigest(), + size=zip_path.stat().st_size) + + +class TestDownloadAndStage(_Base): + def _copy_download(self, src_zip): + def fake(url, dest, user="", password="", timeout=120): + shutil.copy(str(src_zip), str(dest)) + return fake + + def test_stage_flat_zip(self): + z = self.tmp / "n.zip" + _make_release_zip(z, "1.1.0", "NEW") + with patch("services.installer.download", side_effect=self._copy_download(z)): + ver = installer.download_and_stage(self.root, self._info("1.1.0", z)) + self.assertEqual(ver, "1.1.0") + self.assertEqual(installer.staged_version(self.root), "1.1.0") + self.assertTrue((installer.staged_app(self.root) / "CMBot.exe").exists()) + # app untouched until apply + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") + + def test_stage_wrapped_zip(self): + z = self.tmp / "n.zip" + _make_release_zip(z, "1.1.0", "NEW", wrap=True) + with patch("services.installer.download", side_effect=self._copy_download(z)): + ver = installer.download_and_stage(self.root, self._info("1.1.0", z)) + self.assertEqual(ver, "1.1.0") + self.assertEqual(installer.staged_version(self.root), "1.1.0") + + def test_sha_mismatch_raises(self): + z = self.tmp / "n.zip" + _make_release_zip(z, "1.1.0", "NEW") + info = self._info("1.1.0", z) + info.sha256 = "deadbeef" + with patch("services.installer.download", side_effect=self._copy_download(z)): + with self.assertRaises(ValueError): + installer.download_and_stage(self.root, info) + self.assertEqual(installer.staged_version(self.root), "") + + def test_version_mismatch_raises(self): + z = self.tmp / "n.zip" + _make_release_zip(z, "1.1.0", "NEW") + info = self._info("9.9.9", z) # manifest claims a different version + with patch("services.installer.download", side_effect=self._copy_download(z)): + with self.assertRaises(ValueError): + installer.download_and_stage(self.root, info) + + +class TestApplyStaged(_Base): + def _stage(self, version, body): + s = installer.staged_app(self.root) + s.mkdir(parents=True, exist_ok=True) + (s / "CMBot.exe").write_text(body, encoding="ascii") + (s / "version.txt").write_text(version, encoding="ascii") + + def test_apply_swaps(self): + self._stage("1.1.0", "NEW-1.1.0") + applied = installer.apply_staged(self.root) + self.assertEqual(applied, "1.1.0") + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.1.0") + self.assertEqual((self.app / "CMBot.exe").read_text(encoding="ascii"), "NEW-1.1.0") + self.assertEqual((installer.old_dir(self.root) / "version.txt").read_text(encoding="ascii"), "1.0.0") + self.assertEqual(installer.staged_version(self.root), "") + + def test_nothing_staged_returns_empty(self): + self.assertEqual(installer.apply_staged(self.root), "") + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") + + def test_not_newer_staged_is_discarded(self): + self._stage("0.9.0", "OLDER") + self.assertEqual(installer.apply_staged(self.root), "") + self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") + self.assertFalse(installer.staged_app(self.root).exists()) + + +class TestHelpers(_Base): + def test_is_writable(self): + self.assertTrue(installer.is_writable(self.root)) + + def test_install_root_for(self): + self.assertEqual(installer.install_root_for(self.app), self.root.resolve()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_launcher.py b/tests/test_launcher.py index fbde8f8..acdd85d 100644 --- a/tests/test_launcher.py +++ b/tests/test_launcher.py @@ -1,153 +1,86 @@ -"""Tests for launcher — no GUI/network dependency (download is mocked).""" -import hashlib +"""Tests for the launcher — no GUI/network. The launcher only seeds config and +applies a pre-staged update (downloading lives in services.installer/the app).""" import json import os import shutil import sys import tempfile import unittest -import zipfile from pathlib import Path -from unittest.mock import patch sys.path.insert(0, str(Path(__file__).parent.parent / "src")) import launcher -from services.update_service import UpdateInfo +from services import installer -def _make_release_zip(dst_zip, version, exe_body): - """Build a flat release zip (CMBot.exe + version.txt) and return its sha256.""" - tmp = Path(tempfile.mkdtemp()) - try: - (tmp / "CMBot.exe").write_text(exe_body, encoding="ascii") - (tmp / "version.txt").write_text(version, encoding="ascii") - with zipfile.ZipFile(str(dst_zip), "w", zipfile.ZIP_DEFLATED) as z: - z.write(str(tmp / "CMBot.exe"), "CMBot.exe") - z.write(str(tmp / "version.txt"), "version.txt") - finally: - shutil.rmtree(str(tmp), ignore_errors=True) - return hashlib.sha256(dst_zip.read_bytes()).hexdigest() - - -class _LauncherBase(unittest.TestCase): +class _Base(unittest.TestCase): def setUp(self): self.tmp = Path(tempfile.mkdtemp()) - self.install = self.tmp / "install" - self.app = self.install / "app" + self.root = self.tmp / "install" + self.app = self.root / "app" (self.app / "config").mkdir(parents=True) (self.app / "CMBot.exe").write_text("OLD-1.0.0", encoding="ascii") (self.app / "version.txt").write_text("1.0.0", encoding="ascii") - # factory defaults shipped inside app\config (self.app / "config" / "app_config.json").write_text( - json.dumps({"update_source": "http://x/manifest.json", "last_template": "正方形"}), - encoding="utf-8") + json.dumps({"update_source": "http://x"}), encoding="utf-8") (self.app / "config" / "templates.json").write_text("{}", encoding="utf-8") - # data root -> a temp dir via the env override self.data = self.tmp / "data" - self._prev_env = os.environ.get("CMBOT_DATA_DIR") + self._prev = os.environ.get("CMBOT_DATA_DIR") os.environ["CMBOT_DATA_DIR"] = str(self.data) def tearDown(self): - if self._prev_env is None: + if self._prev is None: os.environ.pop("CMBOT_DATA_DIR", None) else: - os.environ["CMBOT_DATA_DIR"] = self._prev_env + os.environ["CMBOT_DATA_DIR"] = self._prev shutil.rmtree(str(self.tmp), ignore_errors=True) + def _stage(self, version, body): + s = installer.staged_app(self.root) + s.mkdir(parents=True, exist_ok=True) + (s / "CMBot.exe").write_text(body, encoding="ascii") + (s / "version.txt").write_text(version, encoding="ascii") -class TestHelpers(_LauncherBase): - def test_seed_defaults_copies_when_missing(self): + +class TestSeed(_Base): + def test_seed_copies_when_missing(self): launcher.seed_defaults(self.app, self.data) self.assertTrue((self.data / "config" / "app_config.json").exists()) self.assertTrue((self.data / "config" / "templates.json").exists()) - def test_seed_defaults_does_not_overwrite(self): + def test_seed_does_not_overwrite(self): (self.data / "config").mkdir(parents=True) - (self.data / "config" / "app_config.json").write_text( - '{"update_source":"USER"}', encoding="utf-8") + (self.data / "config" / "app_config.json").write_text('{"update_source":"USER"}', encoding="utf-8") launcher.seed_defaults(self.app, self.data) kept = json.loads((self.data / "config" / "app_config.json").read_text(encoding="utf-8")) self.assertEqual(kept["update_source"], "USER") - def test_is_writable(self): - self.assertTrue(launcher.is_writable(self.install)) - - def test_find_package_root_flat(self): - d = self.tmp / "flat" - d.mkdir() - (d / "CMBot.exe").write_text("x", encoding="ascii") - self.assertEqual(launcher._find_package_root(d), d) - - def test_find_package_root_wrapped(self): - d = self.tmp / "wrap" - (d / "CMBot-1.1.0").mkdir(parents=True) - (d / "CMBot-1.1.0" / "CMBot.exe").write_text("x", encoding="ascii") - self.assertEqual(launcher._find_package_root(d), d / "CMBot-1.1.0") - - def test_find_package_root_missing(self): - d = self.tmp / "empty" - d.mkdir() - self.assertIsNone(launcher._find_package_root(d)) - - def test_read_version_tolerates_bom(self): - (self.app / "version.txt").write_text("1.0.0", encoding="utf-8-sig") - self.assertEqual(launcher._read_version(self.app), "1.0.0") - - -class TestRun(_LauncherBase): - def _info(self, version, zip_path): - sha = hashlib.sha256(zip_path.read_bytes()).hexdigest() - return UpdateInfo(version=version, source="http://x/{}.zip".format(version), - sha256=sha, size=zip_path.stat().st_size) - - def test_update_applied_and_swapped(self): - new_zip = self.tmp / "new.zip" - _make_release_zip(new_zip, "1.1.0", "NEW-1.1.0") - info = self._info("1.1.0", new_zip) - - def fake_download(url, dest, user="", password="", timeout=120): - shutil.copy(str(new_zip), str(dest)) - - with patch("launcher.check_for_update", return_value=info), \ - patch("launcher.download", side_effect=fake_download): - rc = launcher.run(self.install, no_launch=True) +class TestRun(_Base): + def test_applies_staged_update_then_launches(self): + self._stage("1.1.0", "NEW-1.1.0") + rc = launcher.run(self.root, no_launch=True) self.assertEqual(rc, 0) self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.1.0") self.assertEqual((self.app / "CMBot.exe").read_text(encoding="ascii"), "NEW-1.1.0") - self.assertEqual(((self.install / "app.old") / "version.txt").read_text(encoding="ascii"), "1.0.0") - self.assertFalse((self.install / "staging" / "1.1.0.zip").exists()) + self.assertEqual((installer.old_dir(self.root) / "version.txt").read_text(encoding="ascii"), "1.0.0") - def test_no_update_keeps_local(self): - with patch("launcher.check_for_update", return_value=None): - rc = launcher.run(self.install, no_launch=True) + def test_no_staged_just_launches(self): + rc = launcher.run(self.root, no_launch=True) self.assertEqual(rc, 0) self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") - self.assertFalse((self.install / "app.old").exists()) - - def test_sha256_mismatch_degrades(self): - new_zip = self.tmp / "new.zip" - _make_release_zip(new_zip, "1.1.0", "NEW") - info = self._info("1.1.0", new_zip) - info.sha256 = "deadbeef" # force mismatch - - def fake_download(url, dest, user="", password="", timeout=120): - shutil.copy(str(new_zip), str(dest)) - - with patch("launcher.check_for_update", return_value=info), \ - patch("launcher.download", side_effect=fake_download): - rc = launcher.run(self.install, no_launch=True) - - self.assertEqual(rc, 0) - self.assertEqual((self.app / "version.txt").read_text(encoding="ascii"), "1.0.0") - self.assertFalse((self.install / "app.old").exists()) + self.assertFalse(installer.old_dir(self.root).exists()) def test_seeds_config_into_data_root(self): - with patch("launcher.check_for_update", return_value=None): - launcher.run(self.install, no_launch=True) + launcher.run(self.root, no_launch=True) self.assertTrue((self.data / "config" / "app_config.json").exists()) + def test_missing_exe_returns_error(self): + (self.app / "CMBot.exe").unlink() + rc = launcher.run(self.root, no_launch=True) + self.assertEqual(rc, 1) + if __name__ == "__main__": unittest.main()