"""服务器集成测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from src.server import create_server
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestServerIntegration:
"""服务器集成测试类."""
@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_server_initialization(self, server):
"""测试服务器初始化."""
assert server is not None
assert server.name == "graphitiace"
@pytest.mark.asyncio
async def test_tools_listing(self, server, config_manager):
"""测试工具列表功能."""
from src.tools import get_tools
tools = get_tools(config_manager)
assert tools is not None
assert len(tools) > 0
# 检查关键工具是否存在
tool_names = [tool.name for tool in tools]
assert "configure_neo4j" in tool_names
assert "add_episode" in tool_names
assert "search_entities" in tool_names
@pytest.mark.asyncio
async def test_resources_listing(self, server):
"""测试资源列表功能."""
# 测试资源 URI 格式
expected_resources = [
"graphitiace://recent-episodes",
"graphitiace://entity-counts",
"graphitiace://configuration",
"graphitiace://relationship-stats",
"graphitiace://top-entities",
"graphitiace://statistics",
"graphitiace://strategy-heatmap",
]
for uri in expected_resources:
assert uri.startswith("graphitiace://")
@pytest.mark.asyncio
async def test_prompts_listing(self, server):
"""测试提示列表功能."""
# 测试提示名称格式
expected_prompts = [
"query_user_preferences",
"query_project_info",
"query_recent_learning",
"query_best_practices",
"query_related_entities",
"summarize_knowledge",
"export_data",
"get_statistics",
]
for prompt_name in expected_prompts:
assert len(prompt_name) > 0
@pytest.mark.asyncio
async def test_configuration_flow(self, server, config_manager):
"""测试配置流程."""
from src.tools import handle_tool_call
# 1. 配置 Neo4j
result1 = await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": "bolt://localhost:7687",
"username": "neo4j",
"password": "test"
},
config_manager=config_manager
)
assert result1 is not None
# 2. 检查配置
result2 = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert result2 is not None
@pytest.mark.asyncio
async def test_episode_workflow(self, server, config_manager, graphiti_client):
"""测试 Episode 工作流程."""
from src.tools import handle_tool_call
# 1. 配置 Neo4j（如果未配置）
await handle_tool_call(
tool_name="configure_neo4j",
arguments={
"uri": "bolt://localhost:7687",
"username": "neo4j",
"password": "test"
},
config_manager=config_manager
)
# 2. 尝试添加 Episode（应该失败，因为未连接）
result = await handle_tool_call(
tool_name="add_episode",
arguments={
"content": "Test episode"
},
config_manager=config_manager,
graphiti_client=graphiti_client if graphiti_client.is_connected() else None
)
assert result is not None
@pytest.mark.asyncio
async def test_search_workflow(self, server, config_manager, graphiti_client):
"""测试搜索工作流程."""
from src.tools import handle_tool_call
# 测试搜索实体（应该失败，因为未连接）
result = await handle_tool_call(
tool_name="search_entities",
arguments={
"query": "test",
"limit": 10
},
config_manager=config_manager,
graphiti_client=graphiti_client if graphiti_client.is_connected() else None
)
assert result is not None
@pytest.mark.asyncio
async def test_health_check_workflow(self, server, config_manager):
"""测试健康检查工作流程."""
from src.tools import handle_tool_call
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
assert len(result) > 0
@pytest.mark.asyncio
async def test_statistics_workflow(self, server, config_manager, graphiti_client):
"""测试统计信息工作流程."""
from src.tools import handle_tool_call
# 测试获取统计信息（应该失败，因为未连接）
result = await handle_tool_call(
tool_name="get_statistics",
arguments={},
config_manager=config_manager,
graphiti_client=graphiti_client if graphiti_client.is_connected() else None
)
assert result is not None
