# -*- coding: utf-8 -*- """配置后台与桌面端同步的本地闭环测试。""" from __future__ import annotations import json import hashlib import shutil import model_protocol import socket import sqlite3 import tempfile import threading import unittest import urllib.error import urllib.request from unittest import mock from pathlib import Path from test_support import local_state_redirect import admin_backend import app_version import backend_client class BackendIntegrationTest(unittest.TestCase): 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_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 class DesktopConfigFallbackTest(unittest.TestCase): """桌面端拉配置：新接口优先，老接口兜底。 客户端和服务端是分批更新的，两边版本对不齐是常态而不是意外。直接把路径改 成 v2，那些服务端还没升级的部署会当场同步失败——而且失败得很安静，界面上 只是配置一直显示旧的，没人会立刻发现。 """ def _fail(self, message: str): return backend_client.BackendError(message) def test_the_new_endpoint_is_tried_first(self) -> None: calls = [] def fake(method, server, path, **kwargs): calls.append(path) return 200, {"config": {}, "version": 3} with mock.patch.object(backend_client, "_request", fake): response, used = backend_client._fetch_desktop_config("http://x", "k", 5.0) self.assertEqual(calls, ["/api/v2/desktop/config"]) self.assertEqual(used, "/api/v2/desktop/config") self.assertEqual(response["version"], 3) def test_a_404_falls_back_to_the_legacy_endpoint(self) -> None: """服务端还是老版本时，必须照常同步，不能报错。""" calls = [] def fake(method, server, path, **kwargs): calls.append(path) if path.startswith("/api/v2"): raise backend_client.BackendError("HTTP 404", status=404) return 200, {"config": {}, "version": 7} with mock.patch.object(backend_client, "_request", fake): response, used = backend_client._fetch_desktop_config("http://x", "k", 5.0) self.assertEqual(calls, ["/api/v2/desktop/config", "/api/v1/desktop/config"]) self.assertEqual(used, "/api/v1/desktop/config") self.assertEqual(response["version"], 7) def test_the_fallback_is_decided_by_status_not_by_wording(self) -> None: """老后台对未知 GET 路径回的是"页面不存在"，未知 POST 才回"接口不存在"。 一个字之差。曾经这里靠文案匹配，结果对着真的老后台跑就整个失效了—— 判据必须是状态码。 """ calls = [] def fake(method, server, path, **kwargs): calls.append(path) if path.startswith("/api/v2"): raise backend_client.BackendError("页面不存在", status=404) return 200, {"config": {}} with mock.patch.object(backend_client, "_request", fake): _, used = backend_client._fetch_desktop_config("http://x", "k", 5.0) self.assertEqual(used, "/api/v1/desktop/config") def test_a_connection_failure_is_not_retried_on_the_old_path(self) -> None: """连不上就是连不上，换个路径同样连不上。 真正的危害不是多发一个请求，而是把"服务器宕了"这个明确故障，伪装成 "接口不存在"往下走——最后一层抛出来的是老接口的错误，人会去查一个根本 没坏的地方。 """ calls = [] def fake(method, server, path, **kwargs): calls.append(path) # 连不上根本没拿到响应，所以没有状态码 raise backend_client.BackendError("无法连接后台：timed out") with mock.patch.object(backend_client, "_request", fake): with self.assertRaises(backend_client.BackendError) as caught: backend_client._fetch_desktop_config("http://x", "k", 5.0) self.assertEqual(calls, ["/api/v2/desktop/config"], "不该退回去重试") self.assertIn("无法连接", str(caught.exception)) def test_a_bad_sync_key_surfaces_instead_of_falling_back(self) -> None: """401/403 说明服务在、只是凭证不对。这是要人去改配置的错误，不能吞。""" calls = [] def fake(method, server, path, **kwargs): calls.append(path) raise backend_client.AuthenticationError("同步凭证无效", status=401) with mock.patch.object(backend_client, "_request", fake): with self.assertRaises(backend_client.AuthenticationError): backend_client._fetch_desktop_config("http://x", "k", 5.0) self.assertEqual(calls, ["/api/v2/desktop/config"]) def test_a_404_on_both_paths_reports_the_last_error(self) -> None: def fake(method, server, path, **kwargs): raise backend_client.BackendError("HTTP 404", status=404) with mock.patch.object(backend_client, "_request", fake): with self.assertRaises(backend_client.BackendError): backend_client._fetch_desktop_config("http://x", "k", 5.0) def test_the_legacy_constant_still_points_at_a_real_path(self) -> None: """DESKTOP_CONFIG_PATH 可能被别处引用，不能变成空字符串。""" self.assertIn(backend_client.DESKTOP_CONFIG_PATH, backend_client.DESKTOP_CONFIG_PATHS) self.assertTrue(backend_client.DESKTOP_CONFIG_PATH.startswith("/api/")) class ModelCallReportFallbackTest(unittest.TestCase): """调用留痕上报：和拉配置同一套 v2 → v1 顺序。 补这组测试是因为老后台（8765）退役前卡在这一条上：配置同步早有 v2， 上报却只有 v1。桌面端只要改指到 8766，配置照常同步、回复照常发，唯独 调用记录一条都不进库——静悄悄地丢掉的正是出事后用来解释"这句话怎么来的" 那份证据。 """ def _settings(self): return {"server_url": "http://x"} def test_the_new_endpoint_is_tried_first(self) -> None: calls = [] def fake(method, server, path, **kwargs): calls.append(path) return 200, {"ok": True} with ( mock.patch.object(backend_client, "_request", fake), mock.patch.object(backend_client, "load_settings", self._settings), mock.patch.object(backend_client, "discover_local_sync_token", lambda: ""), ): ok = backend_client.report_model_call({"task_id": "t1"}) self.assertTrue(ok) self.assertEqual(calls, ["/api/v2/model/calls"], "新接口在就不该再打老的") def test_a_404_falls_back_to_the_legacy_endpoint(self) -> None: """还指着老后台（8765）的客户端必须照常上报，不能悄悄丢数据。""" calls = [] def fake(method, server, path, **kwargs): calls.append(path) if path.startswith("/api/v2"): raise backend_client.BackendError("接口不存在", status=404) return 200, {"ok": True} with ( mock.patch.object(backend_client, "_request", fake), mock.patch.object(backend_client, "load_settings", self._settings), mock.patch.object(backend_client, "discover_local_sync_token", lambda: ""), ): ok = backend_client.report_model_call({"task_id": "t1"}) self.assertTrue(ok) self.assertEqual(calls, ["/api/v2/model/calls", "/api/v1/model/calls"]) def test_a_401_does_not_fall_back(self) -> None: """凭证不对换个路径重试同样不对，只会多打一个请求。""" calls = [] def fake(method, server, path, **kwargs): calls.append(path) raise backend_client.AuthenticationError("同步凭证无效", status=401) with ( mock.patch.object(backend_client, "_request", fake), mock.patch.object(backend_client, "load_settings", self._settings), mock.patch.object(backend_client, "discover_local_sync_token", lambda: ""), ): ok = backend_client.report_model_call({"task_id": "t1"}) self.assertFalse(ok) self.assertEqual(calls, ["/api/v2/model/calls"]) def test_a_failure_never_raises_into_the_reply_path(self) -> None: """上报是观测。它出问题绝不能把已经生成好的回复流程带下水。""" def boom(*_args, **_kwargs): raise RuntimeError("断网") with ( mock.patch.object(backend_client, "_request", boom), mock.patch.object(backend_client, "load_settings", self._settings), mock.patch.object(backend_client, "discover_local_sync_token", lambda: ""), ): self.assertFalse(backend_client.report_model_call({"task_id": "t1"})) class GatewayUrlDerivationTest(unittest.TestCase): """网关地址由后台算出来告诉客户端。 这是"只改一个域名"的关键。桌面端配置里只有后台地址一项，网关在哪、模型怎么 编排全部由后台决定——运维换网关位置只动后台一处，不用挨个改客户端。 """ def test_direct_port_access_points_at_the_local_gateway_port(self) -> None: """本机开发：三个服务各占一个端口，网关就在同机的 8770。""" self.assertEqual( admin_backend.derive_gateway_url("", "http", "127.0.0.1:8765"), "http://127.0.0.1:8770/v1/answer", ) self.assertEqual( admin_backend.derive_gateway_url("", "http", "192.168.1.20:8766"), "http://192.168.1.20:8770/v1/answer", ) def test_a_proxied_domain_gets_a_same_origin_path(self) -> None: """反代后面只有 80/443 对外，公网根本连不到 8770。 这两种推法必须分开。用一套规则套两种部署，总有一边是错的——而且错得很 安静：客户端拿到一个连不上的地址，表现成"模型不回复"。 """ self.assertEqual( admin_backend.derive_gateway_url("", "https", "xchat.example.com"), "https://xchat.example.com/gateway/v1/answer", ) def test_an_absolute_override_is_used_as_is(self) -> None: """网关在别的域名或别的机器上时用这个。""" self.assertEqual( admin_backend.derive_gateway_url( "https://gw.example.com/v1/answer", "https", "xchat.example.com" ), "https://gw.example.com/v1/answer", ) def test_a_path_override_hangs_off_the_backend_domain(self) -> None: self.assertEqual( admin_backend.derive_gateway_url( "/model-gw/v1/answer", "https", "xchat.example.com" ), "https://xchat.example.com/model-gw/v1/answer", ) def test_a_missing_host_still_yields_something_usable(self) -> None: """拿不到 Host 时给本机地址，而不是空串。 空串会让客户端把 gateway 当成"没配"，静默回落到本机单模型——而本机 没有密钥，最后报的是一个看不懂的 401。 """ self.assertTrue( admin_backend.derive_gateway_url("", "http", "").startswith("http://") ) def test_an_override_without_a_scheme_is_refused_on_save(self) -> None: """漏了 http:// 会被拼成 https://你的域名/gw.example.com/v1/answer。 那是个没意义的地址，而且失败要等到桌面端下次发模型请求才暴露。 """ base = { "AI_AGENT_NAME": "甲", "AI_HOSPITAL_NAME": "乙", "AI_CONTEXT_MAX_ROUNDS": "5", "AI_MCP_MAX_ROUNDS": "5", "AI_MCP_SERVERS": "[]", } with self.assertRaises(ValueError) as caught: admin_backend.validate_config_form( {**base, "AI_GATEWAY_URL": "gw.example.com/v1/answer"}, {} ) self.assertIn("http", str(caught.exception)) def test_an_empty_override_is_accepted_and_means_auto(self) -> None: config = admin_backend.validate_config_form( { "AI_AGENT_NAME": "甲", "AI_HOSPITAL_NAME": "乙", "AI_CONTEXT_MAX_ROUNDS": "5", "AI_MCP_MAX_ROUNDS": "5", "AI_MCP_SERVERS": "[]", "AI_GATEWAY_URL": "", }, {}, ) self.assertEqual(config["AI_GATEWAY_URL"], "") class GatewayArrivesBySyncTest(unittest.TestCase): """桌面端不用手工配网关——同步时后台会把地址带过来。""" def _settings(self, response, existing=None): """跑一遍 _apply_config_response，返回它写下的设置。""" saved = {} base = backend_client.default_settings() if existing: base.update(existing) with ( mock.patch.object(backend_client, "load_settings", lambda: dict(base)), mock.patch.object(backend_client, "save_settings", saved.update), mock.patch("ai_config.apply_settings", lambda cfg, persist=True: cfg), ): backend_client._apply_config_response(response, dict(base)) return saved def test_the_synced_gateway_address_lands_in_the_settings(self) -> None: saved = self._settings( { "version": 3, "config": {}, "gateway": {"enabled": True, "url": "https://x.example.com/gateway/v1/answer"}, } ) self.assertEqual( saved["gateway"], {"enabled": True, "url": "https://x.example.com/gateway/v1/answer"}, ) def test_an_old_backend_without_a_gateway_block_keeps_the_last_one(self) -> None: """老版本后端不返回这一段。清空等于把已经能用的客户端打回"没有网关"。""" existing = {"gateway": {"enabled": True, "url": "https://old.example.com/v1/answer"}} saved = self._settings({"version": 4, "config": {}}, existing=existing) self.assertEqual(saved.get("gateway", existing["gateway"]), existing["gateway"]) def test_a_blank_url_from_the_backend_is_ignored(self) -> None: """后台配错回了空地址时，宁可保持原样也不要把客户端弄成没网关。""" existing = {"gateway": {"enabled": True, "url": "https://good.example.com/v1/answer"}} saved = self._settings( {"version": 5, "config": {}, "gateway": {"enabled": True, "url": " "}}, existing=existing, ) self.assertEqual(saved.get("gateway", existing["gateway"]), existing["gateway"]) class EndpointModeTest(unittest.TestCase): """接口地址：自动补全 vs 原样使用。 加这个开关是因为 auto 那套拼接规则只覆盖得了"服务商标准形状"。自建服务的路径 常常不按套路，比如 `https://api.example.com/custom/llm/invoke`——auto 会把它 拼成 `.../invoke/chat/completions`，请求发到一个不存在的地址，报 404，而排查的 人会去怀疑密钥和网络。 为什么用显式开关而不是"更聪明的猜测"：`https://api.example.com/openai` 到底是 前缀还是完整端点，光看地址分不出来。猜错的两种方向都会静默地把请求发歪。 """ def test_auto_completes_the_path_for_each_kind(self) -> None: cases = [ ("openai", "https://api.openai.com/v1", "https://api.openai.com/v1/chat/completions"), ("openai", "https://api.deepseek.com", "https://api.deepseek.com/chat/completions"), ("claude", "https://api.anthropic.com", "https://api.anthropic.com/v1/messages"), ("dify", "https://api.dify.ai/v1", "https://api.dify.ai/v1/chat-messages"), ] for kind, base, expected in cases: with self.subTest(kind=kind, base=base): self.assertEqual(model_protocol.endpoint_url(kind, base), expected) def test_auto_does_not_double_append_a_complete_url(self) -> None: """已经写全的标准地址不该再被拼一次。""" full = "https://api.openai.com/v1/chat/completions" self.assertEqual(model_protocol.endpoint_url("openai", full), full) def test_exact_uses_the_address_verbatim(self) -> None: odd = "https://api.example.com/custom/llm/invoke" self.assertEqual(model_protocol.endpoint_url("openai", odd), f"{odd}/chat/completions") self.assertEqual(model_protocol.endpoint_url("openai", odd, "exact"), odd) def test_exact_wins_for_every_kind(self) -> None: """exact 是"别动我的地址"，不该被接口类型的规则翻掉。""" odd = "https://gw.internal/llm" for kind in ("openai", "claude", "dify", "comfyui"): with self.subTest(kind=kind): self.assertEqual(model_protocol.endpoint_url(kind, odd, "exact"), odd) def test_a_trailing_slash_is_trimmed_in_both_modes(self) -> None: self.assertEqual( model_protocol.endpoint_url("openai", "https://x.com/llm/", "exact"), "https://x.com/llm", ) def test_the_connectivity_test_uses_the_same_rule_as_the_real_call(self) -> None: """两边规则若不一致，测试会去戳一个和实际调用不同的地址。 那是最没用的一种测试：测通了照样用不了，或者反过来。 """ odd = "https://api.example.com/custom/llm/invoke" for mode in ("auto", "exact"): with self.subTest(mode=mode): self.assertEqual( admin_backend._model_endpoint(odd, "openai", mode), model_protocol.endpoint_url("openai", odd, mode), ) def test_dify_upload_root_follows_the_mode(self) -> None: """Dify 传图要拼 /files/upload，根地址得跟着 exact 走。""" self.assertEqual( model_protocol.dify_api_root("https://dify.internal/app/chat", "exact"), "https://dify.internal/app/chat", ) class EndpointModeStorageTest(unittest.TestCase): """存和读。老库要能平滑升级，老记录的行为必须一个字都不变。""" def setUp(self) -> None: self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) self.db = admin_backend.Database(self.root / "t.db") self.db.initialize("Admin@123456") self.uid = self.db.authenticate("admin", "Admin@123456")["id"] def _save(self, **extra): item = { "id": "p1", "name": "出口", "kind": "openai", "base_url": "https://api.example.com/custom/llm/invoke", "api_key": "sk-x", "model": "m", } item.update(extra) return self.db.save_model_provider(item, self.uid, "1.1.1.1") def test_the_mode_round_trips(self) -> None: self._save(endpoint_mode="exact") item = next(x for x in self.db.model_providers() if x["id"] == "p1") self.assertEqual(item["endpoint_mode"], "exact") self.assertEqual(item["endpoint"], "https://api.example.com/custom/llm/invoke") def test_the_default_is_auto(self) -> None: """不填就是老行为。这个默认值是整件事零回归的前提。""" self._save() item = next(x for x in self.db.model_providers() if x["id"] == "p1") self.assertEqual(item["endpoint_mode"], "auto") self.assertTrue(item["endpoint"].endswith("/chat/completions")) def test_an_unknown_mode_is_refused(self) -> None: with self.assertRaises(ValueError) as caught: self._save(endpoint_mode="随便写") self.assertIn("地址模式", str(caught.exception)) def test_an_old_database_without_the_column_is_upgraded_in_place(self) -> None: """已经在用的库不能因为加一列就要重建表——重建的每一步都可能丢数据。""" path = self.root / "old.db" con = sqlite3.connect(path) con.executescript( """ CREATE TABLE model_providers ( id TEXT PRIMARY KEY, name TEXT NOT NULL, kind TEXT NOT NULL, base_url TEXT NOT NULL, api_key_enc TEXT NOT NULL DEFAULT '', model TEXT NOT NULL DEFAULT '', capabilities TEXT NOT NULL DEFAULT 'text', max_tokens INTEGER NOT NULL DEFAULT 500, temperature REAL NOT NULL DEFAULT 0.35, timeout_ms INTEGER NOT NULL DEFAULT 30000, max_inflight INTEGER NOT NULL DEFAULT 32, rpm_limit INTEGER NOT NULL DEFAULT 0, enabled INTEGER NOT NULL DEFAULT 1, health TEXT NOT NULL DEFAULT 'unknown', health_checked_at TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, updated_at TEXT NOT NULL, updated_by INTEGER); INSERT INTO model_providers (id,name,kind,base_url,created_at,updated_at) VALUES ('old-1','老出口','openai','https://api.openai.com/v1','x','x'); """ ) con.commit() con.close() with mock.patch("builtins.print"): admin_backend.Database(path).initialize("Admin@123456") item = next( x for x in admin_backend.Database(path).model_providers() if x["id"] == "old-1" ) self.assertEqual(item["endpoint_mode"], "auto", "老记录必须还是老行为") self.assertEqual(item["endpoint"], "https://api.openai.com/v1/chat/completions") def test_running_the_column_migration_twice_is_harmless(self) -> None: with mock.patch("builtins.print"): admin_backend.Database(self.root / "t.db").initialize("Admin@123456") self.assertTrue(self.db.model_providers() is not None) class ModelCallTraceabilityTest(unittest.TestCase): """`model_calls` 存客户原话、模型原话、审核原因——用来"跟踪问题"。 网关和桌面端各自都会往这张表报一次同一次调用（网关看得到候选和裁判， 桌面端才知道审核规则命中原因）。两边共用 task_id，谁先落盘谁建行，后到 的一方只补自己独有的字段——这里测的就是这道合并逻辑不多写也不少写。 """ def setUp(self) -> None: self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) self.db = admin_backend.Database(self.root / "t.db") with mock.patch("builtins.print"): self.db.initialize("Admin@123456") def _record(self, **extra) -> dict: record = { "device_id": "dev-1", "task_id": "task-1", "roles_version": 1, "judge_mode": "arbitrate", "chosen": "openai-primary", "judge": {"winner": "A", "score": 0.8, "risk": "low"}, "candidates": [{"provider": "openai-primary", "text": "建议您注意休息"}], "total_ms": 900, "customer_text": "我这几天总是失眠", "reply_text": "建议您注意休息", "review_reason": "", } record.update(extra) return record def _row(self, task_id: str) -> sqlite3.Row: with self.db.connect() as con: return con.execute( "SELECT * FROM model_calls WHERE task_id = ?", (task_id,) ).fetchone() def test_a_call_round_trips_customer_and_reply_text(self) -> None: self.db.log_model_call(self._record()) row = self._row("task-1") self.assertEqual(row["customer_text"], "我这几天总是失眠") self.assertEqual(row["reply_text"], "建议您注意休息") self.assertEqual(row["review_reason"], "") def test_a_second_report_with_the_same_task_id_only_merges_review_reason(self) -> None: """网关先落盘（候选、裁判都是权威数据），桌面端后到——不能把网关的数据覆盖掉。""" self.db.log_model_call(self._record()) self.db.log_model_call(self._record( chosen="不该生效的值", candidates=[{"provider": "不该生效的值", "text": "x"}], total_ms=1, review_reason="命中审核规则「诊断」", )) with self.db.connect() as con: rows = con.execute( "SELECT * FROM model_calls WHERE task_id='task-1'" ).fetchall() self.assertEqual(len(rows), 1, "同一个 task_id 只能落一行，不是两行") row = self._row("task-1") self.assertEqual(row["chosen"], "openai-primary", "第二次上报不该覆盖已有的候选数据") self.assertEqual(row["review_reason"], "命中审核规则「诊断」", "审核原因只有桌面端知道，必须生效") def test_blank_task_ids_never_collide(self) -> None: """本地非网关路径目前还给不出 task_id，留空时按老行为各插一行。""" self.db.log_model_call(self._record(task_id="", customer_text="第一条")) self.db.log_model_call(self._record(task_id="", customer_text="第二条")) with self.db.connect() as con: n = con.execute( "SELECT COUNT(*) AS n FROM model_calls WHERE task_id=''" ).fetchone()["n"] self.assertEqual(n, 2) def test_list_model_calls_finds_a_keyword_in_either_side_of_the_exchange(self) -> None: self.db.log_model_call(self._record(task_id="t1", customer_text="想咨询糖尿病用药")) self.db.log_model_call(self._record(task_id="t2", customer_text="今天天气不错", reply_text="是呀")) found = self.db.list_model_calls(days=7, q="糖尿病") self.assertEqual(found["total"], 1) self.assertEqual(found["items"][0]["task_id"], "t1") def test_list_model_calls_parses_candidates_back_into_a_list(self) -> None: self.db.log_model_call(self._record(task_id="t1")) found = self.db.list_model_calls(days=7) self.assertEqual(found["items"][0]["candidates"][0]["provider"], "openai-primary") def test_list_model_calls_respects_the_days_window(self) -> None: self.db.log_model_call(self._record(task_id="t1")) with self.db.connect() as con: con.execute( "UPDATE model_calls SET created_at = '2000-01-01 00:00:00' WHERE task_id='t1'" ) con.commit() found = self.db.list_model_calls(days=7) self.assertEqual(found["total"], 0) def test_an_old_database_with_duplicate_task_ids_is_deduped_before_the_index_goes_on(self) -> None: """老版本拿会话指纹当 task_id，同一个会话十几轮全共享一个值——建唯一索引前必须先清干净。""" path = self.root / "old_calls.db" con = sqlite3.connect(path) con.executescript( """ CREATE TABLE model_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT NOT NULL DEFAULT '', task_id TEXT NOT NULL DEFAULT '', roles_version INTEGER NOT NULL DEFAULT 0, judge_mode TEXT NOT NULL DEFAULT '', chosen TEXT NOT NULL DEFAULT '', judge_winner TEXT NOT NULL DEFAULT '', judge_score REAL NOT NULL DEFAULT 0, judge_risk TEXT NOT NULL DEFAULT '', candidates_json TEXT NOT NULL DEFAULT '[]', total_ms INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL ); INSERT INTO model_calls (device_id,task_id,chosen,created_at) VALUES ('d1','same-session','p1','2026-01-01 00:00:01'); INSERT INTO model_calls (device_id,task_id,chosen,created_at) VALUES ('d1','same-session','p2','2026-01-01 00:00:02'); INSERT INTO model_calls (device_id,task_id,chosen,created_at) VALUES ('d1','same-session','p3','2026-01-01 00:00:03'); """ ) con.commit() con.close() with mock.patch("builtins.print"): admin_backend.Database(path).initialize("Admin@123456") db = admin_backend.Database(path) with db.connect() as con: rows = con.execute( "SELECT task_id, chosen FROM model_calls ORDER BY id" ).fetchall() self.assertEqual([r["task_id"] for r in rows], ["", "", "same-session"], "只留最新一行的 task_id") self.assertEqual(rows[-1]["chosen"], "p3", "保留的必须是最新那一行") # 索引确实建起来了：迁移后再报一次同名 task_id 应该走合并而不是新插一行 db.log_model_call({"task_id": "same-session", "review_reason": "新原因"}) with db.connect() as con: n = con.execute( "SELECT COUNT(*) AS n FROM model_calls WHERE task_id='same-session'" ).fetchone()["n"] merged = con.execute( "SELECT chosen, review_reason FROM model_calls WHERE task_id='same-session'" ).fetchone() self.assertEqual(n, 1, "唯一索引必须已经生效，不能再插出第二行") self.assertEqual(merged["chosen"], "p3", "合并不该覆盖已有数据") self.assertEqual(merged["review_reason"], "新原因") def test_running_the_model_calls_migration_twice_is_harmless(self) -> None: with mock.patch("builtins.print"): admin_backend.Database(self.root / "t.db").initialize("Admin@123456") self.db.log_model_call(self._record(task_id="tx")) self.assertEqual(self._row("tx")["chosen"], "openai-primary") class GuardCallsStayOutOfTheCustomerLogTest(unittest.TestCase): """界面识别调用不能混进"客服对话"的调用日志和调用统计。 这是用户真实报上来的 bug：调用记录点开一看，客户消息和模型回复全是「（空）」， 候选里全是 `{"state":"unknown","navigation_right_ratio":0,...}`——那是机器人 自己在看企业微信窗口，不是在回客户。守卫每轮轮询都要问一次模型，条数远多于 真实对话，不分开的话这张表根本没法用来查问题，裁判分布也在给错误的东西打分。 """ def setUp(self) -> None: self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, ignore_errors=True) self.db = admin_backend.Database(self.root / "t.db") with mock.patch("builtins.print"): self.db.initialize("Admin@123456") def _log(self, task_id: str, purpose: str, **extra) -> None: record = { "device_id": "d1", "task_id": task_id, "chosen": "p1", "judge": {"winner": "A", "score": 0.9, "risk": "low"}, "candidates": [{"provider": "p1", "text": "文本"}], "total_ms": 100, "customer_text": "客户说的", "reply_text": "模型答的", "purpose": purpose, } record.update(extra) self.db.log_model_call(record) def test_the_default_log_shows_only_customer_chats(self) -> None: self._log("c1", "chat") self._log("g1", "guard") self._log("g2", "guard") found = self.db.list_model_calls(days=7) self.assertEqual(found["total"], 1) self.assertEqual(found["items"][0]["task_id"], "c1") def test_guard_calls_can_still_be_listed_on_purpose(self) -> None: """内部调用照样在花钱，需要的时候得能翻出来看，不能查无此物。""" self._log("c1", "chat") self._log("g1", "guard") self.assertEqual(self.db.list_model_calls(days=7, purpose="guard")["total"], 1) self.assertEqual(self.db.list_model_calls(days=7, purpose="")["total"], 2) def test_stats_ignore_guard_calls(self) -> None: """裁判分布要量的是"发给客户的回复够不够好"，不是"布局认得准不准"。""" self._log("c1", "chat", judge={"winner": "A", "score": 0.9, "risk": "low"}) for index in range(5): self._log( f"g{index}", "guard", judge={"winner": "B", "score": 0.1, "risk": "high"}, ) stats = self.db.model_call_stats(7) self.assertEqual(stats["total"], 1, "5 条界面识别不该被算成客服调用") self.assertEqual(stats["judged"], 1) self.assertAlmostEqual(stats["avg_score"], 0.9, places=3) self.assertNotIn("high", stats["risk"], "守卫的低分不该污染风险占比") def test_a_call_without_an_explicit_purpose_counts_as_a_customer_chat(self) -> None: """桌面端补报的那条路不带 purpose——默认必须是 chat，不能凭空消失。""" self.db.log_model_call({ "task_id": "c9", "chosen": "p1", "judge": {}, "candidates": [], "customer_text": "在吗", "reply_text": "在的", }) self.assertEqual(self.db.list_model_calls(days=7)["total"], 1) def test_legacy_guard_rows_are_reclassified_by_their_fingerprint(self) -> None: """老库里的守卫调用落的是默认 chat，要按候选内容的指纹认出来并改判。 指纹用的是分类器专用的 JSON 字段名（布局守卫的 navigation_right_ratio、 会话行分类器的 reply_capable）——正常客服回复里不可能出现这些词。 """ path = self.root / "legacy.db" con = sqlite3.connect(path) con.executescript( """ CREATE TABLE model_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT NOT NULL DEFAULT '', task_id TEXT NOT NULL DEFAULT '', roles_version INTEGER NOT NULL DEFAULT 0, judge_mode TEXT NOT NULL DEFAULT '', chosen TEXT NOT NULL DEFAULT '', judge_winner TEXT NOT NULL DEFAULT '', judge_score REAL NOT NULL DEFAULT 0, judge_risk TEXT NOT NULL DEFAULT '', candidates_json TEXT NOT NULL DEFAULT '[]', total_ms INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL ); INSERT INTO model_calls (task_id,candidates_json,created_at) VALUES ('L1','[{"provider":"a","text":"{\\"state\\":\\"unknown\\",\\"navigation_right_ratio\\":0}"}]','2026-01-01 00:00:01'), ('L2','[{"provider":"a","text":"{\\"kind\\":\\"customer_chat\\",\\"reply_capable\\":true}"}]','2026-01-01 00:00:02'), ('L3','[{"provider":"a","text":"忙完了，歇会儿吧，最近还好吗"}]','2026-01-01 00:00:03'); """ ) con.commit() con.close() with mock.patch("builtins.print"): admin_backend.Database(path).initialize("Admin@123456") db = admin_backend.Database(path) with db.connect() as con: marks = { row["task_id"]: row["purpose"] for row in con.execute("SELECT task_id, purpose FROM model_calls") } self.assertEqual(marks["L1"], "guard", "布局守卫的记录要改判") self.assertEqual(marks["L2"], "guard", "会话行分类器的记录也要改判") self.assertEqual(marks["L3"], "chat", "真实客服回复绝不能被误判成内部调用") def test_the_reclassification_never_deletes_anything(self) -> None: self._log("g1", "chat", candidates=[ {"provider": "a", "text": '{"navigation_right_ratio":0.1}'} ]) before = self.db.list_model_calls(days=7, purpose="")["total"] with self.db.connect() as con: admin_backend.Database._backfill_guard_purpose(con) con.commit() after = self.db.list_model_calls(days=7, purpose="")["total"] self.assertEqual(before, after, "只改分类标记，一行都不能少") self.assertEqual(self.db.list_model_calls(days=7, purpose="guard")["total"], 1) def setUpModule(): # 别让测试读到开发机上的真实桌面端配置——配过模型网关的机器会让 # `ai_chat.current_provider()` 整体改走网关分支，一大片无关测试跟着变行为。 local_state_redirect.start() def tearDownModule(): local_state_redirect.stop() if __name__ == "__main__": unittest.main()