chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(update): add startup health rollback fuse

Browse Source
This commit is contained in:
chengma
2026-07-13 12:23:46 +08:00
parent 04124060fc
commit f6db4f0d62
16 changed files with 481 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_gui.py
+34
View File
@@ -1646,6 +1646,7 @@ class GuiTests(TempDirMixin, unittest.TestCase):
staged = SimpleNamespace(
version="1.2.0",
staging_dir=os.path.abspath(staging_root),
sha256="a" * 64,
)
def prepare(_metadata, _install_root, **callbacks):
@@ -1738,6 +1739,39 @@ class GuiTests(TempDirMixin, unittest.TestCase):
self.assertEqual("重试", dialog.action_button.text())
self.assertIsNone(dialog.thread)
def test_startup_update_gate_fuses_a_previously_failed_release(self):
captured = []
class FakeDialog:
def __init__(self, result, parent=None):
captured.append(result)
def exec(self):
return 0
result = update_check.UpdateCheckResult(
current_version="1.0.0",
checked=True,
forced=True,
latest_version="1.2.0",
download_url="https://cm.833729.com/cmshopee.zip",
sha256="a" * 64,
size_bytes=100,
package_format="cmshopee-portable-v1",
updater_protocol=1,
)
with mock.patch(
"app.gui.update_health.get_failed_release",
return_value={"reason": "新版未就绪"},
):
allowed = gui._run_startup_update_gate(
checker=lambda: result,
dialog_factory=FakeDialog,
)
self.assertFalse(allowed)
self.assertIn("已停止重复安装", captured[0].automatic_update_error)
def test_startup_update_gate_check_failure_allows_entry_and_logs(self):
result = update_check.UpdateCheckResult(
current_version="1.0.0",
tests/test_update_health.py
+90
View File
@@ -0,0 +1,90 @@
import json
import tempfile
import unittest
from pathlib import Path
from app import update_health
class UpdateHealthTests(unittest.TestCase):
def test_context_and_atomic_health_markers(self):
with tempfile.TemporaryDirectory() as temp_dir:
context = update_health.context_from_argv(
[
"cmshopee.exe",
"--update-transaction",
"transaction-1234",
"--update-target-version",
"2.0.0",
],
temp_dir,
"2.0.0",
)
self.assertIsNotNone(context)
update_health.write_health(context, "process_started")
update_health.write_health(context, "main_window_ready")
payload = json.loads(context.health_path.read_text(encoding="utf-8"))
self.assertEqual("main_window_ready", payload["status"])
self.assertEqual("2.0.0", payload["target_version"])
def test_context_rejects_wrong_version_or_transaction(self):
self.assertIsNone(
update_health.context_from_argv(
["app", "--update-transaction", "bad", "--update-target-version", "2.0.0"],
".",
"2.0.0",
)
)
self.assertIsNone(
update_health.context_from_argv(
[
"app",
"--update-transaction",
"transaction-1234",
"--update-target-version",
"9.0.0",
],
".",
"2.0.0",
)
)
def test_failed_release_fuse_is_version_and_hash_specific(self):
with tempfile.TemporaryDirectory() as temp_dir:
first = update_health.record_failed_release(
temp_dir,
"2.0.0",
"a" * 64,
"新版程序未就绪",
)
second = update_health.record_failed_release(
temp_dir,
"2.0.0",
"a" * 64,
"再次失败",
)
self.assertEqual(1, first["attempts"])
self.assertEqual(2, second["attempts"])
self.assertIsNotNone(
update_health.get_failed_release(temp_dir, "2.0.0", "a" * 64)
)
self.assertIsNone(
update_health.get_failed_release(temp_dir, "2.0.0", "b" * 64)
)
self.assertIsNone(
update_health.get_failed_release(temp_dir, "2.0.1", "a" * 64)
)
def test_failed_release_file_does_not_touch_data(self):
with tempfile.TemporaryDirectory() as temp_dir:
data_file = Path(temp_dir) / "data" / "cmshopee.db"
data_file.parent.mkdir()
data_file.write_bytes(b"business")
update_health.record_failed_release(temp_dir, "2.0.0", "a" * 64, "失败")
self.assertEqual(b"business", data_file.read_bytes())
if __name__ == "__main__":
unittest.main()
tests/test_update_installer.py
+1
View File
@@ -44,6 +44,7 @@ class UpdateInstallerTests(unittest.TestCase):
(package / "_internal").mkdir(parents=True)
(package / "cmshopee.exe").write_bytes(b"new-exe")
(package / "_internal" / "runtime.dll").write_bytes(b"runtime")
(package / "_internal" / "empty.marker").write_bytes(b"")
(package / "version.txt").write_text(version, encoding="ascii")
(package / "README.txt").write_text("说明", encoding="utf-8")
release_manifest.write_package_manifest(package, version)
tests/test_updater_entry.py
+74 -2
View File
@@ -27,6 +27,7 @@ class UpdaterEntryTests(unittest.TestCase):
(staging / "_internal").mkdir(parents=True)
(staging / "cmshopee.exe").write_bytes(b"new-exe")
(staging / "_internal" / "new.dll").write_bytes(b"new")
(staging / "_internal" / "empty.marker").write_bytes(b"")
(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")
@@ -38,6 +39,7 @@ class UpdaterEntryTests(unittest.TestCase):
target_version="2.0.0",
transaction_id="transaction-1234",
log_path=(install / ".cmshopee-update/logs/update.log").resolve(),
package_sha256="b" * 64,
)
return install, staging, plan
@@ -51,6 +53,7 @@ class UpdaterEntryTests(unittest.TestCase):
plan,
wait_parent=lambda _pid: None,
launcher=lambda executable, args: launched.append((executable, args)),
health_waiter=lambda _plan, _process: "main_window_ready",
)
self.assertEqual(b"new-exe", (install / "cmshopee.exe").read_bytes())
@@ -61,7 +64,7 @@ class UpdaterEntryTests(unittest.TestCase):
before_hash,
hashlib.sha256((install / "data/cmshopee.db").read_bytes()).hexdigest(),
)
self.assertTrue((backup / "cmshopee.exe").is_file())
self.assertFalse(backup.exists())
self.assertEqual(1, len(launched))
def test_move_failure_rolls_back_old_program(self):
@@ -113,6 +116,41 @@ class UpdaterEntryTests(unittest.TestCase):
self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes())
self.assertTrue((install / "_internal/old.dll").is_file())
def test_health_failure_rolls_back_and_records_failed_release(self):
with tempfile.TemporaryDirectory() as temp_dir:
install, _staging, plan = self.make_trees(temp_dir)
def fail_health(_plan, _process):
raise updater_entry.UpdaterError("新版程序在主窗口就绪前退出")
with self.assertRaisesRegex(updater_entry.UpdaterError, "已恢复旧版"):
updater_entry.apply_update(
plan,
wait_parent=lambda _pid: None,
launcher=lambda *_args: object(),
health_waiter=fail_health,
)
self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes())
failed = json.loads(
(install / ".cmshopee-update/failed-versions.json").read_text(encoding="utf-8")
)
self.assertIn("2.0.0:%s" % ("b" * 64), failed["releases"])
def test_environment_block_keeps_new_program_and_backup(self):
with tempfile.TemporaryDirectory() as temp_dir:
install, _staging, plan = self.make_trees(temp_dir)
backup = updater_entry.apply_update(
plan,
wait_parent=lambda _pid: None,
launcher=lambda *_args: object(),
health_waiter=lambda _plan, _process: "environment_blocked",
)
self.assertEqual(b"new-exe", (install / "cmshopee.exe").read_bytes())
self.assertTrue((backup / "cmshopee.exe").is_file())
self.assertFalse((install / ".cmshopee-update/pending.json").exists())
def test_lock_blocks_concurrent_transaction(self):
with tempfile.TemporaryDirectory() as temp_dir:
install, _staging, plan = self.make_trees(temp_dir)
@@ -139,14 +177,48 @@ class UpdaterEntryTests(unittest.TestCase):
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)
plan_path = updater_entry.create_plan(
install,
staging,
"2.0.0",
4321,
"c" * 64,
)
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.assertEqual("c" * 64, loaded.package_sha256)
self.assertNotIn("data", payload)
def test_health_waiter_accepts_ready_and_environment_markers(self):
with tempfile.TemporaryDirectory() as temp_dir:
install, _staging, plan = self.make_trees(temp_dir)
health_path = (
install
/ ".cmshopee-update"
/ "transactions"
/ plan.transaction_id
/ "health.json"
)
health_path.parent.mkdir(parents=True)
for status in ("main_window_ready", "environment_blocked"):
health_path.write_text(
json.dumps(
{
"transaction_id": plan.transaction_id,
"target_version": plan.target_version,
"status": status,
}
),
encoding="utf-8",
)
self.assertEqual(
status,
updater_entry.wait_for_health(plan, process=None, timeout=0.1),
)
if __name__ == "__main__":
unittest.main()