"""测试 ACEManager 的学习趋势功能."""
import pytest
from unittest.mock import MagicMock, patch
from src.ace_manager import ACEManager
from src.config_manager import ConfigManager
class TestACEManagerLearningTrends:
"""确保学习趋势 API 行为正确."""
@pytest.fixture
def config_manager(self, temp_config_dir):
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def _manager(self, config_manager, client):
with patch.object(ACEManager, "_initialize_ace", lambda self: None):
manager = ACEManager(config_manager, client)
manager.enabled = True
return manager
def test_get_trends_returns_data(self, config_manager):
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = MagicMock()
session.run.return_value = [
{
"t": {
"date": "2025-12-01",
"tool_name": "search_entities",
"arguments_hash": "abc123",
"usage_count": 10.0,
"success_count": 8.0,
"failure_count": 2.0,
"feedback_count": 2.0,
"rating_sum": 8.0,
"sample_count": 3,
}
}
]
graphiti_client.driver.session.return_value.__enter__.return_value = session
manager = self._manager(config_manager, graphiti_client)
trends = manager.get_learning_trends(tool_name="search_entities", days=7)
assert len(trends) == 1
assert trends[0]["success_rate"] == pytest.approx(0.8)
assert trends[0]["avg_rating"] == pytest.approx(4.0)
def test_get_trends_no_connection(self, config_manager):
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = False
manager = self._manager(config_manager, graphiti_client)
assert manager.get_learning_trends() == []
def test_get_trends_exception(self, config_manager):
graphiti_client = MagicMock()
graphiti_client.is_connected.return_value = True
session = MagicMock()
session.run.side_effect = Exception("trend error")
graphiti_client.driver.session.return_value.__enter__.return_value = session
manager = self._manager(config_manager, graphiti_client)
trends = manager.get_learning_trends()
assert trends == []
