新增

wechat_rpa/test_backend.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import json
+import hashlib
 import socket
 import tempfile
 import threading
@@ -11,10 +12,114 @@ import unittest
 from pathlib import Path
 
 import admin_backend
+import app_version
 import backend_client
 
 
 class BackendIntegrationTest(unittest.TestCase):
+    def test_release_status_detects_optional_and_forced_updates(self) -> None:
+        optional = app_version.release_status(
+            {"latest_version": "1.0.1", "force_upgrade": False}
+        )
+        self.assertTrue(optional["update_available"])
+        self.assertFalse(optional["force_upgrade"])
+
+        forced = app_version.release_status(
+            {"latest_version": "1.0.1", "force_upgrade": True}
+        )
+        self.assertTrue(forced["force_upgrade"])
+
+        current = app_version.release_status(
+            {"latest_version": app_version.APP_VERSION, "force_upgrade": True}
+        )
+        self.assertFalse(current["update_available"])
+        self.assertFalse(current["force_upgrade"])
+
+    def test_release_form_validation(self) -> None:
+        release = admin_backend.validate_release_form(
+            {
+                "latest_version": "v1.2.3",
+                "download_url": "https://example.com/client.exe",
+                "release_notes": "修复已知问题",
+                "force_upgrade": "1",
+            }
+        )
+        self.assertEqual(release["latest_version"], "1.2.3")
+        self.assertTrue(release["force_upgrade"])
+        with self.assertRaises(ValueError):
+            admin_backend.validate_release_form(
+                {"latest_version": "1.2", "force_upgrade": "1"}
+            )
+        with self.assertRaises(ValueError):
+            admin_backend.validate_release_form(
+                {"latest_version": "1.2.3", "force_upgrade": "1"}
+            )
+
+    def test_desktop_sync_key_matches_server(self) -> None:
+        self.assertEqual(
+            backend_client.DESKTOP_SYNC_KEY,
+            admin_backend.DEFAULT_DESKTOP_SYNC_KEY,
+        )
+
+    def test_pbkdf2_fallback_matches_standard_library(self) -> None:
+        password = b"FallbackPassword123"
+        salt = b"0123456789abcdef"
+        expected = hashlib.pbkdf2_hmac("sha256", password, salt, 1_000)
+        native_pbkdf2 = admin_backend.hashlib.pbkdf2_hmac
+        try:
+            admin_backend.hashlib.pbkdf2_hmac = None
+            actual = admin_backend.pbkdf2_sha256(password, salt, 1_000)
+        finally:
+            admin_backend.hashlib.pbkdf2_hmac = native_pbkdf2
+        self.assertEqual(actual, expected)
+
+    def test_stale_runtime_file_is_ignored(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            runtime_file = root / "backend_runtime.json"
+            runtime_file.write_text(
+                json.dumps(
+                    {
+                        "pid": 0,
+                        "host": "127.0.0.1",
+                        "port": 8765,
+                        "server_url": "http://127.0.0.1:8765",
+                        "local_sync_token": "stale-token",
+                    }
+                ),
+                encoding="utf-8",
+            )
+            old_runtime_file = backend_client.RUNTIME_FILE
+            try:
+                backend_client.RUNTIME_FILE = runtime_file
+                self.assertEqual(backend_client.discover_local_runtime(), {})
+                self.assertEqual(backend_client.discover_local_sync_token(), "")
+            finally:
+                backend_client.RUNTIME_FILE = old_runtime_file
+
+    def test_local_agent_name_overrides_cloud_and_blank_follows_cloud(self) -> None:
+        import ai_config
+
+        old_agent = ai_config.AI_AGENT_NAME
+        old_cloud = ai_config.AI_CLOUD_AGENT_NAME
+        old_local = ai_config.AI_LOCAL_AGENT_NAME
+        old_prompt = ai_config.AI_SYSTEM_PROMPT
+        try:
+            ai_config.AI_CLOUD_AGENT_NAME = "旧云端客服"
+            ai_config.AI_LOCAL_AGENT_NAME = ""
+            ai_config.AI_AGENT_NAME = "旧云端客服"
+            ai_config.set_local_agent_name_override("本机客服", persist=False)
+            ai_config.apply_settings({"AI_AGENT_NAME": "新云端客服"}, persist=False)
+            self.assertEqual(ai_config.AI_CLOUD_AGENT_NAME, "新云端客服")
+            self.assertEqual(ai_config.AI_AGENT_NAME, "本机客服")
+            ai_config.set_local_agent_name_override("", persist=False)
+            self.assertEqual(ai_config.AI_AGENT_NAME, "新云端客服")
+        finally:
+            ai_config.AI_AGENT_NAME = old_agent
+            ai_config.AI_CLOUD_AGENT_NAME = old_cloud
+            ai_config.AI_LOCAL_AGENT_NAME = old_local
+            ai_config.AI_SYSTEM_PROMPT = old_prompt
+
     def test_occupied_port_automatically_uses_next_port(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
@@ -28,6 +133,7 @@ class BackendIntegrationTest(unittest.TestCase):
             server = None
             old_runtime_file = backend_client.RUNTIME_FILE
             old_connection_file = backend_client.CONNECTION_FILE
+            old_default_server_url = backend_client.DEFAULT_SERVER_URL
             server_thread = None
             try:
                 server, actual_port = admin_backend.create_server(
@@ -62,6 +168,7 @@ class BackendIntegrationTest(unittest.TestCase):
                     target=server.serve_forever, daemon=True
                 )
                 server_thread.start()
+                backend_client.DEFAULT_SERVER_URL = f"http://127.0.0.1:{actual_port}"
                 import ai_config
 
                 old_settings_file = ai_config._SETTINGS_FILE
@@ -75,6 +182,7 @@ class BackendIntegrationTest(unittest.TestCase):
             finally:
                 backend_client.RUNTIME_FILE = old_runtime_file
                 backend_client.CONNECTION_FILE = old_connection_file
+                backend_client.DEFAULT_SERVER_URL = old_default_server_url
                 if server is not None:
                     if server_thread is not None:
                         server.shutdown()
@@ -104,6 +212,17 @@ class BackendIntegrationTest(unittest.TestCase):
             config["AI_MODEL"] = "integration-test-model"
             version = database.save_config(config, admin["id"], "127.0.0.1")
             self.assertEqual(version, 2)
+            database.save_release(
+                "1.1.0",
+                "https://example.com/client.exe",
+                "测试升级",
+                True,
+                admin["id"],
+                "127.0.0.1",
+            )
+            release = database.release()
+            self.assertEqual(release["latest_version"], "1.1.0")
+            self.assertEqual(release["force_upgrade"], 1)
 
             server = admin_backend.AdminServer(("127.0.0.1", 0), database)
             thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -127,10 +246,16 @@ class BackendIntegrationTest(unittest.TestCase):
                     result = backend_client.sync_config(force=True)
                     self.assertTrue(result["synced"])
                     self.assertEqual(result["version"], 2)
+                    self.assertTrue(result["update_available"])
+                    self.assertTrue(result["force_upgrade"])
+                    self.assertEqual(result["release"]["latest_version"], "1.1.0")
                     synced = json.loads(
                         Path(ai_config._SETTINGS_FILE).read_text(encoding="utf-8")
                     )
                     self.assertEqual(synced["AI_MODEL"], "integration-test-model")
+                    self.assertTrue(
+                        backend_client.cached_release_status()["force_upgrade"]
+                    )
                 finally:
                     ai_config._SETTINGS_FILE = old_settings_file
                 backend_client.logout()