"""服务器错误处理完整测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from mcp.types import CallToolRequest, CallToolResult, ReadResourceRequest, ReadResourceResult
from src.server import create_server
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestServerErrorHandlingComplete:
"""服务器错误处理完整测试类."""
@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_tool_call_exception_handling(self, server, config_manager):
"""测试工具调用异常处理."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.add_episode = AsyncMock(side_effect=Exception("Database error"))
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
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
@pytest.mark.asyncio
async def test_tool_call_missing_arguments(self, server, config_manager):
"""测试工具调用缺少参数."""
result = await handle_tool_call(
tool_name="add_episode",
arguments={}, # 缺少必需的 content 参数
config_manager=config_manager,
graphiti_client=None
)
assert result is not None
assert isinstance(result, list)
# 应该包含错误信息或使用默认值
@pytest.mark.asyncio
async def test_tool_call_invalid_arguments(self, server, config_manager):
"""测试工具调用无效参数."""
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": "", # 空 URI
"username": "neo4j",
"password": "test"
},
config_manager=config_manager,
graphiti_client=None
)
assert result is not None
assert isinstance(result, list)
@pytest.mark.asyncio
async def test_tool_call_database_not_connected(self, server, config_manager):
"""测试数据库未连接时的处理."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = False
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
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 "connect" in result_text.lower() or "配置" in result_text
@pytest.mark.asyncio
async def test_tool_call_graphiti_not_initialized(self, server, config_manager):
"""测试 Graphiti 未初始化时的处理."""
from src.graphiti_client import GraphitiClient
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.graphiti = None # Graphiti 未初始化
mock_client.semantic_search = AsyncMock(return_value={
"success": True,
"search_type": "enhanced_keyword",
"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
assert isinstance(result, list)
# 应该包含降级处理或错误信息
@pytest.mark.asyncio
async def test_resource_read_exception_handling(self, server):
"""测试资源读取异常处理."""
# 测试读取资源时的异常处理
# 这需要模拟 read_resource 函数
pass # 资源读取测试在其他文件中
@pytest.mark.asyncio
async def test_prompt_get_exception_handling(self, server):
"""测试提示获取异常处理."""
# 测试获取提示时的异常处理
# 这需要模拟 get_prompt 函数
pass # 提示获取测试在其他文件中
