zhaoxi
4aa1dab283
- control_node 标注 DEPRECATED：保留目录壳子供未来远程探针节点复用，删除调用路径与相关测试
- 新增 task 表：极简元数据持久化 regulatory_node 完成的短任务（出报告/写文件/查询整理）
- regulatory_node 自标注：MessageResponse 扩展 task_action/title/summary，_run 末尾非阻塞落库
- query_task_list 改查 task 表，符合用户对"任务列表"的直觉，与 workflow 体系解耦
- 新增 /api/v1/task/list|/{id} 只读 API（task 由 regulatory 内部触发，不开放对外创建）
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
"""``data/toolset/`` manifest.json 加载正确性测试。
|
|
|
|
验证 GlobalToolManager 从 manifest.json 正确读取工具元数据。
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
_toolset_dir = Path(__file__).parent.parent.parent / "data" / "toolset"
|
|
_base_toolset_dir = _toolset_dir / "base_toolset"
|
|
_interactive_toolset_dir = _toolset_dir / "interactive_toolset"
|
|
_regulatory_toolset_dir = _toolset_dir / "regulatory_toolset"
|
|
|
|
|
|
def _read_manifest(toolset_dir=_base_toolset_dir):
|
|
with open(toolset_dir / "manifest.json", "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _get_tool_def(manifest, name):
|
|
for tool in manifest["tools"]:
|
|
if tool["name"] == name:
|
|
return tool
|
|
return None
|
|
|
|
|
|
def test_manifest_json_exists():
|
|
assert (_base_toolset_dir / "manifest.json").exists()
|
|
assert (_interactive_toolset_dir / "manifest.json").exists()
|
|
assert (_regulatory_toolset_dir / "manifest.json").exists()
|
|
|
|
|
|
def test_manifest_has_all_tools():
|
|
base_names = {t["name"] for t in _read_manifest(_base_toolset_dir)["tools"]}
|
|
interactive_names = {
|
|
t["name"] for t in _read_manifest(_interactive_toolset_dir)["tools"]
|
|
}
|
|
regulatory_names = {t["name"] for t in _read_manifest(_regulatory_toolset_dir)["tools"]}
|
|
assert base_names == {
|
|
"shell_executor", "file_reader", "edit_file", "write_file",
|
|
"search_file", "python_executor", "tavily_search",
|
|
}
|
|
assert interactive_names == {"approval"}
|
|
assert regulatory_names == {"query_workflow_status", "query_task_list", "send_file"}
|
|
|
|
|
|
def test_approval_metadata():
|
|
manifest = _read_manifest(_interactive_toolset_dir)
|
|
tool = _get_tool_def(manifest, "approval")
|
|
assert tool["is_system"] is True
|
|
assert tool["category"] == "system"
|
|
assert tool["action_scope"] == []
|
|
|
|
|
|
def test_regulatory_toolset_scope():
|
|
manifest = _read_manifest(_regulatory_toolset_dir)
|
|
for name in ("query_workflow_status", "query_task_list", "send_file"):
|
|
tool = _get_tool_def(manifest, name)
|
|
assert tool is not None
|
|
assert tool["is_system"] is True
|
|
assert tool["action_scope"] == ["regulatory_node"]
|
|
|
|
|
|
def test_tavily_search_metadata():
|
|
manifest = _read_manifest()
|
|
tool = _get_tool_def(manifest, "tavily_search")
|
|
assert tool["is_system"] is False
|
|
assert tool["category"] == "search"
|
|
assert "consciousness_node" in tool["action_scope"]
|
|
assert "regulatory_node" in tool["action_scope"]
|
|
assert "control_node" not in tool["action_scope"]
|
|
assert "api_key" in tool["config_args"]
|
|
|
|
|
|
def test_all_tool_files_exist():
|
|
for toolset_dir in (_base_toolset_dir, _interactive_toolset_dir, _regulatory_toolset_dir):
|
|
manifest = _read_manifest(toolset_dir)
|
|
for tool in manifest["tools"]:
|
|
file_path = toolset_dir / tool["file"]
|
|
assert file_path.exists(), f"Missing tool file: {tool['file']}"
|
|
|
|
|
|
def test_tool_manager_loads_all_tools():
|
|
from kilostar.core.global_state_machine.tool_manager import GlobalToolManager
|
|
|
|
tm = GlobalToolManager()
|
|
# base_toolset(7) + interactive_toolset(1) + regulatory_toolset(3) = 11
|
|
assert len(tm.tool_metadata) == 11
|
|
assert "shell_executor" in tm.tool_metadata
|
|
assert "tavily_search" in tm.tool_metadata
|
|
assert "approval" in tm.tool_metadata
|
|
assert "query_workflow_status" in tm.tool_metadata
|
|
assert "query_task_list" in tm.tool_metadata
|
|
assert "send_file" in tm.tool_metadata
|
|
|
|
|
|
def test_tool_manager_system_vs_third_party():
|
|
from kilostar.core.global_state_machine.tool_manager import GlobalToolManager
|
|
|
|
tm = GlobalToolManager()
|
|
system_tools = tm.get_system_tools()
|
|
third_party = tm.get_third_party_tools()
|
|
system_names = {t["name"] for t in system_tools}
|
|
tp_names = {t["name"] for t in third_party}
|
|
assert "shell_executor" in system_names
|
|
assert "tavily_search" in tp_names
|
|
assert "tavily_search" not in system_names
|