feat: 实现 Client 在线更新与安全回退 (#93)

client/test/test_packaging.py

@@ -2,12 +2,20 @@
 
 import hashlib
 import json
+import os
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 
-from build_tools.launcher import app_executable, install_root
+from build_tools.launcher import (
+    UpdateApplyError,
+    app_executable,
+    apply_pending_update,
+    finalize_applied_update,
+    install_root,
+    is_main_program_running,
+)
 from build_tools.release_manifest import MANIFEST_FILE_NAME, write_manifest
 from src.db import data_dir
 
@@ -27,6 +35,96 @@ class LauncherPathTest(unittest.TestCase):
         )
 
 
+class LauncherUpdateTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary_directory = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary_directory.name)
+        self.current_app = self.root / "app"
+        self.staged_app = self.root / "data" / "update" / "app.new"
+        self.current_app.mkdir(parents=True)
+        self.staged_app.mkdir(parents=True)
+        (self.current_app / "CMAutoBuy.exe").write_bytes(b"old")
+        (self.staged_app / "CMAutoBuy.exe").write_bytes(b"new")
+        self.pending_path = self.root / "data" / "update" / "pending.json"
+        self.pending_path.write_text(
+            json.dumps(
+                {
+                    "schema_version": 1,
+                    "state": "ready",
+                    "version": "0.2.0",
+                }
+            ),
+            encoding="utf-8",
+        )
+
+    def tearDown(self):
+        self.temporary_directory.cleanup()
+
+    def test_apply_keeps_old_app_until_health_is_confirmed(self):
+        version = apply_pending_update(self.root)
+
+        self.assertEqual(version, "0.2.0")
+        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"new")
+        self.assertEqual((self.root / "app.old" / "CMAutoBuy.exe").read_bytes(), b"old")
+        pending = json.loads(self.pending_path.read_text(encoding="utf-8"))
+        self.assertEqual(pending["state"], "applied")
+
+        healthy_path = self.root / "data" / "update" / "healthy.json"
+        healthy_path.write_text('{"version":"0.2.0"}', encoding="utf-8")
+        finalize_applied_update(self.root)
+        self.assertFalse(self.pending_path.exists())
+        self.assertTrue((self.root / "app.old").exists())
+
+    def test_applied_without_health_rolls_back_on_next_launch(self):
+        apply_pending_update(self.root)
+
+        result = apply_pending_update(self.root)
+
+        self.assertIsNone(result)
+        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"old")
+        self.assertFalse(self.pending_path.exists())
+        self.assertTrue((self.root / "data" / "update" / "app.failed").exists())
+
+    def test_move_failure_restores_current_app(self):
+        real_replace = os.replace
+
+        def fail_for_staged_app(source, destination):
+            if Path(source).name == "app.new":
+                raise OSError("simulated lock")
+            return real_replace(source, destination)
+
+        with patch("build_tools.launcher.os.replace", side_effect=fail_for_staged_app):
+            with self.assertRaises(UpdateApplyError):
+                apply_pending_update(self.root)
+
+        self.assertEqual((self.root / "app" / "CMAutoBuy.exe").read_bytes(), b"old")
+        self.assertEqual((self.staged_app / "CMAutoBuy.exe").read_bytes(), b"new")
+
+    def test_running_pid_must_belong_to_expected_executable(self):
+        running_path = self.root / "data" / "update" / "running.json"
+        running_path.write_text('{"pid":123}', encoding="utf-8")
+
+        self.assertTrue(
+            is_main_program_running(
+                self.root,
+                checker=lambda pid, executable: pid == 123
+                and executable == self.root / "app" / "CMAutoBuy.exe",
+            )
+        )
+        self.assertFalse(
+            is_main_program_running(self.root, checker=lambda _pid, _path: False)
+        )
+        self.assertFalse(running_path.exists())
+
+    def test_corrupt_pending_state_stops_update_instead_of_being_ignored(self):
+        self.pending_path.write_text("not-json", encoding="utf-8")
+
+        with self.assertRaises(UpdateApplyError):
+            apply_pending_update(self.root)
+
+        self.assertEqual((self.current_app / "CMAutoBuy.exe").read_bytes(), b"old")
+
+
 class PackagedDataPathTest(unittest.TestCase):
     def test_packaged_main_program_uses_root_data_directory(self):
         with tempfile.TemporaryDirectory() as directory: