feat(aibox): interactive model profile manager (model_config.py)

Menu-driven list/add/delete/connectivity-test for config.json models
section - no hand-written JSON needed. Key convention preserved: env
var name only, never plaintext (or empty = keyless local endpoint).
Delete auto-cleans dangling fallback refs; default profile via
api.default_model; save failure reported. 17 unit tests, 441 passed.
This commit is contained in:
qing committed 2026-08-18 03:26:53 +00:00
1 parent 4cb97318b5
commit d513c5f892
3 files changed
+631 -2

No files matched your search

+18 -2
View File
@@ -57,6 +57,20 @@ python aibox.py
启动后,WebSocket 服务器将运行在 `ws://localhost:8765`(地址与端口可在 `config.json` 的 `system.websocket_host` / `system.websocket_port` 中修改)。
### 管理大模型接入点(不用手写配置)
不想手写 `config.json` 的 `models` 节?用交互式菜单工具:
```bash
python model_config.py # 交互式菜单:列出 / 添加 / 删除 / 连通性验证
python model_config.py --list # 非交互:列出当前接入点
```
- **添加接入点**:按提示填写档案名、API 地址(附常见厂商端点参考)、模型名;Key 两种方式——环境变量名(推荐,Key 不落盘)或免密本地端点(Ollama 等);可选显示名/温度/超时/思考模式/失败转移(从已有接入点多选)/设为默认
- **删除接入点**:编号选择 + 确认,自动清理其他档案的失败转移引用
- **连通性验证**:对选中接入点发一次最小请求,直接给出成败原因
- 添加后可立即使用:`python aibox.py --model <档案名>` 指定,或设为默认后无需 `--model`
> 注意:本模块入口文件为 `aibox.py`(当前约 3408 行,集成 WebSocket 服务器、AI 引擎、工具/技能系统、任务调度器于一体;自 v3.0 起工具类与子系统已抽离至 `tools/`、`subsystems/`、`common/`、`tests/` 包,aibox.py 仅保留核心类与编排逻辑)。`client.py` 为示例 WebSocket 客户端,`voice_client.py` 为语音客户端,`llm_client.py` 为多厂商大模型客户端抽象层,`_run.sh` 为被 `core` 拉起的启动脚本。
## 📁 目录结构
@@ -64,6 +78,7 @@ python aibox.py
```
aibox/
├── aibox.py # 服务主程序(WebSocket 服务器 + AI 引擎 + 任务调度器 + 核心类 DeepSeekChat/WebSocketHandler/EnhancedAIManager)
├── model_config.py # 大模型接入点交互式管理工具(列表/添加/删除/连通性验证,不用手写配置)
├── client.py # 示例 WebSocket 客户端
├── voice_client.py # 语音客户端
├── llm_client.py # 大模型客户端抽象层(多厂商切换,DeepSeekClient 兼容 OpenAI 协议)
@@ -92,7 +107,7 @@ aibox/
│ ├── prompt_tools.py # B 批:UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation
│ ├── proxy_tools.py # B 批:SetProxy/GetProxy
│ └── misc_tools.py # B 批:AskUserTool
├── tests/ # 单元测试套件(424 项全过)
├── tests/ # 单元测试套件(441 项全过)
│ ├── conftest.py # sys.path 配置 + websockets 桩
│ ├── test_basic_tools.py # A 批工具 43 项
│ ├── test_b_tools.py # B 批工具 58 项
@@ -108,7 +123,8 @@ aibox/
│ ├── test_f2_fixes.py # 超时清理/线程收尾(F2 5 项)
│ ├── test_g_fixes.py # 第三轮审计 Batch G(20 项)
│ ├── test_h_fixes.py # 第四轮审计 Batch H(26 项)
│ └── test_i_fixes.py # 遗留收尾 Batch I(16 项)
│ ├── test_i_fixes.py # 遗留收尾 Batch I(16 项)
│ └── test_model_config.py # 模型接入点管理工具(17 项)
├── skills/ # 内置扩展技能(*.py 可执行脚本 / *.md Markdown 技能)
│ ├── hello-world.py
│ ├── aes-crypto.py
+415
View File
@@ -0,0 +1,415 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
大模型接入点(模型档案)交互式管理工具
不用手写 config.json:菜单式列出 / 添加 / 删除 / 连通性验证
`models` 节的模型接入点(OpenAI Chat Completions 兼容端点:
DeepSeek / Kimi / Qwen / GLM / 本地 Ollama 等)。
用法:
python model_config.py # 交互式菜单
python model_config.py --list # 非交互:列出当前接入点
安全约定(与 aibox 审计一致):API Key 一律不写入配置文件——
只保存环境变量名(api_key_env),或留空表示免密本地端点。
"""
import argparse
import os
import re
import sys
from typing import Dict, List, Optional
# 允许从仓库根目录或 aibox/ 目录启动
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import ConfigManager # noqa: E402
PROFILE_NAME_RE = re.compile(r"^[A-Za-z0-9_-]+$")
# 常见厂商端点提示(首行是默认建议值)
PROVIDER_HINTS = {
"DeepSeek": "https://api.deepseek.com/v1/chat/completions",
"Kimi (Moonshot)": "https://api.moonshot.cn/v1/chat/completions",
"Qwen (阿里云百炼)": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
"GLM (智谱)": "https://open.bigmodel.cn/api/paas/v4/chat/completions",
"Ollama 本地": "http://localhost:11434/v1/chat/completions",
}
# ==================== 核心逻辑(可单测,不依赖 input) ====================
def get_models(config: ConfigManager) -> Dict:
"""取 models 节(不存在则初始化为空 dict)。"""
models = config.get("models") or {}
if not isinstance(models, dict):
models = {}
return models
def list_profiles(config: ConfigManager) -> List[Dict]:
"""列出所有档案的归一化展示数据。"""
models = get_models(config)
default = config.get("api.default_model") or ""
profiles = []
for name, profile in models.items():
if not isinstance(profile, dict):
continue
profiles.append({
"name": name,
"is_default": name == default,
"display_name": profile.get("display_name") or name,
"api_url": profile.get("api_url", ""),
"model": profile.get("model", ""),
"api_key_env": profile.get("api_key_env", ""),
"reasoning_enabled": bool(profile.get("reasoning_enabled", False)),
"temperature": profile.get("temperature"),
"timeout": profile.get("timeout"),
"fallback": profile.get("fallback") or [],
})
return profiles
def validate_new_profile(config: ConfigManager, name: str, api_url: str,
model: str) -> Optional[str]:
"""校验新档案字段,返回错误信息(合法返回 None)。"""
if not name:
return "档案名不能为空"
if not PROFILE_NAME_RE.match(name):
return f"档案名只能包含字母/数字/下划线/短横线(当前: {name})"
models = get_models(config)
if name in models:
return f"档案名 '{name}' 已存在"
if not api_url or not str(api_url).strip().lower().startswith(("http://", "https://")):
return "API 地址必须是 http:// 或 https:// 开头的完整 URL"
if not model or not str(model).strip():
return "模型名不能为空(如 deepseek-chat / gpt-4o / qwen-plus)"
return None
def add_profile(config: ConfigManager, name: str, api_url: str, model: str,
api_key_env: str = "", display_name: str = None,
temperature: float = None, timeout: int = None,
reasoning_enabled: bool = False,
fallback: Optional[List[str]] = None,
set_default: bool = False) -> List[str]:
"""添加档案。返回提示信息列表;字段非法时返回带 ❌ 的错误信息。"""
err = validate_new_profile(config, name, api_url, model)
if err:
return [f"❌ {err}"]
models = get_models(config)
profile = {
"api_url": str(api_url).strip(),
"model": str(model).strip(),
}
if api_key_env:
profile["api_key_env"] = api_key_env.strip()
else:
# 空串 = 免密本地端点(Ollama/vLLM),显式声明
profile["api_key_env"] = ""
if display_name:
profile["display_name"] = display_name.strip()
if temperature is not None:
try:
profile["temperature"] = max(0.0, min(2.0, float(temperature)))
except (TypeError, ValueError):
return ["❌ 温度必须是 0.0~2.0 的数字"]
if timeout is not None:
try:
profile["timeout"] = max(5, int(timeout))
except (TypeError, ValueError):
return ["❌ 超时必须是整数秒"]
if reasoning_enabled:
profile["reasoning_enabled"] = True
if fallback:
# 只保留已存在的档案,避免悬空引用(L4a 启动期硬性报错)
known = set(models.keys())
invalid = [f for f in fallback if f not in known or f == name]
if invalid:
return [f"❌ 失败转移包含不存在的档案或自身: {invalid},请重新选择"]
profile["fallback"] = list(fallback)
models[name] = profile
config._config["models"] = models
if set_default:
config._config.setdefault("api", {})["default_model"] = name
saved = config._save_config()
msgs = [f"✅ 接入点 '{name}' 已添加({profile['api_url']},模型 {profile['model']})"]
if not saved:
msgs.append("⚠️ 保存失败(仅本次运行生效,重启后丢失)——请检查配置文件目录权限")
if set_default:
msgs.append(f"⭐ 已设为默认接入点(--model 不指定时使用)")
if not api_key_env:
msgs.append("ℹ️ 免密端点:请求不携带 API Key(适用于 Ollama/vLLM 等本地服务)")
else:
msgs.append(f"🔑 Key 从环境变量 {api_key_env} 读取(未设置时该档案不可用)")
return msgs
def delete_profile(config: ConfigManager, name: str) -> List[str]:
"""删除档案,并自动清理其他档案 fallback 中的悬空引用。"""
models = get_models(config)
if name not in models:
return [f"❌ 档案 '{name}' 不存在"]
del models[name]
# 清理其他档案对该档案的 fallback 引用(悬空引用会导致启动期报错)
cleaned = []
for other, profile in models.items():
if not isinstance(profile, dict):
continue
fb = profile.get("fallback") or []
if name in fb:
profile["fallback"] = [f for f in fb if f != name]
cleaned.append(other)
config._config["models"] = models
# 若删除的是默认档案,清除默认标记(回退旧 api 节语义)
if config.get("api.default_model") == name:
config._config.setdefault("api", {}).pop("default_model", None)
saved = config._save_config()
msgs = [f"✅ 接入点 '{name}' 已删除"]
if cleaned:
msgs.append(f"ℹ️ 已同时清理 {len(cleaned)} 个档案的失败转移引用: {', '.join(cleaned)}")
if not saved:
msgs.append("⚠️ 保存失败——请检查配置文件目录权限")
return msgs
def test_profile(config: ConfigManager, name: str, logger=None) -> str:
"""连通性验证:向该档案端点发一次最小请求。"""
models = get_models(config)
if name not in models:
return f"❌ 档案 '{name}' 不存在"
try:
from llm_client import create_llm_client
client = create_llm_client(config, api_key=None, logger=logger, model=name)
result = client.chat_completion(
[{"role": "user", "content": "ping"}], max_tokens=5)
if result is None or not str(result).strip():
return f"❌ 档案 '{name}' 验证失败:端点返回空响应(多为 4xx 认证/参数错误)"
return f"✅ 档案 '{name}' 连通正常:{str(result).strip()[:60]}"
except Exception as e:
return f"❌ 档案 '{name}' 验证失败:{str(e)}"
# ==================== 交互式菜单 ====================
def _ask(prompt: str, default: str = "") -> str:
if default:
val = input(f"{prompt}(回车默认: {default}): ").strip()
return val if val else default
return input(f"{prompt}: ").strip()
def _ask_yes_no(prompt: str, default: bool = False) -> bool:
hint = "y/n" if not default else "Y/n"
while True:
val = input(f"{prompt} [{hint}] ").strip().lower()
if not val:
return default
if val in ("y", "yes", "是"):
return True
if val in ("n", "no", "否"):
return False
print(" 请输入 y 或 n")
def _choose(prompt: str, options: List[str], multi: bool = False) -> List[str]:
"""编号选择(单选或多选)。多选用逗号分隔,直接回车 = 不选。"""
for i, opt in enumerate(options, 1):
print(f" {i}. {opt}")
if multi:
while True:
raw = input(f"{prompt}(多选,逗号分隔,直接回车不选): ").strip()
if not raw:
return []
try:
idxs = [int(x) for x in raw.replace(",", ",").split(",") if x.strip()]
except ValueError:
print(" 请输入数字编号,逗号分隔")
continue
if all(1 <= i <= len(options) for i in idxs):
return [options[i - 1] for i in idxs]
print(f" 编号超出范围(1~{len(options)})")
else:
while True:
raw = input(f"{prompt}(1~{len(options)},直接回车不选): ").strip()
if not raw:
return []
try:
idx = int(raw)
except ValueError:
print(" 请输入数字编号")
continue
if 1 <= idx <= len(options):
return [options[idx - 1]]
print(f" 编号超出范围(1~{len(options)})")
def _print_profiles(profiles: List[Dict]):
if not profiles:
print(" (当前没有任何接入点。旧版 api 节仍可用,或选 2 添加)")
return
for p in profiles:
mark = " ⭐默认" if p["is_default"] else ""
key_desc = f"环境变量 {p['api_key_env']}" if p["api_key_env"] else "免密"
fb = f",失败转移 → {', '.join(p['fallback'])}" if p["fallback"] else ""
print(f" {p['name']}{mark} [{p['display_name']}]")
print(f" 地址: {p['api_url']} 模型: {p['model']}")
print(f" Key: {key_desc} 温度: {p['temperature'] or '默认'} 超时: {p['timeout'] or '默认'}{fb}")
def _add_flow(config: ConfigManager):
print("\n── 添加接入点 ──")
print("常见厂商端点参考:")
for k, v in PROVIDER_HINTS.items():
print(f" {k}: {v}")
name = _ask("档案名(英文,如 deepseek/kimi/local)")
if not name:
print("❌ 已取消(档案名不能为空)")
return
if not PROFILE_NAME_RE.match(name):
print(f"❌ 档案名只能包含字母/数字/下划线/短横线")
return
if name in get_models(config):
print(f"❌ 档案名 '{name}' 已存在")
return
api_url = _ask("API 地址")
model = _ask("模型名(如 deepseek-chat / qwen-plus)")
print("Key 方式:")
print(" 1. 环境变量名(推荐,如 DEEPSEEK_API_KEY——Key 不落盘,重启不丢)")
print(" 2. 免密本地端点(如 Ollama,不需要 Key)")
key_mode = _ask("选择 [1/2]", "1")
api_key_env = ""
if key_mode == "1":
api_key_env = _ask("环境变量名", "DEEPSEEK_API_KEY")
elif key_mode != "2":
print("❌ 已取消(Key 方式无效)")
return
display_name = _ask("显示名(可选,回车用档案名)")
temperature_raw = _ask("温度 0.0~2.0(可选,越低越确定)")
timeout_raw = _ask("超时秒数(可选)")
reasoning = _ask_yes_no("启用思考模式?", default=False)
profiles = list_profiles(config)
fallback = []
if profiles:
print("失败转移:主接入点故障时按顺序尝试的备用接入点(可选)")
fallback = _choose("选择备用接入点", [p["name"] for p in profiles], multi=True)
set_default = _ask_yes_no("设为默认接入点?", default=(len(profiles) == 0))
print("\n── 确认 ──")
print(f" 档案名: {name}")
print(f" 地址: {api_url} 模型: {model}")
print(f" Key: {'环境变量 ' + api_key_env if api_key_env else '免密(不携带)'}")
if display_name:
print(f" 显示名: {display_name}")
if temperature_raw:
print(f" 温度: {temperature_raw}")
if timeout_raw:
print(f" 超时: {timeout_raw}s")
if reasoning:
print(" 思考模式: 开启")
if fallback:
print(f" 失败转移: {', '.join(fallback)}")
if not _ask_yes_no("确认保存?", default=True):
print("已取消")
return
msgs = add_profile(config, name, api_url, model,
api_key_env=api_key_env, display_name=display_name,
temperature=float(temperature_raw) if temperature_raw else None,
timeout=int(timeout_raw) if timeout_raw else None,
reasoning_enabled=reasoning, fallback=fallback,
set_default=set_default)
for m in msgs:
print(m)
def _delete_flow(config: ConfigManager):
profiles = list_profiles(config)
if not profiles:
print("\n(当前没有任何接入点)")
return
print("\n── 删除接入点 ──")
picked = _choose("选择要删除的接入点", [p["name"] for p in profiles])
if not picked:
print("已取消")
return
name = picked[0]
if not _ask_yes_no(f"确认删除 '{name}'?(其他档案的失败转移引用会自动清理)", default=False):
print("已取消")
return
for m in delete_profile(config, name):
print(m)
def _test_flow(config: ConfigManager):
profiles = list_profiles(config)
if not profiles:
print("\n(当前没有任何接入点)")
return
print("\n── 连通性验证 ──")
picked = _choose("选择要验证的接入点", [p["name"] for p in profiles])
if not picked:
print("已取消")
return
print(" 正在请求端点(最长 20 秒)...")
print(test_profile(config, picked[0]))
def run_interactive(config: ConfigManager = None):
"""交互式主循环。config 可注入(测试用)。"""
config = config or ConfigManager()
config_path = getattr(config, "_config_path", None) or "(默认位置)"
print("=" * 56)
print(" aibox 大模型接入点管理工具")
print(f" 配置文件: {config_path}")
print(" 说明:API Key 只存环境变量名,不落盘;免密端点适合 Ollama 等本地服务")
print("=" * 56)
while True:
print("\n请选择操作:")
print(" 1. 📋 列出接入点")
print(" 2. ➕ 添加接入点")
print(" 3. 🗑️ 删除接入点")
print(" 4. 🔍 验证接入点连通性")
print(" 0. 退出")
choice = input("> ").strip()
if choice == "1":
_print_profiles(list_profiles(config))
elif choice == "2":
_add_flow(config)
elif choice == "3":
_delete_flow(config)
elif choice == "4":
_test_flow(config)
elif choice in ("0", "q", "exit", "退出"):
print("再见")
return
else:
print("无效选择,请输入 0~4")
def main():
parser = argparse.ArgumentParser(description="aibox 大模型接入点管理工具")
parser.add_argument("--list", action="store_true", help="非交互:列出当前接入点")
args = parser.parse_args()
config = ConfigManager()
if args.list:
_print_profiles(list_profiles(config))
return
run_interactive(config)
if __name__ == "__main__":
main()
+198
View File
@@ -0,0 +1,198 @@
"""aibox 模型接入点管理工具(model_config.py)单元测试(2026-08-18)
覆盖:
- 列表归一化(默认标记/免密/失败转移展示字段)
- 添加:字段校验(档案名/地址/模型名/重复)、Key 两种方式、温度/超时钳制、
失败转移引用校验、设为默认、落盘(含脱敏)
- 删除:物理移除、fallback 悬空引用清理、默认档案标记清除
- 连通性验证:成功 / 失败 / 档案不存在
"""
import sys
import types
from unittest import mock
import pytest
if "websockets" not in sys.modules:
sys.modules["websockets"] = types.ModuleType("websockets")
import model_config # noqa: E402
from common import ConfigManager # noqa: E402
def _cfg(tmp_path, initial: dict = None) -> ConfigManager:
"""构造指向临时配置文件的 ConfigManager(避免碰真实 ~/.aibox)"""
cm = ConfigManager()
cm._config_path = str(tmp_path / "config.json")
cm._config = initial if initial is not None else {}
return cm
def _base_models():
return {
"models": {
"deepseek": {
"api_url": "https://api.deepseek.com/v1/chat/completions",
"model": "deepseek-chat",
"api_key_env": "DEEPSEEK_API_KEY",
},
"local": {
"api_url": "http://localhost:11434/v1/chat/completions",
"model": "qwen3",
"api_key_env": "",
},
},
"api": {"default_model": "deepseek"},
}
# ==================== 列表 ====================
class TestListProfiles:
def test_list_normalized(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
profiles = model_config.list_profiles(cm)
assert len(profiles) == 2
ds = next(p for p in profiles if p["name"] == "deepseek")
assert ds["is_default"] is True
assert ds["api_key_env"] == "DEEPSEEK_API_KEY"
local = next(p for p in profiles if p["name"] == "local")
assert local["api_key_env"] == ""
assert local["is_default"] is False
def test_list_empty_models_section(self, tmp_path):
cm = _cfg(tmp_path, {})
assert model_config.list_profiles(cm) == []
def test_list_ignores_malformed_entries(self, tmp_path):
cm = _cfg(tmp_path, {"models": {"bad": "not-a-dict"}})
assert model_config.list_profiles(cm) == []
# ==================== 校验 ====================
class TestValidateNewProfile:
def test_name_rules(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
assert model_config.validate_new_profile(cm, "", "https://x", "m")
assert model_config.validate_new_profile(cm, "坏名字!", "https://x", "m")
assert model_config.validate_new_profile(cm, "deepseek", "https://x", "m") # 已存在
assert model_config.validate_new_profile(cm, "ok_name", "https://x", "m") is None
def test_url_scheme_and_model_required(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
assert model_config.validate_new_profile(cm, "kimi", "ftp://x", "m")
assert model_config.validate_new_profile(cm, "kimi", "https://x", "")
# ==================== 添加 ====================
class TestAddProfile:
def test_add_with_env_key_and_default(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
msgs = model_config.add_profile(
cm, "kimi", "https://api.moonshot.cn/v1/chat/completions",
"moonshot-v1-8k", api_key_env="KIMI_KEY",
display_name="Kimi", set_default=True)
assert any("已添加" in m for m in msgs)
assert any("设为默认" in m for m in msgs)
assert cm.get("api.default_model") == "kimi"
assert cm.get("models.kimi.api_key_env") == "KIMI_KEY"
assert cm.get("models.kimi.api_url").startswith("https://")
# 落盘文件存在且可解析
import json
disk = json.loads((tmp_path / "config.json").read_text(encoding="utf-8"))
assert disk["models"]["kimi"]["model"] == "moonshot-v1-8k"
def test_add_without_key_is_exempt_endpoint(self, tmp_path):
cm = _cfg(tmp_path, {})
msgs = model_config.add_profile(
cm, "local", "http://localhost:11434/v1/chat/completions",
"qwen3", api_key_env="")
assert any("免密" in m for m in msgs)
assert cm.get("models.local.api_key_env") == ""
def test_add_rejects_invalid_fallback(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
msgs = model_config.add_profile(
cm, "kimi", "https://x/v1", "m1",
fallback=["ghost", "deepseek"])
assert any("失败转移" in m for m in msgs)
assert cm.get("models.kimi") is None # 未写入
def test_add_temperature_timeout_clamped(self, tmp_path):
cm = _cfg(tmp_path, {})
model_config.add_profile(cm, "a", "https://x/v1", "m1",
temperature=9.9, timeout=1)
assert cm.get("models.a.temperature") == 2.0
assert cm.get("models.a.timeout") == 5
def test_add_invalid_temperature_rejected(self, tmp_path):
cm = _cfg(tmp_path, {})
msgs = model_config.add_profile(cm, "a", "https://x/v1", "m1",
temperature="abc")
assert any("温度" in m for m in msgs)
assert cm.get("models.a") is None
# ==================== 删除 ====================
class TestDeleteProfile:
def test_delete_cleans_fallback_refs(self, tmp_path):
cm = _cfg(tmp_path, {
"models": {
"main": {"api_url": "https://a", "model": "m1",
"fallback": ["backup"]},
"backup": {"api_url": "https://b", "model": "m2"},
},
})
msgs = model_config.delete_profile(cm, "backup")
assert any("已删除" in m for m in msgs)
assert any("失败转移引用" in m for m in msgs)
assert cm.get("models.main.fallback") == []
def test_delete_default_clears_default_flag(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
model_config.delete_profile(cm, "deepseek")
assert cm.get("api.default_model") is None
def test_delete_nonexistent(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
msgs = model_config.delete_profile(cm, "ghost")
assert any("不存在" in m for m in msgs)
# ==================== 连通性验证 ====================
class TestTestProfile:
def test_connectivity_ok(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
fake = mock.MagicMock()
fake.chat_completion.return_value = "pong"
with mock.patch("llm_client.create_llm_client", return_value=fake):
r = model_config.test_profile(cm, "deepseek")
assert "连通正常" in r and "pong" in r
def test_connectivity_failure(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
fake = mock.MagicMock()
fake.chat_completion.side_effect = RuntimeError("连接被拒绝")
with mock.patch("llm_client.create_llm_client", return_value=fake):
r = model_config.test_profile(cm, "deepseek")
assert "验证失败" in r and "连接被拒绝" in r
def test_connectivity_missing_profile(self, tmp_path):
cm = _cfg(tmp_path, _base_models())
assert "不存在" in model_config.test_profile(cm, "ghost")
# ==================== 落盘失败感知 ====================
class TestSaveFailure:
def test_add_reports_save_failure(self, tmp_path):
cm = _cfg(tmp_path, {})
with mock.patch("builtins.open", side_effect=PermissionError("disk ro")):
msgs = model_config.add_profile(
cm, "a", "https://x/v1", "m1", api_key_env="K")
assert any("保存失败" in m for m in msgs)