"""工具批量添加 Episodes 测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from src.tools import handle_tool_call
from src.config_manager import ConfigManager
class TestToolsAddEpisodesBulk:
"""工具批量添加 Episodes 测试类."""
@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_add_episodes_bulk_client_not_initialized(self, config_manager):
"""测试批量添加 Episodes 时客户端未初始化."""
result = await handle_tool_call(
tool_name="add_episodes_bulk",
arguments={
"episodes": [
{"content": "Episode 1"},
{"content": "Episode 2"}
]
},
config_manager=config_manager,
graphiti_client=None
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "错误" in result_text or "error" in result_text.lower() or "未初始化" in result_text
@pytest.mark.asyncio
async def test_add_episodes_bulk_success(self, config_manager):
"""测试批量添加 Episodes 成功."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.add_episodes_bulk = AsyncMock(return_value={
"success": True,
"message": "批量添加成功",
"total": 2,
"success_count": 2,
"fail_count": 0,
"results": [
{"success": True, "episode_id": 1},
{"success": True, "episode_id": 2}
]
})
result = await handle_tool_call(
tool_name="add_episodes_bulk",
arguments={
"episodes": [
{"content": "Episode 1"},
{"content": "Episode 2"}
]
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "成功" in result_text or "success" in result_text.lower() or "总计" in result_text
@pytest.mark.asyncio
async def test_add_episodes_bulk_with_failures(self, config_manager):
"""测试批量添加 Episodes 部分失败."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.add_episodes_bulk = AsyncMock(return_value={
"success": True,
"message": "批量添加完成",
"total": 2,
"success_count": 1,
"fail_count": 1,
"results": [
{"success": True, "episode_id": 1},
{"success": False, "message": "添加失败"}
]
})
result = await handle_tool_call(
tool_name="add_episodes_bulk",
arguments={
"episodes": [
{"content": "Episode 1"},
{"content": "Episode 2"}
]
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "失败" in result_text or "fail" in result_text.lower() or "失败的 Episode" in result_text
@pytest.mark.asyncio
async def test_add_episodes_bulk_failure(self, config_manager):
"""测试批量添加 Episodes 失败."""
from src.graphiti_client import GraphitiClient
from unittest.mock import AsyncMock
mock_client = Mock(spec=GraphitiClient)
mock_client.is_connected.return_value = True
mock_client.add_episodes_bulk = AsyncMock(return_value={
"success": False,
"message": "批量添加失败"
})
result = await handle_tool_call(
tool_name="add_episodes_bulk",
arguments={
"episodes": [
{"content": "Episode 1"}
]
},
config_manager=config_manager,
graphiti_client=mock_client
)
assert result is not None
assert isinstance(result, list)
result_text = "".join([str(item.text) if hasattr(item, 'text') else str(item) for item in result])
assert "失败" in result_text or "error" in result_text.lower() or "❌" in result_text
