From 48f5765c1f1762c6954d0657f46c581aa91d8cb9 Mon Sep 17 00:00:00 2001 From: xiaoli-cli Date: Sat, 25 Apr 2026 21:18:35 +0800 Subject: [PATCH] =?UTF-8?q?cleanup:=20=E5=88=A0=E9=99=A4=E6=97=A0=E7=94=A8?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=A4=B9=E5=92=8C=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 tests/(v2测试,引用已删除的xiaoli/core模块) - 删除 开发文档/(v2开发文档) - 删除 __pycache__/(Python缓存) --- tests/__init__.py | 1 - tests/conftest.py | 183 --------- tests/test_config.py | 137 ------- tests/test_conversation.py | 226 ------------ tests/test_engines.py | 107 ------ tests/test_history.py | 114 ------ tests/test_integration.py | 154 -------- tests/test_plugins.py | 240 ------------ tests/test_sandbox.py | 576 ----------------------------- 开发文档/开发文档.md | 738 ------------------------------------- 10 files changed, 2476 deletions(-) delete mode 100644 tests/__init__.py delete mode 100644 tests/conftest.py delete mode 100644 tests/test_config.py delete mode 100644 tests/test_conversation.py delete mode 100644 tests/test_engines.py delete mode 100644 tests/test_history.py delete mode 100644 tests/test_integration.py delete mode 100644 tests/test_plugins.py delete mode 100644 tests/test_sandbox.py delete mode 100644 开发文档/开发文档.md diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index 1e9b826..0000000 --- a/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# xiaoli-cli test suite diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index c47a4ea..0000000 --- a/tests/conftest.py +++ /dev/null @@ -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", "", "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 diff --git a/tests/test_config.py b/tests/test_config.py deleted file mode 100644 index df0a421..0000000 --- a/tests/test_config.py +++ /dev/null @@ -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 diff --git a/tests/test_conversation.py b/tests/test_conversation.py deleted file mode 100644 index a978785..0000000 --- a/tests/test_conversation.py +++ /dev/null @@ -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 = '{"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 = '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 deeplyHere 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 = '' - result = process_thinking(resp) - assert result == '' or result is not None - - def test_custom_markers(self): - class FakeEngine: - thinking_start_marker = '' - thinking_end_marker = '' - - resp = 'deep thoughtThe 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"} diff --git a/tests/test_engines.py b/tests/test_engines.py deleted file mode 100644 index f9c7a02..0000000 --- a/tests/test_engines.py +++ /dev/null @@ -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) - # 至少应该尝试加载(可能因缺少依赖而失败) - # 但不应崩溃 diff --git a/tests/test_history.py b/tests/test_history.py deleted file mode 100644 index 028162c..0000000 --- a/tests/test_history.py +++ /dev/null @@ -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 diff --git a/tests/test_integration.py b/tests/test_integration.py deleted file mode 100644 index b30edc0..0000000 --- a/tests/test_integration.py +++ /dev/null @@ -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 diff --git a/tests/test_plugins.py b/tests/test_plugins.py deleted file mode 100644 index f26a60b..0000000 --- a/tests/test_plugins.py +++ /dev/null @@ -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 diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py deleted file mode 100644 index 4d3e4d5..0000000 --- a/tests/test_sandbox.py +++ /dev/null @@ -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", "", "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", "", "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() diff --git a/开发文档/开发文档.md b/开发文档/开发文档.md deleted file mode 100644 index 391ecc5..0000000 --- a/开发文档/开发文档.md +++ /dev/null @@ -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="", # 思考开始标记 - end_marker="", # 思考结束标记 - 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, '', '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 代码规范 -- [ ] 进行了基本测试