# -*- coding: utf-8 -*- """配置后台与桌面端同步的本地闭环测试。""" from __future__ import annotations import json import socket import tempfile import threading import unittest from pathlib import Path import admin_backend import ai_config import backend_client class BackendIntegrationTest(unittest.TestCase): def test_ai_config_ignores_legacy_chat_keys_and_saves_grok_agent_settings( self, ) -> None: original = ai_config.export_settings() first = { "GROK_CUSTOMER_SERVICE_ENABLED": True, "GROK_CUSTOMER_SERVICE_TIMEOUT": 181, "GROK_CUSTOMER_SERVICE_MAX_TURNS": 9, "GROK_CUSTOMER_SERVICE_EFFORT": "medium", "CHAT_API_BASE": "http://legacy.invalid/api", "CHAT_API_ACCOUNT": "legacy-account", "CHAT_API_PASSWORD": "legacy-secret", } second = { "GROK_CUSTOMER_SERVICE_TIMEOUT": 182, } old_settings_file = ai_config._SETTINGS_FILE try: with tempfile.TemporaryDirectory() as directory: settings_file = Path(directory) / "settings.json" ai_config._SETTINGS_FILE = str(settings_file) applied = ai_config.apply_settings(first, persist=True) first_revision = ai_config.get_settings_revision() exported = ai_config.export_settings() persisted = json.loads(settings_file.read_text(encoding="utf-8")) self.assertEqual(181, exported["GROK_CUSTOMER_SERVICE_TIMEOUT"]) self.assertEqual(9, exported["GROK_CUSTOMER_SERVICE_MAX_TURNS"]) self.assertEqual("medium", exported["GROK_CUSTOMER_SERVICE_EFFORT"]) self.assertFalse(any(key.startswith("CHAT_API_") for key in applied)) self.assertFalse(any(key.startswith("CHAT_API_") for key in exported)) self.assertFalse(any(key.startswith("CHAT_API_") for key in persisted)) ai_config.apply_settings(first, persist=False) self.assertEqual( first_revision, ai_config.get_settings_revision(), "re-applying the same effective values must not create a revision", ) ai_config._SETTINGS_FILE = str( Path(directory) / "missing-parent" / "settings.json" ) with self.assertRaises(OSError): ai_config.apply_settings(second, persist=True) self.assertEqual(first_revision, ai_config.get_settings_revision()) self.assertEqual( 181, ai_config.export_settings()["GROK_CUSTOMER_SERVICE_TIMEOUT"], "failed persistence must not publish partial runtime values", ) finally: ai_config._SETTINGS_FILE = old_settings_file ai_config.apply_settings(original, persist=False) def test_admin_html_only_exposes_local_grok_customer_agent_controls( self, ) -> None: config = admin_backend.load_initial_config() config["CHAT_API_PASSWORD"] = "must-never-appear-in-html" rendered = admin_backend.AdminHandler.config_card( {"role": "admin"}, "csrf-token", { "version": 1, "updated_at": "2026-07-23T10:00:00+08:00", "updated_by_name": "admin", }, config, ) self.assertIn("GROK_CUSTOMER_SERVICE_TIMEOUT", rendered) self.assertIn("GROK_CUSTOMER_SERVICE_MAX_TURNS", rendered) self.assertIn("GROK_CUSTOMER_SERVICE_EFFORT", rendered) self.assertNotIn("CHAT_API_", rendered) self.assertNotIn("must-never-appear-in-html", rendered) def test_backend_database_migration_drops_legacy_chat_keys(self) -> None: with tempfile.TemporaryDirectory() as directory: database = admin_backend.Database(Path(directory) / "test.db") database.initialize("InitialAdmin123") row = database.config() legacy = json.loads(row["config_json"]) legacy["CHAT_API_BASE"] = "http://legacy.invalid/api" legacy["CHAT_API_PASSWORD"] = "legacy-secret" legacy.pop("GROK_CUSTOMER_SERVICE_TIMEOUT") with database.connect() as connection: connection.execute( "UPDATE model_config SET config_json=? WHERE id=1", (json.dumps(legacy, ensure_ascii=False),), ) connection.commit() database.initialize("InitialAdmin123") migrated_row = database.config() migrated = json.loads(migrated_row["config_json"]) self.assertFalse(any(key.startswith("CHAT_API_") for key in migrated)) self.assertEqual(180, migrated["GROK_CUSTOMER_SERVICE_TIMEOUT"]) self.assertGreater(migrated_row["version"], row["version"]) def test_customer_service_pages_do_not_embed_a_browser(self) -> None: for filename in ("wechat_gui_qt.py", "wechat_gui.py"): source = (Path(__file__).resolve().parent / filename).read_text( encoding="utf-8" ) with self.subTest(filename=filename): if filename == "wechat_gui_qt.py": self.assertIn("build_headless_args", source) self.assertIn("custom_model_only=True", source) self.assertIn("new_session_id=session_id", source) self.assertIn("resume_session=", source) else: self.assertIn("customer_agent_status", source) self.assertNotIn("CUSTOMER_SERVICE_URL", source) self.assertNotIn("QWebEngine", source) self.assertNotIn("--app=", source) def test_grok_coding_model_form_is_validated_and_published(self) -> None: current = admin_backend.load_initial_config() current["GROK_API_KEY"] = "existing-coding-secret" current["CHAT_API_BASE"] = "http://legacy.invalid/api" current["CHAT_API_PASSWORD"] = "legacy-secret" form = { "AI_ENABLED": "1", "AI_CONTEXT_ENABLED": "1", "AI_API_BASE": "https://customer.example.test/v1", "AI_MODEL": "customer-model", "AI_AGENT_NAME": "客服", "AI_HOSPITAL_NAME": "测试医院", "AI_CONTEXT_MAX_ROUNDS": "5", "AI_MAX_TOKENS": "500", "AI_TEMPERATURE": "0.35", "AI_TIMEOUT": "120", "GROK_CUSTOMER_SERVICE_ENABLED": "1", "GROK_CUSTOMER_SERVICE_TIMEOUT": "180", "GROK_CUSTOMER_SERVICE_MAX_TURNS": "8", "GROK_CUSTOMER_SERVICE_EFFORT": "low", "AI_MCP_MAX_ROUNDS": "5", "AI_MCP_SERVERS": "[]", "GROK_MODEL_ENABLED": "1", "GROK_API_BASE": "https://coding.example.test/v1", "GROK_MODEL": "qwen-coder", "GROK_API_BACKEND": "chat_completions", "GROK_AUTH_SCHEME": "bearer", "GROK_CONTEXT_WINDOW": "131072", "GROK_MAX_TOKENS": "8192", "GROK_TEMPERATURE": "0.2", } config = admin_backend.validate_config_form(form, current) self.assertTrue(config["GROK_MODEL_ENABLED"]) self.assertEqual("https://coding.example.test/v1", config["GROK_API_BASE"]) self.assertEqual("qwen-coder", config["GROK_MODEL"]) self.assertEqual("chat_completions", config["GROK_API_BACKEND"]) self.assertEqual("bearer", config["GROK_AUTH_SCHEME"]) self.assertEqual(131072, config["GROK_CONTEXT_WINDOW"]) self.assertEqual(8192, config["GROK_MAX_TOKENS"]) self.assertEqual(0.2, config["GROK_TEMPERATURE"]) self.assertEqual("existing-coding-secret", config["GROK_API_KEY"]) self.assertTrue(config["GROK_CUSTOMER_SERVICE_ENABLED"]) self.assertEqual(180, config["GROK_CUSTOMER_SERVICE_TIMEOUT"]) self.assertEqual(8, config["GROK_CUSTOMER_SERVICE_MAX_TURNS"]) self.assertEqual("low", config["GROK_CUSTOMER_SERVICE_EFFORT"]) self.assertFalse(any(key.startswith("CHAT_API_") for key in config)) for key, value in ( ("GROK_CUSTOMER_SERVICE_TIMEOUT", "29"), ("GROK_CUSTOMER_SERVICE_TIMEOUT", "601"), ("GROK_CUSTOMER_SERVICE_MAX_TURNS", "1"), ("GROK_CUSTOMER_SERVICE_MAX_TURNS", "31"), ): invalid_customer_setting = dict(form) invalid_customer_setting[key] = value with self.subTest(key=key, value=value): with self.assertRaisesRegex(ValueError, key): admin_backend.validate_config_form( invalid_customer_setting, current, ) invalid_effort = dict(form) invalid_effort["GROK_CUSTOMER_SERVICE_EFFORT"] = "minimal" with self.assertRaisesRegex(ValueError, "推理强度"): admin_backend.validate_config_form(invalid_effort, current) no_agent_model = dict(form) no_agent_model.pop("GROK_MODEL_ENABLED") with self.assertRaisesRegex(ValueError, "不会回退到 Grok/xAI"): admin_backend.validate_config_form(no_agent_model, current) invalid = dict(form) invalid["GROK_API_BASE"] = "https://dify.example.test/v1/chat-messages" with self.assertRaisesRegex(ValueError, "Dify"): admin_backend.validate_config_form(invalid, current) dify_form = dict(form) dify_form["GROK_API_BASE"] = ( "https://dify.example.test/v1/chat-messages" ) dify_form["GROK_API_BACKEND"] = "dify" dify_form["GROK_AUTH_SCHEME"] = "auto" dify_form["GROK_MODEL"] = "" dify_form["GROK_DIFY_INPUTS"] = '{"department":"糖尿病"}' dify_config = admin_backend.validate_config_form(dify_form, current) self.assertEqual("dify", dify_config["GROK_API_BACKEND"]) self.assertEqual("auto", dify_config["GROK_AUTH_SCHEME"]) self.assertEqual("dify-app", dify_config["GROK_MODEL"]) self.assertEqual( {"department": "糖尿病"}, dify_config["GROK_DIFY_INPUTS"], ) invalid_query = dict(form) invalid_query["GROK_API_BASE"] = ( "https://coding.example.test/v1/chat/completions?api-version=1" ) with self.assertRaisesRegex(ValueError, "query"): admin_backend.validate_config_form(invalid_query, current) xai_model = dict(form) xai_model["GROK_API_BASE"] = "https://api.x.ai/v1" with self.assertRaisesRegex(ValueError, "不能配置 xAI/Grok"): admin_backend.validate_config_form(xai_model, current) invalid_auth = dict(form) invalid_auth["GROK_AUTH_SCHEME"] = "basic" with self.assertRaisesRegex(ValueError, "认证方式"): admin_backend.validate_config_form(invalid_auth, current) missing_key_current = dict(current) missing_key_current["GROK_API_KEY"] = "" with self.assertRaisesRegex(ValueError, "独立 API Key"): admin_backend.validate_config_form(form, missing_key_current) def test_occupied_port_automatically_uses_next_port(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) database = admin_backend.Database(root / "test.db") database.initialize("InitialAdmin123") blocker = socket.socket(socket.AF_INET, socket.SOCK_STREAM) blocker.bind(("127.0.0.1", 0)) blocker.listen(1) occupied_port = int(blocker.getsockname()[1]) self.assertLess(occupied_port, 65535) server = None old_runtime_file = backend_client.RUNTIME_FILE old_connection_file = backend_client.CONNECTION_FILE server_thread = None try: server, actual_port = admin_backend.create_server( "127.0.0.1", occupied_port, database, max_attempts=5 ) self.assertGreater(actual_port, occupied_port) self.assertLessEqual(actual_port, occupied_port + 4) runtime_file = root / "backend_runtime.json" admin_backend.write_runtime_info( runtime_file, "127.0.0.1", actual_port, local_sync_token=server.local_sync_token, ) backend_client.RUNTIME_FILE = runtime_file backend_client.CONNECTION_FILE = root / "connection.json" self.assertEqual( backend_client.discover_local_server_url(), f"http://127.0.0.1:{actual_port}", ) self.assertTrue(backend_client.discover_local_sync_token()) self.assertEqual( backend_client.default_settings()["server_url"], f"http://127.0.0.1:{actual_port}", ) config = json.loads(database.config()["config_json"]) config["AI_MODEL"] = "startup-detected-model" admin = database.authenticate("admin", "InitialAdmin123") database.save_config(config, admin["id"], "127.0.0.1") server_thread = threading.Thread( target=server.serve_forever, daemon=True ) server_thread.start() import ai_config old_settings_file = ai_config._SETTINGS_FILE try: ai_config._SETTINGS_FILE = str(root / "startup_ai_settings.json") result = backend_client.startup_sync_config(timeout=3.0) self.assertTrue(result["synced"]) self.assertEqual(ai_config.AI_MODEL, "startup-detected-model") finally: ai_config._SETTINGS_FILE = old_settings_file finally: backend_client.RUNTIME_FILE = old_runtime_file backend_client.CONNECTION_FILE = old_connection_file if server is not None: if server_thread is not None: server.shutdown() server.server_close() if server_thread is not None: server_thread.join(timeout=2) blocker.close() def test_login_roles_publish_and_desktop_sync(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) database = admin_backend.Database(root / "test.db") self.assertTrue(database.initialize("InitialAdmin123")) admin = database.authenticate("admin", "InitialAdmin123") self.assertIsNotNone(admin) database.change_password( admin["id"], "InitialAdmin123", "ChangedAdmin123", "127.0.0.1" ) database.create_user( "readonly.user", "ViewerPassword123", "viewer", admin["id"], "127.0.0.1" ) viewer = database.authenticate("readonly.user", "ViewerPassword123") self.assertEqual(viewer["role"], "viewer") config_row = database.config() config = json.loads(config_row["config_json"]) config["AI_MODEL"] = "integration-test-model" version = database.save_config(config, admin["id"], "127.0.0.1") self.assertEqual(version, 2) server = admin_backend.AdminServer(("127.0.0.1", 0), database) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() old_connection_file = backend_client.CONNECTION_FILE old_runtime_file = backend_client.RUNTIME_FILE try: backend_client.CONNECTION_FILE = root / "connection.json" backend_client.RUNTIME_FILE = root / "runtime.json" port = server.server_address[1] response = backend_client.login( f"http://127.0.0.1:{port}", "admin", "ChangedAdmin123", ) self.assertEqual(response["user"]["role"], "admin") import ai_config old_settings_file = ai_config._SETTINGS_FILE try: ai_config._SETTINGS_FILE = str(root / "synced_ai_settings.json") result = backend_client.sync_config(force=True) self.assertTrue(result["synced"]) self.assertEqual(result["version"], 2) synced = json.loads( Path(ai_config._SETTINGS_FILE).read_text(encoding="utf-8") ) self.assertEqual(synced["AI_MODEL"], "integration-test-model") finally: ai_config._SETTINGS_FILE = old_settings_file backend_client.logout() self.assertFalse(backend_client.is_configured()) finally: backend_client.CONNECTION_FILE = old_connection_file backend_client.RUNTIME_FILE = old_runtime_file server.shutdown() server.server_close() thread.join(timeout=2) if __name__ == "__main__": unittest.main()