"""服务器资源读取测试."""
import pytest
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 TestServerReadResource:
"""服务器资源读取测试类."""
@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(self, server, graphiti_client):
"""测试读取最近 Episodes 资源."""
# 测试资源读取逻辑（不实际连接数据库）
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(self, server, graphiti_client):
"""测试读取实体统计资源."""
# 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": [
{"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 'results' in result
@pytest.mark.asyncio
async def test_read_resource_configuration(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()
# 配置可能为空，但方法应该存在
assert neo4j_config is not None or neo4j_config is None
assert api_config is not None or api_config is None
assert group_id is not None
@pytest.mark.asyncio
async def test_read_resource_relationship_stats(self, server, graphiti_client):
"""测试读取关系统计资源."""
# 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": [
{"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 'results' in result
@pytest.mark.asyncio
async def test_read_resource_top_entities(self, server, graphiti_client):
"""测试读取热门实体资源."""
# 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": [
{"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 'results' in result
@pytest.mark.asyncio
async def test_read_resource_statistics(self, server, graphiti_client):
"""测试读取完整统计信息资源."""
# Mock 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
@pytest.mark.asyncio
async def test_read_resource_unknown(self, server):
"""测试读取未知资源."""
# 测试未知资源处理
unknown_uri = "graphitiace://unknown-resource"
assert unknown_uri.startswith("graphitiace://")
assert "unknown" in unknown_uri
