"""服务器错误处理测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
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 TestServerErrorHandling:
"""服务器错误处理测试类."""
@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_with_invalid_arguments(self, server, config_manager):
"""测试使用无效参数调用工具."""
# 测试缺少必需参数
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments={}, # 缺少必需参数
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_with_malformed_arguments(self, server, config_manager):
"""测试使用格式错误的参数调用工具."""
# 测试参数类型错误
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": 123, # 应该是字符串
"username": None,
"password": []
},
config_manager=config_manager
)
assert result is not None
@pytest.mark.asyncio
async def test_read_resource_with_invalid_uri(self, server):
"""测试使用无效 URI 读取资源."""
invalid_uri = "graphitiace://invalid-resource"
assert invalid_uri.startswith("graphitiace://")
assert "invalid" in invalid_uri
@pytest.mark.asyncio
async def test_read_resource_with_malformed_uri(self, server):
"""测试使用格式错误的 URI 读取资源."""
malformed_uri = "not-a-valid-uri"
assert not malformed_uri.startswith("graphitiace://")
@pytest.mark.asyncio
async def test_get_prompt_with_invalid_name(self, server):
"""测试使用无效名称获取提示."""
invalid_prompt = "nonexistent_prompt"
assert invalid_prompt not in [
"query_user_preferences",
"query_project_info",
"query_recent_learning",
"query_best_practices",
"query_related_entities",
"summarize_knowledge",
"export_data",
"get_statistics",
]
@pytest.mark.asyncio
async def test_call_tool_database_error(self, server, config_manager, graphiti_client):
"""测试数据库错误处理."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'add_episode', side_effect=Exception("Database error")):
result = await handle_tool_call(
tool_name="add_episode",
arguments={"content": "Test"},
config_manager=config_manager,
graphiti_client=graphiti_client
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_read_resource_database_error(self, server, graphiti_client):
"""测试资源读取时数据库错误."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'query_knowledge_graph', side_effect=Exception("Query failed")):
# 测试错误处理
try:
result = graphiti_client.query_knowledge_graph("MATCH (n) RETURN n")
assert False, "应该抛出异常"
except Exception:
assert True
@pytest.mark.asyncio
async def test_call_tool_configuration_error(self, server, config_manager):
"""测试配置错误处理."""
# 测试配置管理器错误
with patch.object(config_manager, 'configure_neo4j', side_effect=Exception("Config error")):
try:
config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
assert False, "应该抛出异常"
except Exception:
assert True
@pytest.mark.asyncio
async def test_call_tool_client_initialization_error(self, server, graphiti_client):
"""测试客户端初始化错误."""
with patch.object(graphiti_client, 'connect', side_effect=Exception("Connection failed")):
try:
graphiti_client.connect()
assert False, "应该抛出异常"
except Exception:
assert True
