"""服务器工具调用完整测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from mcp.types import CallToolRequest, CallToolResult
from src.server import create_server
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestServerCallToolComplete:
"""服务器工具调用完整测试类."""
@pytest.fixture
def server(self):
"""创建服务器实例."""
return create_server()
@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_call_tool_configure_neo4j_complete(self, server, config_manager):
"""测试调用配置 Neo4j 工具（完整流程）."""
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": "bolt://localhost:7687",
"username": "neo4j",
"password": "test_password",
"database": "neo4j"
},
config_manager=config_manager,
graphiti_client=None
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_call_tool_configure_api_complete(self, server, config_manager):
"""测试调用配置 API 工具（完整流程）."""
result = await handle_tool_call(
tool_name="configure_api",
arguments={
"provider": "openai",
"api_key": "test_key",
"model": "gpt-4"
},
config_manager=config_manager,
graphiti_client=None
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_call_tool_add_episode_complete(self, server, config_manager):
"""测试调用添加 Episode 工具（完整流程）."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.add_episode.return_value = {
"success": True,
"message": "Episode 添加成功",
"episode_id": 1
}
result = await handle_tool_call(
tool_name="add_episode",
arguments={
"content": "测试内容",
"metadata": {"type": "test"}
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_call_tool_search_entities_complete(self, server, config_manager):
"""测试调用搜索实体工具（完整流程）."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.search_entities.return_value = {
"success": True,
"results": [{"name": "Entity1"}]
}
result = await handle_tool_call(
tool_name="search_entities",
arguments={
"query": "test",
"limit": 10
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_call_tool_query_knowledge_graph_complete(self, server, config_manager):
"""测试调用查询知识图谱工具（完整流程）."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
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 LIMIT 10"
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_call_tool_get_statistics_complete(self, server, 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,
"statistics": {"nodes": 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)
@pytest.mark.asyncio
async def test_call_tool_unknown_tool_complete(self, server, config_manager):
"""测试调用未知工具（完整流程）."""
result = await handle_tool_call(
tool_name="unknown_tool_xyz",
arguments={},
config_manager=config_manager,
graphiti_client=None
)
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 "unknown" in result_text.lower()
