From d513c5f8924e624762fe1c5e872fec2b629ad1e9 Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Tue, 18 Aug 2026 03:26:53 +0000 Subject: [PATCH] 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. --- aibox/README.md | 20 +- aibox/model_config.py | 415 +++++++++++++++++++++++++++++++ aibox/tests/test_model_config.py | 198 +++++++++++++++ 3 files changed, 631 insertions(+), 2 deletions(-) create mode 100644 aibox/model_config.py create mode 100644 aibox/tests/test_model_config.py diff --git a/aibox/README.md b/aibox/README.md index b498645..bc6f685 100644 --- a/aibox/README.md +++ b/aibox/README.md @@ -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 diff --git a/aibox/model_config.py b/aibox/model_config.py new file mode 100644 index 0000000..1e2abe5 --- /dev/null +++ b/aibox/model_config.py @@ -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() diff --git a/aibox/tests/test_model_config.py b/aibox/tests/test_model_config.py new file mode 100644 index 0000000..7132834 --- /dev/null +++ b/aibox/tests/test_model_config.py @@ -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)