"""handle_tool_call 结果反思阶段的补充测试."""
import asyncio
from unittest.mock import patch
import pytest
from src.config_manager import ConfigManager
from src.tools import handle_tool_call
class DummyAceManager:
"""提供最小化的 ACE Manager 行为以测试反思逻辑."""
def __init__(self, reflection_side_effect=None):
self._reflection_side_effect = reflection_side_effect
def is_enabled(self):
return True
def generate_tool_strategy(self, **_kwargs):
return None
def reflect_on_result(self, *args, **kwargs):
if self._reflection_side_effect:
raise self._reflection_side_effect
@pytest.fixture
def config_manager(temp_config_dir):
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.mark.asyncio
async def test_handle_tool_call_reflection_thread_logs_error(config_manager):
"""无异步方法时应进入线程分支，并吞掉反思错误."""
ace_manager = DummyAceManager(reflection_side_effect=RuntimeError("reflection boom"))
graphiti_client = None
class ImmediateThread:
def __init__(self, target=None, daemon=None):
self._target = target
def start(self):
if self._target:
self._target()
with patch("threading.Thread", side_effect=lambda *args, **kwargs: ImmediateThread(*args, **kwargs)):
with patch.object(config_manager, "configure_api", return_value={"success": True, "message": "API 配置成功"}):
result = await handle_tool_call(
tool_name="configure_api",
arguments={"provider": "openai", "api_key": "demo", "model": "gpt-4o-mini"},
config_manager=config_manager,
graphiti_client=graphiti_client,
ace_manager=ace_manager,
)
assert "API 配置成功" in result[0].text
@pytest.mark.asyncio
async def test_handle_tool_call_reflection_outer_exception(config_manager):
"""当摘要函数抛错时，应触发外围异常日志而不中断响应."""
ace_manager = DummyAceManager()
graphiti_client = None
with patch("src.tools._summarize_result_text", side_effect=ValueError("summary boom")), patch(
"src.tools.logger.warning"
) as mock_warning, patch.object(
config_manager, "configure_api", return_value={"success": True, "message": "API 配置成功"}
):
result = await handle_tool_call(
tool_name="configure_api",
arguments={"provider": "openai", "api_key": "demo", "model": "gpt-4o-mini"},
config_manager=config_manager,
graphiti_client=graphiti_client,
ace_manager=ace_manager,
)
# 函数仍然返回正常文案
assert "API 配置成功" in result[0].text
# 外层异常应该被捕获并记录
assert any("ACE 反思失败" in str(call.args[0]) for call in mock_warning.call_args_list)
