"""所有工具处理函数完整测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestToolsAllHandlers:
"""所有工具处理函数完整测试类."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def mock_client(self):
"""创建模拟客户端."""
client = MagicMock(spec=GraphitiClient)
client.is_connected.return_value = True
return client
@pytest.mark.asyncio
async def test_configure_neo4j_handler(self, config_manager):
"""测试配置 Neo4j 处理."""
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": "bolt://localhost:7687",
"username": "neo4j",
"password": "test"
},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_configure_api_handler(self, config_manager):
"""测试配置 API 处理."""
result = await handle_tool_call(
tool_name="configure_api",
arguments={
"provider": "openai",
"api_key": "test_key"
},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_check_configuration_handler(self, config_manager):
"""测试检查配置处理."""
result = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_reset_configuration_handler(self, config_manager):
"""测试重置配置处理."""
result = await handle_tool_call(
tool_name="reset_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_set_group_id_handler(self, config_manager):
"""测试设置组 ID 处理."""
result = await handle_tool_call(
tool_name="set_group_id",
arguments={"group_id": "test_group"},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_add_episode_handler(self, config_manager, mock_client):
"""测试添加 Episode 处理."""
mock_client.add_episode = AsyncMock(return_value={
"success": True,
"episode_id": 1
})
result = await handle_tool_call(
tool_name="add_episode",
arguments={"content": "Test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_add_episodes_bulk_handler(self, config_manager, mock_client):
"""测试批量添加 Episode 处理."""
mock_client.add_episodes_bulk = AsyncMock(return_value={
"success": True,
"total": 2
})
result = await handle_tool_call(
tool_name="add_episodes_bulk",
arguments={"episodes": [{"content": "Test1"}, {"content": "Test2"}]},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_delete_episode_handler(self, config_manager, mock_client):
"""测试删除 Episode 处理."""
mock_client.delete_episode.return_value = {
"success": True
}
result = await handle_tool_call(
tool_name="delete_episode",
arguments={"episode_id": 1},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_search_entities_handler(self, config_manager, mock_client):
"""测试搜索实体处理."""
mock_client.search_entities.return_value = {
"success": True,
"results": []
}
result = await handle_tool_call(
tool_name="search_entities",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_search_relationships_handler(self, config_manager, mock_client):
"""测试搜索关系处理."""
mock_client.search_relationships.return_value = {
"success": True,
"results": []
}
result = await handle_tool_call(
tool_name="search_relationships",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_semantic_search_handler(self, config_manager, mock_client):
"""测试语义搜索处理."""
mock_client.semantic_search = AsyncMock(return_value={
"success": True,
"results": []
})
result = await handle_tool_call(
tool_name="semantic_search",
arguments={"query": "test"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_query_knowledge_graph_handler(self, config_manager, mock_client):
"""测试查询知识图谱处理."""
mock_client.query_knowledge_graph.return_value = {
"success": True,
"results": []
}
result = await handle_tool_call(
tool_name="query_knowledge_graph",
arguments={"query": "MATCH (n) RETURN n"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_query_by_time_range_handler(self, config_manager, mock_client):
"""测试时间范围查询处理."""
mock_client.query_by_time_range.return_value = {
"success": True,
"results": []
}
result = await handle_tool_call(
tool_name="query_by_time_range",
arguments={"days": 7},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_clear_graph_handler(self, config_manager, mock_client):
"""测试清空图谱处理."""
mock_client.clear_graph.return_value = {
"success": True
}
result = await handle_tool_call(
tool_name="clear_graph",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_export_graph_data_handler(self, config_manager, mock_client):
"""测试导出图谱数据处理."""
mock_client.export_graph_data.return_value = {
"success": True,
"data": {}
}
result = await handle_tool_call(
tool_name="export_graph_data",
arguments={"format": "json"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_import_graph_data_handler(self, config_manager, mock_client):
"""测试导入图谱数据处理."""
mock_client.import_graph_data.return_value = {
"success": True
}
result = await handle_tool_call(
tool_name="import_graph_data",
arguments={"data": {}, "format": "json"},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_get_statistics_handler(self, config_manager, mock_client):
"""测试获取统计信息处理."""
mock_client.get_statistics.return_value = {
"success": True,
"statistics": {}
}
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_validate_data_handler(self, config_manager, mock_client):
"""测试数据验证处理."""
mock_client.validate_data.return_value = {
"success": True,
"issues": []
}
result = await handle_tool_call(
tool_name="validate_data",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_clean_orphaned_nodes_handler(self, config_manager, mock_client):
"""测试清理孤立节点处理."""
mock_client.clean_orphaned_nodes.return_value = {
"success": True,
"deleted_count": 5
}
result = await handle_tool_call(
tool_name="clean_orphaned_nodes",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_rebuild_indexes_handler(self, config_manager, mock_client):
"""测试重建索引处理."""
mock_client.rebuild_indexes.return_value = {
"success": True,
"indexes": []
}
result = await handle_tool_call(
tool_name="rebuild_indexes",
arguments={},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
@pytest.mark.asyncio
async def test_get_cache_stats_handler(self, config_manager):
"""测试获取缓存统计处理."""
result = await handle_tool_call(
tool_name="get_cache_stats",
arguments={},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_health_check_handler(self, config_manager):
"""测试健康检查处理."""
with patch('src.health_check.health_check') as mock_health_check:
mock_health_check.return_value = {
"status": "healthy",
"timestamp": "2025-01-01T00:00:00",
"checks": {
"configuration": {"status": "ok"},
"database": {"status": "ok"}
}
}
result = await handle_tool_call(
tool_name="health_check",
arguments={},
config_manager=config_manager
)
assert result is not None
