"""Graphiti 客户端错误处理测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestGraphitiClientErrorHandling:
"""Graphiti 客户端错误处理测试类."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def connected_client(self, config_manager, mock_neo4j_config, mock_driver):
"""创建已连接的客户端."""
config_manager.configure_neo4j(**mock_neo4j_config)
client = GraphitiClient(config_manager)
driver, session = mock_driver
client.driver = driver
client._connected = True
return client, session
@pytest.mark.asyncio
async def test_add_episode_database_error(self, connected_client):
"""测试添加 Episode 时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Database connection failed")
result = await client.add_episode(content="Test")
assert result['success'] is False
assert "失败" in result['message']
def test_search_entities_database_error(self, connected_client):
"""测试搜索实体时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Query execution failed")
result = client.search_entities(query="test")
assert result['success'] is False
assert "失败" in result['message']
def test_query_knowledge_graph_syntax_error(self, connected_client):
"""测试查询知识图谱时语法错误."""
client, session = connected_client
# Mock 抛出语法错误
session.run.side_effect = Exception("Syntax error in Cypher query")
result = client.query_knowledge_graph(cypher_query="INVALID CYPHER")
assert result['success'] is False
assert "失败" in result['message']
def test_delete_episode_database_error(self, connected_client):
"""测试删除 Episode 时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Delete operation failed")
result = client.delete_episode(episode_id=1)
assert result['success'] is False
assert "失败" in result['message']
def test_clear_graph_database_error(self, connected_client):
"""测试清空图谱时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Clear operation failed")
# clear_graph 不需要 confirm 参数
result = client.clear_graph()
assert result['success'] is False
assert "失败" in result['message']
def test_export_graph_data_database_error(self, connected_client):
"""测试导出数据时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Export operation failed")
result = client.export_graph_data(format="json")
assert result['success'] is False
assert "失败" in result['message']
def test_import_graph_data_validation_error(self, connected_client):
"""测试导入数据时验证错误."""
client, session = connected_client
# 测试无效数据格式
invalid_data = {"invalid": "format"}
result = client.import_graph_data(data=invalid_data, format="json")
# 应该返回错误或处理无效数据
assert result is not None
def test_get_statistics_database_error(self, connected_client):
"""测试获取统计信息时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Statistics query failed")
result = client.get_statistics()
assert result['success'] is False
assert "失败" in result['message']
@pytest.mark.asyncio
async def test_semantic_search_database_error(self, connected_client):
"""测试语义搜索时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Search operation failed")
result = await client.semantic_search(query="test")
assert result['success'] is False
assert "失败" in result['message']
def test_query_by_time_range_invalid_time_format(self, connected_client):
"""测试时间范围查询时无效时间格式."""
client, session = connected_client
# Mock 查询结果
mock_result = MagicMock()
mock_result.__iter__.return_value = []
session.run.return_value = mock_result
# 测试无效时间格式（应该被处理或返回错误）
result = client.query_by_time_range(
start_time="invalid-time",
end_time="invalid-time"
)
# 应该返回结果（即使为空）或错误
assert result is not None
def test_validate_data_database_error(self, connected_client):
"""测试数据验证时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Validation query failed")
result = client.validate_data()
assert result['success'] is False
assert "失败" in result['message']
def test_clean_orphaned_nodes_database_error(self, connected_client):
"""测试清理孤立节点时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Clean operation failed")
result = client.clean_orphaned_nodes()
assert result['success'] is False
assert "失败" in result['message']
def test_rebuild_indexes_database_error(self, connected_client):
"""测试重建索引时数据库错误."""
client, session = connected_client
# Mock 抛出数据库错误
session.run.side_effect = Exception("Index rebuild failed")
result = client.rebuild_indexes()
assert result['success'] is False
assert "失败" in result['message']
@pytest.mark.asyncio
async def test_add_episodes_bulk_partial_failure(self, connected_client):
"""测试批量添加 Episode 时部分失败."""
client, session = connected_client
# Mock 第一次成功，第二次失败
mock_result_success = MagicMock()
mock_record = MagicMock()
mock_record.__getitem__.return_value = 1
mock_result_success.single.return_value = mock_record
mock_result_fail = MagicMock()
mock_result_fail.side_effect = Exception("Second episode failed")
session.run.side_effect = [mock_result_success, Exception("Second episode failed")]
episodes = [
{"content": "Test 1"},
{"content": "Test 2"}
]
result = await client.add_episodes_bulk(episodes=episodes)
# 应该返回部分成功的结果
assert result is not None
assert result.get('total') == 2
