"""服务器资源读取完整测试."""
import pytest
import json
from unittest.mock import Mock, MagicMock, patch
from mcp.types import ReadResourceRequest, ReadResourceResult
from src.server import create_server
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestServerReadResourceComplete:
"""服务器资源读取完整测试类."""
@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_read_resource_recent_episodes_flow(self, server, graphiti_client):
"""测试读取最近 Episodes 资源流程."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'query_by_time_range', return_value={
"success": True,
"results": [
{"content": "Episode 1", "created_at": "2025-01-01"},
{"content": "Episode 2", "created_at": "2025-01-02"}
]
}):
# 模拟资源读取逻辑
result = graphiti_client.query_by_time_range(days=30, limit=10)
assert result['success'] is True
assert 'results' in result
@pytest.mark.asyncio
async def test_read_resource_entity_counts_flow(self, server, graphiti_client):
"""测试读取实体统计资源流程."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'query_knowledge_graph', return_value={
"success": True,
"results": [
{"labels": ["Entity"], "count": 5},
{"labels": ["Preference"], "count": 3}
]
}):
result = graphiti_client.query_knowledge_graph(
"MATCH (n) RETURN labels(n) as labels, count(n) as count ORDER BY count DESC"
)
assert result['success'] is True
# 验证统计逻辑
stats = {}
for record in result['results']:
labels = record.get('labels', [])
label = labels[0] if labels else 'Unknown'
count = record.get('count', 0)
stats[label] = count
assert len(stats) > 0
@pytest.mark.asyncio
async def test_read_resource_configuration_flow(self, server, config_manager):
"""测试读取配置资源流程."""
# 测试配置状态获取
status = config_manager.get_config_status()
assert status is not None
assert isinstance(status, dict)
# 验证安全配置信息构建
safe_status = {
"neo4j_configured": status.get("neo4j_configured", False),
"api_configured": status.get("api_configured", False),
"group_id": status.get("group_id", "default"),
}
assert "neo4j_configured" in safe_status
assert "api_configured" in safe_status
@pytest.mark.asyncio
async def test_read_resource_relationship_stats_flow(self, server, graphiti_client):
"""测试读取关系统计资源流程."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'query_knowledge_graph', return_value={
"success": True,
"results": [
{"relationship_type": "RELATES_TO", "count": 10},
{"relationship_type": "CONTAINS", "count": 5}
]
}):
result = graphiti_client.query_knowledge_graph(
"MATCH ()-[r]->() RETURN type(r) as relationship_type, count(r) as count ORDER BY count DESC"
)
assert result['success'] is True
# 验证关系统计逻辑
stats = {}
for record in result['results']:
rel_type = record.get('relationship_type', 'Unknown')
count = record.get('count', 0)
stats[rel_type] = count
assert len(stats) > 0
assert sum(stats.values()) > 0
@pytest.mark.asyncio
async def test_read_resource_top_entities_flow(self, server, graphiti_client):
"""测试读取热门实体资源流程."""
with patch.object(graphiti_client, 'is_connected', return_value=True):
with patch.object(graphiti_client, 'query_knowledge_graph', return_value={
"success": True,
"results": [
{
"labels": ["Entity"],
"name": "Entity1",
"content": "Content1",
"connection_count": 10
}
]
}):
result = graphiti_client.query_knowledge_graph(
"MATCH (n)-[r]-() WITH n, count(r) as connection_count RETURN labels(n) as labels, n.name as name, n.content as content, connection_count ORDER BY connection_count DESC LIMIT 20"
)
assert result['success'] is True
# 验证实体列表构建
entities = []
for record in result['results']:
entities.append({
"labels": record.get('labels', []),
"name": record.get('name'),
"content": record.get('content'),
"connection_count": record.get('connection_count', 0)
})
assert len(entities) > 0
@pytest.mark.asyncio
async def test_read_resource_statistics_flow(self, server, 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},
"episodes": {"total": 20}
}
}):
result = graphiti_client.get_statistics()
assert result['success'] is True
assert 'statistics' in result
# 验证统计信息结构
stats = result['statistics']
assert 'nodes' in stats
assert 'relationships' in stats
assert 'episodes' in stats
@pytest.mark.asyncio
async def test_read_resource_error_handling_flow(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 error")):
try:
result = graphiti_client.query_knowledge_graph("INVALID QUERY")
assert False, "应该抛出异常"
except Exception as e:
assert "Query error" in str(e) or True
