225 lines
9.8 KiB
Python
225 lines
9.8 KiB
Python
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 / "_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")
|
|
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(),
|
|
package_sha256="b" * 64,
|
|
)
|
|
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)),
|
|
health_waiter=lambda _plan, _process: "main_window_ready",
|
|
)
|
|
|
|
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.assertFalse(backup.exists())
|
|
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_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)
|
|
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,
|
|
"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()
|