"""服务器资源读取详细测试."""
import pytest
import json
from unittest.mock import Mock, MagicMock, patch
from src.server import create_server
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestServerReadResourceDetailed:
"""服务器资源读取详细测试类."""
@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_connected(self, server, graphiti_client):
"""测试读取最近 Episodes 资源（已连接）."""
# Mock 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": [
{"e": {"content": "Episode 1", "created_at": "2025-01-01"}},
{"e": {"content": "Episode 2", "created_at": "2025-01-02"}}
]
}):
assert graphiti_client.is_connected()
result = graphiti_client.query_knowledge_graph("MATCH (e:Episode) RETURN e ORDER BY e.created_at DESC LIMIT 10")
assert result['success'] is True
@pytest.mark.asyncio
async def test_read_resource_recent_episodes_not_connected(self, server, graphiti_client):
"""测试读取最近 Episodes 资源（未连接）."""
# Mock graphiti_client 未连接
with patch.object(graphiti_client, 'is_connected', return_value=False):
assert not graphiti_client.is_connected()
@pytest.mark.asyncio
async def test_read_resource_entity_counts_connected(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")
assert result['success'] is True
assert len(result['results']) > 0
@pytest.mark.asyncio
async def test_read_resource_entity_counts_not_connected(self, server, graphiti_client):
"""测试读取实体统计资源（未连接）."""
with patch.object(graphiti_client, 'is_connected', return_value=False):
assert not graphiti_client.is_connected()
@pytest.mark.asyncio
async def test_read_resource_configuration_details(self, server, config_manager):
"""测试读取配置资源详细信息."""
# 测试配置读取逻辑
neo4j_config = config_manager.get_neo4j_config()
api_config = config_manager.get_api_config()
group_id = config_manager.get_group_id()
# 构建配置信息
config_info = {
"neo4j_configured": neo4j_config is not None,
"api_configured": api_config is not None,
"group_id": group_id
}
assert "neo4j_configured" in config_info
assert "api_configured" in config_info
assert "group_id" in config_info
@pytest.mark.asyncio
async def test_read_resource_relationship_stats_connected(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")
assert result['success'] is True
assert len(result['results']) > 0
@pytest.mark.asyncio
async def test_read_resource_relationship_stats_not_connected(self, server, graphiti_client):
"""测试读取关系统计资源（未连接）."""
with patch.object(graphiti_client, 'is_connected', return_value=False):
assert not graphiti_client.is_connected()
@pytest.mark.asyncio
async def test_read_resource_top_entities_connected(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": [
{"name": "Entity1", "degree": 10},
{"name": "Entity2", "degree": 8}
]
}):
result = graphiti_client.query_knowledge_graph("MATCH (n) RETURN n.name as name, size((n)--()) as degree ORDER BY degree DESC LIMIT 10")
assert result['success'] is True
assert len(result['results']) > 0
@pytest.mark.asyncio
async def test_read_resource_top_entities_not_connected(self, server, graphiti_client):
"""测试读取热门实体资源（未连接）."""
with patch.object(graphiti_client, 'is_connected', return_value=False):
assert not graphiti_client.is_connected()
@pytest.mark.asyncio
async def test_read_resource_statistics_connected(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
assert 'nodes' in result['statistics']
@pytest.mark.asyncio
async def test_read_resource_statistics_not_connected(self, server, graphiti_client):
"""测试读取完整统计信息资源（未连接）."""
with patch.object(graphiti_client, 'is_connected', return_value=False):
assert not graphiti_client.is_connected()
@pytest.mark.asyncio
async def test_read_resource_error_handling(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": False,
"message": "Query failed"
}):
result = graphiti_client.query_knowledge_graph("INVALID QUERY")
assert result['success'] is False
assert 'message' in result
