387 lines
17 KiB
Python
387 lines
17 KiB
Python
# -*- 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()
|