"""Tools 额外覆盖率测试 - 覆盖未测试的分支."""
import asyncio
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestToolsAdditionalCoverage:
"""覆盖 tools.py 中未测试的分支."""
@pytest.fixture
def config_manager(self, temp_config_dir):
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def _call_handle_tool(self, tool_name, arguments, graphiti_client, ace_manager, config_manager):
"""辅助方法：执行异步 handle_tool_call 并返回文本结果。"""
async def _invoke():
result = await handle_tool_call(
tool_name=tool_name,
arguments=arguments,
config_manager=config_manager,
graphiti_client=graphiti_client,
ace_manager=ace_manager
)
texts = []
for item in result or []:
text = getattr(item, "text", None)
if text:
texts.append(text)
return "\n".join(texts)
return asyncio.run(_invoke())
@pytest.fixture
def graphiti_client(self):
client = MagicMock()
client.is_connected.return_value = True
session = MagicMock()
client.driver.session.return_value.__enter__.return_value = session
return client
@pytest.fixture
def ace_manager(self, config_manager, graphiti_client):
from src.ace_manager import ACEManager
with patch.object(ACEManager, "_initialize_ace", lambda self: None):
manager = ACEManager(config_manager, graphiti_client)
manager.enabled = True
manager.ace_agent = MagicMock()
# 将所有方法都 mock 为 MagicMock
manager.query_strategies = MagicMock()
manager.list_strategy_versions = MagicMock()
manager.get_learning_trends = MagicMock()
manager.bulk_update_strategies = MagicMock()
manager.bulk_export_strategies = MagicMock()
manager.export_strategies = MagicMock()
manager.import_strategies = MagicMock()
manager.toggle_strategy = MagicMock()
manager.validate_strategies = MagicMock()
manager.get_strategy_heatmap = MagicMock()
return manager
# ----- query_strategies 工具处理测试 -----
def test_query_strategies_with_results(self, config_manager, graphiti_client, ace_manager):
"""测试 query_strategies 工具返回结果。"""
strategies = [
{
"tool_name": "search_entities",
"success_rate": 0.9,
"usage_count": 10,
"success_count": 9,
"failure_count": 1,
"content": "Test strategy content" * 10 # 超过200字符
}
]
ace_manager.query_strategies.return_value = strategies
result = self._call_handle_tool(
"query_strategies",
{"tool_name": "search_entities", "limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "找到 1 个策略" in result
assert "search_entities" in result
def test_query_strategies_applies_ace_optimized_arguments(self, config_manager, graphiti_client, ace_manager):
"""ACE 返回优化参数时，应覆盖原始 arguments 并传递给查询方法。"""
strategies = [
{
"tool_name": "search_entities",
"success_rate": 0.95,
"usage_count": 5,
"success_count": 5,
"failure_count": 0,
"content": "Optimized strategy"
}
]
ace_manager.query_strategies.return_value = strategies
ace_manager.generate_tool_strategy = MagicMock(
return_value={"optimized_arguments": {"limit": 3}}
)
result = self._call_handle_tool(
"query_strategies",
{"tool_name": "search_entities", "limit": 10},
graphiti_client,
ace_manager,
config_manager
)
# 确认文本输出正常
assert "找到 1 个策略" in result
# 确认优化后的 limit 已传递给 ACEManager.query_strategies
kwargs = ace_manager.query_strategies.call_args.kwargs
assert kwargs["limit"] == 3
def test_query_strategies_no_results(self, config_manager, graphiti_client, ace_manager):
"""测试 query_strategies 工具无结果。"""
ace_manager.query_strategies.return_value = []
result = self._call_handle_tool(
"query_strategies",
{"tool_name": "test", "limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "暂无策略记录" in result
# ----- list_strategy_versions 工具处理测试 -----
def test_list_strategy_versions_with_results(self, config_manager, graphiti_client, ace_manager):
"""测试 list_strategy_versions 工具返回结果。"""
versions = [
{
"version": 2,
"is_latest": True,
"success_rate": 0.9,
"usage_count": 10,
"success_count": 9,
"failure_count": 1,
"arguments_hash": "hash123",
"updated_at": "2025-01-01"
},
{
"version": 1,
"is_latest": False,
"success_rate": 0.8,
"usage_count": 5,
"success_count": 4,
"failure_count": 1,
"arguments_hash": None,
"updated_at": None
}
]
ace_manager.list_strategy_versions.return_value = versions
result = self._call_handle_tool(
"list_strategy_versions",
{
"tool_name": "test_tool",
"arguments_hash": "hash123",
"limit": 10
},
graphiti_client,
ace_manager,
config_manager
)
assert "v2 (最新)" in result
assert "v1" in result
assert "hash123" in result
def test_list_strategy_versions_no_results(self, config_manager, graphiti_client, ace_manager):
"""测试 list_strategy_versions 工具无结果。"""
ace_manager.list_strategy_versions.return_value = []
result = self._call_handle_tool(
"list_strategy_versions",
{"tool_name": "test", "limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "暂无历史版本记录" in result
# ----- get_learning_trends 工具处理测试 -----
def test_get_learning_trends_with_results(self, config_manager, graphiti_client, ace_manager):
"""测试 get_learning_trends 工具返回结果。"""
trends = [
{
"date": "2025-12-01",
"success_rate": 0.9,
"usage_count": 10.0,
"success_count": 9.0,
"failure_count": 1.0,
"avg_rating": 4.5
},
{
"date": "2025-12-02",
"success_rate": 0.8,
"usage_count": 5.0,
"success_count": 4.0,
"failure_count": 1.0,
"avg_rating": None
}
]
ace_manager.get_learning_trends.return_value = trends
result = self._call_handle_tool(
"get_learning_trends",
{"tool_name": "test", "days": 7},
graphiti_client,
ace_manager,
config_manager
)
assert "学习趋势" in result
assert "平均评分 4.50" in result
assert "2025-12-01" in result
def test_get_learning_trends_no_results(self, config_manager, graphiti_client, ace_manager):
"""测试 get_learning_trends 工具无结果。"""
ace_manager.get_learning_trends.return_value = []
result = self._call_handle_tool(
"get_learning_trends",
{"tool_name": "test", "days": 7},
graphiti_client,
ace_manager,
config_manager
)
assert "暂无学习趋势数据" in result
def test_get_learning_trends_all_tools(self, config_manager, graphiti_client, ace_manager):
"""测试 get_learning_trends 工具查询所有工具。"""
trends = [{"date": "2025-12-01", "success_rate": 0.9, "usage_count": 10.0}]
ace_manager.get_learning_trends.return_value = trends
result = self._call_handle_tool(
"get_learning_trends",
{"days": 7},
graphiti_client,
ace_manager,
config_manager
)
assert "所有工具" in result
# ----- bulk_update_strategies 工具处理测试 -----
def test_bulk_update_strategies_with_outcome(self, config_manager, graphiti_client, ace_manager):
"""测试 bulk_update_strategies 工具有结果。"""
outcome = {
"action": "disable",
"affected": 5,
"limit": 10
}
ace_manager.bulk_update_strategies.return_value = outcome
result = self._call_handle_tool(
"bulk_update_strategies",
{
"action": "disable",
"success_rate_min": 0.0,
"success_rate_max": 0.5,
"limit": 10
},
graphiti_client,
ace_manager,
config_manager
)
assert "已批量禁用策略" in result
assert "5" in result
def test_bulk_update_strategies_no_outcome(self, config_manager, graphiti_client, ace_manager):
"""测试 bulk_update_strategies 工具无结果。"""
ace_manager.bulk_update_strategies.return_value = None
result = self._call_handle_tool(
"bulk_update_strategies",
{"action": "enable", "limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "未执行任何批量操作" in result
# ----- bulk_export_strategies 工具处理测试 -----
def test_bulk_export_strategies_with_file_path(self, config_manager, graphiti_client, ace_manager):
"""测试 bulk_export_strategies 工具导出到文件。"""
exported = {
"file_path": "/path/to/export.json",
"count": 10
}
ace_manager.bulk_export_strategies.return_value = exported
result = self._call_handle_tool(
"bulk_export_strategies",
{
"tool_name": "test",
"file_path": "/path/to/export.json",
"limit": 10
},
graphiti_client,
ace_manager,
config_manager
)
assert "已导出 10 个策略到文件" in result
assert "/path/to/export.json" in result
def test_bulk_export_strategies_in_memory(self, config_manager, graphiti_client, ace_manager):
"""测试 bulk_export_strategies 工具内存预览。"""
exported = {
"count": 5,
"strategies": [
{"tool_name": "tool1", "success_rate": 0.9, "usage_count": 10},
{"tool_name": "tool2", "success_rate": 0.8, "usage_count": 5},
{"tool_name": "tool3", "success_rate": 0.7, "usage_count": 3},
{"tool_name": "tool4", "success_rate": 0.6, "usage_count": 2},
{"tool_name": "tool5", "success_rate": 0.5, "usage_count": 1}
]
}
ace_manager.bulk_export_strategies.return_value = exported
result = self._call_handle_tool(
"bulk_export_strategies",
{"limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "匹配 5 个策略" in result
assert "前 3 个预览" in result
assert "tool1" in result
def test_bulk_export_strategies_no_results(self, config_manager, graphiti_client, ace_manager):
"""测试 bulk_export_strategies 工具无结果。"""
ace_manager.bulk_export_strategies.return_value = None
result = self._call_handle_tool(
"bulk_export_strategies",
{"limit": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "未导出任何策略" in result
# ----- export_strategies 工具处理测试 -----
def test_export_strategies_success(self, config_manager, graphiti_client, ace_manager):
"""测试 export_strategies 工具成功。"""
result_data = {
"file_path": "/path/to/export.json",
"count": 5,
"size": 1024
}
ace_manager.export_strategies.return_value = result_data
result = self._call_handle_tool(
"export_strategies",
{"tool_name": "test", "file_path": "/path/to/export.json"},
graphiti_client,
ace_manager,
config_manager
)
assert "策略导出成功" in result
assert "5" in result
assert "1024" in result
def test_export_strategies_failure(self, config_manager, graphiti_client, ace_manager):
"""测试 export_strategies 工具失败。"""
ace_manager.export_strategies.return_value = None
result = self._call_handle_tool(
"export_strategies",
{"file_path": "/path/to/export.json"},
graphiti_client,
ace_manager,
config_manager
)
assert "策略导出失败" in result
# ----- import_strategies 工具处理测试 -----
def test_import_strategies_success(self, config_manager, graphiti_client, ace_manager):
"""测试 import_strategies 工具成功。"""
result_data = {
"count": 5,
"overwritten": 2,
"created": 3,
"errors": []
}
ace_manager.import_strategies.return_value = result_data
result = self._call_handle_tool(
"import_strategies",
{"file_path": "/path/to/import.json", "overwrite": True},
graphiti_client,
ace_manager,
config_manager
)
assert "策略导入成功" in result
assert "5" in result
assert "2" in result
assert "3" in result
def test_import_strategies_with_errors(self, config_manager, graphiti_client, ace_manager):
"""测试 import_strategies 工具有错误。"""
result_data = {
"count": 5,
"overwritten": 2,
"created": 3,
"errors": ["Error 1", "Error 2"]
}
ace_manager.import_strategies.return_value = result_data
result = self._call_handle_tool(
"import_strategies",
{"file_path": "/path/to/import.json", "overwrite": True},
graphiti_client,
ace_manager,
config_manager
)
assert "导入错误: 2 个" in result
def test_import_strategies_failure(self, config_manager, graphiti_client, ace_manager):
"""测试 import_strategies 工具失败。"""
ace_manager.import_strategies.return_value = None
result = self._call_handle_tool(
"import_strategies",
{"file_path": "/path/to/import.json"},
graphiti_client,
ace_manager,
config_manager
)
assert "策略导入失败" in result
# ----- toggle_strategy 工具处理测试 -----
def test_toggle_strategy_enable(self, config_manager, graphiti_client, ace_manager):
"""测试 toggle_strategy 工具启用策略。"""
result_data = {"count": 1}
ace_manager.toggle_strategy.return_value = result_data
result = self._call_handle_tool(
"toggle_strategy",
{
"tool_name": "test_tool",
"arguments_hash": "hash123",
"enabled": True
},
graphiti_client,
ace_manager,
config_manager
)
assert "策略已启用" in result
assert "test_tool" in result
assert "hash123" in result
def test_toggle_strategy_disable(self, config_manager, graphiti_client, ace_manager):
"""测试 toggle_strategy 工具禁用策略。"""
result_data = {"count": 1}
ace_manager.toggle_strategy.return_value = result_data
result = self._call_handle_tool(
"toggle_strategy",
{
"tool_name": "test_tool",
"enabled": False
},
graphiti_client,
ace_manager,
config_manager
)
assert "策略已禁用" in result
assert "hash123" not in result # 没有提供 arguments_hash
def test_toggle_strategy_failure(self, config_manager, graphiti_client, ace_manager):
"""测试 toggle_strategy 工具失败。"""
ace_manager.toggle_strategy.return_value = None
result = self._call_handle_tool(
"toggle_strategy",
{"tool_name": "test", "enabled": True},
graphiti_client,
ace_manager,
config_manager
)
assert "策略操作失败" in result
# ----- validate_strategies 工具处理测试 -----
def test_validate_strategies_valid(self, config_manager, graphiti_client, ace_manager):
"""测试 validate_strategies 工具验证通过。"""
result_data = {
"valid": True,
"total": 10,
"enabled": 8,
"disabled": 2,
"avg_success_rate": 0.85
}
ace_manager.validate_strategies.return_value = result_data
result = self._call_handle_tool(
"validate_strategies",
{},
graphiti_client,
ace_manager,
config_manager
)
assert "策略验证通过" in result
assert "10" in result
assert "85.00%" in result
def test_validate_strategies_with_issues(self, config_manager, graphiti_client, ace_manager):
"""测试 validate_strategies 工具发现问题。"""
result_data = {
"valid": False,
"total": 10,
"enabled": 8,
"disabled": 2,
"avg_success_rate": 0.5,
"issues": ["Issue 1", "Issue 2"]
}
ace_manager.validate_strategies.return_value = result_data
result = self._call_handle_tool(
"validate_strategies",
{},
graphiti_client,
ace_manager,
config_manager
)
assert "策略验证发现问题" in result
assert "Issue 1" in result
assert "Issue 2" in result
def test_validate_strategies_with_error(self, config_manager, graphiti_client, ace_manager):
"""测试 validate_strategies 工具有错误。"""
result_data = {
"valid": False,
"total": 0,
"enabled": 0,
"disabled": 0,
"avg_success_rate": 0.0,
"error": "Database connection failed"
}
ace_manager.validate_strategies.return_value = result_data
result = self._call_handle_tool(
"validate_strategies",
{},
graphiti_client,
ace_manager,
config_manager
)
assert "错误: Database connection failed" in result
def test_validate_strategies_failure(self, config_manager, graphiti_client, ace_manager):
"""测试 validate_strategies 工具失败。"""
ace_manager.validate_strategies.return_value = None
result = self._call_handle_tool(
"validate_strategies",
{},
graphiti_client,
ace_manager,
config_manager
)
assert "策略验证失败" in result
# ----- render_strategy_insights 工具处理测试 -----
def test_render_strategy_insights_mermaid_format(self, config_manager, graphiti_client, ace_manager):
"""测试 render_strategy_insights 工具 Mermaid 格式。"""
heatmap = {
"success": True,
"generated_at": "2025-12-02T10:00:00",
"entries": [
{"tool_name": "tool1", "bucket": "90-100%", "total_usage": 10, "avg_success_rate": 0.95},
{"tool_name": "tool2", "bucket": "80-90%", "total_usage": 5, "avg_success_rate": 0.85}
],
"summary": {
"total_usage": 15,
"total_entries": 2,
"unique_tools": 2
}
}
ace_manager.get_strategy_heatmap.return_value = heatmap
result = self._call_handle_tool(
"render_strategy_insights",
{"format": "mermaid", "top_n": 10, "group_by": "tool"},
graphiti_client,
ace_manager,
config_manager
)
assert "策略热力图洞察" in result
assert "```mermaid" in result
assert "pie showData" in result
def test_render_strategy_insights_ascii_format(self, config_manager, graphiti_client, ace_manager):
"""测试 render_strategy_insights 工具 ASCII 格式。"""
heatmap = {
"success": True,
"generated_at": "2025-12-02T10:00:00",
"entries": [
{"tool_name": "tool1", "bucket": "90-100%", "strategy_count": 5, "total_usage": 10, "avg_success_rate": 0.95}
] * 50, # 超过40个，应该使用 ASCII
"summary": {
"total_usage": 500,
"total_entries": 50,
"unique_tools": 1
}
}
ace_manager.get_strategy_heatmap.return_value = heatmap
result = self._call_handle_tool(
"render_strategy_insights",
{"format": "auto", "top_n": 50, "group_by": "tool"},
graphiti_client,
ace_manager,
config_manager
)
assert "策略热力图洞察" in result
assert "Tool" in result # ASCII 表格头部
def test_render_strategy_insights_no_data(self, config_manager, graphiti_client, ace_manager):
"""测试 render_strategy_insights 工具无数据。"""
heatmap = {
"success": False,
"message": "暂无策略数据"
}
ace_manager.get_strategy_heatmap.return_value = heatmap
result = self._call_handle_tool(
"render_strategy_insights",
{"format": "auto", "top_n": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "暂无策略数据" in result
def test_render_strategy_insights_empty_entries(self, config_manager, graphiti_client, ace_manager):
"""测试 render_strategy_insights 工具空条目。"""
heatmap = {
"success": True,
"entries": [],
"summary": {}
}
ace_manager.get_strategy_heatmap.return_value = heatmap
result = self._call_handle_tool(
"render_strategy_insights",
{"format": "auto", "top_n": 10},
graphiti_client,
ace_manager,
config_manager
)
assert "暂无策略数据" in result
def test_render_strategy_insights_invalid_format(self, config_manager, graphiti_client, ace_manager):
"""测试 render_strategy_insights 工具无效格式。"""
heatmap = {
"success": True,
"generated_at": "2025-12-02T10:00:00",
"entries": [
{"tool_name": "tool1", "bucket": "90-100%", "total_usage": 10, "avg_success_rate": 0.95}
],
"summary": {"total_usage": 10, "total_entries": 1, "unique_tools": 1}
}
ace_manager.get_strategy_heatmap.return_value = heatmap
result = self._call_handle_tool(
"render_strategy_insights",
{"format": "invalid_format", "top_n": 10},
graphiti_client,
ace_manager,
config_manager
)
# 应该回退到 auto 模式
assert "策略热力图洞察" in result
