"""ACE 相关统计、告警与评分类 MCP 工具的 handler 测试."""
import pytest
from unittest.mock import Mock
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestToolsAceStatsAndFeedback:
"""覆盖 get_strategy_stats / get_strategy_alerts / rate_result 等工具分支."""
@pytest.fixture
def config_manager(self, temp_config_dir):
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
# ----- get_strategy_stats -----
@pytest.mark.asyncio
async def test_get_strategy_stats_success_with_by_tool(self, config_manager):
"""ACE 返回全局统计数据且包含 by_tool 时的输出."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.get_strategy_stats.return_value = {
"total_strategies": 3,
"avg_success_rate": 0.75,
"total_usage": 30,
"total_success": 24,
"total_failure": 6,
"by_tool": {
"search_entities": {
"count": 2,
"avg_success_rate": 0.8,
"usage": 20,
"success": 16,
"failure": 4,
}
},
}
result = await handle_tool_call(
tool_name="get_strategy_stats",
arguments={"tool_name": None},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "ACE 策略统计信息" in text
assert "总策略数: 3" in text
assert "平均成功率" in text
assert "按工具分类" in text
assert "search_entities" in text
@pytest.mark.asyncio
async def test_get_strategy_stats_no_data(self, config_manager):
"""无统计数据时返回提示信息."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.get_strategy_stats.return_value = None
result = await handle_tool_call(
tool_name="get_strategy_stats",
arguments={"tool_name": "search_entities"},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "暂无策略统计信息" in text
# ----- get_strategy_alerts -----
@pytest.mark.asyncio
async def test_get_strategy_alerts_with_results(self, config_manager):
"""有告警结果时应输出详细列表."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.get_strategy_alerts.return_value = [
{
"tool_name": "search_entities",
"arguments_hash": "hash1",
"failures": 12.0,
"usage": 20.0,
"failure_rate": 0.6,
"last_date": "2025-12-01",
"severity": "warning",
}
]
result = await handle_tool_call(
tool_name="get_strategy_alerts",
arguments={"failure_threshold": 5, "days": 7, "tool_name": None, "arguments_hash": None},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "策略告警" in text
assert "search_entities" in text
assert "失败次数" in text
assert "级别" in text
@pytest.mark.asyncio
async def test_get_strategy_alerts_no_results(self, config_manager):
"""无告警结果时返回一切正常提示."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.get_strategy_alerts.return_value = []
result = await handle_tool_call(
tool_name="get_strategy_alerts",
arguments={"failure_threshold": 5, "days": 7, "tool_name": None, "arguments_hash": None},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "近期未发现策略告警" in text
# ----- rate_result -----
@pytest.mark.asyncio
async def test_rate_result_success(self, config_manager):
"""评分成功时返回感谢文案."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.rate_result.return_value = True
result = await handle_tool_call(
tool_name="rate_result",
arguments={
"tool_name": "search_entities",
"rating": 5,
"feedback": "很好用",
"context": {"foo": "bar"},
},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "感谢您的反馈" in text
assert "search_entities" in text
assert "⭐" in text
@pytest.mark.asyncio
async def test_rate_result_ace_disabled(self, config_manager):
"""ACE 未启用时返回错误提示."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = False
result = await handle_tool_call(
tool_name="rate_result",
arguments={
"tool_name": "search_entities",
"rating": 3,
"feedback": None,
"context": None,
},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "ACE Manager 未启用" in text
@pytest.mark.asyncio
async def test_rate_result_failure_from_manager(self, config_manager):
"""ACE Manager 返回 False 时，应提示保存反馈失败."""
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
ace_manager = Mock()
ace_manager.is_enabled.return_value = True
ace_manager.optimize_tool_input.return_value = {}
ace_manager.rate_result.return_value = False
result = await handle_tool_call(
tool_name="rate_result",
arguments={
"tool_name": "search_entities",
"rating": 4,
"feedback": "一般",
"context": None,
},
config_manager=config_manager,
graphiti_client=mock_client,
ace_manager=ace_manager,
)
text = result[0].text
assert "保存反馈失败" in text
