diff --git a/CODEBUDDY.md b/CODEBUDDY.md index 8316754..8f15c80 100644 --- a/CODEBUDDY.md +++ b/CODEBUDDY.md @@ -27,16 +27,32 @@ cmake --build --preset R243 --config Release ### 测试 -工程内**没有单元测试框架**。所谓“测试”是根目录下的 Python 脚本，它们是 MCP 客户端的联调工具，**必须先启动已加载插件的 CAD，并确认 8080 端口在监听**： +分两层，**互不替代**： + +**快层 —— pytest 离线单元测试**（`tests/`，无需 CAD、不联网、秒级）： ```bash -python test.py # 最小示例：截图 -> 发给 DeepSeek 视觉模型 -> 回答 -python test_autoagent.py # 仅拉取一次视口截图并保存为 test_screenshot.png -python countOfCircle.py # 伪造一次 zoom_window_normalized 调用，验证 TCP 链路 -python agent_loop.py # 完整 Agent 循环（多回合 截图/缩放/结论），需 openai 包与 API Key +pip install -r requirements-dev.txt # 仅 pytest +python -m pytest # 期望全绿 ``` -脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。`agent_loop.py` / `test.py` 中硬编码了 DeepSeek API Key 与模型名，属于调试脚本。 +全部使用假 LLM + 假 MCP 基座（`tests/fakes.py`），锁定 `dwg_review` 流水线既有行为。 +配置见 `pytest.ini`（`testpaths = tests`）与根 `conftest.py`（把仓库根加入 `sys.path`， +并提供 `RENDER_WAIT=0` / `REPORT_DIR=tmp` 两个 fixture）。**注意**：`dwg_review.config` +的常量在导入期定值，测试改 `os.environ` 无效，须 `monkeypatch.setattr(config, ...)`。 + +**慢层 —— 端到端冒烟脚本**（仓库根 `verify_*.py`，**必须先启动已加载插件的 CAD，并确认 +8080 端口在监听**）： + +```bash +python verify_mcp_frame.py # 基座最小集：分帧 / id / isError 映射 +python verify_mcp_tcp_hardening.py # TCP 收发路径加固（超长帧 / 粘包 / 半关闭） +python verify_mcp_fault_injection.py # 工具异常保护（类型错误不再崩 CAD） +python verify_stdio_bridge.py # stdio ↔ TCP 桥（协议协商 / 转发 / 通知） +python verify_annotate.py # 批注回写链路 +``` + +这些脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。 ## 架构 diff --git a/agent_loop.py b/agent_loop.py deleted file mode 100644 index 0e66e39..0000000 --- a/agent_loop.py +++ /dev/null @@ -1,163 +0,0 @@ -import socket -import json -import time -from openai import OpenAI - -# 初始化 DeepSeek 客户端 -client = OpenAI( - api_key="sk-420190f448fe41158c4e2ccff90e35ce", - base_url="https://api.deepseek.com" -) -MODEL_NAME = "deepseek-chat" - -def fetch_and_translate_tools(): - """向 CAD 基座发送 tools/list，并转换为大模型认识的格式""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - - # 1. 向 MCP Server 发送标准的 tools/list 请求 - req = { - "jsonrpc": "2.0", - "id": 100, - "method": "tools/list" - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - - response_data = cad_socket.recv(8192) - cad_socket.close() - - mcp_response = json.loads(response_data.decode('utf-8')) - - if "result" not in mcp_response or "tools" not in mcp_response["result"]: - print("[Error] 无法从 CAD 获取工具列表") - return [] - - llm_tools = [] - # 2. 遍历 CAD 返回的工具，将其翻译为 DeepSeek/OpenAI 格式 - for mcp_tool in mcp_response["result"]["tools"]: - llm_tools.append({ - "type": "function", - "function": { - "name": mcp_tool["name"], - "description": mcp_tool["description"], - # 核心转换：MCP 的 inputSchema 等价于 LLM 的 parameters - "parameters": mcp_tool["inputSchema"] - } - }) - - print(f"[Init] 成功从 CAD 动态加载了 {len(llm_tools)} 个工具！") - return llm_tools - - except Exception as e: - print(f"[Error] 获取工具列表失败: {e}") - return [] - -def call_cad_mcp_server(tool_name, arguments): - """底层 TCP 通信保持不变""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - req = { - "jsonrpc": "2.0", "id": 1, "method": "tools/call", - "params": { "name": tool_name, "arguments": arguments } - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - response_data = b"" - while True: - chunk = cad_socket.recv(8192) - response_data += chunk - if len(chunk) < 8192: break - cad_socket.close() - return json.loads(response_data.decode('utf-8')) - except Exception as e: - print(f"[Error] 连接 CAD 基座失败: {e}") - return None - -def run_agent(user_prompt): - print("=== 🚀 CAD 具身智能 Agent 启动 ===") - - # 动态向 CAD 请求可用工具，彻底解耦！ - tools = fetch_and_translate_tools() - - if not tools: - print("没有可用的工具，Agent 退出。") - return - - # 核心心智设定：教它怎么形成视觉闭环 - system_prompt = """你是一个高阶 AutoCAD 视觉检查专家。 -请严格遵循以下工作流： -1. 第一步永远是调用 `get_viewport_screenshot` 观察当前画面。 -2. 仔细评估图像。如果目标物体（如图元、文字）太小导致无法确信细节，请调用 `zoom_window_normalized` 放大该局部区域。 -3. 【关键指令】：每次执行缩放操作（zoom）后，你必须在下一回合再次调用 `get_viewport_screenshot` 获取放大后的新截图！ -4. 反复观察和放大，直到你 100% 看清细节，再用自然语言向用户输出最终结论。""" - - messages = [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_prompt} - ] - - MAX_TURNS = 8 # 熔断机制：最多允许思考 8 个回合，防止无限套娃 - - for turn in range(MAX_TURNS): - print(f"\n--- [第 {turn + 1} 回合] 思考中 ---") - - response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages, - tools=tools - ) - - assistant_message = response.choices[0].message - messages.append(assistant_message) # 记录思维路径 - - # 1. 检查是否需要执行物理动作 - if assistant_message.tool_calls: - for tool_call in assistant_message.tool_calls: - tool_name = tool_call.function.name - args = json.loads(tool_call.function.arguments) - - print(f"[动作决定] ⚡ 调用工具: {tool_name}") - if args: print(f" 参数: {args}") - - # 执行 CAD 通信 - cad_res = call_cad_mcp_server(tool_name, args) - time.sleep(0.5) # 给予 CAD 渲染刷新窗口的时间缓冲 - - # 处理执行结果，并准备发回给 LLM 的观测报告 - tool_result_content = [] - - if cad_res and "result" in cad_res: - for item in cad_res["result"]["content"]: - if item["type"] == "text": - tool_result_content.append({"type": "text", "text": item["text"]}) - elif item["type"] == "image": - base64_img = item["data"] - mime = item.get("mimeType", "image/png") - tool_result_content.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{base64_img}"} - }) - print("[观测反馈] 📸 已将最新 CAD 屏幕画面传回视觉中枢。") - else: - tool_result_content.append({"type": "text", "text": "CAD 工具执行失败或无响应。"}) - - # 必须向大模型提交 Tool 返回结果 - messages.append({ - "role": "tool", - "tool_call_id": tool_call.id, - "content": tool_result_content - }) - - # 2. 如果不调工具，说明任务已完成，输出最终自然语言结论 - else: - print("\n==================================") - print(f"🎯 [最终结论]:\n{assistant_message.content}") - print("==================================") - break - - else: - print("\n[警告] 达到最大思考回合数限制，Agent 强行中止。") - -if __name__ == "__main__": - run_agent("请仔细检查当前图纸，告诉我图纸中心那些小圆的内部，是否还包含了更小的同心圆？如果看不清，请务必放大确认。") \ No newline at end of file diff --git a/ai.memory/MEMORY.md b/ai.memory/MEMORY.md index ba6012c..afade36 100644 --- a/ai.memory/MEMORY.md +++ b/ai.memory/MEMORY.md @@ -5,6 +5,7 @@ > 组织方式：本文件是索引（一行一条），各主题内容放在独立文件中。 - [CADProject 图纸审查 Agent 工作](project_dwg_review_agent.md) — 基于 MCP 框架的图纸审查工具集与逐图框审查流程 +- [DeepSeek API Key 泄露与轮换（已处理）](project_api_key_rotation.md) — 旧 Key 曾入 git 历史；2026-10-14 已吊销并换新 Key，验证可用 - [LLM 识图审图 Agent 架构决策](project_review_agent_impl.md) — 混合式自研 agent、基座 B 最小集、标准化先走 stdio 桥；含 dwg_review 包结构 - [stdio↔TCP 桥（MCP 标准化）](project_stdio_bridge.md) — mcp_bridge 包：基座协议缺口、读线程+单工作线程设计、MCP Inspector 验证方法与踩坑 - [MCP 基座 c++_dll 加载与执行的坑](project_mcp_frame_dll_loader.md) — acrxGetApiVersion 有状态不可比对、写库需文档锁、同一 DLL 只加载一次 diff --git a/ai.memory/project_api_key_rotation.md b/ai.memory/project_api_key_rotation.md new file mode 100644 index 0000000..1ce59c6 --- /dev/null +++ b/ai.memory/project_api_key_rotation.md @@ -0,0 +1,29 @@ +--- +name: DeepSeek API Key 泄露与轮换（已处理） +description: 旧 Key 曾硬编码并进入 git 历史；2026-10-14 已由用户在控制台吊销并换新 Key，验证可用 +type: project +--- + +# DeepSeek API Key 泄露与轮换（已处理） + +**状态：已处理（2026-10-14）。** + +**经过**：DeepSeek API Key 曾被**硬编码**在三个根目录脚本里 +（`agent_loop.py:8`、`countOfCircle.py:8`、`dwg_review_agent.py:38`），随提交 `bdd77af` +进入 **git 历史并推送到远程 Gitea**（`gjm/cad-agent`）。三个脚本已于 2026-10-14 删除。 + +**处置（2026-10-14，用户操作）**：用户已在 DeepSeek 控制台**吊销旧 Key**，并在 +Windows **系统级环境变量** `DEEPSEEK_API_KEY` 中设置新 Key。AI 侧已验证： +`config._read_env("DEEPSEEK_API_KEY")` 读到非空值（35 字符），且一次真实 +`chat_text` 调用成功返回 → 新 Key 可用。 + +**Why（保留此记录的理由）**：旧 Key 仍留在 git 历史（`bdd77af`）里，但**已被吊销**， +残留风险很低；删除文件从来不能清除历史，真正生效的处置是**轮换**而非删除。 + +**How to apply**： +- 不要再把 Key 写进代码：只经环境变量 `DEEPSEEK_API_KEY` 注入； + `dwg_review/config.py` 已实现「环境变量读不到时回退读注册表」（因新设的系统变量 + 不会传播到已启动进程）。 +- 若日后仍要从 git 历史中彻底抹除旧 Key（`git filter-repo` / BFG + 强推），属于 + **破坏性且影响共享远程**的操作，**必须单独取得用户明确同意**后才可执行。 +- 复核历史残留：`git show bdd77af`（或 `git log --all -S "<旧 key 片段>"`）。 diff --git a/ai.memory/project_dwg_review_agent.md b/ai.memory/project_dwg_review_agent.md index f2b598c..aa706bf 100644 --- a/ai.memory/project_dwg_review_agent.md +++ b/ai.memory/project_dwg_review_agent.md @@ -100,8 +100,10 @@ TCP 线程只 `recv`，把请求封装成 `TaskData` 后 `PostMessage(WM_MCP_EXE ### 审查脚本与清单 -- `dwg_review_agent.py`：逐图框审查流程。`MODEL_NAME`（默认 `deepseek-flash`）、 - `CHECKLIST_PATH`（默认 `review_checklist.json`）。 +> 注：原单文件脚本 `dwg_review_agent.py` 已由 `dwg_review/` 包取代（见 +> [LLM 识图审图 Agent 架构决策](project_review_agent_impl.md)），并于 2026-10-14 删除。 + +- 逐图框审查流程：`MODEL_NAME`（默认 `deepseek-flash`）、`CHECKLIST_PATH`（默认 `review_checklist.json`）。 流程：`detect_drawing_frames` → 每框 `zoom_to_window` + `query_entities(window)` + `get_text_content(window)` + `get_viewport_screenshot` → LLM 审查 → 汇总。 - `review_checklist.json`：**可配置审查清单**，5 大类 @@ -144,6 +146,7 @@ TCP 线程只 `recv`，把请求封装成 `TaskData` 后 `PostMessage(WM_MCP_EXE ## 遗留 / 待办 -1. `dwg_review_agent.py` 仍硬编码 DeepSeek API Key（已推送远程），建议改为环境变量 `DEEPSEEK_API_KEY`。 +1. ~~`dwg_review_agent.py` 仍硬编码 DeepSeek API Key~~ —— 该脚本已于 2026-10-14 删除； + 但 Key 已随提交 `bdd77af` 进入 git 历史并推到远程，**必须去 DeepSeek 控制台轮换**。 2. `review_checklist.json` 可补充「检查方法 / 阈值」；可加跨图框统计工具；可把清单移入 MCP 侧。 3. 后期整理建筑/工业管道等审图规范并接入，让 LLM 参照规范审核。 diff --git a/ai.memory/project_review_agent_impl.md b/ai.memory/project_review_agent_impl.md index d3a8966..38392bd 100644 --- a/ai.memory/project_review_agent_impl.md +++ b/ai.memory/project_review_agent_impl.md @@ -40,7 +40,12 @@ type: project （唯一 TCP 封装：单飞锁 + 换行分帧 + 重试 + 兼容 `error`/`isError`）、`llm_client.py`、 `cad_tools.py`（动态工具表 + 只读白名单 `READ_ONLY_TOOLS`）、`checklist.py`、 `phases/{planning,evidence,review,report}.py`、`annotator.py`、`__main__.py`。 - **取代**根目录散落的 `dwg_review_agent.py` / `agent_loop.py` 等重复脚本。 + **取代**根目录散落的 `dwg_review_agent.py` / `agent_loop.py` 等重复脚本 + （这 5 个遗留脚本已于 2026-10-14 删除，见 [环境与外部资源](reference_environment.md)）。 +- `tests/`（pytest 回归网）：假 LLM + 假 MCP 基座的**离线快层**，`tests/fakes.py` 提供 + `FakeLlmClient` / `FakeMcpClient` / `make_ctx`。锁定 `dwg_review` 既有行为。 + **陷阱**：绝不构造真 `LlmClient`（无 key 抛错）、无参 `CadTools()` 或 `ReviewContext.create()` + （构造即 `list_tools()` 触网）；一律 `make_ctx()` 注入假件。 - `verify_mcp_frame.py`：基座改动验证脚本（分帧 / 大报文 / string id / 通知 / isError）。 ## 环境事实 diff --git a/ai.memory/reference_environment.md b/ai.memory/reference_environment.md index b97b48f..7a18f43 100644 --- a/ai.memory/reference_environment.md +++ b/ai.memory/reference_environment.md @@ -1,6 +1,6 @@ --- name: 环境与外部资源 -description: CadProject 的构建/运行要点、CAD SDK 路径、测试脚本等外部资源指引 +description: CadProject 的构建/运行要点、CAD SDK 路径、测试分层（pytest 快层 + verify_* 慢层）等外部资源指引 type: reference --- @@ -26,16 +26,28 @@ type: reference - 先启动 **已加载插件的 CAD**，确认 `127.0.0.1:8080` 在监听。 - 脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。 -## 测试脚本（根目录，非单元测试） +## 测试（两层，互不替代） + +**快层 —— pytest 离线单元测试**（`tests/`，假 LLM + 假基座，无需 CAD、不联网、秒级）： + +- `pip install -r requirements-dev.txt` → `python -m pytest`。 +- 配置：`pytest.ini`（`testpaths = tests`）、根 `conftest.py`（加 `sys.path` + `RENDER_WAIT=0` / `REPORT_DIR=tmp` fixture）。 +- 陷阱：`dwg_review.config` 常量导入期定值，改 `os.environ` 无效，须 `monkeypatch.setattr(config, ...)`。 + +**慢层 —— 端到端冒烟脚本**（根目录 `verify_*.py`，需 CAD 在线且 `127.0.0.1:8080` 监听）： | 脚本 | 作用 | | --- | --- | -| `test.py` | 截图 → 发 DeepSeek 视觉模型 → 回答 | -| `test_autoagent.py` | 拉一次视口截图存 `test_screenshot.png` | -| `countOfCircle.py` | 伪造 `zoom_window_normalized` 调用，验证 TCP 链路 | -| `agent_loop.py` | 完整 Agent 循环（多回合），需 openai 包与 API Key | +| `verify_mcp_frame.py` | 基座最小集：分帧 / id / isError 映射 | +| `verify_mcp_tcp_hardening.py` | TCP 收发路径加固（超长帧 / 粘包 / 半关闭） | +| `verify_mcp_fault_injection.py` | 工具异常保护（类型错误不再崩 CAD） | +| `verify_stdio_bridge.py` | stdio ↔ TCP 桥（协议协商 / 转发 / 通知） | | `verify_annotate.py` | 批注工具运行时验证（绘制/校验/改属性/清理）；`--keep` 保留批注 | +> 历史：根目录 `test.py` / `test_autoagent.py` / `countOfCircle.py` / `agent_loop.py` / +> `dwg_review_agent.py` 五个遗留调试脚本已于 2026-10-14 删除（前三个含硬编码 API Key， +> 后两个是重复实现）。`test_screenshot.png` 是 `test_autoagent.py` 的残留产物。 + ## 两个构建目标 | 目标 | 产物 | 作用 | diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..dc9becc --- /dev/null +++ b/conftest.py @@ -0,0 +1,31 @@ +"""pytest 全局配置（位于仓库根，保证 `import dwg_review` 可用）。 + +注意：`dwg_review.config` 的常量在**导入期**就已定值，改 `os.environ` 不会生效， +所以测试里一律用 `monkeypatch.setattr(config, "XXX", ...)` 覆盖模块属性。 +""" + +import pathlib +import sys + +import pytest + +_ROOT = pathlib.Path(__file__).resolve().parent +if str(_ROOT) not in sys.path: + sys.path.insert(0, str(_ROOT)) + + +@pytest.fixture(autouse=True) +def _no_render_wait(monkeypatch): + """消除 collect_frame 里每个图框的 RENDER_WAIT 等待（单元测试不需要真实渲染）。""" + from dwg_review import config + + monkeypatch.setattr(config, "RENDER_WAIT", 0) + + +@pytest.fixture +def tmp_reports(tmp_path, monkeypatch): + """把报告输出目录指向临时目录，避免测试污染仓库的 reports/。""" + from dwg_review import config + + monkeypatch.setattr(config, "REPORT_DIR", tmp_path) + return tmp_path diff --git a/countOfCircle.py b/countOfCircle.py deleted file mode 100644 index 6defc86..0000000 --- a/countOfCircle.py +++ /dev/null @@ -1,127 +0,0 @@ -import socket -import json -from openai import OpenAI - -# 1. 初始化 DeepSeek 客户端 -# 注意：务必确保已经执行过 pip install openai -client = OpenAI( - api_key="sk-420190f448fe41158c4e2ccff90e35ce", - base_url="https://api.deepseek.com" # 核心修改：将网关指向 DeepSeek 服务器 -) - -# 核心修改：使用 DeepSeek 的主模型。 -# (注: DeepSeek-V4/V4.1 的视觉支持已集成，具体模型名请以你 DeepSeek 后台显示的可用模型为准) -MODEL_NAME = "deepseek-flash" - -def call_cad_mcp_server(tool_name, arguments): - """封装与 AutoCAD MCP 基座的 TCP 通信""" - try: - cad_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cad_socket.connect(("127.0.0.1", 8080)) - - req = { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": { - "name": tool_name, - "arguments": arguments - } - } - cad_socket.sendall(json.dumps(req).encode('utf-8')) - - response_data = b"" - while True: - chunk = cad_socket.recv(8192) - response_data += chunk - if len(chunk) < 8192: - break - - cad_socket.close() - return json.loads(response_data.decode('utf-8')) - except Exception as e: - print(f"[Error] 连接 CAD 基座失败: {e}") - return None - -def main(): - print("=== DeepSeek CAD Vision Agent 启动 ===") - - # 2. 定义 MCP 工具 (Tool Calling) - tools = [ - { - "type": "function", - "function": { - "name": "get_viewport_screenshot", - "description": "获取当前 AutoCAD 视口的实时截图。当用户询问图纸上的视觉特征、数量或位置时，必须先调用此工具获取画面。" - } - } - ] - - # 3. 初始化历史 - messages = [ - {"role": "system", "content": "你是一个专业的 AutoCAD 视觉审查助手。必须先使用截图工具观察图纸，再回答用户问题。"}, - {"role": "user", "content": "请看看当前 CAD 屏幕上，我一共画了几个圆？"} - ] - - print("\n[DeepSeek] 正在思考如何完成任务...") - - # === 第一回合：DeepSeek 思考并决定调用工具 === - response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages, - tools=tools - ) - - assistant_message = response.choices[0].message - messages.append(assistant_message) - - # 检查 DeepSeek 是否发起了 Tool Call - if assistant_message.tool_calls: - tool_call = assistant_message.tool_calls[0] - tool_name = tool_call.function.name - - print(f"[DeepSeek] 决定调用 CAD 工具: {tool_name}") - print(f"[CAD] 正在执行截图，请稍候...") - - # === 与 CAD 基座通信 === - cad_res = call_cad_mcp_server(tool_name, {}) - - if cad_res and "result" in cad_res: - content_array = cad_res["result"]["content"] - base64_img = "" - mime_type = "image/png" - for item in content_array: - if item["type"] == "image": - base64_img = item["data"] - mime_type = item.get("mimeType", "image/png") - - print("[CAD] 截图成功！正在将视觉数据传回大模型神经中枢...") - - # 4. 将 Base64 图片按标准多模态格式塞回历史 - messages.append({ - "role": "tool", - "tool_call_id": tool_call.id, - "content": [ - {"type": "text", "text": "这是 AutoCAD 当前界面的实时截图。请根据图片回答用户的问题。"}, - {"type": "image_url", "image_url": {"url": f"data:{mime_type};base64,{base64_img}"}} - ] - }) - - # === 第二回合：DeepSeek “看”图并回答 === - print("[DeepSeek] 正在进行视觉分析...") - final_response = client.chat.completions.create( - model=MODEL_NAME, - messages=messages - ) - - print("\n==================================") - print(f"[DeepSeek 最终回答]:\n{final_response.choices[0].message.content}") - print("==================================") - - else: - print("[Error] CAD 工具执行失败。") - else: - print(f"[DeepSeek 盲猜]: {assistant_message.content}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/dwg_review_agent.py b/dwg_review_agent.py deleted file mode 100644 index ed8461e..0000000 --- a/dwg_review_agent.py +++ /dev/null @@ -1,310 +0,0 @@ -""" -dwg_review_agent.py -==================== -逐图框图纸审查 Agent（联调脚本） - -流程： - 1. detect_drawing_frames 识别图纸中的所有图框 - 2. 对每个图框： - a. zoom_to_window 缩放到该图框范围 - b. query_entities 取图框内图元摘要（按 window 过滤） - c. get_text_content 取图框内文字 - d. get_viewport_screenshot 截取当前视口 - e. 交给 LLM 审查，产出该图框的审查意见 - 3. 汇总所有图框意见，生成总报告 - -依赖： - pip install openai -环境变量： - DEEPSEEK_API_KEY （不设置时回退到脚本内默认值） - DEEPSEEK_MODEL （默认 deepseek-chat） - -使用前提： - CAD 已打开并加载了 cad_mcp_frame.arx，8080 端口在监听。 -""" - -import os -import sys -import json -import time -import socket - -from openai import OpenAI - -HOST = "127.0.0.1" -PORT = 8080 - -# 与仓库既有脚本保持一致；生产环境请改为从环境变量注入 -API_KEY = os.environ.get("DEEPSEEK_API_KEY", "sk-420190f448fe41158c4e2ccff90e35ce") -BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com") -MODEL_NAME = os.environ.get("DEEPSEEK_MODEL", "deepseek-flash") # 该模型已确认支持视觉 - -# 每个图框审查时，取数据的条数上限 -FRAME_ENTITY_LIMIT = int(os.environ.get("FRAME_ENTITY_LIMIT", "200")) -# 最多审查多少个图框（0 = 全部） -MAX_FRAMES = int(os.environ.get("MAX_FRAMES", "0")) -# 审查清单文件路径 -CHECKLIST_PATH = os.environ.get("REVIEW_CHECKLIST", "review_checklist.json") - -client = OpenAI(api_key=API_KEY, base_url=BASE_URL) - - -# --------------------------------------------------------------------------- -# MCP / TCP 通信 -# --------------------------------------------------------------------------- -def call_tool(tool_name, arguments, recv_timeout=30.0): - """调用 CAD 基座的 tools/call，返回解析后的 JSON-RPC 响应（失败返回 None）""" - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(recv_timeout) - sock.connect((HOST, PORT)) - req = { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/call", - "params": {"name": tool_name, "arguments": arguments}, - } - sock.sendall(json.dumps(req).encode("utf-8")) - - data = b"" - while True: - chunk = sock.recv(65536) - if not chunk: - break - data += chunk - # 基座是单次 send 返回整包；收到后尝试解析即可结束 - try: - json.loads(data.decode("utf-8")) - break - except json.JSONDecodeError: - continue - sock.close() - return json.loads(data.decode("utf-8")) - except Exception as e: - print(f"[Error] 调用工具 {tool_name} 失败: {e}") - return None - - -def mcp_text(resp): - """从 MCP 响应中取出文本内容；若文本是 JSON 则解析为对象返回""" - if not resp or "result" not in resp: - return None - for item in resp["result"].get("content", []): - if item.get("type") == "text": - text = item.get("text", "") - try: - return json.loads(text) - except (json.JSONDecodeError, TypeError): - return text - return None - - -def mcp_image(resp): - """从 MCP 响应中取出图片，返回 (base64, mimeType)""" - if not resp or "result" not in resp: - return None, None - for item in resp["result"].get("content", []): - if item.get("type") == "image": - return item.get("data"), item.get("mimeType", "image/png") - return None, None - - -def mcp_error(resp): - """若响应是错误，返回错误信息字符串，否则返回 None""" - if resp and "error" in resp: - err = resp["error"] - return f"code={err.get('code')} message={err.get('message')}" - return None - - -def bbox_to_window(bbox): - """[minx,miny,maxx,maxy] -> window 对象""" - return {"min_x": bbox[0], "min_y": bbox[1], "max_x": bbox[2], "max_y": bbox[3]} - - -# --------------------------------------------------------------------------- -# LLM 审查（审查清单可配置） -# --------------------------------------------------------------------------- -def load_checklist(path): - """加载审查清单 JSON；失败时回退到内置通用清单""" - try: - with open(path, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, dict) and data.get("categories"): - return data - print(f"[Warn] 审查清单 {path} 结构异常，使用内置默认清单。") - except Exception as e: - print(f"[Warn] 无法加载审查清单 {path}: {e}，使用内置默认清单。") - - return { - "name": "内置通用清单", - "categories": [ - {"title": "图层规范性", "severity": "medium", - "checks": ["图元是否在正确图层", "是否存在空图层/命名异常图层"]}, - {"title": "文字", "severity": "medium", - "checks": ["空文字", "错别字", "字高异常", "文字重叠"]}, - {"title": "标注", "severity": "high", - "checks": ["关键尺寸缺失", "标注值异常"]}, - {"title": "几何", "severity": "medium", - "checks": ["零长度/退化实体", "重复重叠实体", "未闭合轮廓"]}, - {"title": "图面整体", "severity": "low", - "checks": ["图面完整性", "内容是否超出图框"]}, - ], - } - - -def build_system_prompt(checklist): - """根据审查清单动态生成 LLM 系统提示词""" - lines = [ - "你是一名资深建筑与工业厂房图纸审查专家。", - "你将逐张收到某个图框（一张图纸）的：结构化图元数据、文字内容，以及该图框的视口截图。", - "", - f"请依据审查清单《{checklist.get('name', '审查清单')}》逐项审查，" - "指出**具体问题**（尽量引用实体 handle 或文字内容），并标注严重程度：", - "", - ] - for cat in checklist.get("categories", []): - lines.append(f"【{cat.get('title', '')}】（严重度：{cat.get('severity', 'medium')}）") - for chk in cat.get("checks", []): - lines.append(f" - {chk}") - lines.append("") - lines += [ - "要求：", - "- 只报告你**有依据**的问题，不要臆测；数据与截图冲突时以数据为准。", - "- 按清单条目组织结论，未命中的条目可省略。", - "- 若该图框无任何明显问题，明确说“未发现明显问题”。", - "- 不要复述输入数据。", - ] - return "\n".join(lines) - - -def review_frame(idx, frame, entities, texts, img_b64, mime, system_prompt): - """对单个图框调用 LLM 审查，返回审查意见文本""" - sheet = frame.get("sheet_size") or "未知图幅" - kind = frame.get("kind") - layer = frame.get("layer") - - summary = { - "图框序号": idx, - "图幅": sheet, - "形态": kind, - "图层": layer, - "包围盒": frame.get("bbox"), - "图元总数": (entities or {}).get("total") if isinstance(entities, dict) else None, - "图元(截断)": (entities or {}).get("items") if isinstance(entities, dict) else None, - "文字": (texts or {}).get("items") if isinstance(texts, dict) else None, - } - - user_content = [ - {"type": "text", "text": "以下是该图框的审查输入数据(JSON)：\n" + - json.dumps(summary, ensure_ascii=False, indent=2)} - ] - if img_b64: - user_content.append({ - "type": "image_url", - "image_url": {"url": f"data:{mime};base64,{img_b64}"}, - }) - - messages = [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_content}, - ] - - try: - resp = client.chat.completions.create(model=MODEL_NAME, messages=messages) - return resp.choices[0].message.content - except Exception as e: - return f"[LLM 审查失败] {e}" - - -# --------------------------------------------------------------------------- -# 主流程 -# --------------------------------------------------------------------------- -def main(): - print("=== 🏗️ 逐图框图纸审查 Agent 启动 ===") - - # 0. 载入可配置审查清单 - checklist = load_checklist(CHECKLIST_PATH) - system_prompt = build_system_prompt(checklist) - print(f"[Init] 审查清单：{checklist.get('name')}" - f"（{len(checklist.get('categories', []))} 类），模型：{MODEL_NAME}\n") - - # 1. 识别图框 - resp = call_tool("detect_drawing_frames", {"min_confidence": 0.3}) - err = mcp_error(resp) - if err: - print(f"[Error] 识别图框失败: {err}") - return - frames_data = mcp_text(resp) - items = frames_data.get("items", []) if isinstance(frames_data, dict) else [] - if not items: - print("未识别到任何图框，请确认图纸中存在图框，或调整 detect_drawing_frames 参数。") - return - - total = len(items) - if MAX_FRAMES > 0: - items = items[:MAX_FRAMES] - print(f"识别到 {total} 个图框，本次审查 {len(items)} 个。\n") - - findings = [] - - # 2. 逐图框审查 - for idx, frame in enumerate(items, 1): - bbox = frame.get("bbox") - if not bbox or len(bbox) != 4: - print(f"[跳过] 图框 {idx} 缺少有效包围盒") - continue - - print(f"--- 图框 {idx}/{len(items)} kind={frame.get('kind')} " - f"sheet={frame.get('sheet_size')} bbox={bbox}") - - win = bbox_to_window(bbox) - - # a. 缩放到图框 - call_tool("zoom_to_window", dict(win, margin=0.02)) - time.sleep(0.6) # 给 CAD 渲染留出时间 - - # b/c. 取图框内数据 - entities = mcp_text(call_tool("query_entities", { - "window": win, "limit": FRAME_ENTITY_LIMIT - })) - texts = mcp_text(call_tool("get_text_content", { - "window": win, "limit": FRAME_ENTITY_LIMIT - })) - - # d. 截图 - shot = call_tool("get_viewport_screenshot", {}) - img_b64, mime = mcp_image(shot) - print(f" 图元 {(entities or {}).get('total')} 个，" - f"文字 {(texts or {}).get('total')} 条，" - f"截图 {'已获取' if img_b64 else '无'}") - - # e. LLM 审查 - finding = review_frame(idx, frame, entities, texts, img_b64, mime, system_prompt) - findings.append({"frame": idx, "sheet": frame.get("sheet_size"), "finding": finding}) - print(f" 审查结论：{finding}\n") - - # 3. 汇总总报告 - print("=== 📋 汇总总报告 ===") - report_input = "\n\n".join( - f"【图框 {f['frame']}（图幅 {f['sheet']}）】\n{f['finding']}" for f in findings - ) - try: - resp = client.chat.completions.create( - model=MODEL_NAME, - messages=[ - {"role": "system", "content": - "你是图纸审查负责人。请把下面各图框的审查意见汇总为一份" - "结构化的审查报告：先给整体结论，再按图框列出问题，最后给出整改建议。" - "语言简洁、条目化。"}, - {"role": "user", "content": report_input}, - ], - ) - print(resp.choices[0].message.content) - except Exception as e: - print(f"[汇总失败] {e}") - print(report_input) - - -if __name__ == "__main__": - main() diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..9452f83 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +# 只扫描 tests/：既覆盖全部单元测试，又避开根目录的 verify_*.py（需 CAD 在线的端到端冒烟） +testpaths = tests +python_files = test_*.py +addopts = -q diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..b4cc416 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,4 @@ +# dwg_review 回归测试网（离线快层）的开发依赖 +# 安装：pip install -r requirements-dev.txt +# 运行：python -m pytest +pytest>=8.3 diff --git a/review_checklist.json b/review_checklist.json index 70978c7..242bad5 100644 --- a/review_checklist.json +++ b/review_checklist.json @@ -1,7 +1,7 @@ { "version": "1.0.0", "name": "通用图纸审查清单", - "description": "审查清单（可替换）。dwg_review_agent.py 会把 categories[].checks[] 注入 LLM 审查提示词。后期可在此追加建筑标准、工业管道标准等规范条目。", + "description": "审查清单（可替换）。dwg_review 包会把 categories[].checks[] 注入 LLM 审查提示词。后期可在此追加建筑标准、工业管道标准等规范条目。", "categories": [ { "id": "layers", diff --git a/test.py b/test.py deleted file mode 100644 index 76bd46c..0000000 --- a/test.py +++ /dev/null @@ -1,41 +0,0 @@ -import socket -import json - -def test_fake_llm_zoom(): - print("=== 模拟 LLM 发起缩放指令测试 ===") - - try: - client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - client.connect(("127.0.0.1", 8080)) - - # 伪造大模型生成的 tool_call 报文 - req = { - "jsonrpc": "2.0", - "id": 999, - "method": "tools/call", - "params": { - "name": "zoom_window_normalized", - "arguments": { - "x1": 0.4, - "y1": 0.4, - "x2": 0.6, - "y2": 0.6 - } - } - } - - print(f"\n[发送] 请求参数:\n{json.dumps(req['params'], indent=2)}") - client.sendall(json.dumps(req).encode('utf-8')) - - response_data = client.recv(4096) - - print(f"\n[接收] CAD 基座返回:") - print(json.dumps(json.loads(response_data.decode('utf-8')), indent=2, ensure_ascii=False)) - - client.close() - - except Exception as e: - print(f"测试失败: {e}") - -if __name__ == "__main__": - test_fake_llm_zoom() \ No newline at end of file diff --git a/test_autoagent.py b/test_autoagent.py deleted file mode 100644 index 0241a2f..0000000 --- a/test_autoagent.py +++ /dev/null @@ -1,43 +0,0 @@ -import socket -import json -import base64 - -def test_screenshot(): - client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - client.connect(("127.0.0.1", 8080)) - - req = { - "jsonrpc": "2.0", - "id": 5, - "method": "tools/call", - "params": { - "name": "get_viewport_screenshot", - "arguments": {} - } - } - client.sendall(json.dumps(req).encode('utf-8')) - - # 图片 base64 报文可能很大，需要循环接收直到拿完 - response_data = b"" - while True: - chunk = client.recv(8192) - response_data += chunk - if len(chunk) < 8192: - break - - client.close() - - resp_json = json.loads(response_data.decode('utf-8')) - - # 解析并保存图片 - content_list = resp_json["result"]["content"] - for item in content_list: - if item["type"] == "image": - base64_data = item["data"] - image_bytes = base64.b64decode(base64_data) - with open("test_screenshot.png", "wb") as f: - f.write(image_bytes) - print("截图已保存为 test_screenshot.png，请在当前目录查看！") - -if __name__ == "__main__": - test_screenshot() \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..d36f2c4 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1,5 @@ +"""dwg_review 回归测试网（离线快层）。 + +与仓库根目录的 `verify_*.py` 不同：这里全部使用假 LLM + 假 MCP 基座， +不需要 CAD 在线、不联网、秒级完成，用于锁定流水线的既有行为。 +""" diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..c69849f --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,337 @@ +"""离线测试假件：假 LLM、假 MCP 基座、响应构造器。 + +设计约束（踩过的坑，务必遵守）： + - 绝不构造真实的 `LlmClient`——没有 API Key 时它的 `__init__` 直接抛 `LlmError`。 + - 绝不无参构造 `CadTools()` / `ReviewContext.create()`——二者会立刻调用 + `list_tools()` 去连 127.0.0.1:8080。一律用 `make_ctx()` 直接拼装。 + +本文件名不带 `test_` 前缀，故不会被 pytest 收集为用例。 +""" + +import json + +from dwg_review.cad_tools import CadTools +from dwg_review.checklist import DEFAULT_CHECKLIST +from dwg_review.context import ReviewContext +from dwg_review.mcp_client import McpClient, McpError, McpToolError + +__all__ = [ + "FakeLlmClient", + "FakeMcpClient", + "McpError", + "McpToolError", + "STANDARD_TOOLS", + "err", + "frame", + "llm_calls", + "llm_text", + "make_ctx", + "make_evidence", + "ok_image", + "ok_mixed", + "ok_text", + "rpc_error", + "tool_schema", +] + + +# --------------------------------------------------------------------------- +# OpenAI 风格响应对象（模仿 openai SDK 的 choices[0].message 结构） +# --------------------------------------------------------------------------- +class _Fn: + """tool_call.function""" + + def __init__(self, name, arguments): + self.name = name + self.arguments = arguments + + +class _TC: + """一个 tool_call。""" + + def __init__(self, call_id, name, args): + self.id = call_id + self.type = "function" + self.function = _Fn( + name, args if isinstance(args, str) else json.dumps(args, ensure_ascii=False) + ) + + +class _Msg: + """assistant 消息。""" + + def __init__(self, content=None, tool_calls=None): + self.role = "assistant" + self.content = content + self.tool_calls = tool_calls or [] + + +class _Choice: + def __init__(self, message): + self.message = message + + +class _Resp: + def __init__(self, message): + self.choices = [_Choice(message)] + + +def llm_text(text): + """构造一个「纯文本回复」的 LLM 响应。""" + return _Resp(_Msg(content=text)) + + +def llm_calls(pairs, content=None): + """构造一个「请求调用工具」的 LLM 响应。 + + pairs: [(工具名, 参数字典), ...] + content: 可选的附带文本。 + """ + tool_calls = [ + _TC(str(i + 1), name, args) for i, (name, args) in enumerate(pairs) + ] + return _Resp(_Msg(content=content, tool_calls=tool_calls)) + + +# --------------------------------------------------------------------------- +# 假 LLM 客户端 +# --------------------------------------------------------------------------- +class FakeLlmClient: + """按脚本回放 LLM 响应，并记录每次调用。 + + - `script` ：`chat()` 依次弹出的响应队列；弹空后回落到 `text`。 + - `text` ：`chat()` / `chat_text()` 的兜底文本。 + - `text_script` ：`chat_text()` 专用队列（便于同时喂 summarize 与 extract_issues）。 + """ + + def __init__(self, script=None, text="", text_script=None, model="fake-model"): + self.script = list(script or []) + self.text = text + self.text_script = list(text_script or []) + self.model = model + self.calls = [] + + def chat(self, messages, *, tools=None, timeout=None, retries=None): + self.calls.append({"messages": list(messages), "tools": tools}) + if self.script: + return self.script.pop(0) + return llm_text(self.text) + + def chat_text(self, system, user, **kwargs): + self.calls.append({"system": system, "user": user}) + if self.text_script: + return self.text_script.pop(0) + return self.text + + +# --------------------------------------------------------------------------- +# JSON-RPC / MCP 响应构造器 +# --------------------------------------------------------------------------- +def ok_text(obj, *, req_id=1): + """成功响应，content 为一段文本（非字符串会先 JSON 序列化）。""" + text = obj if isinstance(obj, str) else json.dumps(obj, ensure_ascii=False) + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "text", "text": text}]}, + } + + +def ok_image(b64, mime="image/png", *, req_id=1): + """成功响应，content 为一张图片。""" + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "image", "data": b64, "mimeType": mime}]}, + } + + +def ok_mixed(objs, images=(), *, req_id=1): + """成功响应，content 同时含文本与图片。""" + content = [] + for obj in objs: + text = obj if isinstance(obj, str) else json.dumps(obj, ensure_ascii=False) + content.append({"type": "text", "text": text}) + for b64, mime in images: + content.append({"type": "image", "data": b64, "mimeType": mime}) + return {"jsonrpc": "2.0", "id": req_id, "result": {"content": content}} + + +def err(text, *, req_id=1): + """工具执行错误：result.isError == true。""" + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "text", "text": text}], "isError": True}, + } + + +def rpc_error(code, message, *, req_id=1): + """协议级错误：JSON-RPC 顶层 error。""" + return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}} + + +# --------------------------------------------------------------------------- +# 工具 schema +# --------------------------------------------------------------------------- +def tool_schema(name, description="", properties=None, required=None): + """构造一条 tools/list 条目。""" + return { + "name": name, + "description": description, + "inputSchema": { + "type": "object", + "properties": properties or {}, + "required": required or [], + }, + } + + +STANDARD_TOOLS = [ + tool_schema("get_drawing_info", "图纸信息"), + tool_schema("get_layers", "图层列表"), + tool_schema("get_layouts", "布局列表"), + tool_schema("detect_drawing_frames", "识别图框"), + tool_schema("zoom_to_window", "缩放到窗口"), + tool_schema("zoom_to_entity", "缩放到实体"), + tool_schema("query_entities", "查询图元"), + tool_schema("get_text_content", "文字内容"), + tool_schema("get_dimensions", "标注"), + tool_schema("get_block_references", "块参照"), + tool_schema("get_polylines_info", "多段线"), + tool_schema("get_viewport_screenshot", "视口截图"), + tool_schema("get_entity_detail", "实体详情"), + tool_schema("find_duplicate_entities", "重复实体"), + tool_schema("find_zero_length", "零长度实体"), + tool_schema("add_leader_annotation", "添加引线批注"), + tool_schema("highlight_entities", "高亮实体"), + tool_schema("delete_entities", "删除实体"), +] + + +# --------------------------------------------------------------------------- +# 假 MCP 客户端 +# --------------------------------------------------------------------------- +def _as_result(value, arguments): + """把注入值解释为「客户端返回值」或「异常」。""" + if isinstance(value, Exception): + raise value + if callable(value): + value = value(arguments) + if isinstance(value, Exception): + raise value + # 原始 JSON-RPC 响应 → 走真实客户端的归一化（error → McpError，isError → McpToolError） + if isinstance(value, dict) and ("result" in value or "error" in value): + return McpClient._normalize(value) + return value + + +class FakeMcpClient: + """假 MCP 基座客户端。 + + - `tools` ：tools/list 返回的工具列表（None = STANDARD_TOOLS）。 + - `responses` ：工具名 → 注入值。注入值可以是 + * 原始响应 dict（用 ok_text/ok_image/err/rpc_error 构造）——按真实客户端语义归一化； + * 异常实例（McpError/McpToolError）——直接抛出； + * 可调用对象 f(arguments)——返回上面任意一种。 + 未命中的工具默认返回 `ok_text({})`。 + """ + + def __init__(self, tools=None, responses=None, available=True): + self._tools = list(STANDARD_TOOLS if tools is None else tools) + self.responses = dict(responses or {}) + self.available = available + self.calls = [] # call_tool 调用日志 + self.raw_calls = [] # raw_rpc 调用日志 + + # -- 客户端 API -------------------------------------------------------- + def list_tools(self, *, use_cache=True): + return list(self._tools) + + def call_tool(self, name, arguments=None, *, timeout=None, retries=None, is_write=False): + self.calls.append({"name": name, "arguments": arguments, "is_write": is_write}) + if name not in self.responses: + return ok_text({}) + return _as_result(self.responses[name], arguments) + + def raw_rpc(self, method, params=None, *, timeout=None, retries=None): + self.raw_calls.append({"method": method, "params": params}) + if method in self.responses: + return _as_result(self.responses[method], params) + return {"jsonrpc": "2.0", "id": 1, "result": {}} + + def is_available(self, timeout=1.5): + return self.available + + # -- 断言辅助 ---------------------------------------------------------- + def called(self, name): + return [c for c in self.calls if c["name"] == name] + + def called_names(self): + return [c["name"] for c in self.calls] + + def arguments_of(self, name): + """返回某工具**最后一次**调用的参数。""" + hits = self.called(name) + return hits[-1]["arguments"] if hits else None + + +# --------------------------------------------------------------------------- +# 组合辅助 +# --------------------------------------------------------------------------- +def make_ctx(client=None, llm=None, checklist=None, **over): + """直接拼装一个 ReviewContext（不触碰网络、不读密钥）。""" + client = client if client is not None else FakeMcpClient() + params = dict( + client=client, + tools=CadTools(client), + llm=llm if llm is not None else FakeLlmClient(), + checklist=checklist if checklist is not None else DEFAULT_CHECKLIST, + max_frames=0, + frame_limit=200, + max_iters=4, + ) + params.update(over) + return ReviewContext(**params) + + +def frame(bbox, kind="title_block", sheet_size="A1", layer="0", confidence=0.9, **extra): + """构造一个图框描述（detect_drawing_frames 返回的 items 元素）。""" + data = { + "bbox": list(bbox), + "kind": kind, + "sheet_size": sheet_size, + "layer": layer, + "confidence": confidence, + } + data.update(extra) + return data + + +def make_evidence( + index=1, + bbox=(0.0, 0.0, 100.0, 50.0), + image_b64=None, + image_mime=None, + **over, +): + """构造 collect_frame 的返回值（evidence）。""" + ev = { + "index": index, + "frame": frame(bbox), + "window": { + "min_x": float(bbox[0]), + "min_y": float(bbox[1]), + "max_x": float(bbox[2]), + "max_y": float(bbox[3]), + }, + "entities": {"total": 0, "items": []}, + "texts": {"total": 0, "items": []}, + "dimensions": {"total": 0, "items": []}, + "blocks": {"total": 0, "items": []}, + "polylines": {"total": 0, "items": []}, + "image_b64": image_b64, + "image_mime": image_mime, + } + ev.update(over) + return ev diff --git a/tests/test_annotator.py b/tests/test_annotator.py new file mode 100644 index 0000000..46dbe3a --- /dev/null +++ b/tests/test_annotator.py @@ -0,0 +1,166 @@ +"""`dwg_review.annotator` 单元测试。""" + +import pytest + +from dwg_review.annotator import _pick_point, annotate, clear +from dwg_review.mcp_client import McpError, McpToolError +from tests.fakes import FakeMcpClient, make_ctx, ok_text + + +# --------------------------------------------------------------------------- +# _pick_point +# --------------------------------------------------------------------------- +def test_pick_point_from_bbox_center(): + assert _pick_point({"bbox": [0, 0, 10, 20]}) == [5.0, 10.0, 0.0] + + +def test_pick_point_from_position(): + assert _pick_point({"position": [1, 2, 3]}) == [1.0, 2.0, 3.0] + + +def test_pick_point_from_position_without_z(): + assert _pick_point({"position": [1, 2]}) == [1.0, 2.0, 0.0] + + +def test_pick_point_prefers_bbox_over_position(): + assert _pick_point({"bbox": [0, 0, 10, 10], "position": [99, 99]}) == [5.0, 5.0, 0.0] + + +def test_pick_point_returns_none_when_unlocatable(): + assert _pick_point({}) is None + assert _pick_point({"bbox": [0, 0, 1]}) is None + assert _pick_point("not-a-dict") is None + + +# --------------------------------------------------------------------------- +# annotate +# --------------------------------------------------------------------------- +def build_client(detail=None, created=None, fail_leader=False): + responses = {"get_entity_detail": detail if detail is not None else ok_text({"bbox": [0, 0, 10, 10]})} + if fail_leader: + responses["add_leader_annotation"] = McpToolError("引线失败") + else: + responses["add_leader_annotation"] = ok_text( + created if created is not None else {"created": [{"handle": "NEW1"}]} + ) + responses["highlight_entities"] = ok_text({}) + return FakeMcpClient(responses=responses) + + +def test_annotate_dry_run_does_not_call_cad(): + logs = [] + ctx = make_ctx(client=FakeMcpClient()) + created, highlighted = annotate( + ctx, [{"handle": "A", "severity": "high", "message": "x"}], dry_run=True, log=logs.append + ) + assert created == [] and highlighted == [] + assert ctx.client.calls == [] + assert any("[dry-run]" in line for line in logs) + + +def test_annotate_skips_issues_without_handle(): + ctx = make_ctx(client=FakeMcpClient()) + created, highlighted = annotate( + ctx, [{"severity": "high", "message": "无 handle"}], log=lambda *a: None + ) + assert created == [] and highlighted == [] + assert ctx.client.calls == [] + + +def test_annotate_creates_leader_and_highlights(): + ctx = make_ctx(client=build_client()) + created, highlighted = annotate( + ctx, [{"handle": "A", "severity": "high", "message": "重叠"}], log=lambda *a: None + ) + assert created == ["NEW1"] + assert highlighted == ["A"] + names = ctx.client.called_names() + assert "get_entity_detail" in names + assert "add_leader_annotation" in names + assert "highlight_entities" in names + + +def test_annotate_leader_payload_shape(): + ctx = make_ctx(client=build_client()) + annotate(ctx, [{"handle": "A", "severity": "high", "message": "m"}], log=lambda *a: None) + args = ctx.client.arguments_of("add_leader_annotation") + assert args["layer"] == "AI_REVIEW" + assert len(args["points"]) == 2 + assert args["points"][0] == [5.0, 5.0, 0.0] # 实体定位点 + assert args["text_position"] == args["points"][1] + assert args["text"].startswith("[high]") + + +def test_annotate_skips_when_unlocatable(): + logs = [] + ctx = make_ctx(client=build_client(detail=ok_text({}))) + created, highlighted = annotate( + ctx, [{"handle": "A", "message": "x"}], log=logs.append + ) + assert created == [] and highlighted == [] + assert "add_leader_annotation" not in ctx.client.called_names() + assert any("无法定位坐标" in line for line in logs) + + +def test_annotate_tool_error_does_not_highlight(): + logs = [] + ctx = make_ctx(client=build_client(fail_leader=True)) + created, highlighted = annotate( + ctx, [{"handle": "A", "message": "x"}], log=logs.append + ) + assert created == [] and highlighted == [] + assert "highlight_entities" not in ctx.client.called_names() + assert any("批注失败" in line for line in logs) + + +def test_annotate_handles_multiple_created_entities(): + client = build_client(created={"created": [{"handle": "N1"}, {"handle": "N2"}]}) + ctx = make_ctx(client=client) + created, _ = annotate(ctx, [{"handle": "A", "message": "x"}], log=lambda *a: None) + assert created == ["N1", "N2"] + + +def test_annotate_created_not_deduped_known_gap(): + """已知瑕疵：同一 handle 出现两次会创建两条批注（created 无去重，仅 highlighted 去重）。""" + ctx = make_ctx(client=build_client()) + created, highlighted = annotate( + ctx, + [ + {"handle": "A", "severity": "high", "message": "第一次"}, + {"handle": "A", "severity": "high", "message": "第二次"}, + ], + log=lambda *a: None, + ) + assert created == ["NEW1", "NEW1"] # 重复批注（期望去重） + assert highlighted == ["A"] # 高亮已去重 + + +def test_annotate_propagates_protocol_error_on_detail(): + ctx = make_ctx(client=FakeMcpClient(responses={"get_entity_detail": McpError(-32602, "bad")})) + # annotate 只兜 add_leader/highlight 的错；detail 的协议错误会向上冒泡（现状行为） + with pytest.raises(McpError): + annotate(ctx, [{"handle": "A", "message": "x"}], log=lambda *a: None) + + +# --------------------------------------------------------------------------- +# clear +# --------------------------------------------------------------------------- +def test_clear_deletes_and_restores(): + ctx = make_ctx(client=FakeMcpClient()) + clear(ctx, handles=["N1"], highlighted=["A"], log=lambda *a: None) + assert ctx.client.arguments_of("delete_entities") == {"handles": ["N1"]} + assert ctx.client.arguments_of("highlight_entities") == {"handles": ["A"], "restore": True} + + +def test_clear_noop_without_arguments(): + ctx = make_ctx(client=FakeMcpClient()) + clear(ctx, log=lambda *a: None) + assert ctx.client.calls == [] + + +def test_clear_swallows_tool_errors(): + logs = [] + client = FakeMcpClient(responses={"delete_entities": McpToolError("删不掉")}) + ctx = make_ctx(client=client) + clear(ctx, handles=["N1"], log=logs.append) + assert any("撤销失败" in line for line in logs) diff --git a/tests/test_cad_tools.py b/tests/test_cad_tools.py new file mode 100644 index 0000000..0a23173 --- /dev/null +++ b/tests/test_cad_tools.py @@ -0,0 +1,102 @@ +"""`dwg_review.cad_tools` 单元测试。""" + +from dwg_review.cad_tools import CadTools, is_write_tool +from dwg_review.mcp_client import result_json +from tests.fakes import FakeMcpClient, ok_text, tool_schema + + +# --------------------------------------------------------------------------- +# 写类工具判定 +# --------------------------------------------------------------------------- +def test_write_prefixes_detected(): + for name in ("add_leader_annotation", "delete_entities", "draw_line", "set_layer"): + assert is_write_tool(name) is True + + +def test_write_exact_names_detected(): + assert is_write_tool("set_entity_properties") is True + assert is_write_tool("highlight_entities") is True + + +def test_readonly_tools_are_not_write(): + for name in ("query_entities", "get_text_content", "zoom_to_window", "find_zero_length"): + assert is_write_tool(name) is False + + +# --------------------------------------------------------------------------- +# 注册表 +# --------------------------------------------------------------------------- +def test_registry_built_from_client(): + client = FakeMcpClient() + tools = CadTools(client) + assert "query_entities" in tools.names + assert tools.has("get_viewport_screenshot") is True + assert tools.has("__nope__") is False + assert tools.names == sorted(tools.names) + + +def test_describe_returns_tool_dict(): + client = FakeMcpClient() + tools = CadTools(client) + desc = tools.describe("query_entities") + assert desc["name"] == "query_entities" + assert "inputSchema" in desc + + +def test_refresh_picks_up_new_tools(): + client = FakeMcpClient(tools=[tool_schema("a")]) + tools = CadTools(client) + assert tools.names == ["a"] + client._tools = [tool_schema("a"), tool_schema("b")] + tools.refresh() + assert tools.names == ["a", "b"] + + +# --------------------------------------------------------------------------- +# openai_tools +# --------------------------------------------------------------------------- +def test_openai_tools_format(): + client = FakeMcpClient(tools=[tool_schema("query_entities", "查询", {"window": {"type": "object"}})]) + entry = CadTools(client).openai_tools()[0] + assert entry["type"] == "function" + assert entry["function"]["name"] == "query_entities" + assert entry["function"]["description"] == "查询" + assert entry["function"]["parameters"]["properties"] == {"window": {"type": "object"}} + + +def test_openai_tools_defaults_schema_when_missing(): + client = FakeMcpClient(tools=[{"name": "bare", "description": ""}]) + entry = CadTools(client).openai_tools()[0] + assert entry["function"]["parameters"] == {"type": "object", "properties": {}} + + +def test_openai_tools_names_filter(): + client = FakeMcpClient() + names = CadTools(client).openai_tools(names={"query_entities", "get_layers"}) + assert sorted(n["function"]["name"] for n in names) == ["get_layers", "query_entities"] + + +# --------------------------------------------------------------------------- +# call +# --------------------------------------------------------------------------- +def test_call_returns_normalized_response(): + client = FakeMcpClient(responses={"query_entities": ok_text({"total": 7})}) + tools = CadTools(client) + assert result_json(tools.call("query_entities", {"window": {}})) == {"total": 7} + assert client.calls[-1]["name"] == "query_entities" + assert client.calls[-1]["arguments"] == {"window": {}} + + +def test_call_marks_write_tools(): + client = FakeMcpClient() + tools = CadTools(client) + tools.call("add_leader_annotation", {"text": "x"}) + tools.call("query_entities", {"window": {}}) + assert client.calls[0]["is_write"] is True + assert client.calls[1]["is_write"] is False + + +def test_call_explicit_is_write_overrides_inference(): + client = FakeMcpClient() + CadTools(client).call("query_entities", {}, is_write=True) + assert client.calls[-1]["is_write"] is True diff --git a/tests/test_checklist.py b/tests/test_checklist.py new file mode 100644 index 0000000..b90dbbb --- /dev/null +++ b/tests/test_checklist.py @@ -0,0 +1,72 @@ +"""`dwg_review.checklist` 单元测试。""" + +import json + +from dwg_review.checklist import ( + DEFAULT_CHECKLIST, + build_system_prompt, + checklist_summary, + load_checklist, +) + + +def test_load_valid_checklist(tmp_path): + data = { + "name": "自定义清单", + "categories": [{"title": "类别A", "severity": "high", "checks": ["检查1"]}], + } + path = tmp_path / "cl.json" + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") + loaded = load_checklist(path) + assert loaded["name"] == "自定义清单" + assert loaded["categories"][0]["title"] == "类别A" + + +def test_load_missing_file_falls_back(tmp_path, capsys): + loaded = load_checklist(tmp_path / "nope.json") + assert loaded is DEFAULT_CHECKLIST + assert "无法加载审查清单" in capsys.readouterr().out + + +def test_load_malformed_structure_falls_back(tmp_path, capsys): + path = tmp_path / "cl.json" + path.write_text(json.dumps({"name": "缺 categories"}), encoding="utf-8") + assert load_checklist(path) is DEFAULT_CHECKLIST + assert "结构异常" in capsys.readouterr().out + + +def test_load_invalid_json_falls_back(tmp_path, capsys): + path = tmp_path / "cl.json" + path.write_text("{ not json", encoding="utf-8") + assert load_checklist(path) is DEFAULT_CHECKLIST + assert "无法加载审查清单" in capsys.readouterr().out + + +def test_load_empty_categories_falls_back(tmp_path): + path = tmp_path / "cl.json" + path.write_text(json.dumps({"categories": []}), encoding="utf-8") + assert load_checklist(path) is DEFAULT_CHECKLIST + + +def test_build_system_prompt_includes_categories_and_checks(): + prompt = build_system_prompt(DEFAULT_CHECKLIST) + assert DEFAULT_CHECKLIST["name"] in prompt + for cat in DEFAULT_CHECKLIST["categories"]: + assert cat["title"] in prompt + assert f"严重度：{cat['severity']}" in prompt + for chk in cat["checks"]: + assert chk in prompt + + +def test_build_system_prompt_tool_hint_toggle(): + with_tools = build_system_prompt(DEFAULT_CHECKLIST, allow_tools=True) + without = build_system_prompt(DEFAULT_CHECKLIST, allow_tools=False) + assert "只读" in with_tools and "取证工具" in with_tools + assert "只读" not in without + + +def test_checklist_summary_mentions_name_and_count(): + summary = checklist_summary(DEFAULT_CHECKLIST) + assert DEFAULT_CHECKLIST["name"] in summary + assert str(len(DEFAULT_CHECKLIST["categories"])) in summary + assert DEFAULT_CHECKLIST["categories"][0]["title"] in summary diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..2477279 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,68 @@ +"""`dwg_review.config` 单元测试（只验证纯函数与常量类型，不依赖真实环境变量）。""" + +import os + +from dwg_review import config + + +def test_resolve_path_relative_uses_project_root(): + assert config._resolve_path("reports") == config.PROJECT_ROOT / "reports" + + +def test_resolve_path_absolute_unchanged(tmp_path): + assert config._resolve_path(str(tmp_path)) == tmp_path + + +def test_project_root_is_repo_root(): + assert (config.PROJECT_ROOT / "dwg_review" / "config.py").is_file() + + +def test_read_env_prefers_environ(monkeypatch): + monkeypatch.setenv("CAD_TEST_ENV_PROBE", "hello") + assert config._read_env("CAD_TEST_ENV_PROBE") == "hello" + + +def test_read_env_missing_returns_empty(monkeypatch): + monkeypatch.delenv("CAD_TEST_DEFINITELY_MISSING_XYZ", raising=False) + assert config._read_env("CAD_TEST_DEFINITELY_MISSING_XYZ") == "" + + +def test_has_api_key_reflects_module_value(monkeypatch): + monkeypatch.setattr(config, "API_KEY", "") + assert config.has_api_key() is False + monkeypatch.setattr(config, "API_KEY", "sk-x") + assert config.has_api_key() is True + + +def test_has_api_key_ignores_whitespace(monkeypatch): + monkeypatch.setattr(config, "API_KEY", " ") + assert config.has_api_key() is False + + +def test_constants_have_expected_types(): + assert isinstance(config.HOST, str) + assert isinstance(config.PORT, int) + assert isinstance(config.CALL_TIMEOUT, float) + assert isinstance(config.CALL_RETRIES, int) + assert isinstance(config.MAX_FRAME_BYTES, int) + assert isinstance(config.IMAGE_MAX_EDGE, int) + assert isinstance(config.IMAGE_FORMAT, str) + assert isinstance(config.CHECKLIST_PATH, type(config.PROJECT_ROOT)) + + +def test_checklist_path_resolves_relative_to_root(): + assert config.CHECKLIST_PATH.name == "review_checklist.json" + assert config.CHECKLIST_PATH.parent == config.PROJECT_ROOT + + +def test_port_default_matches_baseline(): + # 基座固定监听 127.0.0.1:8080；若此值被改动，请同步更新基座与文档。 + assert config.HOST == "127.0.0.1" + assert config.PORT == 8080 + + +def test_os_name_guard_in_read_env(monkeypatch): + # 非 Windows 平台不应触碰注册表 + monkeypatch.setattr(os, "name", "posix") + monkeypatch.delenv("CAD_TEST_ENV_PROBE2", raising=False) + assert config._read_env("CAD_TEST_ENV_PROBE2") == "" diff --git a/tests/test_evidence.py b/tests/test_evidence.py new file mode 100644 index 0000000..1964a3d --- /dev/null +++ b/tests/test_evidence.py @@ -0,0 +1,91 @@ +"""`dwg_review.phases.evidence` 单元测试。""" + +import pytest + +from dwg_review.phases.evidence import bbox_to_window, collect_frame +from tests.fakes import FakeMcpClient, frame, make_ctx, ok_image, ok_text + + +def test_bbox_to_window_mapping(): + assert bbox_to_window([1, 2, 3, 4]) == { + "min_x": 1.0, + "min_y": 2.0, + "max_x": 3.0, + "max_y": 4.0, + } + + +def test_bbox_to_window_non_numeric_known_gap(): + """已知瑕疵：4 元素但含非数值的 bbox 会抛 ValueError，上游 collect_frame 只判长度。""" + with pytest.raises(ValueError): + bbox_to_window([0, "a", 10, 10]) + + +def test_collect_frame_returns_none_for_missing_bbox(): + ctx = make_ctx(client=FakeMcpClient()) + assert collect_frame(ctx, 1, {}) is None + assert collect_frame(ctx, 1, {"bbox": None}) is None + + +def test_collect_frame_returns_none_for_bad_bbox_length(): + ctx = make_ctx(client=FakeMcpClient()) + assert collect_frame(ctx, 1, {"bbox": [0, 0, 10]}) is None + + +def build_evidence_client(image=ok_image("QUJD")): + return FakeMcpClient( + responses={ + "query_entities": ok_text({"total": 5, "items": [{"handle": "A"}]}), + "get_text_content": ok_text({"total": 1, "items": [{"text": "你好"}]}), + "get_dimensions": ok_text({"total": 2, "items": []}), + "get_block_references": ok_text({"total": 0, "items": []}), + "get_polylines_info": ok_text({"total": 3, "items": []}), + "get_viewport_screenshot": image, + } + ) + + +def test_collect_frame_assembles_evidence(): + ctx = make_ctx(client=build_evidence_client()) + ev = collect_frame(ctx, 2, frame([0, 0, 100, 50])) + assert ev["index"] == 2 + assert ev["window"] == {"min_x": 0.0, "min_y": 0.0, "max_x": 100.0, "max_y": 50.0} + assert ev["entities"]["total"] == 5 + assert ev["texts"]["total"] == 1 + assert ev["dimensions"]["total"] == 2 + assert ev["blocks"]["total"] == 0 + assert ev["polylines"]["total"] == 3 + assert ev["image_b64"] == "QUJD" + assert ev["image_mime"] == "image/png" + + +def test_collect_frame_calls_expected_tools(): + ctx = make_ctx(client=build_evidence_client(), frame_limit=99) + collect_frame(ctx, 1, frame([1, 2, 3, 4])) + client = ctx.client + + win = {"min_x": 1.0, "min_y": 2.0, "max_x": 3.0, "max_y": 4.0} + assert client.arguments_of("zoom_to_window") == dict(win, margin=0.02) + for name in ( + "query_entities", + "get_text_content", + "get_dimensions", + "get_block_references", + "get_polylines_info", + ): + assert client.arguments_of(name) == {"window": win, "limit": 99} + assert client.arguments_of("get_viewport_screenshot") == {} + + +def test_collect_frame_without_image(): + ctx = make_ctx(client=build_evidence_client(image=ok_text({}))) + ev = collect_frame(ctx, 1, frame([0, 0, 10, 10])) + assert ev["image_b64"] is None + assert ev["image_mime"] is None + + +def test_collect_frame_tolerates_empty_structured_data(): + ctx = make_ctx(client=FakeMcpClient()) # 全部默认 ok_text({}) + ev = collect_frame(ctx, 1, frame([0, 0, 10, 10])) + assert ev["entities"] == {} + assert ev["image_b64"] is None diff --git a/tests/test_llm_client.py b/tests/test_llm_client.py new file mode 100644 index 0000000..f4e554c --- /dev/null +++ b/tests/test_llm_client.py @@ -0,0 +1,188 @@ +"""`dwg_review.llm_client` 单元测试（绝不真实调用 LLM）。""" + +import base64 +import io + +import pytest + +from dwg_review import config +from dwg_review import llm_client as llm_module +from dwg_review.llm_client import LlmClient, LlmError, downsample_image_b64 +from tests.fakes import llm_text + + +# --------------------------------------------------------------------------- +# 假 OpenAI SDK +# --------------------------------------------------------------------------- +class _Completions: + def __init__(self, outer): + self.outer = outer + + def create(self, **kwargs): + self.outer.kwargs.append(kwargs) + if self.outer.raises: + raise RuntimeError("upstream boom") + return self.outer.result + + +class _Chat: + def __init__(self, outer): + self.completions = _Completions(outer) + + +class _FakeOpenAI: + def __init__(self, **kwargs): + self.init_kwargs = kwargs + self.kwargs = [] + self.raises = False + self.result = None + self.chat = _Chat(self) + + +@pytest.fixture +def fake_openai(monkeypatch): + monkeypatch.setattr(llm_module, "OpenAI", _FakeOpenAI) + monkeypatch.setattr(llm_module.time, "sleep", lambda *_: None) + return _FakeOpenAI + + +def build_client(fake_openai, **kw): + return LlmClient(api_key="sk-test", **kw) + + +# --------------------------------------------------------------------------- +# 构造 +# --------------------------------------------------------------------------- +def test_init_requires_api_key(monkeypatch): + monkeypatch.setattr(config, "API_KEY", "") + with pytest.raises(LlmError): + LlmClient() + + +def test_init_passes_key_and_base_url(fake_openai): + client = LlmClient(api_key="sk-abc", base_url="https://example.test") + assert client._client.init_kwargs["api_key"] == "sk-abc" + assert client._client.init_kwargs["base_url"] == "https://example.test" + + +def test_init_uses_config_defaults(fake_openai, monkeypatch): + monkeypatch.setattr(config, "MODEL_NAME", "test-model") + monkeypatch.setattr(config, "BASE_URL", "https://cfg.test") + client = LlmClient(api_key="sk-abc") + assert client.model == "test-model" + assert client._client.init_kwargs["base_url"] == "https://cfg.test" + + +# --------------------------------------------------------------------------- +# chat / chat_text +# --------------------------------------------------------------------------- +def test_chat_returns_raw_response(fake_openai): + client = build_client(fake_openai) + client._client.result = llm_text("hi") + resp = client.chat([{"role": "user", "content": "x"}]) + assert resp.choices[0].message.content == "hi" + + +def test_chat_includes_tools_only_when_present(fake_openai): + client = build_client(fake_openai) + client._client.result = llm_text("x") + client.chat([{"role": "user", "content": "x"}]) + assert "tools" not in client._client.kwargs[0] + + client.chat([{"role": "user", "content": "x"}], tools=[{"type": "function"}]) + assert client._client.kwargs[1]["tools"] == [{"type": "function"}] + + +def test_chat_wraps_exception_into_llmerror(fake_openai): + client = build_client(fake_openai) + client._client.raises = True + with pytest.raises(LlmError): + client.chat([{"role": "user", "content": "x"}], retries=0) + + +def test_chat_retries_then_succeeds(fake_openai): + client = build_client(fake_openai) + + class Flaky(_Completions): + def __init__(self, outer): + super().__init__(outer) + self.n = 0 + + def create(self, **kwargs): + self.n += 1 + if self.n < 3: + raise RuntimeError("transient") + return llm_text("ok") + + client._client.chat.completions = Flaky(client._client) + resp = client.chat([{"role": "user", "content": "x"}], retries=2) + assert resp.choices[0].message.content == "ok" + assert client._client.chat.completions.n == 3 + + +def test_chat_text_returns_assistant_text(fake_openai): + client = build_client(fake_openai) + client._client.result = llm_text("报告正文") + assert client.chat_text("sys", "user") == "报告正文" + + +def test_image_content_format(): + block = LlmClient.image_content("QUJD", "image/jpeg") + assert block == { + "type": "image_url", + "image_url": {"url": "data:image/jpeg;base64,QUJD"}, + } + + +# --------------------------------------------------------------------------- +# 截图下采样 +# --------------------------------------------------------------------------- +def test_downsample_passthrough_without_pil(monkeypatch): + monkeypatch.setattr(llm_module, "HAS_PIL", False) + assert downsample_image_b64("QUJD", "image/png", max_edge=10) == ("QUJD", "image/png") + + +def test_downsample_passthrough_when_max_edge_zero(): + assert downsample_image_b64("QUJD", "image/png", max_edge=0) == ("QUJD", "image/png") + + +def test_downsample_passthrough_when_no_resize_needed(): + pytest.importorskip("PIL") + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (100, 80), "white").save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode("ascii") + out, mime = downsample_image_b64(b64, "image/png", max_edge=1280) + assert out == b64 and mime == "image/png" + + +def test_downsample_resizes_large_image(): + pytest.importorskip("PIL") + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (2000, 1000), "white").save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode("ascii") + out, mime = downsample_image_b64(b64, "image/png", max_edge=500) + assert mime == "image/png" + img = Image.open(io.BytesIO(base64.b64decode(out))) + assert max(img.size) <= 500 + + +def test_downsample_converts_to_jpeg(): + pytest.importorskip("PIL") + from PIL import Image + + buf = io.BytesIO() + Image.new("RGB", (2000, 1000), "white").save(buf, format="PNG") + b64 = base64.b64encode(buf.getvalue()).decode("ascii") + out, mime = downsample_image_b64(b64, "image/png", max_edge=400, fmt="jpeg") + assert mime == "image/jpeg" + assert Image.open(io.BytesIO(base64.b64decode(out))).format == "JPEG" + + +def test_downsample_bad_base64_returns_original(): + pytest.importorskip("PIL") + bad = base64.b64encode(b"this is not an image").decode("ascii") + assert downsample_image_b64(bad, "image/png", max_edge=10) == (bad, "image/png") diff --git a/tests/test_mcp_client.py b/tests/test_mcp_client.py new file mode 100644 index 0000000..1287d60 --- /dev/null +++ b/tests/test_mcp_client.py @@ -0,0 +1,249 @@ +"""`dwg_review.mcp_client` 单元测试。 + +真实 `McpClient` + 打桩的 `_send_frame`（绝不真连 8080）。 +""" + +import socket + +import pytest + +from dwg_review import config +from dwg_review.mcp_client import ( + McpClient, + McpError, + McpToolError, + result_images, + result_json, + result_texts, +) +from tests.fakes import err, ok_mixed, ok_text, rpc_error + + +def make_client(monkeypatch, response): + """构造一个 McpClient，其 _send_frame 被替换为回放 response。""" + client = McpClient() + sent = [] + + def fake_send(payload, timeout=None): + sent.append(payload) + return response(payload) if callable(response) else response + + monkeypatch.setattr(client, "_send_frame", fake_send) + return client, sent + + +# --------------------------------------------------------------------------- +# 响应辅助函数 +# --------------------------------------------------------------------------- +def test_result_texts_extracts_only_text_blocks(): + resp = ok_mixed(["hello", "world"], [("AAA", "image/png")]) + assert result_texts(resp) == ["hello", "world"] + + +def test_result_images_extracts_data_and_mime(): + resp = ok_mixed([], [("AAA", "image/png"), ("BBB", "image/jpeg")]) + assert result_images(resp) == [("AAA", "image/png"), ("BBB", "image/jpeg")] + + +def test_result_images_defaults_mime_to_png(): + resp = {"result": {"content": [{"type": "image", "data": "X"}]}} + assert result_images(resp) == [("X", "image/png")] + + +def test_result_json_parses_json_text(): + assert result_json(ok_text({"a": 1})) == {"a": 1} + + +def test_result_json_returns_raw_text_when_not_json(): + assert result_json(ok_text("未发现明显问题")) == "未发现明显问题" + + +def test_result_json_returns_default_when_no_text(): + assert result_json({"result": {"content": []}}, "D") == "D" + assert result_json(None, "D") == "D" + + +def test_result_json_null_text_known_gap(): + """已知瑕疵：文本 'null' 是合法 JSON，解析为 None，与「无 text 返回 default」语义易混。""" + assert result_json(ok_text("null"), "D") is None + + +def test_result_texts_tolerates_malformed_response(): + assert result_texts(None) == [] + assert result_texts({}) == [] + + +# --------------------------------------------------------------------------- +# 请求 / 响应 +# --------------------------------------------------------------------------- +def test_call_tool_returns_normalized_response(monkeypatch): + client, sent = make_client(monkeypatch, ok_text({"total": 3})) + resp = client.call_tool("query_entities", {"window": {}}) + assert result_json(resp) == {"total": 3} + assert sent[0]["method"] == "tools/call" + assert sent[0]["params"] == {"name": "query_entities", "arguments": {"window": {}}} + assert sent[0]["jsonrpc"] == "2.0" + + +def test_call_tool_arguments_default_to_empty_object(monkeypatch): + client, sent = make_client(monkeypatch, ok_text({})) + client.call_tool("get_layers") + assert sent[0]["params"]["arguments"] == {} + + +def test_request_ids_increment(monkeypatch): + client, sent = make_client(monkeypatch, ok_text({})) + client.call_tool("a") + client.call_tool("b") + assert [p["id"] for p in sent] == [1, 2] + assert client._next_id == 3 + + +def test_rpc_error_raises_mcp_error(monkeypatch): + client, _ = make_client(monkeypatch, rpc_error(-32602, "Unknown tool: x")) + with pytest.raises(McpError) as ei: + client.call_tool("x") + assert ei.value.code == -32602 + assert "Unknown tool" in ei.value.message + + +def test_is_error_raises_mcp_tool_error(monkeypatch): + client, _ = make_client(monkeypatch, err("工具炸了")) + with pytest.raises(McpToolError) as ei: + client.call_tool("x") + assert ei.value.message == "工具炸了" + + +def test_is_error_without_text_uses_fallback_message(monkeypatch): + resp = {"result": {"content": [], "isError": True}} + client, _ = make_client(monkeypatch, resp) + with pytest.raises(McpToolError) as ei: + client.call_tool("x") + assert ei.value.message == "tool error" + + +def test_raw_rpc_does_not_normalize_is_error(monkeypatch): + client, _ = make_client(monkeypatch, err("boom")) + out = client.raw_rpc("tools/call", {"name": "x"}) + assert out["result"]["isError"] is True # 未抛异常，原样透传 + + +def test_raw_rpc_does_not_normalize_rpc_error(monkeypatch): + client, _ = make_client(monkeypatch, rpc_error(-32700, "parse error")) + out = client.raw_rpc("tools/call", {"name": "x"}) + assert out["error"]["code"] == -32700 + + +def test_normalize_passthrough_on_success(): + resp = ok_text({}) + assert McpClient._normalize(resp) is resp + + +# --------------------------------------------------------------------------- +# 重试 +# --------------------------------------------------------------------------- +def test_retries_on_oserror(monkeypatch): + monkeypatch.setattr("dwg_review.mcp_client.time.sleep", lambda *_: None) + + def boom(payload, timeout=None): + raise OSError("connection refused") + + client = McpClient() + monkeypatch.setattr(client, "_send_frame", boom) + with pytest.raises(McpError) as ei: + client.call_tool("x", retries=2) + assert ei.value.code == -32603 + + +def test_retry_count_matches_config(monkeypatch): + monkeypatch.setattr("dwg_review.mcp_client.time.sleep", lambda *_: None) + attempts = {"n": 0} + + def boom(payload, timeout=None): + attempts["n"] += 1 + raise OSError("nope") + + client = McpClient() + monkeypatch.setattr(client, "_send_frame", boom) + with pytest.raises(McpError): + client.call_tool("x", retries=2) + assert attempts["n"] == 3 # 1 次 + 2 次重试 + + +def test_write_tool_never_retries(monkeypatch): + monkeypatch.setattr("dwg_review.mcp_client.time.sleep", lambda *_: None) + attempts = {"n": 0} + + def boom(payload, timeout=None): + attempts["n"] += 1 + raise OSError("nope") + + client = McpClient() + monkeypatch.setattr(client, "_send_frame", boom) + with pytest.raises(McpError): + client.call_tool("delete_entities", {"handles": []}, retries=5, is_write=True) + assert attempts["n"] == 1 # 写类工具强制不重试 + + +def test_protocol_error_not_retried(monkeypatch): + attempts = {"n": 0} + + def respond(payload, timeout=None): + attempts["n"] += 1 + return rpc_error(-32602, "bad") + + client = McpClient() + monkeypatch.setattr(client, "_send_frame", respond) + with pytest.raises(McpError): + client.call_tool("x", retries=3) + assert attempts["n"] == 1 + + +# --------------------------------------------------------------------------- +# 工具表缓存 / 帧大小 / 可用性 +# --------------------------------------------------------------------------- +def test_list_tools_caches(monkeypatch): + payload = {"jsonrpc": "2.0", "id": 1, "result": {"tools": [{"name": "a"}]}} + client, sent = make_client(monkeypatch, payload) + assert client.list_tools() == [{"name": "a"}] + assert client.list_tools() == [{"name": "a"}] + assert len(sent) == 1 # 第二次走缓存 + client.list_tools(use_cache=False) + assert len(sent) == 2 + + +def test_send_frame_rejects_oversize_request(monkeypatch): + monkeypatch.setattr(config, "MAX_FRAME_BYTES", 16) + client = McpClient() + with pytest.raises(McpError) as ei: + client._send_frame({"blob": "x" * 100}) + assert ei.value.code == -32600 + + +def test_is_available_true_when_connect_ok(monkeypatch): + client = McpClient() + monkeypatch.setattr(client, "_connect", lambda timeout=None: type("S", (), {"close": lambda self: None})()) + assert client.is_available() is True + + +def test_is_available_false_on_oserror(monkeypatch): + client = McpClient() + + def boom(timeout=None): + raise OSError("refused") + + monkeypatch.setattr(client, "_connect", boom) + assert client.is_available() is False + + +def test_timeout_error_wrapped_as_mcp_error(monkeypatch): + monkeypatch.setattr("dwg_review.mcp_client.time.sleep", lambda *_: None) + + def boom(payload, timeout=None): + raise socket.timeout("timed out") + + client = McpClient() + monkeypatch.setattr(client, "_send_frame", boom) + with pytest.raises(McpError) as ei: + client.call_tool("x", retries=0) + assert ei.value.code == -32603 diff --git a/tests/test_planning.py b/tests/test_planning.py new file mode 100644 index 0000000..c48685b --- /dev/null +++ b/tests/test_planning.py @@ -0,0 +1,93 @@ +"""`dwg_review.phases.planning` 单元测试。""" + +from dwg_review.phases.planning import _sort_frames, plan +from tests.fakes import FakeMcpClient, frame, make_ctx, ok_text + + +# --------------------------------------------------------------------------- +# 阅读顺序排序 +# --------------------------------------------------------------------------- +def test_sort_frames_top_to_bottom_then_left_to_right(): + # bbox = [min_x, min_y, max_x, max_y] + top_left = frame([0, 100, 10, 200]) + top_right = frame([50, 100, 60, 200]) + bottom_left = frame([0, 0, 10, 50]) + bottom_right = frame([50, 0, 60, 50]) + + ordered = _sort_frames([bottom_right, top_left, bottom_left, top_right]) + assert ordered == [top_left, top_right, bottom_left, bottom_right] + + +def test_sort_frames_same_row_by_x(): + a = frame([90, 0, 100, 10]) + b = frame([10, 0, 20, 10]) + assert _sort_frames([a, b]) == [b, a] + + +def test_sort_frames_handles_missing_bbox(): + with_bbox = frame([0, 0, 10, 10]) + without = {"kind": "x"} # 缺 bbox -> 视作全 0 + ordered = _sort_frames([without, with_bbox]) + assert with_bbox in ordered and without in ordered + + +# --------------------------------------------------------------------------- +# plan() +# --------------------------------------------------------------------------- +def build_plan_client(frames): + return FakeMcpClient( + responses={ + "get_drawing_info": ok_text({"units": "mm"}), + "get_layers": ok_text({"layers": [{"name": "0"}]}), + "get_layouts": ok_text({"layouts": [{"name": "Model"}]}), + "detect_drawing_frames": ok_text({"items": frames}), + } + ) + + +def test_plan_returns_all_sections(): + frames = [frame([0, 0, 10, 10]), frame([0, 20, 10, 30])] + ctx = make_ctx(client=build_plan_client(frames)) + result = plan(ctx) + assert result["drawing_info"] == {"units": "mm"} + assert result["layers"] == {"layers": [{"name": "0"}]} + assert result["layouts"] == {"layouts": [{"name": "Model"}]} + assert len(result["frames"]) == 2 + + +def test_plan_sorts_frames(): + lower = frame([0, 0, 10, 10]) + upper = frame([0, 50, 10, 60]) + ctx = make_ctx(client=build_plan_client([lower, upper])) + assert plan(ctx)["frames"] == [upper, lower] + + +def test_plan_respects_max_frames(): + frames = [frame([0, i * 10, 10, i * 10 + 5]) for i in range(5)] + ctx = make_ctx(client=build_plan_client(frames), max_frames=2) + assert len(plan(ctx)["frames"]) == 2 + + +def test_plan_calls_with_expected_arguments(): + ctx = make_ctx(client=build_plan_client([])) + plan(ctx) + client = ctx.client + assert client.arguments_of("get_layers") == {"count_entities": True} + assert client.arguments_of("detect_drawing_frames") == {"min_confidence": 0.3} + + +def test_plan_tolerates_non_dict_frames_data(): + client = FakeMcpClient( + responses={ + "get_drawing_info": ok_text({}), + "get_layers": ok_text({}), + "get_layouts": ok_text({}), + "detect_drawing_frames": ok_text("oops"), # 非 dict + } + ) + assert plan(make_ctx(client=client))["frames"] == [] + + +def test_plan_tolerates_missing_items_key(): + client = FakeMcpClient(responses={"detect_drawing_frames": ok_text({"other": 1})}) + assert plan(make_ctx(client=client))["frames"] == [] diff --git a/tests/test_report.py b/tests/test_report.py new file mode 100644 index 0000000..aaa17c1 --- /dev/null +++ b/tests/test_report.py @@ -0,0 +1,141 @@ +"""`dwg_review.phases.report` 单元测试。""" + +import json + +import pytest + +from dwg_review.phases.report import extract_issues, save_report, summarize +from tests.fakes import FakeLlmClient, make_ctx + + +def findings_fixture(): + return [ + {"frame": 1, "sheet": "A1", "finding": "图框1：文字重叠"}, + {"frame": 2, "sheet": "A3", "finding": "图框2：未发现明显问题"}, + ] + + +# --------------------------------------------------------------------------- +# summarize +# --------------------------------------------------------------------------- +def test_summarize_empty_returns_message_without_llm(): + llm = FakeLlmClient(text="不应被调用") + ctx = make_ctx(llm=llm) + out = summarize(ctx, [], log=lambda *a: None) + assert "未识别到可审查的图框" in out + assert llm.calls == [] + + +def test_summarize_uses_chat_text(): + llm = FakeLlmClient(text="## 总报告") + ctx = make_ctx(llm=llm) + assert summarize(ctx, findings_fixture(), log=lambda *a: None) == "## 总报告" + assert len(llm.calls) == 1 + assert "图框1" in llm.calls[0]["user"] + + +def test_summarize_falls_back_to_raw_input_on_error(): + class Boom(FakeLlmClient): + def chat_text(self, system, user, **kwargs): + raise RuntimeError("汇总炸了") + + logs = [] + ctx = make_ctx(llm=Boom()) + out = summarize(ctx, findings_fixture(), log=logs.append) + assert "图框1" in out and "图框2" in out + assert any("汇总失败" in line for line in logs) + + +def test_summarize_missing_frame_key_known_gap(): + """已知瑕疵：summarize 用硬下标 f['frame']，findings 缺该键直接 KeyError。""" + ctx = make_ctx(llm=FakeLlmClient(text="x")) + with pytest.raises(KeyError): + summarize(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None) + + +# --------------------------------------------------------------------------- +# extract_issues +# --------------------------------------------------------------------------- +def test_extract_issues_empty_returns_empty_list(): + llm = FakeLlmClient(text_script=["[]"]) + ctx = make_ctx(llm=llm) + assert extract_issues(ctx, [], log=lambda *a: None) == [] + assert llm.calls == [] + + +def test_extract_issues_parses_json_array(): + payload = '[{"frame": 1, "handle": "A", "severity": "high", "message": "x"}]' + llm = FakeLlmClient(text_script=[payload]) + ctx = make_ctx(llm=llm) + issues = extract_issues(ctx, findings_fixture(), log=lambda *a: None) + assert issues == [{"frame": 1, "handle": "A", "severity": "high", "message": "x"}] + + +def test_extract_issues_strips_code_fence(): + fenced = '```json\n[{"frame": 1, "message": "y"}]\n```' + llm = FakeLlmClient(text_script=[fenced]) + ctx = make_ctx(llm=llm) + assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == [ + {"frame": 1, "message": "y"} + ] + + +def test_extract_issues_returns_empty_on_invalid_json(): + logs = [] + llm = FakeLlmClient(text_script=["这不是 JSON"]) + ctx = make_ctx(llm=llm) + assert extract_issues(ctx, findings_fixture(), log=logs.append) == [] + assert any("结构化抽取失败" in line for line in logs) + + +def test_extract_issues_returns_empty_when_not_a_list(): + llm = FakeLlmClient(text_script=['{"frame": 1}']) + ctx = make_ctx(llm=llm) + assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == [] + + +def test_extract_issues_single_line_fence_known_gap(): + """已知瑕疵：围栏剥离依赖换行，单行 ```json[...]``` 会残留 json 前缀 → 解析失败。""" + llm = FakeLlmClient(text_script=['```json[{"frame": 1, "message": "z"}]```']) + ctx = make_ctx(llm=llm) + assert extract_issues(ctx, findings_fixture(), log=lambda *a: None) == [] + + +def test_extract_issues_missing_frame_key_known_gap(): + """已知瑕疵：extract_issues 同样用硬下标 f['frame']，findings 缺该键直接 KeyError。""" + ctx = make_ctx(llm=FakeLlmClient(text="[]")) + with pytest.raises(KeyError): + extract_issues(ctx, [{"finding": "缺 frame 键"}], log=lambda *a: None) + + +# --------------------------------------------------------------------------- +# save_report +# --------------------------------------------------------------------------- +def test_save_report_writes_md_and_json(tmp_reports): + findings = findings_fixture() + plan = {"frames": [{"bbox": [0, 0, 10, 10]}]} + issues = [{"frame": 1, "handle": "A", "message": "x"}] + + md_path, json_path = save_report("# 报告", findings, plan, issues, log=lambda *a: None) + + assert md_path.exists() and json_path.exists() + assert md_path.read_text(encoding="utf-8") == "# 报告" + data = json.loads(json_path.read_text(encoding="utf-8")) + assert data["findings"] == findings + assert data["issues"] == issues + assert data["frames"] == plan["frames"] + assert "generated_at" in data + + +def test_save_report_issues_default_to_empty(tmp_reports): + _, json_path = save_report("md", findings_fixture(), {"frames": []}, log=lambda *a: None) + assert json.loads(json_path.read_text(encoding="utf-8"))["issues"] == [] + + +def test_save_report_creates_missing_directory(tmp_path, monkeypatch): + from dwg_review import config + + target = tmp_path / "deep" / "nested" + monkeypatch.setattr(config, "REPORT_DIR", target) + save_report("md", [], {"frames": []}, log=lambda *a: None) + assert target.is_dir() diff --git a/tests/test_review.py b/tests/test_review.py new file mode 100644 index 0000000..6360758 --- /dev/null +++ b/tests/test_review.py @@ -0,0 +1,209 @@ +"""`dwg_review.phases.review` 单元测试。""" + +import json + +from dwg_review.cad_tools import READ_ONLY_TOOLS +from dwg_review.mcp_client import McpToolError +from dwg_review.phases.review import ( + _run_readonly_tool, + evidence_payload, + review_frame, +) +from tests.fakes import ( + FakeLlmClient, + FakeMcpClient, + llm_calls, + llm_text, + make_ctx, + make_evidence, + ok_image, + ok_mixed, + ok_text, +) + + +# --------------------------------------------------------------------------- +# evidence_payload +# --------------------------------------------------------------------------- +def test_evidence_payload_strips_image(): + ev = make_evidence(image_b64="QUJD", image_mime="image/png") + payload = evidence_payload(ev) + assert "image_b64" not in payload + assert "image_mime" not in payload + assert payload["frame_index"] == 1 + + +def test_evidence_payload_keeps_only_frame_metadata(): + ev = make_evidence() + ev["frame"]["secret"] = "should-not-leak" + payload = evidence_payload(ev) + assert "secret" not in payload["frame"] + assert set(payload["frame"]) == {"sheet_size", "kind", "layer", "bbox", "confidence"} + + +def test_evidence_payload_carries_structured_data(): + ev = make_evidence(entities={"total": 9, "items": []}, texts={"total": 2}) + payload = evidence_payload(ev) + assert payload["entities"] == {"total": 9, "items": []} + assert payload["texts"] == {"total": 2} + + +# --------------------------------------------------------------------------- +# _run_readonly_tool +# --------------------------------------------------------------------------- +def test_readonly_rejects_non_whitelisted_tool(): + client = FakeMcpClient() + ctx = make_ctx(client=client) + content, budget = _run_readonly_tool(ctx, "get_some_unknown_tool", {}, [3]) + assert "不在只读取证白名单内" in content[0]["text"] + assert client.calls == [] + assert budget == [3] + + +def test_readonly_rejects_write_tool(): + client = FakeMcpClient() + ctx = make_ctx(client=client) + content, _ = _run_readonly_tool(ctx, "add_leader_annotation", {}, [3]) + assert "不在只读取证白名单内" in content[0]["text"] + assert client.calls == [] + + +def test_readonly_whitelist_covers_expected_tools(): + assert {"query_entities", "get_entity_detail", "zoom_to_window"} <= READ_ONLY_TOOLS + + +def test_readonly_handles_tool_error(): + client = FakeMcpClient(responses={"query_entities": McpToolError("炸了")}) + ctx = make_ctx(client=client) + content, _ = _run_readonly_tool(ctx, "query_entities", {"window": {}}, [3]) + assert "执行失败" in content[0]["text"] + + +def test_readonly_text_and_image_content(): + client = FakeMcpClient( + responses={"query_entities": ok_mixed(["hi"], [("QUJD", "image/png")])} + ) + ctx = make_ctx(client=client) + content, budget = _run_readonly_tool(ctx, "query_entities", {}, [3]) + assert content[0] == {"type": "text", "text": "hi"} + assert content[1]["type"] == "image_url" + assert budget == [2] + + +def test_readonly_image_budget_exhausted(): + client = FakeMcpClient( + responses={"query_entities": ok_mixed(["t"], [("QUJD", "image/png")])} + ) + ctx = make_ctx(client=client) + content, budget = _run_readonly_tool(ctx, "query_entities", {}, [0]) + assert budget == [0] + assert any("已省略" in c.get("text", "") for c in content) + + +def test_readonly_empty_response_marker(): + empty = {"jsonrpc": "2.0", "id": 1, "result": {"content": []}} + client = FakeMcpClient(responses={"query_entities": empty}) + ctx = make_ctx(client=client) + content, _ = _run_readonly_tool(ctx, "query_entities", {}, [3]) + assert content == [{"type": "text", "text": "（无返回内容）"}] + + +# --------------------------------------------------------------------------- +# review_frame +# --------------------------------------------------------------------------- +def test_review_frame_returns_text_when_no_tools(): + llm = FakeLlmClient(script=[llm_text("结论A")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + out = review_frame(ctx, make_evidence(), "系统提示", log=lambda *a: None) + assert out == "结论A" + assert len(llm.calls) == 1 + assert llm.calls[0]["tools"] is not None # 传了只读工具 schema + + +def test_review_frame_includes_image_block_when_present(): + llm = FakeLlmClient(script=[llm_text("ok")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + ev = make_evidence(image_b64="QUJD", image_mime="image/png") + review_frame(ctx, ev, "sys", log=lambda *a: None) + user_content = llm.calls[0]["messages"][1]["content"] + assert any(b.get("type") == "image_url" for b in user_content) + + +def test_review_frame_omits_image_block_when_absent(): + llm = FakeLlmClient(script=[llm_text("ok")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + user_content = llm.calls[0]["messages"][1]["content"] + assert all(b.get("type") == "text" for b in user_content) + + +def test_review_frame_runs_tool_then_concludes(): + llm = FakeLlmClient( + script=[ + llm_calls([("query_entities", {"window": {}})]), + llm_text("结论B"), + ] + ) + client = FakeMcpClient(responses={"query_entities": ok_text({"total": 1})}) + ctx = make_ctx(client=client, llm=llm) + out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + assert out == "结论B" + assert len(llm.calls) == 2 + assert client.called_names() == ["query_entities"] + + +def test_review_frame_feeds_observation_back_as_tool_message(): + llm = FakeLlmClient(script=[llm_calls([("query_entities", {})]), llm_text("done")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + tool_msgs = [m for m in llm.calls[1]["messages"] if isinstance(m, dict) and m.get("role") == "tool"] + assert tool_msgs and tool_msgs[0]["tool_call_id"] == "1" + + +def test_review_frame_invalid_tool_arguments_become_empty(): + llm = FakeLlmClient( + script=[llm_calls([("query_entities", "not-json")]), llm_text("done")] + ) + client = FakeMcpClient() + ctx = make_ctx(client=client, llm=llm) + review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + assert client.arguments_of("query_entities") == {} + + +def test_review_frame_force_closes_at_iteration_limit(): + llm = FakeLlmClient( + script=[ + llm_calls([("query_entities", {})]), + llm_text("收口结论"), + ] + ) + ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=1) + out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + assert out == "收口结论" + last_msg = llm.calls[-1]["messages"][-1] + assert last_msg["role"] == "user" + assert "上限" in last_msg["content"] + + +def test_review_frame_with_zero_iters_forces_immediate_close(): + llm = FakeLlmClient(script=[llm_text("直接结论")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm, max_iters=0) + out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + assert out == "直接结论" + assert len(llm.calls) == 1 + + +def test_review_frame_logs_tool_usage(): + logs = [] + llm = FakeLlmClient(script=[llm_calls([("get_layers", {})]), llm_text("x")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + review_frame(ctx, make_evidence(), "sys", log=logs.append) + assert any("取证" in line and "get_layers" in line for line in logs) + + +def test_review_frame_payload_is_json_serializable(): + llm = FakeLlmClient(script=[llm_text("ok")]) + ctx = make_ctx(client=FakeMcpClient(), llm=llm) + review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + text = llm.calls[0]["messages"][1]["content"][0]["text"] + json.loads(text.split("：\n", 1)[1]) # 不抛异常即通过