"""服务器工具调用详细测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
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
from src.graphiti_client import GraphitiClient
class TestServerCallToolDetailed:
"""服务器工具调用详细测试类."""
@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.fixture
def graphiti_client(self, config_manager):
"""创建 Graphiti 客户端."""
return GraphitiClient(config_manager)
@pytest.mark.asyncio
async def test_call_tool_configure_neo4j_detailed(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
)
assert result is not None
assert len(result) > 0
# 检查结果文本中是否包含成功或配置相关的信息
result_text = " ".join([r.text for r in result])
assert len(result_text) > 0
@pytest.mark.asyncio
async def test_call_tool_configure_api_detailed(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
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_check_configuration(self, server, config_manager):
"""测试调用检查配置工具."""
result = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_reset_configuration(self, server, config_manager):
"""测试调用重置配置工具."""
result = await handle_tool_call(
tool_name="reset_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_set_group_id(self, server, 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
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_add_episode_no_client(self, server, config_manager):
"""测试调用添加 Episode 工具（无客户端）."""
result = await handle_tool_call(
tool_name="add_episode",
arguments={
"content": "Test episode",
"metadata": {"type": "test"}
},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_add_episode_with_client(self, server, config_manager, graphiti_client):
"""测试调用添加 Episode 工具（有客户端）."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'add_episode', return_value={
"success": True,
"message": "Episode 已添加",
"episode_id": 1
}):
result = await handle_tool_call(
tool_name="add_episode",
arguments={
"content": "Test episode"
},
config_manager=config_manager,
graphiti_client=graphiti_client
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_search_entities_no_client(self, server, config_manager):
"""测试调用搜索实体工具（无客户端）."""
result = await handle_tool_call(
tool_name="search_entities",
arguments={
"query": "test",
"limit": 10
},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_search_entities_with_client(self, server, config_manager, graphiti_client):
"""测试调用搜索实体工具（有客户端）."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'search_entities', return_value={
"success": True,
"message": "找到 1 个结果",
"results": [{"name": "Test"}]
}):
result = await handle_tool_call(
tool_name="search_entities",
arguments={
"query": "test",
"limit": 10
},
config_manager=config_manager,
graphiti_client=graphiti_client
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_health_check_detailed(self, server, config_manager):
"""测试调用健康检查工具（详细）."""
result = await handle_tool_call(
tool_name="health_check",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_get_statistics_no_client(self, server, config_manager):
"""测试调用获取统计信息工具（无客户端）."""
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_get_statistics_with_client(self, server, config_manager, graphiti_client):
"""测试调用获取统计信息工具（有客户端）."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'get_statistics', return_value={
"success": True,
"statistics": {
"nodes": {"total": 100},
"relationships": {"total": 50}
}
}):
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=graphiti_client
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_error_handling(self, server, config_manager):
"""测试工具调用错误处理."""
# 测试无效工具名称
result = await handle_tool_call(
tool_name="nonexistent_tool",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
# 应该包含错误信息
assert any("未知" in r.text or "unknown" in r.text.lower() for r in result)
