gr/kefu
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

更新bug

Browse Source
This commit is contained in:
Your Name
2026-07-31 11:48:16 +08:00
parent f913a57529
commit f22cc1a70d
109 changed files with 37586 additions and 927 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wechat_rpa/test_backend.py
+296 -3
View File
@@ -9,6 +9,9 @@ import socket
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from unittest import mock
from pathlib import Path
import admin_backend
@@ -17,6 +20,290 @@ import backend_client
class BackendIntegrationTest(unittest.TestCase):
def test_admin_page_contains_non_saving_model_test_button(self) -> None:
config = admin_backend.load_initial_config()
card = admin_backend.AdminHandler.config_card(
{"role": "admin"},
"csrf-token",
{"version": 1, "updated_at": "now", "updated_by_name": "admin"},
config,
)
self.assertIn("formaction='/admin/model/test'", card)
self.assertIn("formtarget='_blank'", card)
self.assertIn("测试模型连通性", card)
self.assertIn("value='openai'", card)
self.assertIn("value='dify'", card)
self.assertIn("value='comfyui'", card)
self.assertIn("name='AI_DEVELOPMENT_MODE'", card)
self.assertIn("开启开发模式", card)
self.assertIn("name='AI_UI_GUARD_ENABLED'", card)
self.assertIn("启用 AI 页面守护", card)
def test_development_diagnostics_are_cloud_controlled_and_redacted(self) -> None:
config = {
"AI_DEVELOPMENT_MODE": True,
"AI_PROVIDER_TYPE": "openai",
"AI_API_BASE": "https://api.example/v1",
"AI_API_KEY": "top-secret-key",
"AI_MODEL": "test-model",
"AI_MCP_SERVERS": [
{
"headers": {"Authorization": "Bearer hidden-auth"},
"env": {"ACCESS_TOKEN": "hidden-token"},
"args": [
"--token",
"hidden-argument",
"--password=hidden-inline",
"Bearer hidden-bearer",
],
}
],
}
response = {"version": 8, "updated_at": "now"}
diagnostics = backend_client._config_diagnostics(
config,
response,
"https://cloud.example/api/v1/desktop/config?api_key=query-secret",
)
rendered = "\n".join(diagnostics)
self.assertIn("https://cloud.example/api/v1/desktop/config", rendered)
self.assertIn("云端配置版本: v8", rendered)
self.assertIn('"AI_MODEL": "test-model"', rendered)
for secret in (
"top-secret-key",
"hidden-auth",
"hidden-token",
"hidden-argument",
"hidden-inline",
"hidden-bearer",
"query-secret",
):
self.assertNotIn(secret, rendered)
self.assertEqual(
backend_client._config_diagnostics(
{**config, "AI_DEVELOPMENT_MODE": False}, response, "https://cloud.example/config"
),
[],
)
def test_model_request_diagnostics_never_print_the_api_key(self) -> None:
import ai_chat
import ai_config
previous = {
"AI_DEVELOPMENT_MODE": getattr(ai_config, "AI_DEVELOPMENT_MODE", False),
"AI_API_BASE": ai_config.AI_API_BASE,
"AI_API_KEY": ai_config.AI_API_KEY,
"AI_MODEL": ai_config.AI_MODEL,
}
try:
ai_config.AI_DEVELOPMENT_MODE = True
ai_config.AI_API_BASE = "https://api.example/v1"
ai_config.AI_API_KEY = "never-print-this-key"
ai_config.AI_MODEL = "diagnostic-model"
with mock.patch("builtins.print") as printer:
ai_chat._log_request_diagnostics(
"https://api.example/v1/chat/completions", "OpenAI 兼容"
)
rendered = "\n".join(
" ".join(str(item) for item in call.args)
for call in printer.call_args_list
)
self.assertIn("https://api.example/v1/chat/completions", rendered)
self.assertIn("diagnostic-model", rendered)
self.assertNotIn("never-print-this-key", rendered)
finally:
for key, value in previous.items():
setattr(ai_config, key, value)
def test_model_connection_uses_unsaved_values_and_saved_key(self) -> None:
current = {
"AI_API_BASE": "https://saved.example/v1",
"AI_API_KEY": "saved-secret",
"AI_MODEL": "saved-model",
"AI_TIMEOUT": 120,
}
config = admin_backend.model_test_config(
{
"AI_API_BASE": "https://new.example/v1",
"AI_API_KEY": "",
"AI_MODEL": "new-model",
"AI_TIMEOUT": "180",
},
current,
)
self.assertEqual(config["endpoint"], "https://new.example/v1/chat/completions")
self.assertEqual(config["api_key"], "saved-secret")
self.assertEqual(config["model"], "new-model")
self.assertEqual(config["timeout"], 60)
def test_model_connection_success_does_not_expose_key(self) -> None:
config = admin_backend.model_test_config(
{
"AI_API_BASE": "https://api.example/v1",
"AI_API_KEY": "top-secret-key",
"AI_MODEL": "test-model",
"AI_TIMEOUT": 10,
},
{},
)
response = json.dumps(
{"choices": [{"message": {"content": "OK"}}]}
).encode("utf-8")
with mock.patch(
"admin_backend._perform_http_request", return_value=(200, response)
) as call:
result = admin_backend.test_model_connection(config)
self.assertTrue(result["ok"])
self.assertEqual(result["http_status"], 200)
self.assertNotIn("top-secret-key", json.dumps(result, ensure_ascii=False))
self.assertEqual(call.call_args.args[0], config["endpoint"])
self.assertEqual(
call.call_args.kwargs["headers"]["Authorization"], "Bearer top-secret-key"
)
payload = call.call_args.kwargs["payload"]
self.assertEqual(payload["model"], "test-model")
def test_model_connection_error_redacts_key(self) -> None:
config = admin_backend.model_test_config(
{
"AI_API_BASE": "https://api.example/v1",
"AI_API_KEY": "top-secret-key",
"AI_MODEL": "test-model",
"AI_TIMEOUT": 10,
},
{},
)
response = json.dumps(
{"error": {"message": "invalid top-secret-key"}}
).encode("utf-8")
with mock.patch(
"admin_backend._perform_http_request", return_value=(401, response)
):
result = admin_backend.test_model_connection(config)
self.assertFalse(result["ok"])
self.assertEqual(result["http_status"], 401)
self.assertIn("API Key 无效", result["message"])
self.assertNotIn("top-secret-key", json.dumps(result, ensure_ascii=False))
def test_dify_and_comfyui_use_provider_specific_endpoints(self) -> None:
dify = admin_backend.model_test_config(
{
"AI_PROVIDER_TYPE": "dify",
"AI_API_BASE": "https://dify.example/v1",
"AI_API_KEY": "app-secret",
"AI_MODEL": "",
"AI_TIMEOUT": 10,
},
{},
)
self.assertEqual(dify["endpoint"], "https://dify.example/v1/chat-messages")
self.assertEqual(dify["provider_type"], "dify")
comfyui = admin_backend.model_test_config(
{
"AI_PROVIDER_TYPE": "comfyui",
"AI_API_BASE": "http://127.0.0.1:8188",
"AI_API_KEY": "",
"AI_MODEL": "",
"AI_TIMEOUT": 10,
},
{},
)
self.assertEqual(comfyui["endpoint"], "http://127.0.0.1:8188/system_stats")
with mock.patch(
"admin_backend._perform_http_request",
return_value=(200, b'{"system": {"os": "windows"}, "devices": []}'),
) as call:
result = admin_backend.test_model_connection(comfyui)
self.assertTrue(result["ok"])
self.assertEqual(call.call_args.kwargs["method"], "GET")
self.assertIsNone(call.call_args.kwargs["payload"])
def test_desktop_ai_respects_explicit_dify_provider(self) -> None:
import ai_chat
import ai_config
old_provider = ai_config.AI_PROVIDER_TYPE
old_base = ai_config.AI_API_BASE
try:
ai_config.AI_PROVIDER_TYPE = "dify"
ai_config.AI_API_BASE = "https://dify.example/v1"
self.assertTrue(ai_chat._is_dify_endpoint())
self.assertEqual(
ai_chat._completions_url(),
"https://dify.example/v1/chat-messages",
)
ai_config.AI_PROVIDER_TYPE = "openai"
self.assertFalse(ai_chat._is_dify_endpoint())
finally:
ai_config.AI_PROVIDER_TYPE = old_provider
ai_config.AI_API_BASE = old_base
def test_model_test_api_requires_edit_role(self) -> None:
with tempfile.TemporaryDirectory() as directory:
database = admin_backend.Database(Path(directory) / "test.db")
database.initialize("InitialAdmin123")
admin = database.authenticate("admin", "InitialAdmin123")
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")
database.change_password(
viewer["id"], "ViewerPassword123", "ViewerChanged123", "127.0.0.1"
)
admin_token, _ = database.create_token(admin["id"], "api", "test", 3600)
viewer_token, _ = database.create_token(viewer["id"], "api", "test", 3600)
server = admin_backend.AdminServer(("127.0.0.1", 0), database)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
url = f"http://127.0.0.1:{server.server_address[1]}/api/v1/model/test"
payload = json.dumps(
{
"AI_API_BASE": "https://api.example/v1",
"AI_MODEL": "test-model",
"AI_TIMEOUT": 10,
}
).encode("utf-8")
def request(token: str):
return urllib.request.Request(
url,
data=payload,
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
},
method="POST",
)
fake_result = {
"ok": True,
"provider": "OpenAI 兼容",
"model": "test-model",
"endpoint": "https://api.example/v1/chat/completions",
"http_status": 200,
"latency_ms": 8,
"message": "连接成功，模型回复：OK",
}
try:
with mock.patch(
"admin_backend.test_model_connection", return_value=fake_result
):
with urllib.request.urlopen(request(admin_token), timeout=3) as response:
data = json.loads(response.read().decode("utf-8"))
self.assertTrue(data["ok"])
with self.assertRaises(urllib.error.HTTPError) as context:
urllib.request.urlopen(request(viewer_token), timeout=3)
self.assertEqual(context.exception.code, 403)
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
def test_release_status_detects_optional_and_forced_updates(self) -> None:
optional = app_version.release_status(
{"latest_version": "1.0.1", "force_upgrade": False}
@@ -193,6 +480,10 @@ class BackendIntegrationTest(unittest.TestCase):
def test_login_roles_publish_and_desktop_sync(self) -> None:
with tempfile.TemporaryDirectory() as directory:
major, minor, patch = (
int(part) for part in app_version.APP_VERSION.split(".")[:3]
)
newer_version = f"{major}.{minor}.{patch + 1}"
root = Path(directory)
database = admin_backend.Database(root / "test.db")
self.assertTrue(database.initialize("InitialAdmin123"))
@@ -213,7 +504,7 @@ class BackendIntegrationTest(unittest.TestCase):
version = database.save_config(config, admin["id"], "127.0.0.1")
self.assertEqual(version, 2)
database.save_release(
"1.1.0",
newer_version,
"https://example.com/client.exe",
"测试升级",
True,
@@ -221,7 +512,7 @@ class BackendIntegrationTest(unittest.TestCase):
"127.0.0.1",
)
release = database.release()
self.assertEqual(release["latest_version"], "1.1.0")
self.assertEqual(release["latest_version"], newer_version)
self.assertEqual(release["force_upgrade"], 1)
server = admin_backend.AdminServer(("127.0.0.1", 0), database)
@@ -248,7 +539,9 @@ class BackendIntegrationTest(unittest.TestCase):
self.assertEqual(result["version"], 2)
self.assertTrue(result["update_available"])
self.assertTrue(result["force_upgrade"])
self.assertEqual(result["release"]["latest_version"], "1.1.0")
self.assertEqual(
result["release"]["latest_version"], newer_version
)
synced = json.loads(
Path(ai_config._SETTINGS_FILE).read_text(encoding="utf-8")
)