"""工具获取统计信息详细测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestToolsGetStatisticsDetailed:
"""工具获取统计信息详细测试类."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.mark.asyncio
async def test_get_statistics_with_node_types(self, config_manager):
"""测试获取统计信息包含节点类型分布."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.get_statistics.return_value = {
"success": True,
"message": "统计信息获取成功",
"statistics": {
"nodes": {
"total": 100,
"by_type": {
"Entity": 50,
"Preference": 30,
"Project": 20
}
},
"relationships": {
"total": 50,
"by_type": {}
},
"episodes": {
"total": 10,
"first_episode": None,
"last_episode": None
}
}
}
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "成功" in result_text or "success" in result_text.lower() or "统计" in result_text
assert "节点类型分布" in result_text or "Entity" in result_text or "Preference" in result_text
@pytest.mark.asyncio
async def test_get_statistics_with_relationship_types(self, config_manager):
"""测试获取统计信息包含关系类型分布."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.get_statistics.return_value = {
"success": True,
"message": "统计信息获取成功",
"statistics": {
"nodes": {
"total": 100,
"by_type": {}
},
"relationships": {
"total": 50,
"by_type": {
"RELATES_TO": 30,
"CONTAINS": 20
}
},
"episodes": {
"total": 10,
"first_episode": None,
"last_episode": None
}
}
}
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "成功" in result_text or "success" in result_text.lower() or "统计" in result_text
assert "关系类型分布" in result_text or "RELATES_TO" in result_text or "CONTAINS" in result_text
@pytest.mark.asyncio
async def test_get_statistics_with_episodes(self, config_manager):
"""测试获取统计信息包含 Episode 信息."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.get_statistics.return_value = {
"success": True,
"message": "统计信息获取成功",
"statistics": {
"nodes": {
"total": 100,
"by_type": {}
},
"relationships": {
"total": 50,
"by_type": {}
},
"episodes": {
"total": 10,
"first_episode": "2025-01-01",
"last_episode": "2025-01-10"
}
}
}
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "成功" in result_text or "success" in result_text.lower() or "统计" in result_text
assert "第一个 Episode" in result_text or "最后一个 Episode" in result_text or "Episode" in result_text
@pytest.mark.asyncio
async def test_get_statistics_failure(self, config_manager):
"""测试获取统计信息失败."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.get_statistics.return_value = {
"success": False,
"message": "获取失败"
}
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "失败" in result_text or "error" in result_text.lower() or "❌" in result_text
