diff --git a/app/updater_entry.py b/app/updater_entry.py new file mode 100644 index 0000000..2a6e1f0 --- /dev/null +++ b/app/updater_entry.py @@ -0,0 +1,364 @@ +"""在主程序退出后执行程序根项目事务替换的独立更新器。""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import shutil +import subprocess +import tempfile +import time +import uuid +from dataclasses import dataclass +from pathlib import Path, PurePosixPath + + +PACKAGE_FORMAT = "cmshopee-portable-v1" +UPDATER_PROTOCOL = 1 +ALLOWED_ROOTS = { + "cmshopee.exe", + "_internal", + "version.txt", + "README.txt", + "package-manifest.json", + "cmshopee-updater.exe", +} +PROTECTED_ROOTS = {"data", ".cmshopee-update"} + + +class UpdaterError(RuntimeError): + """独立更新器拒绝计划或无法安全完成事务。""" + + +@dataclass(frozen=True) +class UpdatePlan: + parent_pid: int + install_root: Path + staging_root: Path + target_version: str + transaction_id: str + log_path: Path + + +def create_plan(install_root, staging_root, target_version, parent_pid): + install_root = Path(install_root).resolve() + transaction_id = uuid.uuid4().hex + update_root = install_root / ".cmshopee-update" + plan = UpdatePlan( + parent_pid=int(parent_pid), + install_root=install_root, + staging_root=Path(staging_root).resolve(), + target_version=str(target_version), + transaction_id=transaction_id, + log_path=(update_root / "logs" / (transaction_id + ".log")).resolve(), + ) + validate_plan(plan) + plan_path = update_root / "plans" / (transaction_id + ".json") + _atomic_json( + plan_path, + { + "parent_pid": plan.parent_pid, + "install_root": str(plan.install_root), + "staging_root": str(plan.staging_root), + "target_version": plan.target_version, + "transaction_id": plan.transaction_id, + "log_path": str(plan.log_path), + }, + ) + return plan_path + + +def _is_child(path, parent): + try: + Path(path).resolve().relative_to(Path(parent).resolve()) + return True + except ValueError: + return False + + +def _atomic_json(path, payload): + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + os.replace(str(temporary), str(path)) + + +def _append_log(path, message): + Path(path).parent.mkdir(parents=True, exist_ok=True) + with Path(path).open("a", encoding="utf-8") as stream: + stream.write("%s %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), message)) + + +def load_plan(plan_path): + plan_path = Path(plan_path).resolve() + try: + payload = json.loads(plan_path.read_text(encoding="utf-8")) + plan = UpdatePlan( + parent_pid=int(payload["parent_pid"]), + install_root=Path(payload["install_root"]).resolve(), + staging_root=Path(payload["staging_root"]).resolve(), + target_version=str(payload["target_version"]), + transaction_id=str(payload["transaction_id"]), + log_path=Path(payload["log_path"]).resolve(), + ) + except (OSError, ValueError, KeyError, TypeError) as exc: + raise UpdaterError("更新计划无效") from exc + validate_plan(plan, plan_path) + return plan + + +def validate_plan(plan, plan_path=None): + if plan.parent_pid <= 0: + raise UpdaterError("更新计划缺少主程序进程编号") + if not plan.install_root.is_absolute() or not plan.install_root.is_dir(): + raise UpdaterError("程序安装目录无效") + update_root = plan.install_root / ".cmshopee-update" + staging_parent = update_root / "staging" + if not plan.staging_root.is_dir() or plan.staging_root.parent != staging_parent.resolve(): + raise UpdaterError("新版暂存目录越界") + if not re.fullmatch(r"\d+(?:\.\d+)*", plan.target_version): + raise UpdaterError("目标版本号格式不正确") + if not re.fullmatch(r"[A-Za-z0-9-]{8,64}", plan.transaction_id): + raise UpdaterError("更新事务编号无效") + if not _is_child(plan.log_path, update_root / "logs"): + raise UpdaterError("更新日志路径越界") + if plan_path is not None and not _is_child(plan_path, update_root): + raise UpdaterError("更新计划文件路径越界") + probe = update_root / (".write-probe-" + plan.transaction_id) + try: + probe.parent.mkdir(parents=True, exist_ok=True) + probe.write_bytes(b"") + probe.unlink() + except OSError as exc: + raise UpdaterError("程序安装目录不可写") from exc + + +def _sha256(path): + digest = hashlib.sha256() + with Path(path).open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def validate_staging(plan): + manifest_path = plan.staging_root / "package-manifest.json" + try: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise UpdaterError("新版程序清单无效") from exc + if ( + manifest.get("package_format") != PACKAGE_FORMAT + or manifest.get("app_version") != plan.target_version + or manifest.get("entry_point") != "cmshopee.exe" + or int(manifest.get("updater_protocol") or 0) != UPDATER_PROTOCOL + ): + raise UpdaterError("新版程序清单与更新计划不一致") + roots = manifest.get("replace_roots") + if not isinstance(roots, list) or not roots: + raise UpdaterError("新版程序替换范围无效") + roots = list(dict.fromkeys(str(root) for root in roots)) + if any(root not in ALLOWED_ROOTS or root in PROTECTED_ROOTS for root in roots): + raise UpdaterError("新版程序包含未授权替换范围") + if "cmshopee.exe" not in roots or "_internal" not in roots: + raise UpdaterError("新版程序缺少必要替换项目") + + declared = {} + for item in manifest.get("files") or []: + relative = PurePosixPath(str(item.get("path") or "")) + if not relative.parts or ".." in relative.parts or relative.parts[0] not in roots: + raise UpdaterError("新版程序清单包含越界文件") + key = relative.as_posix().casefold() + if key in declared: + raise UpdaterError("新版程序清单包含重复文件") + declared[key] = (relative, int(item.get("size_bytes") or -1), str(item.get("sha256") or "")) + actual = {} + for path in plan.staging_root.rglob("*"): + if path.is_symlink(): + raise UpdaterError("新版程序不允许符号链接") + if path.is_file() and path.name != "package-manifest.json": + relative = PurePosixPath(path.relative_to(plan.staging_root).as_posix()) + actual[relative.as_posix().casefold()] = relative + if set(actual) != set(declared): + raise UpdaterError("新版程序文件与清单不一致") + for key, (relative, size_bytes, expected_hash) in declared.items(): + path = plan.staging_root.joinpath(*relative.parts) + if path.stat().st_size != size_bytes or _sha256(path).lower() != expected_hash.lower(): + raise UpdaterError("新版程序文件校验失败") + if (plan.staging_root / "version.txt").read_text(encoding="utf-8-sig").strip() != plan.target_version: + raise UpdaterError("新版程序版本文件不一致") + return roots + + +def _pid_running(pid): + if pid <= 0: + return False + if os.name == "nt": + import ctypes + + handle = ctypes.windll.kernel32.OpenProcess(0x1000, False, pid) + if handle: + ctypes.windll.kernel32.CloseHandle(handle) + return True + return False + try: + os.kill(pid, 0) + return True + except OSError: + return False + + +def wait_for_parent_exit(pid, timeout=60, poll_seconds=0.2): + deadline = time.monotonic() + timeout + while _pid_running(pid): + if time.monotonic() >= deadline: + raise UpdaterError("等待旧版程序退出超时") + time.sleep(poll_seconds) + + +class TransactionLock: + def __init__(self, path): + self.path = Path(path) + self.fd = None + + def __enter__(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + try: + self.fd = os.open(str(self.path), os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(self.fd, str(os.getpid()).encode("ascii")) + except FileExistsError as exc: + raise UpdaterError("已有更新程序正在处理当前安装目录") from exc + return self + + def __exit__(self, *_args): + if self.fd is not None: + os.close(self.fd) + try: + self.path.unlink() + except FileNotFoundError: + pass + + +def _default_launch(executable, args): + creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) + return subprocess.Popen([str(executable), *args], close_fds=True, creationflags=creationflags) + + +def apply_update(plan, *, wait_parent=wait_for_parent_exit, move=os.replace, launcher=_default_launch): + validate_plan(plan) + roots = validate_staging(plan) + update_root = plan.install_root / ".cmshopee-update" + old_version_path = plan.install_root / "version.txt" + old_version = old_version_path.read_text(encoding="utf-8-sig").strip() if old_version_path.exists() else "unknown" + backup_root = update_root / "backup" / (old_version + "-" + plan.transaction_id) + failed_root = update_root / "failed" / plan.transaction_id + journal_path = update_root / "transactions" / (plan.transaction_id + ".json") + operations = [] + + def journal(status, error=""): + _atomic_json( + journal_path, + { + "transaction_id": plan.transaction_id, + "target_version": plan.target_version, + "status": status, + "operations": operations, + "error": error, + }, + ) + + with TransactionLock(update_root / "update.lock"): + wait_parent(plan.parent_pid) + backup_root.mkdir(parents=True, exist_ok=False) + journal("moving_old") + try: + for root in roots: + source = plan.install_root / root + if source.exists(): + destination = backup_root / root + destination.parent.mkdir(parents=True, exist_ok=True) + move(str(source), str(destination)) + operations.append({"kind": "old_to_backup", "root": root}) + journal("moving_old") + for root in roots: + source = plan.staging_root / root + if not source.exists(): + raise UpdaterError("新版程序缺少替换项目：%s" % root) + destination = plan.install_root / root + move(str(source), str(destination)) + operations.append({"kind": "new_to_install", "root": root}) + journal("moving_new") + launcher( + plan.install_root / "cmshopee.exe", + ["--update-transaction", plan.transaction_id, "--update-target-version", plan.target_version], + ) + journal("launched") + _append_log(plan.log_path, "新版程序已替换并启动：%s" % plan.target_version) + return backup_root + except Exception as exc: + journal("rolling_back", str(exc)) + failed_root.mkdir(parents=True, exist_ok=True) + rollback_errors = [] + for operation in reversed(operations): + root = operation["root"] + try: + if operation["kind"] == "new_to_install": + source = plan.install_root / root + if source.exists(): + destination = failed_root / root + destination.parent.mkdir(parents=True, exist_ok=True) + move(str(source), str(destination)) + else: + source = backup_root / root + if source.exists(): + move(str(source), str(plan.install_root / root)) + except Exception as rollback_exc: + rollback_errors.append(str(rollback_exc)) + status = "rollback_failed" if rollback_errors else "rolled_back" + journal(status, "；".join(rollback_errors) or str(exc)) + _append_log(plan.log_path, "更新失败，已尝试恢复旧版：%s" % exc) + if rollback_errors: + raise UpdaterError("更新失败且旧版恢复不完整，请联系管理员") from exc + raise UpdaterError("更新失败，已恢复旧版程序") from exc + + +def copy_and_launch_updater(updater_path, plan_path): + updater_path = Path(updater_path).resolve() + if not updater_path.is_file(): + raise UpdaterError("独立更新程序不存在") + temporary_dir = Path(tempfile.mkdtemp(prefix="cmshopee-update-")) + external_updater = temporary_dir / updater_path.name + shutil.copy2(str(updater_path), str(external_updater)) + creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr( + subprocess, "DETACHED_PROCESS", 0 + ) + return subprocess.Popen( + [str(external_updater), "--plan", str(Path(plan_path).resolve())], + close_fds=True, + creationflags=creationflags, + ) + + +def main(argv=None): + parser = argparse.ArgumentParser(description="蝦皮圈优化助手独立更新程序") + parser.add_argument("--plan", required=True) + args = parser.parse_args(argv) + try: + plan = load_plan(args.plan) + apply_update(plan) + return 0 + except Exception as exc: + try: + if "plan" in locals(): + _append_log(plan.log_path, "自动更新失败：%s" % exc) + except OSError: + pass + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cmshopee-updater.spec b/cmshopee-updater.spec new file mode 100644 index 0000000..454456b --- /dev/null +++ b/cmshopee-updater.spec @@ -0,0 +1,28 @@ +# -*- mode: python ; coding: utf-8 -*- + +a = Analysis( + ["app/updater_entry.py"], + pathex=[], + binaries=[], + datas=[], + hiddenimports=[], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=["PySide6"], + noarchive=False, +) +pyz = PYZ(a.pure) +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.datas, + [], + name="cmshopee-updater", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + console=False, +) diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 1657f44..5345b0f 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -500,6 +500,7 @@ cmshopee/ │ ├── editor.py / ai.py / prompts.py / gui.py / workers.py │ ├── release_manifest.py # 发布包文件清单、zip哈希与服务端元数据模板 │ ├── update_installer.py # 自动升级安全下载、解压、manifest校验与同盘暂存 +│ ├── updater_entry.py # 独立更新器入口、事务根项目切换、journal与回滚 ├── main.py # GUI 启动入口：from app.gui import main ├── shopee待处理任务模板.xlsx # 标准空 Excel 模板，可提交；业务填写后的副本不提交 ├── data/ # 用户本地数据根（整体 gitignore；打包更新时保留） @@ -529,6 +530,7 @@ cmshopee/ - 发布包只能包含程序根项目，`data/` 和 `.cmshopee-update/` 永远在替换边界之外。第一阶段预留签名字段，但 SHA-256 只负责传输完整性，不等同于发布者身份认证。 - T-615 只提供可验证发布契约；启动门禁仍保持 T-544 的人工下载行为，直到后续下载、独立更新器、事务替换和失败熔断任务全部接入。 - T-616 的下载暂存根固定为安装目录下 `.cmshopee-update/`，与 `data/` 完全隔离。远程zip必须经过HTTPS/受信任域名、声明大小、整包SHA-256、安全zip路径和包内manifest逐文件校验，才写 `pending.json`；此阶段不替换任何运行中程序文件。 +- T-617 的独立 `cmshopee-updater.exe` 必须先复制到系统临时目录运行，并等待主程序退出。替换粒度是manifest允许的程序根项目，旧根先整体移动到同盘backup，新根再整体移入；事务锁防止并发更新，journal记录每次移动，任一步失败逆序恢复。`data/`、更新管理目录与未知安装根项目永不进入替换清单。 - CDP 交互事实变化同步第七节。 - 正式代码只放 `app/` 包；根目录只保留 `main.py`、配置/数据目录、文档和原型目录，不新增正式业务模块。 diff --git a/docs/packaging.md b/docs/packaging.md index 5de29a7..ee4ad35 100644 --- a/docs/packaging.md +++ b/docs/packaging.md @@ -178,6 +178,8 @@ T-615 不改变客户端行为：T-544 仍只打开浏览器下载。客户端 T-616 提供无Qt依赖的安全下载暂存层 `app/update_installer.py`。它只接受受信任域名的HTTPS，下载到 `<安装目录>/.cmshopee-update/downloads/`，完成大小与zip SHA-256校验后安全解压到同盘 `staging/`；路径穿越、符号链接、大小写重复、Windows保留名、ADS、异常压缩比、文件数/解压总量超限，以及manifest缺失或逐文件hash不一致都会拒绝。完整验证后才原子写 `pending.json`，当前程序目录和 `data/` 不变。 +T-617 增加独立 `cmshopee-updater.exe`。构建脚本用 `cmshopee-updater.spec` 生成无控制台单文件更新器并放进release和manifest。执行更新前，主程序把它复制到系统临时目录；更新器有上限地等待主程序退出，再按manifest白名单将旧程序根项目整体移动到 `.cmshopee-update/backup/`，把已验证暂存根项目移入安装目录。每一步写事务journal，移动或新版启动失败时逆序恢复；`data/` 与安装根未知文件不扫描、不移动、不删除。 + ## 四、绝不打包的本地数据 发布包里不能包含以下本地数据、密钥、业务数据或登录态： diff --git a/docs/tasks/T-617.md b/docs/tasks/T-617.md index 9821add..9c22c71 100644 --- a/docs/tasks/T-617.md +++ b/docs/tasks/T-617.md @@ -3,7 +3,7 @@ id: T-617 title: 独立更新器事务替换与失败恢复 phase: 8 deps: [T-615] -status: TODO +status: DONE created: 2026-07-13 --- @@ -45,4 +45,6 @@ Windows正在运行的 `cmshopee.exe` 和PyInstaller `_internal/` 可能被锁 ## 执行记录 -（完成后记录实现、验证命令与结果。） +- 2026-07-13：新增标准库独立更新器入口、结构化plan、父进程退出等待、安装目录事务锁、根项目整体切换、逐步journal、故障逆序回滚及临时目录隐藏启动能力。 +- 2026-07-13：新增无控制台单文件 `cmshopee-updater.spec`，构建脚本将更新器纳入release与manifest；Windows 10 / Python 3.10 / PyInstaller 6.11.1实际构建成功，产物约6.7MB且使用windowed bootloader。 +- 2026-07-13：干净worktree验证通过：ruff、compileall、完整unittest（408项）和 `git diff --check`。尚未在正式release上人工执行“锁住运行中主程序后自动退出、替换、重启”的端到端验收，该项留给T-618/T-619集成后执行。 diff --git a/docs/update-check.md b/docs/update-check.md index 28891af..850d93b 100644 --- a/docs/update-check.md +++ b/docs/update-check.md @@ -49,6 +49,8 @@ T-616 已实现但尚未接入GUI的安全暂存层：仅接受受信任域名的 HTTPS 地址，流式下载到安装目录 `.cmshopee-update/`，校验zip大小和SHA-256，安全解压后再按包内manifest逐文件校验。任一步失败都不修改当前程序或 `data/`；GUI接入由T-618完成。 +T-617 已提供独立无控制台更新器和事务回滚能力：更新器从系统临时目录运行，旧主程序退出后才切换程序根项目，并在新版进程无法创建时恢复旧版。启动检查弹窗尚未调用该流程，直到T-618完成GUI编排。 + ## 五、发版约定（服务端据此控制） 自动安装使用的全部字段必须放在同一个 `release` 对象内，不得把版本取自一个对象、hash 取自另一个对象。构建脚本生成的 `release/release-metadata.json` 是服务端录入模板；发布人员只补 HTTPS `download_url`、强制策略和中文发布说明，不得手工改写 hash、大小、包格式或协议版本。 diff --git a/scripts/build_exe.ps1 b/scripts/build_exe.ps1 index a8dcdfb..3fc2d99 100644 --- a/scripts/build_exe.ps1 +++ b/scripts/build_exe.ps1 @@ -120,6 +120,18 @@ if ($LASTEXITCODE -ne 0) { throw "$(Decode-Utf8Base64 "UHlJbnN0YWxsZXIg5omT5YyF5aSx6LSl77yM6YCA5Ye656CB"): $LASTEXITCODE" } +$updaterSpec = Join-Path $repoRoot "cmshopee-updater.spec" +$updaterDist = Join-Path $repoRoot "dist\updater" +$updaterWork = Join-Path $repoRoot "build\updater" +& $pythonExe @pythonArgs -m PyInstaller --noconfirm --clean --distpath $updaterDist --workpath $updaterWork $updaterSpec +if ($LASTEXITCODE -ne 0) { + throw "独立更新器打包失败，退出码: $LASTEXITCODE" +} +$updaterExe = Join-Path $updaterDist "cmshopee-updater.exe" +if (-not (Test-Path -LiteralPath $updaterExe)) { + throw "独立更新器产物不存在: $updaterExe" +} + $distDir = Join-Path $repoRoot "dist\cmshopee" $exePath = Join-Path $distDir "cmshopee.exe" if (-not (Test-Path -LiteralPath $exePath)) { @@ -164,6 +176,7 @@ foreach ($path in @($releaseDir, $portableZip, $releaseMetadata)) { New-Item -ItemType Directory -Force -Path $releaseDir | Out-Null Get-ChildItem -LiteralPath $distDir -Force | Copy-Item -Destination $releaseDir -Recurse -Force +Copy-Item -LiteralPath $updaterExe -Destination (Join-Path $releaseDir "cmshopee-updater.exe") -Force Set-Content -LiteralPath (Join-Path $releaseDir "version.txt") -Value $appVersion -Encoding ascii -NoNewline $readme = @( diff --git a/tests/test_packaging.py b/tests/test_packaging.py index f35416d..a1cfc82 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -96,6 +96,17 @@ class PackagingTests(unittest.TestCase): self.assertIn('-m app.release_manifest manifest', script) self.assertIn('-m app.release_manifest metadata', script) self.assertIn('release-metadata.json', script) + self.assertIn('cmshopee-updater.spec', script) + self.assertIn('cmshopee-updater.exe', script) + + def test_updater_spec_builds_a_windowed_standalone_executable(self): + spec = self.read_text("cmshopee-updater.spec") + normalized = "".join(spec.split()) + + self.assertIn('app/updater_entry.py', spec) + self.assertIn('name="cmshopee-updater"', normalized) + self.assertIn('console=False', normalized) + self.assertNotIn("COLLECT(", spec) def make_release_dir(self, root, app_version="1.2.3"): release_dir = Path(root) / "release" diff --git a/tests/test_updater_entry.py b/tests/test_updater_entry.py new file mode 100644 index 0000000..cd41c69 --- /dev/null +++ b/tests/test_updater_entry.py @@ -0,0 +1,152 @@ +import hashlib +import json +import os +import tempfile +import unittest +from pathlib import Path + +from app import release_manifest, updater_entry + + +class UpdaterEntryTests(unittest.TestCase): + def make_trees(self, root): + install = Path(root) / "install" + staging = install / ".cmshopee-update" / "staging" / "2.0.0-test" + install.mkdir() + (install / "_internal").mkdir() + (install / "cmshopee.exe").write_bytes(b"old-exe") + (install / "_internal" / "old.dll").write_bytes(b"old") + (install / "version.txt").write_text("1.0.0", encoding="ascii") + (install / "README.txt").write_text("旧说明", encoding="utf-8") + (install / "package-manifest.json").write_text("{}", encoding="utf-8") + (install / "cmshopee-updater.exe").write_bytes(b"old-updater") + (install / "data").mkdir() + (install / "data" / "cmshopee.db").write_bytes(b"business-data") + (install / "operator-note.txt").write_text("保留", encoding="utf-8") + + (staging / "_internal").mkdir(parents=True) + (staging / "cmshopee.exe").write_bytes(b"new-exe") + (staging / "_internal" / "new.dll").write_bytes(b"new") + (staging / "version.txt").write_text("2.0.0", encoding="ascii") + (staging / "README.txt").write_text("新说明", encoding="utf-8") + (staging / "cmshopee-updater.exe").write_bytes(b"new-updater") + release_manifest.write_package_manifest(staging, "2.0.0") + plan = updater_entry.UpdatePlan( + parent_pid=12345, + install_root=install.resolve(), + staging_root=staging.resolve(), + target_version="2.0.0", + transaction_id="transaction-1234", + log_path=(install / ".cmshopee-update/logs/update.log").resolve(), + ) + return install, staging, plan + + def test_transaction_replaces_roots_and_preserves_data_and_unknown_files(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + before_hash = hashlib.sha256((install / "data/cmshopee.db").read_bytes()).hexdigest() + launched = [] + + backup = updater_entry.apply_update( + plan, + wait_parent=lambda _pid: None, + launcher=lambda executable, args: launched.append((executable, args)), + ) + + self.assertEqual(b"new-exe", (install / "cmshopee.exe").read_bytes()) + self.assertTrue((install / "_internal/new.dll").is_file()) + self.assertFalse((install / "_internal/old.dll").exists()) + self.assertEqual("保留", (install / "operator-note.txt").read_text(encoding="utf-8")) + self.assertEqual( + before_hash, + hashlib.sha256((install / "data/cmshopee.db").read_bytes()).hexdigest(), + ) + self.assertTrue((backup / "cmshopee.exe").is_file()) + self.assertEqual(1, len(launched)) + + def test_move_failure_rolls_back_old_program(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + calls = {"count": 0} + + def failing_move(source, destination): + calls["count"] += 1 + if calls["count"] == 8: + raise OSError("injected move failure") + return os.replace(source, destination) + + with self.assertRaisesRegex(updater_entry.UpdaterError, "已恢复旧版"): + updater_entry.apply_update( + plan, + wait_parent=lambda _pid: None, + move=failing_move, + launcher=lambda *_args: None, + ) + self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes()) + self.assertTrue((install / "_internal/old.dll").is_file()) + self.assertEqual(b"business-data", (install / "data/cmshopee.db").read_bytes()) + + def test_parent_timeout_keeps_install_unchanged(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + + def timeout(_pid): + raise updater_entry.UpdaterError("等待旧版程序退出超时") + + with self.assertRaisesRegex(updater_entry.UpdaterError, "退出超时"): + updater_entry.apply_update(plan, wait_parent=timeout) + self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes()) + + def test_new_process_launch_failure_rolls_back(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + + def fail_launch(*_args): + raise OSError("injected launch failure") + + with self.assertRaisesRegex(updater_entry.UpdaterError, "已恢复旧版"): + updater_entry.apply_update( + plan, + wait_parent=lambda _pid: None, + launcher=fail_launch, + ) + self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes()) + self.assertTrue((install / "_internal/old.dll").is_file()) + + def test_lock_blocks_concurrent_transaction(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + lock_path = install / ".cmshopee-update/update.lock" + with updater_entry.TransactionLock(lock_path): + with self.assertRaisesRegex(updater_entry.UpdaterError, "已有更新程序"): + updater_entry.apply_update(plan, wait_parent=lambda _pid: None) + + def test_plan_path_and_staging_must_stay_inside_update_root(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, _staging, plan = self.make_trees(temp_dir) + escaped = updater_entry.UpdatePlan( + parent_pid=plan.parent_pid, + install_root=install, + staging_root=Path(temp_dir) / "outside", + target_version=plan.target_version, + transaction_id=plan.transaction_id, + log_path=plan.log_path, + ) + escaped.staging_root.mkdir() + with self.assertRaisesRegex(updater_entry.UpdaterError, "暂存目录越界"): + updater_entry.validate_plan(escaped) + + def test_create_and_load_plan_round_trip(self): + with tempfile.TemporaryDirectory() as temp_dir: + install, staging, _plan = self.make_trees(temp_dir) + plan_path = updater_entry.create_plan(install, staging, "2.0.0", 4321) + loaded = updater_entry.load_plan(plan_path) + payload = json.loads(plan_path.read_text(encoding="utf-8")) + + self.assertEqual("2.0.0", loaded.target_version) + self.assertEqual(4321, payload["parent_pid"]) + self.assertNotIn("data", payload) + + +if __name__ == "__main__": + unittest.main()