"""端到端测试 - 完整工作流."""
import pytest
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
from src.tools import handle_tool_call
@pytest.mark.e2e
class TestFullWorkflow:
"""端到端测试类."""
@pytest.fixture
def workflow_setup(self, temp_config_dir):
"""设置工作流测试环境."""
config_manager = ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
return config_manager
@pytest.mark.asyncio
async def test_complete_workflow(self, workflow_setup, mock_neo4j_config):
"""测试完整工作流."""
config_manager = workflow_setup
# 步骤 1: 配置 Neo4j
result = await handle_tool_call(
tool_name="configure_neo4j",
arguments=mock_neo4j_config,
config_manager=config_manager
)
assert len(result) > 0
# 步骤 2: 检查配置
result = await handle_tool_call(
tool_name="check_configuration",
arguments={},
config_manager=config_manager
)
assert len(result) > 0
# 步骤 3: 设置组 ID
result = await handle_tool_call(
tool_name="set_group_id",
arguments={"group_id": "e2e_test"},
config_manager=config_manager
)
assert len(result) > 0
# 步骤 4: 创建客户端并连接（如果数据库可用）
client = GraphitiClient(config_manager)
try:
if client.connect():
# 步骤 5: 添加 Episode
result = client.add_episode(
content="E2E 测试：我喜欢 Python",
metadata={"type": "e2e_test"}
)
if result['success']:
episode_id = result.get('episode_id')
# 步骤 6: 搜索实体
result = client.search_entities(query="Python")
assert result['success'] is True
# 步骤 7: 获取统计信息
result = client.get_statistics()
assert result['success'] is True
# 步骤 8: 清理
if episode_id:
client.delete_episode(episode_id=episode_id)
client.disconnect()
except Exception:
pytest.skip("Neo4j 数据库不可用，跳过 E2E 测试")
@pytest.mark.asyncio
async def test_tool_workflow(self, workflow_setup, mock_neo4j_config):
"""测试工具调用工作流."""
config_manager = workflow_setup
# 配置
await handle_tool_call(
tool_name="configure_neo4j",
arguments=mock_neo4j_config,
config_manager=config_manager
)
# 创建客户端
client = GraphitiClient(config_manager)
try:
if client.connect():
# 通过工具添加 Episode
result = await handle_tool_call(
tool_name="add_episode",
arguments={
"content": "工具测试 Episode",
"metadata": {"type": "tool_test"}
},
config_manager=config_manager,
graphiti_client=client
)
assert len(result) > 0
# 通过工具搜索
result = handle_tool_call(
tool_name="search_entities",
arguments={"query": "工具"},
config_manager=config_manager,
graphiti_client=client
)
assert len(result) > 0
client.disconnect()
except Exception:
pytest.skip("Neo4j 数据库不可用，跳过工具工作流测试")
