"""测试 server.py 资源处理的完整覆盖."""
import pytest
import json
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestServerResourceCoverage:
"""测试 server.py 资源处理的完整覆盖."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def mock_graphiti_client(self, config_manager):
"""创建 mock GraphitiClient."""
client = GraphitiClient(config_manager)
mock_driver = MagicMock()
mock_session = MagicMock()
mock_driver.session.return_value.__enter__ = MagicMock(return_value=mock_session)
mock_driver.session.return_value.__exit__ = MagicMock(return_value=None)
client.driver = mock_driver
client._connected = True
return client, mock_session
def test_read_resource_reconnect_failure(self, config_manager, mock_graphiti_client):
"""测试读取资源时重连失败."""
client, mock_session = mock_graphiti_client
# Mock check_reconnect 抛出异常
client.check_reconnect = MagicMock(side_effect=Exception("reconnect failed"))
# 验证异常被处理
try:
client.check_reconnect()
except Exception as e:
assert "reconnect failed" in str(e)
def test_strategy_heatmap_ace_disabled(self, config_manager):
"""测试 strategy-heatmap 资源，ACE 未启用."""
# ACE Manager 未启用时应该返回错误消息
ace_manager = MagicMock()
ace_manager.is_enabled.return_value = False
# 验证逻辑
if ace_manager is None or not ace_manager.is_enabled():
content = json.dumps({
"success": False,
"message": "ACE Manager 未启用或未安装，无法生成策略热力图。"
}, indent=2, ensure_ascii=False)
result = json.loads(content)
assert result["success"] is False
assert "ACE Manager" in result["message"]
def test_strategy_heatmap_ace_enabled_success(self, config_manager):
"""测试 strategy-heatmap 资源，ACE 启用且成功."""
ace_manager = MagicMock()
ace_manager.is_enabled.return_value = True
ace_manager.get_strategy_heatmap.return_value = {
"success": True,
"heatmap": {"tool1": {"high": 5, "medium": 3}}
}
heatmap = ace_manager.get_strategy_heatmap(limit=30, group_by="tool")
assert heatmap["success"] is True
def test_strategy_heatmap_ace_enabled_no_data(self, config_manager):
"""测试 strategy-heatmap 资源，ACE 启用但无数据."""
ace_manager = MagicMock()
ace_manager.is_enabled.return_value = True
ace_manager.get_strategy_heatmap.return_value = {
"success": False,
"message": "暂无可用的策略数据。"
}
heatmap = ace_manager.get_strategy_heatmap(limit=30, group_by="tool")
assert heatmap["success"] is False
message = heatmap.get("message", "暂无可用的策略数据。")
assert "暂无" in message
def test_read_resource_unknown_uri(self, config_manager):
"""测试读取未知 URI."""
uri = "graphitiace://unknown-resource"
expected_text = f"❌ 未知资源: {uri}"
assert uri in expected_text
def test_read_resource_exception(self, config_manager):
"""测试读取资源时抛出异常."""
error_msg = "some error"
expected_text = f"❌ 读取资源失败: {error_msg}"
assert error_msg in expected_text
class TestServerCallToolCoverage:
"""测试 server.py 工具调用的完整覆盖."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.mark.asyncio
async def test_call_tool_empty_arguments(self, config_manager):
"""测试工具调用时 arguments 为空字典."""
from src.tools import handle_tool_call
# handle_tool_call 不支持 None，server.py 会将 None 转为 {}
# 这里直接测试 {}
result = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert result is not None
assert len(result) > 0
@pytest.mark.asyncio
async def test_call_tool_reconnect_failure(self, config_manager):
"""测试工具调用时重连失败."""
from src.graphiti_client import GraphitiClient
client = GraphitiClient(config_manager)
# Mock check_reconnect 抛出异常
with patch.object(client, 'check_reconnect', side_effect=Exception("reconnect failed")):
# 验证异常被捕获
try:
client.check_reconnect()
assert False, "Should raise exception"
except Exception as e:
assert "reconnect failed" in str(e)
@pytest.mark.asyncio
async def test_call_tool_ace_check_reload(self, config_manager):
"""测试工具调用时 ACE 配置检查."""
ace_manager = MagicMock()
ace_manager.check_reload = MagicMock()
# 调用 check_reload
ace_manager.check_reload()
# 验证被调用
ace_manager.check_reload.assert_called_once()
class TestServerMainFunction:
"""测试 server.py 的 main 函数."""
def test_main_function_exists(self):
"""测试 main 函数存在."""
from src.server import main
assert main is not None
assert callable(main)
def test_create_server_function(self):
"""测试 create_server 函数."""
# Patch src.ace_manager.ACEManager 而不是 src.server.ACEManager
# 因为 server.py 中是从 .ace_manager 导入的
with patch('src.ace_manager.ACEManager') as mock_ace:
mock_ace.return_value = MagicMock()
mock_ace.return_value.is_enabled.return_value = False
from src.server import create_server
server = create_server()
assert server is not None
