cleanup: 删除无用文件夹和缓存
- 删除 tests/（v2测试，引用已删除的xiaoli/core模块） - 删除 开发文档/（v2开发文档） - 删除 __pycache__/（Python缓存）
This commit is contained in:
xiaoli-cli committed
1 parent
4d341a7b55
commit
48f5765c1f
10 files changed
-2476
No files matched your search
@@ -1 +0,0 @@
|
||||
# xiaoli-cli test suite
|
||||
@@ -1,183 +0,0 @@
|
||||
"""共享测试 fixtures"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import tempfile
|
||||
import shutil
|
||||
import pytest
|
||||
|
||||
# 确保项目根目录在 sys.path 中
|
||||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, PROJECT_ROOT)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_dir():
|
||||
"""创建临时目录，测试后清理"""
|
||||
d = tempfile.mkdtemp(prefix='xiaoli_test_')
|
||||
yield d
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_python_code():
|
||||
"""安全的示例 Python 代码"""
|
||||
return '''
|
||||
import json
|
||||
import math
|
||||
|
||||
result = math.sqrt(144)
|
||||
data = {"value": result, "squared": result ** 2}
|
||||
print(json.dumps(data))
|
||||
'''
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dangerous_codes():
|
||||
"""各种危险代码样本"""
|
||||
return {
|
||||
'os_import': 'import os\nos.system("rm -rf /")',
|
||||
'subprocess': 'import subprocess\nsubprocess.run(["ls"])',
|
||||
'exec_call': 'exec("import os")',
|
||||
'eval_call': 'eval("__import__(\'os\')")',
|
||||
'dunder': 'x = (1).__class__.__bases__[0].__subclasses__()',
|
||||
'open_file': 'f = open("/etc/passwd")',
|
||||
'getattr_bypass': 'getattr(__builtins__, "exec")',
|
||||
'socket': 'import socket\ns = socket.socket()',
|
||||
'shutil': 'import shutil\nshutil.rmtree("/")',
|
||||
'compile': 'compile("import os", "<string>", "exec")',
|
||||
'globals': 'globals()["__builtins__"]',
|
||||
'breakpoint': 'breakpoint()',
|
||||
'del_stmt': 'del some_var',
|
||||
'global_stmt': 'global x',
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def safe_codes():
|
||||
"""各种安全代码样本"""
|
||||
return {
|
||||
'math': 'import math\nresult = math.pi',
|
||||
'json': 'import json\nresult = json.dumps({"a": 1})',
|
||||
're': 'import re\nresult = re.findall(r"\\d+", "abc123")',
|
||||
'datetime': 'from datetime import datetime\nresult = datetime.now().year',
|
||||
'collections': 'from collections import Counter\nresult = Counter("aabbc")',
|
||||
'list_comp': 'result = [x**2 for x in range(10)]',
|
||||
'function': 'def add(a, b): return a + b\nresult = add(3, 4)',
|
||||
'string_ops': 'result = "hello world".upper().split()',
|
||||
'sorting': 'result = sorted([3,1,2], reverse=True)',
|
||||
'fizzbuzz': 'result = ["fizz"*(i%3==0)+"buzz"*(i%5==0) or str(i) for i in range(1,16)]',
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_file(tmp_dir):
|
||||
"""创建临时配置文件"""
|
||||
config = {
|
||||
"system": {
|
||||
"default_engine": "ollama",
|
||||
"max_history": 100,
|
||||
"code_execution_timeout": 10,
|
||||
},
|
||||
"api": {
|
||||
"engines": {
|
||||
"ollama": {"base_url": "http://localhost:11434"},
|
||||
"mimo": {"api_key": "test-key", "base_url": "http://test.com"},
|
||||
}
|
||||
},
|
||||
"plugins": {},
|
||||
}
|
||||
path = os.path.join(tmp_dir, "config.json")
|
||||
with open(path, 'w') as f:
|
||||
json.dump(config, f)
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def history_dir(tmp_dir):
|
||||
"""创建临时历史目录"""
|
||||
d = os.path.join(tmp_dir, "chat_history")
|
||||
os.makedirs(d)
|
||||
return d
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plugins_dir(tmp_dir):
|
||||
"""创建带示例插件的目录"""
|
||||
d = os.path.join(tmp_dir, "plugins")
|
||||
os.makedirs(d)
|
||||
|
||||
# __init__.py
|
||||
with open(os.path.join(d, "__init__.py"), 'w') as f:
|
||||
f.write("")
|
||||
|
||||
# 简单插件
|
||||
with open(os.path.join(d, "test_plugin.py"), 'w') as f:
|
||||
f.write('''
|
||||
class Plugin:
|
||||
def __init__(self):
|
||||
self.cli = None
|
||||
|
||||
def set_cli(self, cli):
|
||||
self.cli = cli
|
||||
|
||||
def get_tool_info(self):
|
||||
return {
|
||||
"name": "test_tool",
|
||||
"description": "A test tool",
|
||||
"keywords": ["test"],
|
||||
}
|
||||
|
||||
def get_mcp_definition(self):
|
||||
return {
|
||||
"name": "test_tool",
|
||||
"description": "A test tool",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"operation": {"type": "string"},
|
||||
},
|
||||
"required": ["operation"]
|
||||
}
|
||||
}
|
||||
|
||||
def convert_mcp_args(self, arguments):
|
||||
return arguments.get("operation", "")
|
||||
|
||||
def handle(self, args):
|
||||
return f"test result: {args}"
|
||||
''')
|
||||
|
||||
return d
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def skills_dir(tmp_dir):
|
||||
"""创建带示例 skill 的目录"""
|
||||
d = os.path.join(tmp_dir, "skills")
|
||||
os.makedirs(d)
|
||||
|
||||
with open(os.path.join(d, "__init__.py"), 'w') as f:
|
||||
f.write("")
|
||||
|
||||
with open(os.path.join(d, "test_skill.py"), 'w') as f:
|
||||
f.write('''
|
||||
class Skill:
|
||||
def __init__(self):
|
||||
self.cli = None
|
||||
|
||||
def set_cli(self, cli):
|
||||
self.cli = cli
|
||||
|
||||
def get_tool_info(self):
|
||||
return {
|
||||
"name": "test_skill",
|
||||
"description": "A test skill",
|
||||
"keywords": ["test", "skill"],
|
||||
}
|
||||
|
||||
def handle(self, args):
|
||||
return f"skill result: {args}"
|
||||
''')
|
||||
|
||||
return d
|
||||
@@ -1,137 +0,0 @@
|
||||
"""配置管理测试"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli import config as cfg
|
||||
|
||||
|
||||
class TestConfig:
|
||||
"""配置加载/保存测试"""
|
||||
|
||||
def setup_method(self):
|
||||
"""重置缓存"""
|
||||
cfg._cache = None
|
||||
|
||||
def test_default_config(self):
|
||||
"""无配置文件时使用默认值"""
|
||||
# 临时指向不存在的文件
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = '/tmp/nonexistent_config.json'
|
||||
cfg._cache = None
|
||||
|
||||
result = cfg.load()
|
||||
assert 'system' in result
|
||||
assert result['system']['default_engine'] == 'ollama'
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_load_existing_config(self, config_file):
|
||||
"""加载现有配置文件"""
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_file
|
||||
cfg._cache = None
|
||||
|
||||
result = cfg.load()
|
||||
assert result['system']['default_engine'] == 'ollama'
|
||||
assert result['system']['max_history'] == 100
|
||||
assert 'mimo' in result['api']['engines']
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_get_specific_key(self, config_file):
|
||||
"""获取特定配置项"""
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_file
|
||||
cfg._cache = None
|
||||
|
||||
assert cfg.get('default_engine') == 'ollama'
|
||||
assert cfg.get('max_history') == 100
|
||||
assert cfg.get('nonexistent', 'fallback') == 'fallback'
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_set_and_save(self, tmp_dir):
|
||||
"""设置并保存配置"""
|
||||
config_path = os.path.join(tmp_dir, "config.json")
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_path
|
||||
cfg._cache = None
|
||||
|
||||
cfg.set('default_engine', 'mimo')
|
||||
assert cfg.get('default_engine') == 'mimo'
|
||||
|
||||
# 验证文件持久化
|
||||
with open(config_path, 'r') as f:
|
||||
saved = json.load(f)
|
||||
assert saved['system']['default_engine'] == 'mimo'
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_reload(self, config_file):
|
||||
"""强制重新加载"""
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_file
|
||||
cfg._cache = None
|
||||
|
||||
cfg.load()
|
||||
assert cfg._cache is not None
|
||||
|
||||
reloaded = cfg.reload()
|
||||
assert cfg._cache is not None
|
||||
assert 'system' in reloaded
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_caching(self, config_file):
|
||||
"""配置缓存机制"""
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_file
|
||||
cfg._cache = None
|
||||
|
||||
r1 = cfg.load()
|
||||
r2 = cfg.load()
|
||||
assert r1 is r2 # 同一对象（缓存）
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_corrupted_json(self, tmp_dir):
|
||||
"""损坏的 JSON 文件处理"""
|
||||
config_path = os.path.join(tmp_dir, "config.json")
|
||||
with open(config_path, 'w') as f:
|
||||
f.write("{invalid json")
|
||||
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_path
|
||||
cfg._cache = None
|
||||
|
||||
result = cfg.load()
|
||||
assert 'system' in result # 应回退到默认值
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
|
||||
def test_missing_system_section(self, tmp_dir):
|
||||
"""缺少 system 节时的行为"""
|
||||
config_path = os.path.join(tmp_dir, "config.json")
|
||||
with open(config_path, 'w') as f:
|
||||
json.dump({"api": {}}, f)
|
||||
|
||||
original = cfg.CONFIG_FILE
|
||||
cfg.CONFIG_FILE = config_path
|
||||
cfg._cache = None
|
||||
|
||||
assert cfg.get('default_engine') is None
|
||||
assert cfg.get('default_engine', 'ollama') == 'ollama'
|
||||
|
||||
cfg.CONFIG_FILE = original
|
||||
cfg._cache = None
|
||||
@@ -1,226 +0,0 @@
|
||||
"""对话响应解析器测试"""
|
||||
import os
|
||||
import sys
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli.conversation import parse_response, process_thinking, process_code_blocks, extract_tool_calls
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 1. parse_response 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestParseResponse:
|
||||
"""响应解析测试"""
|
||||
|
||||
def test_pure_text(self):
|
||||
text, data = parse_response("hello world")
|
||||
assert text == "hello world"
|
||||
assert data is None
|
||||
|
||||
def test_empty_response(self):
|
||||
text, data = parse_response("")
|
||||
assert text == ""
|
||||
assert data is None
|
||||
|
||||
def test_pure_json_dict(self):
|
||||
resp = '{"action": "use_tool", "tool": "code_editor", "args": "read foo.py"}'
|
||||
text, data = parse_response(resp)
|
||||
assert text == ""
|
||||
assert isinstance(data, dict)
|
||||
assert data['action'] == 'use_tool'
|
||||
assert data['tool'] == 'code_editor'
|
||||
|
||||
def test_pure_json_list(self):
|
||||
resp = '[{"action": "use_tool", "tool": "t1", "args": "a1"}]'
|
||||
text, data = parse_response(resp)
|
||||
assert text == ""
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_json_in_code_block(self):
|
||||
resp = 'Here is the command:\n```json\n{"action": "use_tool", "tool": "git_tools", "args": "status"}\n```'
|
||||
text, data = parse_response(resp)
|
||||
assert 'command' in text.lower() or 'Here' in text
|
||||
assert isinstance(data, dict)
|
||||
assert data['tool'] == 'git_tools'
|
||||
|
||||
def test_text_before_json(self):
|
||||
resp = 'I will edit the file.\n{"action": "use_tool", "tool": "code_editor", "args": "edit foo.py"}'
|
||||
text, data = parse_response(resp)
|
||||
assert 'edit' in text.lower()
|
||||
assert isinstance(data, dict)
|
||||
assert data['action'] == 'use_tool'
|
||||
|
||||
def test_continue_action(self):
|
||||
resp = '{"action": "continue", "content": "more info"}'
|
||||
text, data = parse_response(resp)
|
||||
assert text == ""
|
||||
assert isinstance(data, dict)
|
||||
assert data['action'] == 'continue'
|
||||
|
||||
def test_inline_json_use_tool(self):
|
||||
resp = 'Let me check. {"action":"use_tool","tool":"code_search","args":"find . test"}'
|
||||
text, data = parse_response(resp)
|
||||
assert isinstance(data, dict)
|
||||
assert data['tool'] == 'code_search'
|
||||
|
||||
def test_inline_json_continue(self):
|
||||
resp = 'Some text. {"action":"continue","content":"extra"}'
|
||||
text, data = parse_response(resp)
|
||||
assert isinstance(data, dict)
|
||||
assert data['action'] == 'continue'
|
||||
|
||||
def test_multiple_json_per_line(self):
|
||||
resp = '{"action":"use_tool","tool":"t1","args":"a1"}\n{"action":"use_tool","tool":"t2","args":"a2"}'
|
||||
text, data = parse_response(resp)
|
||||
# Should pick up at least one
|
||||
assert data is not None
|
||||
|
||||
def test_invalid_json_ignored(self):
|
||||
resp = 'This is not json {broken'
|
||||
text, data = parse_response(resp)
|
||||
assert text == resp
|
||||
assert data is None
|
||||
|
||||
def test_tool_call_tag_format(self):
|
||||
resp = '<tool_call>{"action":"use_tool","tool":"test","args":"arg"}'
|
||||
text, data = parse_response(resp)
|
||||
assert isinstance(data, dict)
|
||||
assert data['tool'] == 'test'
|
||||
|
||||
def test_message_field(self):
|
||||
resp = '{"message": "done processing"}'
|
||||
text, data = parse_response(resp)
|
||||
assert isinstance(data, dict)
|
||||
assert data['message'] == 'done processing'
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 2. process_thinking 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestProcessThinking:
|
||||
"""思考标记处理测试"""
|
||||
|
||||
def test_think_tag(self):
|
||||
resp = '<think/>This is the reply'
|
||||
result = process_thinking(resp)
|
||||
assert result == 'This is the reply'
|
||||
|
||||
def test_think_tag_with_thinking(self):
|
||||
resp = 'I am thinking deeply<think/>Here is my answer'
|
||||
result = process_thinking(resp)
|
||||
assert 'answer' in result
|
||||
|
||||
def test_no_think_tag(self):
|
||||
resp = 'Just a normal response'
|
||||
result = process_thinking(resp)
|
||||
assert result == resp
|
||||
|
||||
def test_empty_after_think(self):
|
||||
resp = '<think/>'
|
||||
result = process_thinking(resp)
|
||||
assert result == '' or result is not None
|
||||
|
||||
def test_custom_markers(self):
|
||||
class FakeEngine:
|
||||
thinking_start_marker = '<thinking>'
|
||||
thinking_end_marker = '</thinking>'
|
||||
|
||||
resp = '<thinking>deep thought</thinking>The answer is 42'
|
||||
result = process_thinking(resp, FakeEngine())
|
||||
assert '42' in result
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 3. process_code_blocks 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestProcessCodeBlocks:
|
||||
"""代码块处理测试"""
|
||||
|
||||
def test_triple_quote_block(self):
|
||||
text = 'before\n"""\ncode here\n"""\nafter'
|
||||
result = process_code_blocks(text)
|
||||
assert 'before' in result
|
||||
assert 'after' in result
|
||||
|
||||
def test_no_code_blocks(self):
|
||||
text = 'plain text no blocks'
|
||||
result = process_code_blocks(text)
|
||||
assert result == text
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 4. extract_tool_calls 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestExtractToolCalls:
|
||||
"""工具调用提取测试"""
|
||||
|
||||
def test_single_use_tool(self):
|
||||
data = {"action": "use_tool", "tool": "code_editor", "args": "edit foo.py"}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 'code_editor'
|
||||
|
||||
def test_list_of_use_tool(self):
|
||||
data = [
|
||||
{"action": "use_tool", "tool": "t1", "args": "a1"},
|
||||
{"action": "use_tool", "tool": "t2", "args": "a2"},
|
||||
]
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 2
|
||||
|
||||
def test_simplified_format(self):
|
||||
data = {"tool": "git_tools", "args": "status"}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 'git_tools'
|
||||
|
||||
def test_openai_function_call(self):
|
||||
data = {"function_call": {"name": "code_editor", "arguments": '{"operation": "read"}'}}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 'code_editor'
|
||||
|
||||
def test_mcp_format(self):
|
||||
data = {"name": "code_search", "arguments": {"operation": "find", "directory": "."}}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 'code_search'
|
||||
|
||||
def test_empty_data(self):
|
||||
assert extract_tool_calls([]) == []
|
||||
assert extract_tool_calls({}) == []
|
||||
assert extract_tool_calls(None) == []
|
||||
assert extract_tool_calls("string") == []
|
||||
|
||||
def test_non_tool_dict(self):
|
||||
data = {"message": "just text", "continue": True}
|
||||
calls = extract_tool_calls(data)
|
||||
# continue action should not be extracted as tool call
|
||||
assert len(calls) == 0
|
||||
|
||||
def test_mixed_list(self):
|
||||
data = [
|
||||
{"action": "use_tool", "tool": "t1", "args": "a1"},
|
||||
{"action": "continue", "content": "more"},
|
||||
]
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 't1'
|
||||
|
||||
def test_function_call_string_args(self):
|
||||
data = {"function_call": {"name": "test", "arguments": '{"key": "val"}'}}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['arguments'] == {"key": "val"}
|
||||
|
||||
def test_function_call_dict_args(self):
|
||||
data = {"function_call": {"name": "test", "arguments": {"key": "val"}}}
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['arguments'] == {"key": "val"}
|
||||
@@ -1,107 +0,0 @@
|
||||
"""AI 引擎管理器测试"""
|
||||
import os
|
||||
import sys
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli.engines import EngineManager
|
||||
|
||||
|
||||
class TestEngineManager:
|
||||
"""EngineManager 测试"""
|
||||
|
||||
def test_init_empty(self):
|
||||
mgr = EngineManager()
|
||||
assert mgr.engines == {}
|
||||
assert mgr.current is None
|
||||
|
||||
def test_set_default_existing(self):
|
||||
mgr = EngineManager()
|
||||
mgr.engines = {'test': 'fake_engine'}
|
||||
assert mgr.set_default('test')
|
||||
assert mgr.current == 'fake_engine'
|
||||
|
||||
def test_set_default_missing(self):
|
||||
mgr = EngineManager()
|
||||
assert not mgr.set_default('nonexistent')
|
||||
|
||||
def test_switch_existing(self):
|
||||
mgr = EngineManager()
|
||||
mgr.engines = {'a': 'engine_a', 'b': 'engine_b'}
|
||||
assert mgr.switch('b')
|
||||
assert mgr.current == 'engine_b'
|
||||
|
||||
def test_switch_missing(self):
|
||||
mgr = EngineManager()
|
||||
assert not mgr.switch('nope')
|
||||
|
||||
def test_current_name_none(self):
|
||||
mgr = EngineManager()
|
||||
assert mgr.current_name() == '未设置'
|
||||
|
||||
def test_current_name_set(self):
|
||||
mgr = EngineManager()
|
||||
mgr.current = type('E', (), {'name': 'ollama'})()
|
||||
assert mgr.current_name() == 'ollama'
|
||||
|
||||
def test_names(self):
|
||||
mgr = EngineManager()
|
||||
mgr.engines = {'a': 1, 'b': 2, 'c': 3}
|
||||
names = mgr.names()
|
||||
assert set(names) == {'a', 'b', 'c'}
|
||||
|
||||
def test_generate_no_engine(self):
|
||||
mgr = EngineManager()
|
||||
result = mgr.generate("hello")
|
||||
assert '错误' in result or 'error' in result.lower()
|
||||
|
||||
def test_generate_with_engine(self):
|
||||
mgr = EngineManager()
|
||||
|
||||
class FakeEngine:
|
||||
name = 'fake'
|
||||
def generate_response(self, user_input, system_prompt=None):
|
||||
return f"echo: {user_input}"
|
||||
|
||||
mgr.current = FakeEngine()
|
||||
result = mgr.generate("test input")
|
||||
assert 'test input' in result
|
||||
|
||||
def test_load_all_missing_dir(self):
|
||||
mgr = EngineManager()
|
||||
result = mgr.load_all('/tmp/nonexistent_engines_dir_12345')
|
||||
assert result == {}
|
||||
|
||||
def test_find_class_patterns(self):
|
||||
mgr = EngineManager()
|
||||
import types
|
||||
mod = types.ModuleType('test')
|
||||
|
||||
class TestAI:
|
||||
def generate_response(self, **kw):
|
||||
pass
|
||||
mod.TestAI = TestAI
|
||||
|
||||
found = mgr._find_class(mod, 'test')
|
||||
assert found == 'TestAI'
|
||||
|
||||
def test_find_class_no_match(self):
|
||||
mgr = EngineManager()
|
||||
import types
|
||||
mod = types.ModuleType('test')
|
||||
mod.Foo = 42 # not a class ending in AI
|
||||
found = mgr._find_class(mod, 'test')
|
||||
assert found is None
|
||||
|
||||
def test_load_all_real_engines(self):
|
||||
"""测试加载实际引擎目录（不验证引擎可用性，只验证不崩溃）"""
|
||||
mgr = EngineManager()
|
||||
engines_dir = os.path.join(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
'ai_engines'
|
||||
)
|
||||
if os.path.exists(engines_dir):
|
||||
mgr.load_all(engines_dir)
|
||||
# 至少应该尝试加载（可能因缺少依赖而失败）
|
||||
# 但不应崩溃
|
||||
@@ -1,114 +0,0 @@
|
||||
"""对话历史管理测试"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli.conversation.history import History
|
||||
|
||||
|
||||
class TestHistory:
|
||||
"""History 类测试"""
|
||||
|
||||
def test_add_message(self, history_dir):
|
||||
h = History(history_dir)
|
||||
h.add("user", "hello")
|
||||
h.add("assistant", "hi there")
|
||||
assert len(h.messages) == 2
|
||||
assert h.messages[0] == {"role": "user", "content": "hello"}
|
||||
assert h.messages[1] == {"role": "assistant", "content": "hi there"}
|
||||
|
||||
def test_get_messages(self, history_dir):
|
||||
h = History(history_dir)
|
||||
h.add("user", "q1")
|
||||
h.add("assistant", "a1")
|
||||
msgs = h.get()
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0]['role'] == 'user'
|
||||
|
||||
def test_clear(self, history_dir):
|
||||
h = History(history_dir)
|
||||
h.add("user", "hello")
|
||||
h.add("assistant", "hi")
|
||||
h.clear()
|
||||
assert len(h.messages) == 0
|
||||
|
||||
def test_max_items(self, history_dir):
|
||||
h = History(history_dir, max_items=3)
|
||||
for i in range(5):
|
||||
h.add("user", f"msg {i}")
|
||||
assert len(h.messages) == 3
|
||||
assert h.messages[0]['content'] == 'msg 2'
|
||||
|
||||
def test_save_and_load(self, history_dir):
|
||||
h = History(history_dir)
|
||||
h.add("user", "hello")
|
||||
h.add("assistant", "hi")
|
||||
assert h.save("test_chat")
|
||||
|
||||
h2 = History(history_dir)
|
||||
assert h2.load("test_chat")
|
||||
assert len(h2.messages) == 2
|
||||
assert h2.messages[0]['content'] == 'hello'
|
||||
|
||||
def test_load_nonexistent(self, history_dir, capsys):
|
||||
h = History(history_dir)
|
||||
result = h.load("nonexistent_chat")
|
||||
assert not result
|
||||
|
||||
def test_list_saved(self, history_dir, capsys):
|
||||
h = History(history_dir)
|
||||
h.add("user", "test")
|
||||
h.save("chat1")
|
||||
h.save("chat2")
|
||||
|
||||
h2 = History(history_dir)
|
||||
h2.list_saved() # should not raise
|
||||
captured = capsys.readouterr()
|
||||
assert 'chat1' in captured.out or 'chat2' in captured.out
|
||||
|
||||
def test_list_saved_empty(self, history_dir, capsys):
|
||||
h = History(history_dir)
|
||||
h.list_saved()
|
||||
captured = capsys.readouterr()
|
||||
assert 'no' in captured.out.lower() or '没有' in captured.out
|
||||
|
||||
def test_set_all(self, history_dir):
|
||||
h = History(history_dir)
|
||||
msgs = [
|
||||
{"role": "user", "content": "q"},
|
||||
{"role": "assistant", "content": "a"},
|
||||
]
|
||||
h.set_all(msgs)
|
||||
assert len(h.messages) == 2
|
||||
assert h.messages == msgs
|
||||
|
||||
def test_save_creates_directory(self, tmp_dir):
|
||||
nested_dir = os.path.join(tmp_dir, "deep", "nested", "history")
|
||||
h = History(nested_dir)
|
||||
h.add("user", "test")
|
||||
h.save("test")
|
||||
assert os.path.exists(os.path.join(nested_dir, "test.json"))
|
||||
|
||||
def test_save_file_format(self, history_dir):
|
||||
h = History(history_dir)
|
||||
h.add("user", "hello")
|
||||
h.save("format_test")
|
||||
|
||||
path = os.path.join(history_dir, "format_test.json")
|
||||
with open(path, 'r') as f:
|
||||
data = json.load(f)
|
||||
assert 'timestamp' in data
|
||||
assert 'conversation' in data
|
||||
assert len(data['conversation']) == 1
|
||||
|
||||
def test_load_corrupted_file(self, history_dir):
|
||||
path = os.path.join(history_dir, "corrupt.json")
|
||||
with open(path, 'w') as f:
|
||||
f.write("{bad json")
|
||||
|
||||
h = History(history_dir)
|
||||
result = h.load("corrupt")
|
||||
assert not result
|
||||
@@ -1,154 +0,0 @@
|
||||
"""集成测试 - 模块间协作"""
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli.sandbox import Sandbox
|
||||
from xiaoli.conversation import parse_response, extract_tool_calls
|
||||
from xiaoli.conversation.history import History
|
||||
|
||||
|
||||
class TestSandboxConversationIntegration:
|
||||
"""沙箱 + 对话解析 集成测试"""
|
||||
|
||||
def test_sandbox_result_parsed_as_tool_output(self):
|
||||
"""沙箱执行结果可以正确传递给对话系统"""
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run('result = [x**2 for x in range(5)]')
|
||||
assert result['success']
|
||||
|
||||
# 模拟工具结果注入对话
|
||||
tool_msg = f"工具执行结果:\n{result['result']}"
|
||||
text, data = parse_response(tool_msg)
|
||||
assert data is None # 纯文本，无工具调用
|
||||
assert '0' in text and '16' in text
|
||||
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_sandbox_failure_triggers_error_handling(self):
|
||||
"""沙箱失败时应正确处理"""
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run('import os') # should fail
|
||||
assert not result['success']
|
||||
|
||||
# 错误信息应可解析
|
||||
assert result['result']
|
||||
assert len(result['result']) > 0
|
||||
|
||||
sandbox.cleanup()
|
||||
|
||||
|
||||
class TestHistoryConversationIntegration:
|
||||
"""历史 + 对话解析 集成测试"""
|
||||
|
||||
def test_full_conversation_roundtrip(self, history_dir):
|
||||
"""完整的对话存取流程"""
|
||||
h = History(history_dir)
|
||||
|
||||
# 模拟对话
|
||||
h.add("user", "帮我写一个排序函数")
|
||||
h.add("assistant", "好的，这是一个快速排序：\n```python\ndef quicksort(arr):\n ...\n```")
|
||||
h.add("user", "能改成归并排序吗？")
|
||||
|
||||
# 保存
|
||||
assert h.save("sorting_chat")
|
||||
|
||||
# 加载到新实例
|
||||
h2 = History(history_dir)
|
||||
assert h2.load("sorting_chat")
|
||||
assert len(h2.messages) == 3
|
||||
assert "quicksort" in h2.messages[1]['content']
|
||||
|
||||
def test_conversation_with_tool_calls(self, history_dir):
|
||||
"""带工具调用的对话历史"""
|
||||
h = History(history_dir)
|
||||
|
||||
# 用户请求
|
||||
h.add("user", "查看 main.py 的第 10-20 行")
|
||||
|
||||
# AI 的工具调用
|
||||
tool_call = '{"action":"use_tool","tool":"code_editor","args":"read_range main.py 10 20"}'
|
||||
text, data = parse_response(tool_call)
|
||||
assert data is not None
|
||||
calls = extract_tool_calls(data)
|
||||
assert len(calls) == 1
|
||||
assert calls[0]['tool'] == 'code_editor'
|
||||
|
||||
# 记录工具结果
|
||||
h.add("assistant", f"工具调用: {json.dumps(data)}")
|
||||
h.add("assistant", "工具执行结果:\n line 10: def main():")
|
||||
|
||||
assert len(h.messages) == 3
|
||||
|
||||
|
||||
class TestSandboxSecurityIntegration:
|
||||
"""沙箱安全 + 工具链 集成测试"""
|
||||
|
||||
def test_safe_code_executes_in_chain(self):
|
||||
"""安全代码在工具链中正常执行"""
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
|
||||
# 步骤 1: 数学计算
|
||||
r1 = sandbox.run('import math\nresult = math.factorial(10)')
|
||||
assert r1['success']
|
||||
assert '3628800' in r1['result']
|
||||
|
||||
# 步骤 2: 使用上一步结果
|
||||
r2 = sandbox.run(
|
||||
'result = f"factorial(10) = {val}"',
|
||||
variables={'val': r1['result']}
|
||||
)
|
||||
assert r2['success']
|
||||
assert '3628800' in r2['result']
|
||||
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_blocked_code_stops_chain(self):
|
||||
"""被阻止的代码应中断工具链"""
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
|
||||
r = sandbox.run('import os\nresult = os.listdir(".")')
|
||||
assert not r['success']
|
||||
# 后续步骤不应执行
|
||||
assert r['result']
|
||||
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_subprocess_isolation(self):
|
||||
"""子进程模式下文件系统隔离"""
|
||||
sandbox = Sandbox(timeout=10, mode='subprocess')
|
||||
result = sandbox.run(
|
||||
'import tempfile\nresult = tempfile.gettempdir()'
|
||||
)
|
||||
# 子进程应使用沙箱临时目录
|
||||
if result['success']:
|
||||
assert 'xiaoli_sandbox' in result['result'] or result['success']
|
||||
|
||||
sandbox.cleanup()
|
||||
|
||||
|
||||
class TestPromptIntegration:
|
||||
"""系统提示词集成测试"""
|
||||
|
||||
def test_prompt_builder(self):
|
||||
from xiaoli import prompt as prompt_builder
|
||||
|
||||
# 无工具
|
||||
p1 = prompt_builder.build()
|
||||
assert '小狸' in p1
|
||||
assert '编程' in p1 or 'code' in p1.lower()
|
||||
|
||||
# 带工具
|
||||
p2 = prompt_builder.build(tool_prompts="- code_editor: edit code")
|
||||
assert 'code_editor' in p2
|
||||
|
||||
# 带 tool_search
|
||||
p3 = prompt_builder.build(has_tool_search=True)
|
||||
assert 'tool_search' in p3
|
||||
|
||||
# Clawli 模式
|
||||
p4 = prompt_builder.build(is_clawli=True)
|
||||
assert 'Clawli' in p4 or 'clawli' in p4 or '远程' in p4
|
||||
@@ -1,240 +0,0 @@
|
||||
"""插件系统测试"""
|
||||
import os
|
||||
import sys
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 1. code_editor 插件测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestCodeEditor:
|
||||
"""代码编辑器插件测试"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup(self, tmp_dir):
|
||||
sys.path.insert(0, os.path.join(PROJECT_ROOT, 'plugins'))
|
||||
from code_editor import Plugin
|
||||
self.plugin = Plugin()
|
||||
self.tmp_dir = tmp_dir
|
||||
|
||||
def test_tool_info(self):
|
||||
info = self.plugin.get_tool_info()
|
||||
assert info['name'] == 'code_editor'
|
||||
assert 'description' in info
|
||||
|
||||
def test_mcp_definition(self):
|
||||
mcp = self.plugin.get_mcp_definition()
|
||||
assert mcp['name'] == 'code_editor'
|
||||
assert 'inputSchema' in mcp
|
||||
assert 'operation' in mcp['inputSchema']['properties']
|
||||
|
||||
def test_convert_mcp_args(self):
|
||||
args = self.plugin.convert_mcp_args({"operation": "edit", "path": "test.py"})
|
||||
assert 'edit' in args
|
||||
|
||||
def test_create_file(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'test.py')
|
||||
result = self.plugin.handle(f'create {filepath}\nprint("hello")')
|
||||
assert os.path.exists(filepath)
|
||||
with open(filepath) as f:
|
||||
assert 'print("hello")' in f.read()
|
||||
|
||||
def test_read_range(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'read_test.py')
|
||||
with open(filepath, 'w') as f:
|
||||
f.write('line1\nline2\nline3\nline4\nline5\n')
|
||||
result = self.plugin.handle(f'read_range {filepath} 2 4')
|
||||
assert 'line2' in result
|
||||
assert 'line3' in result
|
||||
|
||||
def test_edit_file(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'edit_test.py')
|
||||
with open(filepath, 'w') as f:
|
||||
f.write('def old():\n pass\n')
|
||||
result = self.plugin.handle(f'edit {filepath} def old():\n pass <<<>>> def new():\n return True')
|
||||
with open(filepath) as f:
|
||||
content = f.read()
|
||||
assert 'def new():' in content
|
||||
assert 'return True' in content
|
||||
|
||||
def test_diff_files(self):
|
||||
f1 = os.path.join(self.tmp_dir, 'a.py')
|
||||
f2 = os.path.join(self.tmp_dir, 'b.py')
|
||||
with open(f1, 'w') as f:
|
||||
f.write('hello\nworld\n')
|
||||
with open(f2, 'w') as f:
|
||||
f.write('hello\nuniverse\n')
|
||||
result = self.plugin.handle(f'diff {f1} {f2}')
|
||||
assert result is not None # should produce diff output
|
||||
|
||||
def test_search_code(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'search.py')
|
||||
with open(filepath, 'w') as f:
|
||||
f.write('def foo():\n return 42\ndef bar():\n return foo()\n')
|
||||
result = self.plugin.handle(f'search {self.tmp_dir} main *.py')
|
||||
assert 'main' in result
|
||||
|
||||
def test_write_file(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'write.py')
|
||||
result = self.plugin.handle(f'write {filepath}\nresult = 42\n')
|
||||
assert os.path.exists(filepath)
|
||||
with open(filepath) as f:
|
||||
assert 'result = 42' in f.read()
|
||||
|
||||
def test_append_file(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'append.py')
|
||||
with open(filepath, 'w') as f:
|
||||
f.write('line1\n')
|
||||
self.plugin.handle(f'append {filepath}\nline2\n')
|
||||
with open(filepath) as f:
|
||||
content = f.read()
|
||||
assert 'line1' in content
|
||||
assert 'line2' in content
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 2. code_search 插件测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestCodeSearch:
|
||||
"""代码搜索插件测试"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup(self, tmp_dir):
|
||||
sys.path.insert(0, os.path.join(PROJECT_ROOT, 'plugins'))
|
||||
from code_search import Plugin
|
||||
self.plugin = Plugin()
|
||||
self.tmp_dir = tmp_dir
|
||||
|
||||
# 创建测试文件
|
||||
with open(os.path.join(tmp_dir, 'main.py'), 'w') as f:
|
||||
f.write('import os\ndef main():\n print("hello")\n\nif __name__ == "__main__":\n main()\n')
|
||||
with open(os.path.join(tmp_dir, 'utils.py'), 'w') as f:
|
||||
f.write('def helper():\n return 42\ndef process(data):\n return [x*2 for x in data]\n')
|
||||
with open(os.path.join(tmp_dir, 'test_main.py'), 'w') as f:
|
||||
f.write('# TODO: add tests\ndef test_main():\n assert True\n')
|
||||
|
||||
def test_tool_info(self):
|
||||
info = self.plugin.get_tool_info()
|
||||
assert info['name'] == 'code_search'
|
||||
|
||||
def test_structure(self):
|
||||
result = self.plugin.handle(f'structure {self.tmp_dir}')
|
||||
assert 'main.py' in result
|
||||
assert 'utils.py' in result
|
||||
|
||||
def test_find(self):
|
||||
result = self.plugin.handle(f'find {self.tmp_dir} def *.py')
|
||||
assert 'main' in result or 'helper' in result or 'def' in result
|
||||
|
||||
def test_symbols(self):
|
||||
result = self.plugin.handle(f'symbols {self.tmp_dir} main.py')
|
||||
assert 'main' in result
|
||||
|
||||
def test_todo(self):
|
||||
result = self.plugin.handle(f'todo {self.tmp_dir}')
|
||||
assert 'TODO' in result
|
||||
|
||||
def test_stats(self):
|
||||
result = self.plugin.handle(f'stats {self.tmp_dir}')
|
||||
assert result is not None
|
||||
|
||||
def test_context(self):
|
||||
filepath = os.path.join(self.tmp_dir, 'main.py')
|
||||
result = self.plugin.handle(f'context {filepath} 2 2')
|
||||
assert 'main' in result or 'def' in result
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 3. git_tools 插件测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestGitTools:
|
||||
"""Git 工具插件测试"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup(self, tmp_dir):
|
||||
sys.path.insert(0, os.path.join(PROJECT_ROOT, 'plugins'))
|
||||
from git_tools import Plugin
|
||||
self.plugin = Plugin()
|
||||
self.tmp_dir = tmp_dir
|
||||
|
||||
def test_tool_info(self):
|
||||
info = self.plugin.get_tool_info()
|
||||
assert info['name'] == 'git_tools'
|
||||
|
||||
def test_mcp_definition(self):
|
||||
mcp = self.plugin.get_mcp_definition()
|
||||
assert mcp['name'] == 'git_tools'
|
||||
assert 'inputSchema' in mcp
|
||||
|
||||
def test_status_non_git_dir(self):
|
||||
result = self.plugin.handle(f'status')
|
||||
# 在非 git 目录应返回错误或提示
|
||||
assert result is not None
|
||||
|
||||
def test_help_text(self):
|
||||
assert 'git' in self.plugin.usage.lower() or 'Git' in self.plugin.usage
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 4. unified_tool_manager 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestUnifiedToolManager:
|
||||
"""统一工具管理器测试"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup(self, plugins_dir, skills_dir):
|
||||
from unified_tool_manager import UnifiedToolManager
|
||||
self.manager = UnifiedToolManager()
|
||||
self.plugins_dir = plugins_dir
|
||||
self.skills_dir = skills_dir
|
||||
|
||||
def test_initialize(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
assert len(self.manager._tools) > 0
|
||||
|
||||
def test_tools_registered(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
assert 'test_tool' in self.manager._tools
|
||||
|
||||
def test_protocol_map(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
assert 'test_tool' in self.manager._protocol_map
|
||||
|
||||
def test_execute_plugin(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
result = self.manager.execute('test_tool', 'hello')
|
||||
assert result is not None
|
||||
assert 'hello' in str(result) or 'test' in str(result).lower()
|
||||
|
||||
def test_execute_nonexistent(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
result = self.manager.execute('nonexistent_tool', 'args')
|
||||
assert result is None or 'not found' in str(result).lower() or '未找到' in str(result)
|
||||
|
||||
def test_mcp_cache(self):
|
||||
self.manager.initialize(self.plugins_dir, self.skills_dir)
|
||||
cache = self.manager._mcp_tools_cache
|
||||
assert cache is not None
|
||||
assert len(cache) > 0
|
||||
|
||||
def test_empty_dirs(self, tmp_dir):
|
||||
empty_plugins = os.path.join(tmp_dir, 'empty_p')
|
||||
empty_skills = os.path.join(tmp_dir, 'empty_s')
|
||||
os.makedirs(empty_plugins)
|
||||
os.makedirs(empty_skills)
|
||||
with open(os.path.join(empty_plugins, '__init__.py'), 'w') as f:
|
||||
pass
|
||||
with open(os.path.join(empty_skills, '__init__.py'), 'w') as f:
|
||||
pass
|
||||
|
||||
mgr = __import__('unified_tool_manager').UnifiedToolManager()
|
||||
mgr.initialize(empty_plugins, empty_skills)
|
||||
assert len(mgr._tools) == 0
|
||||
@@ -1,576 +0,0 @@
|
||||
"""沙箱安全测试 - 测试所有安全层级"""
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from xiaoli.sandbox import Sandbox, SandboxResult, ASTSecurityChecker
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 1. AST 安全检查器测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestASTSecurityChecker:
|
||||
"""AST 静态分析测试"""
|
||||
|
||||
def setup_method(self):
|
||||
self.checker = ASTSecurityChecker()
|
||||
|
||||
# ── 应该拒绝的 ──
|
||||
|
||||
def test_block_os_import(self):
|
||||
safe, violations = self.checker.check('import os')
|
||||
assert not safe
|
||||
assert any('os' in v for v in violations)
|
||||
|
||||
def test_block_subprocess_import(self):
|
||||
safe, violations = self.checker.check('import subprocess')
|
||||
assert not safe
|
||||
|
||||
def test_block_from_import(self):
|
||||
safe, violations = self.checker.check('from os import system')
|
||||
assert not safe
|
||||
assert any('os' in v for v in violations)
|
||||
|
||||
def test_block_exec_call(self):
|
||||
safe, violations = self.checker.check('exec("print(1)")')
|
||||
assert not safe
|
||||
assert any('exec' in v for v in violations)
|
||||
|
||||
def test_block_eval_call(self):
|
||||
safe, violations = self.checker.check('x = eval("1+1")')
|
||||
assert not safe
|
||||
|
||||
def test_block_compile(self):
|
||||
safe, violations = self.checker.check('compile("x=1", "<s>", "exec")')
|
||||
assert not safe
|
||||
|
||||
def test_block_globals(self):
|
||||
safe, violations = self.checker.check('globals()')
|
||||
assert not safe
|
||||
|
||||
def test_block_locals(self):
|
||||
safe, violations = self.checker.check('locals()')
|
||||
assert not safe
|
||||
|
||||
def test_block_getattr(self):
|
||||
safe, violations = self.checker.check('getattr(obj, "name")')
|
||||
assert not safe
|
||||
|
||||
def test_block_setattr(self):
|
||||
safe, violations = self.checker.check('setattr(obj, "name", 1)')
|
||||
assert not safe
|
||||
|
||||
def test_block_dunder_class(self):
|
||||
safe, violations = self.checker.check('x = object.__class__')
|
||||
assert not safe
|
||||
|
||||
def test_block_dunder_subclasses(self):
|
||||
safe, violations = self.checker.check('x = object.__subclasses__()')
|
||||
assert not safe
|
||||
|
||||
def test_block_open(self):
|
||||
safe, violations = self.checker.check('f = open("/etc/passwd")')
|
||||
assert not safe
|
||||
|
||||
def test_block_breakpoint(self):
|
||||
safe, violations = self.checker.check('breakpoint()')
|
||||
assert not safe
|
||||
|
||||
def test_block_socket_import(self):
|
||||
safe, violations = self.checker.check('import socket')
|
||||
assert not safe
|
||||
|
||||
def test_block_pickle_import(self):
|
||||
safe, violations = self.checker.check('import pickle')
|
||||
assert not safe
|
||||
|
||||
def test_block_ctypes_import(self):
|
||||
safe, violations = self.checker.check('import ctypes')
|
||||
assert not safe
|
||||
|
||||
def test_block_importlib(self):
|
||||
safe, violations = self.checker.check('import importlib')
|
||||
assert not safe
|
||||
|
||||
def test_block_del(self):
|
||||
safe, violations = self.checker.check('del x')
|
||||
assert not safe
|
||||
|
||||
def test_block_global_stmt(self):
|
||||
safe, violations = self.checker.check('global x')
|
||||
assert not safe
|
||||
|
||||
def test_block_nested_dunder(self):
|
||||
safe, violations = self.checker.check(
|
||||
'x = (1).__class__.__bases__[0].__subclasses__()'
|
||||
)
|
||||
assert not safe
|
||||
|
||||
def test_block_sys_import(self):
|
||||
safe, violations = self.checker.check('import sys')
|
||||
assert not safe
|
||||
|
||||
def test_block_shutil(self):
|
||||
safe, violations = self.checker.check('import shutil')
|
||||
assert not safe
|
||||
|
||||
def test_block_urllib(self):
|
||||
safe, violations = self.checker.check('import urllib')
|
||||
assert not safe
|
||||
|
||||
def test_block_http(self):
|
||||
safe, violations = self.checker.check('import http')
|
||||
assert not safe
|
||||
|
||||
# ── 应该允许的 ──
|
||||
|
||||
def test_allow_math(self):
|
||||
safe, _ = self.checker.check('import math\nresult = math.pi')
|
||||
assert safe
|
||||
|
||||
def test_allow_json(self):
|
||||
safe, _ = self.checker.check('import json\nresult = json.dumps({"a": 1})')
|
||||
assert safe
|
||||
|
||||
def test_allow_re(self):
|
||||
safe, _ = self.checker.check('import re\nresult = re.findall(r"\\d+", "abc")')
|
||||
assert safe
|
||||
|
||||
def test_allow_datetime(self):
|
||||
safe, _ = self.checker.check('from datetime import datetime\nresult = datetime.now()')
|
||||
assert safe
|
||||
|
||||
def test_allow_collections(self):
|
||||
safe, _ = self.checker.check('from collections import Counter\nresult = Counter("aab")')
|
||||
assert safe
|
||||
|
||||
def test_allow_list_comprehension(self):
|
||||
safe, _ = self.checker.check('result = [x**2 for x in range(10)]')
|
||||
assert safe
|
||||
|
||||
def test_allow_function_def(self):
|
||||
safe, _ = self.checker.check('def add(a, b): return a + b\nresult = add(1, 2)')
|
||||
assert safe
|
||||
|
||||
def test_allow_string_ops(self):
|
||||
safe, _ = self.checker.check('result = "hello".upper()')
|
||||
assert safe
|
||||
|
||||
def test_allow_sorting(self):
|
||||
safe, _ = self.checker.check('result = sorted([3,1,2])')
|
||||
assert safe
|
||||
|
||||
def test_allow_typing(self):
|
||||
safe, _ = self.checker.check('from typing import List\nresult: List[int] = [1,2,3]')
|
||||
assert safe
|
||||
|
||||
def test_allow_enum(self):
|
||||
safe, _ = self.checker.check('from enum import Enum')
|
||||
assert safe
|
||||
|
||||
def test_allow_decimal(self):
|
||||
safe, _ = self.checker.check('from decimal import Decimal\nresult = Decimal("3.14")')
|
||||
assert safe
|
||||
|
||||
def test_allow_itertools(self):
|
||||
safe, _ = self.checker.check('import itertools\nresult = list(itertools.chain([1],[2]))')
|
||||
assert safe
|
||||
|
||||
# ── 语法错误 ──
|
||||
|
||||
def test_syntax_error(self):
|
||||
safe, violations = self.checker.check('def foo(')
|
||||
assert not safe
|
||||
assert any('syntax' in v.lower() for v in violations)
|
||||
|
||||
def test_empty_code(self):
|
||||
safe, _ = self.checker.check('')
|
||||
assert safe
|
||||
|
||||
# ── 多种违规组合 ──
|
||||
|
||||
def test_multiple_violations(self):
|
||||
code = 'import os\nimport socket\nexec("x=1")\nglobals()'
|
||||
safe, violations = self.checker.check(code)
|
||||
assert not safe
|
||||
assert len(violations) >= 4
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 2. Sandbox 核心功能测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxBasic:
|
||||
"""沙箱基本功能测试"""
|
||||
|
||||
def setup_method(self):
|
||||
self.sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
|
||||
def teardown_method(self):
|
||||
self.sandbox.cleanup()
|
||||
|
||||
def test_simple_execution(self):
|
||||
result = self.sandbox.run('result = 1 + 2')
|
||||
assert result['success']
|
||||
assert result['result'] == '3'
|
||||
|
||||
def test_string_manipulation(self):
|
||||
result = self.sandbox.run('result = "hello world".split()')
|
||||
assert result['success']
|
||||
assert 'hello' in result['result']
|
||||
|
||||
def test_math_operations(self):
|
||||
result = self.sandbox.run('import math\nresult = math.sqrt(144)')
|
||||
assert result['success']
|
||||
assert '12.0' in result['result']
|
||||
|
||||
def test_json_operations(self):
|
||||
result = self.sandbox.run(
|
||||
'import json\nresult = json.dumps({"key": "value"})'
|
||||
)
|
||||
assert result['success']
|
||||
assert 'key' in result['result']
|
||||
|
||||
def test_print_capture(self):
|
||||
result = self.sandbox.run('print("hello")\nresult = "done"')
|
||||
assert result['success']
|
||||
assert 'hello' in result['stdout']
|
||||
|
||||
def test_variables_injection(self):
|
||||
result = self.sandbox.run(
|
||||
'result = x + y',
|
||||
variables={'x': 10, 'y': 20}
|
||||
)
|
||||
assert result['success']
|
||||
assert '30' in result['result']
|
||||
|
||||
def test_function_definition(self):
|
||||
code = '''
|
||||
def factorial(n):
|
||||
if n <= 1:
|
||||
return 1
|
||||
return n * factorial(n - 1)
|
||||
result = factorial(5)
|
||||
'''
|
||||
result = self.sandbox.run(code)
|
||||
assert result['success']
|
||||
assert '120' in result['result']
|
||||
|
||||
def test_list_comprehension(self):
|
||||
result = self.sandbox.run('result = [x**2 for x in range(5)]')
|
||||
assert result['success']
|
||||
assert '[0, 1, 4, 9, 16]' in result['result']
|
||||
|
||||
def test_dict_operations(self):
|
||||
code = '''
|
||||
data = {"a": 1, "b": 2, "c": 3}
|
||||
result = {k: v*2 for k, v in data.items()}
|
||||
'''
|
||||
result = self.sandbox.run(code)
|
||||
assert result['success']
|
||||
|
||||
def test_datetime(self):
|
||||
code = 'from datetime import datetime\nresult = datetime.now().year'
|
||||
result = self.sandbox.run(code)
|
||||
assert result['success']
|
||||
|
||||
def test_collections_counter(self):
|
||||
code = 'from collections import Counter\nresult = Counter("aabbc")'
|
||||
result = self.sandbox.run(code)
|
||||
assert result['success']
|
||||
|
||||
def test_regex(self):
|
||||
code = 'import re\nresult = re.findall(r"\\d+", "abc123def456")'
|
||||
result = self.sandbox.run(code)
|
||||
assert result['success']
|
||||
assert '123' in result['result']
|
||||
|
||||
def test_no_result_variable(self):
|
||||
result = self.sandbox.run('x = 42')
|
||||
assert result['success']
|
||||
assert 'no return' in result['result'].lower() or 'ok' in result['result'].lower()
|
||||
|
||||
def test_syntax_error(self):
|
||||
result = self.sandbox.run('def foo(')
|
||||
assert not result['success']
|
||||
|
||||
def test_runtime_error(self):
|
||||
result = self.sandbox.run('result = 1 / 0')
|
||||
assert not result['success']
|
||||
assert 'error' in result['result'].lower() or 'division' in result['result'].lower()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 3. 安全拒绝测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxSecurity:
|
||||
"""沙箱安全拒绝测试"""
|
||||
|
||||
def setup_method(self):
|
||||
self.sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
|
||||
def teardown_method(self):
|
||||
self.sandbox.cleanup()
|
||||
|
||||
def test_reject_os_import(self):
|
||||
result = self.sandbox.run('import os\nresult = os.getcwd()')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_subprocess(self):
|
||||
result = self.sandbox.run('import subprocess\nsubprocess.run(["ls"])')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_exec(self):
|
||||
result = self.sandbox.run('exec("result = 1")')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_eval(self):
|
||||
result = self.sandbox.run('result = eval("1+1")')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_open(self):
|
||||
result = self.sandbox.run('f = open("/etc/passwd")\nresult = f.read()')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_socket(self):
|
||||
result = self.sandbox.run('import socket\ns = socket.socket()')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_pickle(self):
|
||||
result = self.sandbox.run('import pickle\nresult = pickle.dumps({})')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_sys(self):
|
||||
result = self.sandbox.run('import sys\nresult = sys.version')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_getattr(self):
|
||||
result = self.sandbox.run('result = getattr(str, "upper")')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_dunder(self):
|
||||
result = self.sandbox.run('result = object.__class__')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_breakpoint(self):
|
||||
result = self.sandbox.run('breakpoint()')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_compile(self):
|
||||
result = self.sandbox.run('compile("x=1", "<s>", "exec")')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_globals(self):
|
||||
result = self.sandbox.run('result = globals()')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_importlib(self):
|
||||
result = self.sandbox.run('import importlib')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_shutil(self):
|
||||
result = self.sandbox.run('import shutil')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_ctypes(self):
|
||||
result = self.sandbox.run('import ctypes')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_multiprocessing(self):
|
||||
result = self.sandbox.run('import multiprocessing')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_threading(self):
|
||||
result = self.sandbox.run('import threading')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_http(self):
|
||||
result = self.sandbox.run('import http')
|
||||
assert not result['success']
|
||||
|
||||
def test_reject_urllib(self):
|
||||
result = self.sandbox.run('import urllib')
|
||||
assert not result['success']
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 4. 超时测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxTimeout:
|
||||
"""超时机制测试"""
|
||||
|
||||
def test_timeout_infinite_loop(self):
|
||||
sandbox = Sandbox(timeout=2, mode='threaded')
|
||||
result = sandbox.run('while True: pass')
|
||||
assert not result['success']
|
||||
assert result['killed']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_timeout_sleep(self):
|
||||
sandbox = Sandbox(timeout=2, mode='threaded')
|
||||
result = sandbox.run('import time\ntime.sleep(30)\nresult = "done"')
|
||||
# Should either timeout or fail on import
|
||||
assert not result['success'] or result['killed']
|
||||
sandbox.cleanup()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 5. SandboxResult 测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxResult:
|
||||
"""SandboxResult 数据类测试"""
|
||||
|
||||
def test_success_result(self):
|
||||
r = SandboxResult(success=True, result="42", time=0.1)
|
||||
assert r.success
|
||||
d = r.to_dict()
|
||||
assert d['success']
|
||||
assert d['result'] == '42'
|
||||
assert d['time'] == 0.1
|
||||
|
||||
def test_failure_result(self):
|
||||
r = SandboxResult(success=False, result="error", security_violation="blocked os")
|
||||
assert not r.success
|
||||
d = r.to_dict()
|
||||
assert not d['success']
|
||||
assert d['security_violation'] == 'blocked os'
|
||||
|
||||
def test_str_representation(self):
|
||||
r = SandboxResult(success=True, result="ok", time=0.05)
|
||||
s = str(r)
|
||||
assert 'OK' in s
|
||||
assert '0.050' in s
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 6. 沙箱模式对比测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxModes:
|
||||
"""threaded vs subprocess 模式对比"""
|
||||
|
||||
def test_threaded_mode(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run('result = 2 + 3')
|
||||
assert result['success']
|
||||
assert '5' in result['result']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_subprocess_mode(self):
|
||||
sandbox = Sandbox(timeout=10, mode='subprocess')
|
||||
result = sandbox.run('result = 2 + 3')
|
||||
assert result['success']
|
||||
assert '5' in result['result']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_both_modes_same_result(self):
|
||||
code = 'import json\nresult = json.dumps({"x": 42})'
|
||||
|
||||
t = Sandbox(timeout=10, mode='threaded')
|
||||
s = Sandbox(timeout=10, mode='subprocess')
|
||||
|
||||
r_threaded = t.run(code)
|
||||
r_subprocess = s.run(code)
|
||||
|
||||
assert r_threaded['success'] == r_subprocess['success']
|
||||
assert r_threaded['result'] == r_subprocess['result']
|
||||
|
||||
t.cleanup()
|
||||
s.cleanup()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 7. 禁用沙箱测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxDisabled:
|
||||
"""沙箱禁用时的行为"""
|
||||
|
||||
def test_disabled_rejects_all(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
sandbox.enabled = False
|
||||
result = sandbox.run('result = 1')
|
||||
assert not result['success']
|
||||
assert 'disabled' in result['result'].lower() or 'off' in result['security_violation'].lower()
|
||||
sandbox.cleanup()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 8. 历史记录测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxHistory:
|
||||
"""沙箱执行历史记录"""
|
||||
|
||||
def test_history_recorded(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
sandbox.run('result = 1')
|
||||
sandbox.run('result = 2')
|
||||
assert len(sandbox.history) == 2
|
||||
assert sandbox.history[0]['success']
|
||||
assert sandbox.history[1]['success']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_failed_history_recorded(self):
|
||||
sandbox = Sandbox(timeout=10, mode="threaded")
|
||||
result = sandbox.run("x = 1 / 0") # runtime error, not security
|
||||
assert not result["success"]
|
||||
assert len(sandbox.history) == 1
|
||||
assert not sandbox.history[0]["success"]
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_security_violation_no_history(self):
|
||||
sandbox = Sandbox(timeout=10, mode="threaded")
|
||||
result = sandbox.run("import os") # security violation
|
||||
assert not result["success"]
|
||||
# security violations may not record history
|
||||
sandbox.cleanup()
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════
|
||||
# 9. 变量注入安全测试
|
||||
# ═══════════════════════════════════════════════
|
||||
|
||||
class TestSandboxVariables:
|
||||
"""变量注入测试"""
|
||||
|
||||
def test_inject_numbers(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run('result = a + b', variables={'a': 100, 'b': 200})
|
||||
assert result['success']
|
||||
assert '300' in result['result']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_inject_strings(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run('result = name.upper()', variables={'name': 'hello'})
|
||||
assert result['success']
|
||||
assert 'HELLO' in result['result']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_inject_dict(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run(
|
||||
'result = data["key"]',
|
||||
variables={'data': {'key': 'found'}}
|
||||
)
|
||||
assert result['success']
|
||||
assert 'found' in result['result']
|
||||
sandbox.cleanup()
|
||||
|
||||
def test_inject_list(self):
|
||||
sandbox = Sandbox(timeout=10, mode='threaded')
|
||||
result = sandbox.run(
|
||||
'result = sum(items)',
|
||||
variables={'items': [1, 2, 3, 4, 5]}
|
||||
)
|
||||
assert result['success']
|
||||
assert '15' in result['result']
|
||||
sandbox.cleanup()
|
||||
-738
@@ -1,738 +0,0 @@
|
||||
# 小狸 Pro-CLI 开发文档
|
||||
|
||||
> 版本：2.0 | 更新日期：2026年3月
|
||||
|
||||
---
|
||||
|
||||
## 目录
|
||||
|
||||
1. [系统架构概述](#1-系统架构概述)
|
||||
2. [AI引擎开发规范](#2-ai引擎开发规范)
|
||||
3. [Plugin协议开发](#3-plugin协议开发)
|
||||
4. [Skill协议开发](#4-skill协议开发)
|
||||
5. [MCP协议支持](#5-mcp协议支持)
|
||||
6. [WebSocket API](#6-websocket-api)
|
||||
7. [已加载模块列表](#7-已加载模块列表)
|
||||
8. [最佳实践](#8-最佳实践)
|
||||
9. [调试与测试](#9-调试与测试)
|
||||
10. [常见问题](#10-常见问题)
|
||||
|
||||
---
|
||||
|
||||
## 1. 系统架构概述
|
||||
|
||||
### 1.1 三大协议支持
|
||||
|
||||
小狸 Pro-CLI 的工具系统支持三种协议：
|
||||
|
||||
| 协议 | 目录 | 说明 |
|
||||
|------|------|------|
|
||||
| **Plugin 协议** | `plugins/` | 小狸自有协议，简单易用 |
|
||||
| **Skill 协议** | `skills/` | Agent Skills 开源协议，支持 Markdown 和 Python 格式 |
|
||||
| **MCP 协议** | - | Model Context Protocol，Anthropic/OpenAI 标准 |
|
||||
|
||||
### 1.2 统一工具管理器
|
||||
|
||||
`unified_tool_manager.py` 协调三种协议，提供统一的工具调用接口：
|
||||
|
||||
```python
|
||||
from unified_tool_manager import UnifiedToolManager
|
||||
|
||||
manager = UnifiedToolManager(cli_instance)
|
||||
manager.initialize("plugins", "skills")
|
||||
|
||||
# Plugin 格式调用
|
||||
result = manager.execute("file_manager", "read test.txt")
|
||||
|
||||
# MCP 格式调用
|
||||
result = manager.mcp_call("file_manager", {"operation": "read", "path": "test.txt"})
|
||||
```
|
||||
|
||||
### 1.3 路由策略
|
||||
|
||||
- 优先使用 Skill 协议（如果存在同名技能）
|
||||
- 回退到 Plugin 协议
|
||||
- 支持强制指定协议
|
||||
- MCP 格式调用统一入口
|
||||
|
||||
---
|
||||
|
||||
## 2. AI引擎开发规范
|
||||
|
||||
### 2.1 文件与类命名规范
|
||||
|
||||
| 规范 | 说明 |
|
||||
|------|------|
|
||||
| 文件命名 | 必须以 `_engine.py` 结尾 |
|
||||
| 类命名 | 使用驼峰命名，如 `Qwen3CoderPlusAI` |
|
||||
|
||||
**示例：**
|
||||
- 文件：`qwen3-coder-plus_engine.py`
|
||||
- 类名：`Qwen3CoderPlusAI`
|
||||
|
||||
### 2.2 引擎配置声明
|
||||
|
||||
```python
|
||||
class MyEngineAI:
|
||||
# 配置属性
|
||||
requires_api_key = True # 是否需要API密钥
|
||||
default_base_url = "https://api.example.com/v1"
|
||||
default_model = "my-model"
|
||||
|
||||
def __init__(self):
|
||||
self.name = "my_engine"
|
||||
# 初始化代码...
|
||||
```
|
||||
|
||||
### 2.3 核心接口
|
||||
|
||||
```python
|
||||
def generate_response(self, user_input, tool_results=None, system_prompt=None):
|
||||
"""
|
||||
生成AI响应
|
||||
|
||||
参数:
|
||||
user_input: 用户输入文本
|
||||
tool_results: 工具执行结果（可选）
|
||||
system_prompt: 系统提示词（可选，由主程序生成）
|
||||
|
||||
返回:
|
||||
AI响应字符串
|
||||
"""
|
||||
pass
|
||||
|
||||
def get_help_info(self):
|
||||
"""返回引擎帮助信息"""
|
||||
return "引擎帮助信息..."
|
||||
```
|
||||
|
||||
### 2.4 禁止设置超时
|
||||
|
||||
```python
|
||||
# ✅ 正确
|
||||
response = requests.post(url, json=body, headers=headers, timeout=None)
|
||||
|
||||
# ❌ 错误
|
||||
response = requests.post(url, json=body, headers=headers, timeout=30)
|
||||
```
|
||||
|
||||
### 2.5 SDK快速开发
|
||||
|
||||
使用 SDK，只需几行代码就能完成引擎开发：
|
||||
|
||||
```python
|
||||
from SDK_openAI import *
|
||||
|
||||
class MyAIAI:
|
||||
def __init__(self):
|
||||
engine_Online("on") # 云模式
|
||||
ai_name("my_ai")
|
||||
ai_api_url("https://api.example.com/v1")
|
||||
ai_api_key("your_api_key")
|
||||
ai_model_id("gpt-3.5-turbo")
|
||||
|
||||
self.name = "my_ai"
|
||||
self.cli = None
|
||||
self.shared_conversation_history = None
|
||||
|
||||
def set_cli(self, cli):
|
||||
set_cli(cli)
|
||||
self.cli = cli
|
||||
|
||||
def generate_response(self, user_input, tool_results=None, system_prompt=None):
|
||||
set_shared_history(self.shared_conversation_history)
|
||||
return generate_response(user_input, tool_results, system_prompt)
|
||||
```
|
||||
|
||||
### 2.6 SDK方法说明
|
||||
|
||||
| 方法 | 说明 |
|
||||
|------|------|
|
||||
| `engine_Online("on"/"off")` | 云模式/本地模式 |
|
||||
| `ai_name(name)` | 设置引擎名称 |
|
||||
| `ai_api_url(url)` | 设置API地址 |
|
||||
| `ai_api_key(key)` | 设置API密钥 |
|
||||
| `ai_model_id(model)` | 设置模型名称 |
|
||||
| `ai_thinking_engine(...)` | 深度思考引擎配置 |
|
||||
|
||||
### 2.7 深度思考引擎配置
|
||||
|
||||
适用于 QwQ、DeepSeek-R1 等会输出思考过程的模型：
|
||||
|
||||
```python
|
||||
ai_thinking_engine(
|
||||
enable=True, # 启用深度思考
|
||||
start_marker="<thinking>", # 思考开始标记
|
||||
end_marker="</thinking>", # 思考结束标记
|
||||
show_thinking=True # 显示思考过程
|
||||
)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 3. Plugin协议开发
|
||||
|
||||
### 3.1 基本结构
|
||||
|
||||
```python
|
||||
class Plugin:
|
||||
"""插件描述"""
|
||||
|
||||
def __init__(self):
|
||||
self.usage = "使用方法说明"
|
||||
self.cli = None
|
||||
|
||||
def set_cli(self, cli):
|
||||
"""设置CLI引用（可选）"""
|
||||
self.cli = cli
|
||||
|
||||
def get_tool_info(self):
|
||||
"""返回工具信息（必需）"""
|
||||
return {
|
||||
"name": "tool_name",
|
||||
"description": "工具描述",
|
||||
"keywords": ["关键词1", "关键词2"],
|
||||
"usage": self.usage
|
||||
}
|
||||
|
||||
def get_mcp_definition(self):
|
||||
"""返回MCP格式定义（可选）"""
|
||||
return {
|
||||
"name": "tool_name",
|
||||
"description": "工具描述",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"operation": {"type": "string"},
|
||||
"path": {"type": "string"}
|
||||
},
|
||||
"required": ["operation"]
|
||||
}
|
||||
}
|
||||
|
||||
def convert_mcp_args(self, arguments):
|
||||
"""转换MCP参数（可选）"""
|
||||
return f"{arguments.get('operation', '')} {arguments.get('path', '')}"
|
||||
|
||||
def handle(self, args: str) -> str:
|
||||
"""处理工具调用（必需）"""
|
||||
try:
|
||||
# 实现逻辑
|
||||
return "处理结果"
|
||||
except Exception as e:
|
||||
return f"错误: {str(e)}"
|
||||
```
|
||||
|
||||
### 3.2 完整示例：文件管理器
|
||||
|
||||
```python
|
||||
import os
|
||||
from typing import Dict, Any, List
|
||||
|
||||
class Plugin:
|
||||
"""文件管理器插件"""
|
||||
|
||||
def __init__(self):
|
||||
self.usage = """文件管理器使用方法：
|
||||
file_manager <操作> <参数>
|
||||
|
||||
操作类型:
|
||||
- list <路径> - 列出目录内容
|
||||
- read <文件路径> - 读取文件内容
|
||||
- write <文件路径> -c "内容" - 写入文件
|
||||
- delete <路径> - 删除文件或目录
|
||||
"""
|
||||
self.cli = None
|
||||
|
||||
def set_cli(self, cli):
|
||||
self.cli = cli
|
||||
|
||||
def get_tool_info(self):
|
||||
return {
|
||||
"name": "file_manager",
|
||||
"description": "文件和目录管理工具",
|
||||
"keywords": ["文件", "目录", "管理", "file"],
|
||||
"usage": self.usage
|
||||
}
|
||||
|
||||
def get_mcp_definition(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"name": "file_manager",
|
||||
"description": "文件和目录管理工具",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"operation": {
|
||||
"type": "string",
|
||||
"enum": ["list", "read", "write", "delete"],
|
||||
"description": "操作类型"
|
||||
},
|
||||
"path": {"type": "string", "description": "路径"},
|
||||
"content": {"type": "string", "description": "内容"}
|
||||
},
|
||||
"required": ["operation"]
|
||||
}
|
||||
}
|
||||
|
||||
def convert_mcp_args(self, arguments: Dict[str, Any]) -> str:
|
||||
op = arguments.get("operation", "")
|
||||
path = arguments.get("path", "")
|
||||
content = arguments.get("content", "")
|
||||
if content:
|
||||
return f'{op} {path} -c "{content}"'
|
||||
return f"{op} {path}"
|
||||
|
||||
def handle(self, args: str) -> str:
|
||||
parts = args.strip().split(maxsplit=2)
|
||||
if not parts:
|
||||
return "错误：请提供操作类型"
|
||||
|
||||
operation = parts[0].lower()
|
||||
|
||||
if operation == "list":
|
||||
return self._op_list(parts[1:] if len(parts) > 1 else ["."])
|
||||
elif operation == "read":
|
||||
return self._op_read(parts[1:] if len(parts) > 1 else [])
|
||||
elif operation == "write":
|
||||
return self._op_write(parts[1:])
|
||||
elif operation == "delete":
|
||||
return self._op_delete(parts[1:])
|
||||
|
||||
return f"错误：不支持的操作 '{operation}'"
|
||||
|
||||
def _op_list(self, args: List[str]) -> str:
|
||||
path = args[0] if args else "."
|
||||
if not os.path.exists(path):
|
||||
return f"错误：路径不存在: {path}"
|
||||
# 实现列出目录逻辑...
|
||||
return f"目录内容: {path}"
|
||||
|
||||
def _op_read(self, args: List[str]) -> str:
|
||||
if not args:
|
||||
return "错误：请提供文件路径"
|
||||
# 实现读取文件逻辑...
|
||||
return "文件内容..."
|
||||
|
||||
def _op_write(self, args: List[str]) -> str:
|
||||
if len(args) < 2:
|
||||
return "错误：请提供文件路径和内容"
|
||||
# 实现写入文件逻辑...
|
||||
return "写入成功"
|
||||
|
||||
def _op_delete(self, args: List[str]) -> str:
|
||||
if not args:
|
||||
return "错误：请提供路径"
|
||||
# 实现删除逻辑...
|
||||
return "删除成功"
|
||||
```
|
||||
|
||||
### 3.3 参数格式规范
|
||||
|
||||
**多操作插件格式：**
|
||||
```
|
||||
<操作类型> <目标对象> [可选内容]
|
||||
```
|
||||
示例：`read file.txt`、`write file.txt 内容`
|
||||
|
||||
**单参数插件格式：**
|
||||
```
|
||||
<待处理内容>
|
||||
```
|
||||
示例：`2+3`、`hello world`
|
||||
|
||||
---
|
||||
|
||||
## 4. Skill协议开发
|
||||
|
||||
### 4.1 Python格式技能
|
||||
|
||||
```python
|
||||
# skills/my_skill/__init__.py
|
||||
from skills.base import Skill, SkillResult
|
||||
|
||||
class MySkill(Skill):
|
||||
"""我的技能"""
|
||||
|
||||
name = "my_skill"
|
||||
description = "技能描述"
|
||||
tags = ["标签1", "标签2"]
|
||||
|
||||
def get_tool_definition(self):
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": self.name,
|
||||
"description": self.description,
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"operation": {"type": "string"}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
def run(self, **kwargs) -> SkillResult:
|
||||
try:
|
||||
operation = kwargs.get("operation", "")
|
||||
# 实现逻辑
|
||||
return SkillResult(
|
||||
success=True,
|
||||
message="执行成功",
|
||||
data={"result": "..."}
|
||||
)
|
||||
except Exception as e:
|
||||
return SkillResult(
|
||||
success=False,
|
||||
error=str(e)
|
||||
)
|
||||
```
|
||||
|
||||
### 4.2 Markdown格式技能
|
||||
|
||||
```markdown
|
||||
# skills/my_skill/SKILL.md
|
||||
|
||||
---
|
||||
name: my_skill
|
||||
description: 技能描述
|
||||
tags: [标签1, 标签2]
|
||||
---
|
||||
|
||||
# 技能说明
|
||||
|
||||
这是技能的详细使用说明...
|
||||
|
||||
## 使用方法
|
||||
|
||||
1. 步骤一
|
||||
2. 步骤二
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 5. MCP协议支持
|
||||
|
||||
### 5.1 MCP工具定义格式
|
||||
|
||||
```python
|
||||
{
|
||||
"name": "tool_name",
|
||||
"description": "工具描述",
|
||||
"inputSchema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"param1": {"type": "string", "description": "参数说明"},
|
||||
"param2": {"type": "integer", "description": "参数说明"}
|
||||
},
|
||||
"required": ["param1"]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 5.2 MCP调用格式
|
||||
|
||||
**请求：**
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": "tools/call",
|
||||
"params": {
|
||||
"name": "file_manager",
|
||||
"arguments": {
|
||||
"operation": "read",
|
||||
"path": "test.txt"
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**响应：**
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"result": {
|
||||
"content": [
|
||||
{"type": "text", "text": "文件内容..."}
|
||||
],
|
||||
"isError": false
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 5.3 统一工具管理器MCP接口
|
||||
|
||||
```python
|
||||
# 获取MCP工具列表
|
||||
tools = manager.get_mcp_tools()
|
||||
|
||||
# MCP格式调用
|
||||
result = manager.mcp_call("file_manager", {"operation": "read", "path": "test.txt"})
|
||||
|
||||
# 处理MCP请求
|
||||
response = manager.handle_mcp_request({
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"method": "tools/call",
|
||||
"params": {...}
|
||||
})
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 6. WebSocket API
|
||||
|
||||
### 6.1 消息类型
|
||||
|
||||
| 类型 | 方向 | 说明 |
|
||||
|------|------|------|
|
||||
| `heartbeat` | 双向 | 心跳检测 |
|
||||
| `auth` | 客户端→服务端 | 认证请求 |
|
||||
| `auth_success` / `auth_failed` | 服务端→客户端 | 认证结果 |
|
||||
| `message` | 双向 | 文本消息 |
|
||||
| `file` | 客户端→服务端 | 文件上传 |
|
||||
| `image` | 服务端→客户端 | 图片发送 |
|
||||
| `tool_status` | 服务端→客户端 | 工具状态 |
|
||||
|
||||
### 6.2 消息格式示例
|
||||
|
||||
**认证请求：**
|
||||
```json
|
||||
{"type": "auth", "password": "your_password"}
|
||||
```
|
||||
|
||||
**发送消息：**
|
||||
```json
|
||||
{"type": "message", "content": "你好"}
|
||||
```
|
||||
|
||||
**上传文件：**
|
||||
```json
|
||||
{
|
||||
"type": "file",
|
||||
"file": "base64_encoded_data",
|
||||
"filename": "example.txt",
|
||||
"text": "附加说明"
|
||||
}
|
||||
```
|
||||
|
||||
### 6.3 启动远程服务
|
||||
|
||||
```bash
|
||||
# 本地模式
|
||||
/remote start 9079 your_password
|
||||
|
||||
# 代理模式
|
||||
/remote proxy relay.example.com:9079 server_pwd 12345 pc_pwd
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 7. 已加载模块列表
|
||||
|
||||
### 7.1 AI引擎
|
||||
|
||||
| 引擎名 | 类型 | 说明 |
|
||||
|--------|------|------|
|
||||
| ollama | 本地 | Ollama本地模型，无需API密钥 |
|
||||
| glm | 云端 | 智谱GLM |
|
||||
| spark | 云端 | 讯飞星火 |
|
||||
| qwq | 云端 | QwQ深度思考引擎 |
|
||||
| qwen | 云端 | 通义千问 |
|
||||
|
||||
### 7.2 插件
|
||||
|
||||
| 插件名 | 功能 |
|
||||
|--------|------|
|
||||
| file_manager | 文件管理（14种操作） |
|
||||
| ai_search | AI搜索 |
|
||||
| audio_player | 音频播放 |
|
||||
| cmd_executor | 命令执行 |
|
||||
| frontend_tester | 前端测试 |
|
||||
| glados_tts_plugin | GLaDOS语音合成 |
|
||||
| napcat_qq | QQ机器人 |
|
||||
| network_tools | 网络工具 |
|
||||
| ollama_image | Ollama图像 |
|
||||
| send_image | 发送图片 |
|
||||
| speech_recognition | 语音识别 |
|
||||
| task_manager | 任务管理 |
|
||||
| tool_search | 工具搜索 |
|
||||
|
||||
---
|
||||
|
||||
## 8. 最佳实践
|
||||
|
||||
### 8.1 安全性
|
||||
|
||||
```python
|
||||
# ❌ 避免使用危险函数
|
||||
result = eval(user_input) # 危险！
|
||||
|
||||
# ✅ 安全的替代方案
|
||||
import ast
|
||||
try:
|
||||
tree = ast.parse(user_input, mode='eval')
|
||||
result = eval(compile(tree, '<string>', 'eval'))
|
||||
except:
|
||||
return "无效表达式"
|
||||
```
|
||||
|
||||
### 8.2 输入验证
|
||||
|
||||
```python
|
||||
def handle(self, args: str) -> str:
|
||||
# 参数验证
|
||||
if not args:
|
||||
return "错误：参数不能为空"
|
||||
|
||||
# 长度限制
|
||||
if len(args) > 10000:
|
||||
return "错误：输入过长"
|
||||
|
||||
# 格式验证
|
||||
if not re.match(r'^[\w\s\-\.]+$', args):
|
||||
return "错误：包含非法字符"
|
||||
|
||||
# 继续处理...
|
||||
```
|
||||
|
||||
### 8.3 错误处理
|
||||
|
||||
```python
|
||||
def handle(self, args: str) -> str:
|
||||
try:
|
||||
# 业务逻辑
|
||||
return "结果"
|
||||
except FileNotFoundError:
|
||||
return "错误：文件不存在"
|
||||
except PermissionError:
|
||||
return "错误：权限不足"
|
||||
except Exception as e:
|
||||
return f"错误: {str(e)}"
|
||||
```
|
||||
|
||||
### 8.4 资源管理
|
||||
|
||||
```python
|
||||
# 使用上下文管理器
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
|
||||
# 或确保资源释放
|
||||
try:
|
||||
f = open(file_path, 'r')
|
||||
content = f.read()
|
||||
finally:
|
||||
f.close()
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 9. 调试与测试
|
||||
|
||||
### 9.1 日志记录
|
||||
|
||||
```python
|
||||
def handle(self, args: str) -> str:
|
||||
if self.cli:
|
||||
print(f"[DEBUG] 插件被调用: {args}")
|
||||
# 处理逻辑
|
||||
return "结果"
|
||||
```
|
||||
|
||||
### 9.2 单元测试
|
||||
|
||||
```python
|
||||
import unittest
|
||||
from plugins.file_manager import Plugin
|
||||
|
||||
class TestFileManager(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.plugin = Plugin()
|
||||
|
||||
def test_get_tool_info(self):
|
||||
info = self.plugin.get_tool_info()
|
||||
self.assertIn("name", info)
|
||||
self.assertEqual(info["name"], "file_manager")
|
||||
|
||||
def test_handle_list(self):
|
||||
result = self.plugin.handle("list .")
|
||||
self.assertIsInstance(result, str)
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
```
|
||||
|
||||
### 9.3 集成测试
|
||||
|
||||
```python
|
||||
# integration_test.py
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.dirname(__file__))
|
||||
|
||||
from ai_cli import AICLI
|
||||
|
||||
def test_integration():
|
||||
cli = AICLI()
|
||||
# 测试插件调用...
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 10. 常见问题
|
||||
|
||||
### Q1: 插件不加载？
|
||||
|
||||
检查以下几点：
|
||||
- 文件是否放在 `plugins/` 目录
|
||||
- 类名是否为 `Plugin`
|
||||
- 是否实现了 `get_tool_info()` 和 `handle()` 方法
|
||||
- 文件名是否为 `.py` 结尾
|
||||
|
||||
### Q2: 插件不被调用？
|
||||
|
||||
检查：
|
||||
- 关键词是否匹配用户输入
|
||||
- `keywords` 列表是否正确设置
|
||||
- 插件是否成功加载（查看启动日志）
|
||||
|
||||
### Q3: MCP格式调用失败？
|
||||
|
||||
确保：
|
||||
- 实现了 `get_mcp_definition()` 方法
|
||||
- 实现了 `convert_mcp_args()` 方法（如果需要）
|
||||
- `inputSchema` 格式正确
|
||||
|
||||
### Q4: 如何调试？
|
||||
|
||||
- 在 `handle()` 方法中添加 `print()` 语句
|
||||
- 检查返回值是否为字符串
|
||||
- 查看主程序日志
|
||||
|
||||
### Q5: 如何处理大文件？
|
||||
|
||||
```python
|
||||
# 限制输出长度
|
||||
MAX_LENGTH = 10000
|
||||
if len(content) > MAX_LENGTH:
|
||||
content = content[:MAX_LENGTH] + "\n... (内容已截断)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 附录：插件开发检查清单
|
||||
|
||||
- [ ] 插件类名是 `Plugin`
|
||||
- [ ] 实现了 `get_tool_info()` 方法
|
||||
- [ ] 实现了 `handle(args)` 方法
|
||||
- [ ] 返回字符串结果
|
||||
- [ ] 包含 `usage` 属性
|
||||
- [ ] 可选：实现 `get_mcp_definition()` 支持 MCP
|
||||
- [ ] 可选：实现 `convert_mcp_args()` 转换参数
|
||||
- [ ] 所有异常都被捕获
|
||||
- [ ] 遵循参数格式规范
|
||||
- [ ] 遵循 PEP 8 代码规范
|
||||
- [ ] 进行了基本测试
|
||||
Reference in new issue
Block a user