"""测试 server.py 中未覆盖的代码行."""
import pytest
import json
from unittest.mock import patch, MagicMock
from mcp.types import (
ReadResourceRequest,
ReadResourceRequestParams,
ReadResourceResult,
CallToolRequest,
CallToolRequestParams,
CallToolResult,
)
from src.server import create_server, TextResourceContentsWithContent
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
@pytest.fixture
def server_with_mocks():
"""创建带有 Mock ConfigManager/GraphitiClient 的服务器实例."""
mock_config_manager = MagicMock(spec=ConfigManager)
mock_config_manager.get_group_id.return_value = "test-group"
mock_graphiti_client = MagicMock(spec=GraphitiClient)
mock_graphiti_client.is_connected.return_value = True
with patch("src.server.ConfigManager", return_value=mock_config_manager), \
patch("src.server.GraphitiClient", return_value=mock_graphiti_client):
server = create_server()
return server, mock_graphiti_client, mock_config_manager
async def invoke_read_resource(handler, uri: str):
"""构造 ReadResourceRequest 并调用 handler."""
request = ReadResourceRequest(
params=ReadResourceRequestParams(uri=uri)
)
response = await handler(request)
if hasattr(response, "root"):
return response.root
return response
async def invoke_call_tool(handler, name: str, arguments=None):
"""构造 CallToolRequest 并调用 handler."""
request = CallToolRequest(
params=CallToolRequestParams(name=name, arguments=arguments)
)
response = await handler(request)
if hasattr(response, "root"):
return response.root
return response
class TestTextResourceContentsWithContent:
"""测试 TextResourceContentsWithContent 的 __getattr__ 方法."""
def test_getattr_content(self):
"""测试获取 content 属性."""
wrapper = TextResourceContentsWithContent(uri="test://uri", text="test text")
assert wrapper.content == "test text"
assert wrapper.text == "test text"
def test_getattr_other_attribute_error(self):
"""测试获取不存在的属性时抛出 AttributeError."""
wrapper = TextResourceContentsWithContent(uri="test://uri", text="test text")
with pytest.raises(AttributeError) as exc_info:
_ = wrapper.nonexistent
assert "TextResourceContentsWithContent" in str(exc_info.value)
assert "nonexistent" in str(exc_info.value)
class TestReadResourceEntityCounts:
"""测试 read_resource 中 entity-counts 资源的所有分支."""
@pytest.mark.asyncio
async def test_read_resource_entity_counts_connected(self, server_with_mocks):
"""测试 entity-counts 资源读取（已连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.query_knowledge_graph.return_value = {
'success': True,
'results': [
{'labels': ['Person'], 'count': 5},
{'labels': ['Project'], 'count': 3},
{'labels': [], 'count': 2} # 测试空 labels
]
}
result = await invoke_read_resource(handler, "graphitiace://entity-counts")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is True
assert content['statistics']['Person'] == 5
assert content['statistics']['Project'] == 3
assert content['statistics']['Unknown'] == 2
assert content['total_nodes'] == 10
@pytest.mark.asyncio
async def test_read_resource_entity_counts_not_connected(self, server_with_mocks):
"""测试 entity-counts 资源读取（未连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.is_connected.return_value = False
result = await invoke_read_resource(handler, "graphitiace://entity-counts")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is False
assert '未连接到数据库' in content['message']
@pytest.mark.asyncio
async def test_read_resource_entity_counts_query_failed(self, server_with_mocks):
"""测试 entity-counts 资源读取（查询失败）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.query_knowledge_graph.return_value = {
'success': False,
'results': []
}
result = await invoke_read_resource(handler, "graphitiace://entity-counts")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is True
assert content['total_nodes'] == 0
class TestReadResourceConfiguration:
"""测试 read_resource 中 configuration 资源."""
@pytest.mark.asyncio
async def test_read_resource_configuration(self, server_with_mocks):
"""测试 configuration 资源读取."""
server, _, mock_config_manager = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_config_manager.get_config_status.return_value = {
"neo4j_configured": True,
"api_configured": True,
"neo4j": {"uri": "bolt://localhost:7687", "username": "neo4j"},
"api": {"provider": "openai", "has_api_key": True},
"group_id": "test-group"
}
result = await invoke_read_resource(handler, "graphitiace://configuration")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['neo4j_configured'] is True
assert content['api_configured'] is True
assert content['group_id'] == "test-group"
assert content['neo4j']['uri'] == "bolt://localhost:7687"
assert content['api']['provider'] == "openai"
assert content['api']['has_api_key'] is True
@pytest.mark.asyncio
async def test_read_resource_configuration_no_neo4j(self, server_with_mocks):
"""测试 configuration 资源读取（未配置 Neo4j）."""
server, _, mock_config_manager = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_config_manager.get_config_status.return_value = {
"neo4j_configured": False,
"api_configured": False,
"neo4j": None,
"api": None,
"group_id": None
}
result = await invoke_read_resource(handler, "graphitiace://configuration")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['neo4j_configured'] is False
assert content['neo4j'] is None
class TestReadResourceRelationshipStats:
"""测试 read_resource 中 relationship-stats 资源."""
@pytest.mark.asyncio
async def test_read_resource_relationship_stats_connected(self, server_with_mocks):
"""测试 relationship-stats 资源读取（已连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.query_knowledge_graph.return_value = {
'success': True,
'results': [
{'relationship_type': 'KNOWS', 'count': 10},
{'relationship_type': 'WORKS_WITH', 'count': 5},
{'relationship_type': None, 'count': 2} # 测试 None 类型
]
}
result = await invoke_read_resource(handler, "graphitiace://relationship-stats")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is True
assert content['statistics']['KNOWS'] == 10
assert content['statistics']['WORKS_WITH'] == 5
assert content['statistics']['Unknown'] == 2
assert content['total_relationships'] == 17
@pytest.mark.asyncio
async def test_read_resource_relationship_stats_not_connected(self, server_with_mocks):
"""测试 relationship-stats 资源读取（未连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.is_connected.return_value = False
result = await invoke_read_resource(handler, "graphitiace://relationship-stats")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is False
assert '未连接到数据库' in content['message']
class TestReadResourceTopEntities:
"""测试 read_resource 中 top-entities 资源."""
@pytest.mark.asyncio
async def test_read_resource_top_entities_connected(self, server_with_mocks):
"""测试 top-entities 资源读取（已连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.query_knowledge_graph.return_value = {
'success': True,
'results': [
{
'labels': ['Person'],
'name': 'Alice',
'content': 'Test content',
'connection_count': 15
},
{
'labels': ['Project'],
'name': None,
'content': None,
'connection_count': 10
}
]
}
result = await invoke_read_resource(handler, "graphitiace://top-entities")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is True
assert len(content['top_entities']) == 2
assert content['top_entities'][0]['name'] == 'Alice'
assert content['top_entities'][0]['connection_count'] == 15
assert content['top_entities'][1]['name'] is None
@pytest.mark.asyncio
async def test_read_resource_top_entities_not_connected(self, server_with_mocks):
"""测试 top-entities 资源读取（未连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.is_connected.return_value = False
result = await invoke_read_resource(handler, "graphitiace://top-entities")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is False
assert '未连接到数据库' in content['message']
class TestReadResourceStatistics:
"""测试 read_resource 中 statistics 资源."""
@pytest.mark.asyncio
async def test_read_resource_statistics_connected_success(self, server_with_mocks):
"""测试 statistics 资源读取（已连接且成功）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.get_statistics.return_value = {
'success': True,
'statistics': {
'total_nodes': 100,
'total_relationships': 50
}
}
result = await invoke_read_resource(handler, "graphitiace://statistics")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['total_nodes'] == 100
assert content['total_relationships'] == 50
@pytest.mark.asyncio
async def test_read_resource_statistics_connected_failed(self, server_with_mocks):
"""测试 statistics 资源读取（已连接但失败）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.get_statistics.return_value = {
'success': False,
'message': '查询失败'
}
result = await invoke_read_resource(handler, "graphitiace://statistics")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is False
assert content['message'] == '查询失败'
@pytest.mark.asyncio
async def test_read_resource_statistics_not_connected(self, server_with_mocks):
"""测试 statistics 资源读取（未连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.is_connected.return_value = False
result = await invoke_read_resource(handler, "graphitiace://statistics")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert content['success'] is False
assert '未连接到数据库' in content['message']
class TestReadResourceUnknown:
"""测试 read_resource 中未知资源."""
@pytest.mark.asyncio
async def test_read_resource_unknown_uri(self, server_with_mocks):
"""测试未知资源 URI."""
server, _, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
result = await invoke_read_resource(handler, "graphitiace://unknown-resource")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
assert "未知资源" in result.contents[0].text
class TestCallToolArgumentsNone:
"""测试 call_tool 中 arguments is None 的分支."""
@pytest.mark.asyncio
async def test_call_tool_arguments_none(self, server_with_mocks):
"""测试 call_tool 中 arguments 为 None 的情况."""
server, _, mock_config_manager = server_with_mocks
handler = server.request_handlers.get(CallToolRequest)
assert handler is not None
with patch('src.server.handle_tool_call') as mock_handle:
mock_handle.return_value = [{"type": "text", "text": "Test result"}]
result = await invoke_call_tool(handler, "health_check", None)
assert isinstance(result, CallToolResult)
assert isinstance(result.content, list)
mock_handle.assert_called_once()
call_args = mock_handle.call_args
assert call_args[1]['arguments'] == {}
class TestReadResourceRecentEpisodes:
"""测试 read_resource 中 recent-episodes 资源（未连接时的分支）."""
@pytest.mark.asyncio
async def test_read_resource_recent_episodes_not_connected(self, server_with_mocks):
"""测试 recent-episodes 资源读取（未连接状态）."""
server, mock_graphiti_client, _ = server_with_mocks
handler = server.request_handlers.get(ReadResourceRequest)
assert handler is not None
mock_graphiti_client.is_connected.return_value = False
mock_graphiti_client.query_by_time_range.return_value = {"results": []}
result = await invoke_read_resource(handler, "graphitiace://recent-episodes")
assert isinstance(result, ReadResourceResult)
assert len(result.contents) == 1
content = json.loads(result.contents[0].text)
assert 'results' in content
