"""Graphiti 客户端集成测试."""
import pytest
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
@pytest.mark.integration
class TestGraphitiClient:
"""Graphiti 客户端集成测试类."""
@pytest.fixture
def configured_client(self, config_manager, mock_neo4j_config):
"""创建已配置的客户端."""
config_manager.configure_neo4j(**mock_neo4j_config)
client = GraphitiClient(config_manager)
return client
def test_connect_success(self, configured_client, mock_neo4j_config):
"""测试成功连接（需要真实的 Neo4j 数据库）."""
# 注意：这个测试需要真实的 Neo4j 数据库
# 如果数据库不可用，测试会跳过
# 使用较短的超时避免产生大量认证错误日志
try:
result = configured_client.connect()
if result:
assert configured_client.is_connected() is True
configured_client.disconnect()
else:
# 连接失败时快速跳过，避免产生认证错误日志
pytest.skip("Neo4j 数据库连接失败（可能密码不匹配或数据库未运行）")
except Exception as e:
# 如果是认证错误，快速跳过而不是重试
if "authentication" in str(e).lower() or "password" in str(e).lower():
pytest.skip(f"Neo4j 认证失败: {e}")
pytest.skip(f"Neo4j 数据库不可用: {e}")
def test_connect_failure(self, config_manager):
"""测试连接失败."""
config_manager.configure_neo4j(
uri="bolt://invalid:7687",
username="neo4j",
password="wrong"
)
client = GraphitiClient(config_manager)
result = client.connect()
assert result is False
assert client.is_connected() is False
def test_add_episode_without_connection(self, graphiti_client):
"""测试未连接时添加 Episode."""
result = graphiti_client.add_episode(
content="Test content",
metadata={"type": "test"}
)
assert result['success'] is False
assert "未连接" in result['message']
def test_search_entities_without_connection(self, graphiti_client):
"""测试未连接时搜索实体."""
result = graphiti_client.search_entities(query="test")
assert result['success'] is False
assert "未连接" in result['message']
def test_get_statistics_without_connection(self, graphiti_client):
"""测试未连接时获取统计信息."""
result = graphiti_client.get_statistics()
assert result['success'] is False
assert "未连接" in result['message']
def test_cache_integration(self, configured_client, cache_manager):
"""测试缓存集成."""
# 配置并连接
try:
if configured_client.connect():
# 获取统计信息（应该使用缓存）
result1 = configured_client.get_statistics(use_cache=True)
if result1['success']:
# 再次获取（应该从缓存）
result2 = configured_client.get_statistics(use_cache=True)
assert result2['success'] is True
configured_client.disconnect()
else:
# 连接失败时快速跳过，避免产生认证错误日志
pytest.skip("Neo4j 数据库连接失败（可能密码不匹配或数据库未运行）")
except Exception as e:
# 如果是认证错误，快速跳过而不是重试
if "authentication" in str(e).lower() or "password" in str(e).lower():
pytest.skip(f"Neo4j 认证失败: {e}")
pytest.skip(f"Neo4j 数据库不可用: {e}")
