"""Graphiti 客户端方法单元测试（不连接数据库）."""
import pytest
from unittest.mock import Mock, patch, MagicMock
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient
class TestGraphitiClientMethods:
"""Graphiti 客户端方法测试类."""
def test_is_connected_false(self, graphiti_client):
"""测试未连接状态."""
assert graphiti_client.is_connected() is False
def test_disconnect_when_not_connected(self, graphiti_client):
"""测试断开未连接的客户端."""
# 不应该抛出异常
graphiti_client.disconnect()
assert True
@patch('src.graphiti_client.GraphDatabase')
def test_connect_success(self, mock_graph_db, config_manager, mock_neo4j_config):
"""测试成功连接（模拟）."""
config_manager.configure_neo4j(**mock_neo4j_config)
client = GraphitiClient(config_manager)
# 模拟驱动和会话
mock_driver = MagicMock()
mock_session = MagicMock()
mock_driver.session.return_value.__enter__.return_value = mock_session
mock_driver.session.return_value.__exit__.return_value = None
mock_session.run.return_value = None
mock_graph_db.driver.return_value = mock_driver
# Mock _initialize_graphiti 以避免实际初始化
with patch.object(client, '_initialize_graphiti', return_value=True):
result = client.connect()
assert result is True
assert client.is_connected() is True
client.disconnect()
@patch('src.graphiti_client.GraphDatabase')
def test_connect_failure(self, mock_graph_db, config_manager, mock_neo4j_config):
"""测试连接失败（模拟）."""
config_manager.configure_neo4j(**mock_neo4j_config)
client = GraphitiClient(config_manager)
# 模拟连接失败
mock_graph_db.driver.side_effect = Exception("Connection failed")
result = client.connect()
assert result is False
assert client.is_connected() is False
@pytest.mark.asyncio
async def test_add_episode_not_connected(self, graphiti_client):
"""测试未连接时添加 Episode."""
result = await graphiti_client.add_episode(
content="Test",
metadata={"type": "test"}
)
assert result['success'] is False
def test_search_entities_not_connected(self, graphiti_client):
"""测试未连接时搜索实体."""
result = graphiti_client.search_entities(query="test")
assert result['success'] is False
def test_search_relationships_not_connected(self, graphiti_client):
"""测试未连接时搜索关系."""
result = graphiti_client.search_relationships(query="test")
assert result['success'] is False
def test_query_knowledge_graph_not_connected(self, graphiti_client):
"""测试未连接时查询知识图谱."""
result = graphiti_client.query_knowledge_graph(cypher_query="MATCH (n) RETURN n")
assert result['success'] is False
def test_delete_episode_not_connected(self, graphiti_client):
"""测试未连接时删除 Episode."""
result = graphiti_client.delete_episode(episode_id=1)
assert result['success'] is False
def test_clear_graph_not_connected(self, graphiti_client):
"""测试未连接时清空图谱."""
result = graphiti_client.clear_graph()
assert result['success'] is False
def test_query_by_time_range_not_connected(self, graphiti_client):
"""测试未连接时时间范围查询."""
result = graphiti_client.query_by_time_range(days=7)
assert result['success'] is False
@pytest.mark.asyncio
async def test_semantic_search_not_connected(self, graphiti_client):
"""测试未连接时语义搜索."""
result = await graphiti_client.semantic_search(query="test")
assert result['success'] is False
@pytest.mark.asyncio
async def test_add_episodes_bulk_not_connected(self, graphiti_client):
"""测试未连接时批量添加."""
result = await graphiti_client.add_episodes_bulk(episodes=[])
assert result['success'] is False
def test_export_graph_data_not_connected(self, graphiti_client):
"""测试未连接时导出数据."""
result = graphiti_client.export_graph_data(format="json")
assert result['success'] is False
def test_import_graph_data_not_connected(self, graphiti_client):
"""测试未连接时导入数据."""
result = graphiti_client.import_graph_data(data={})
assert result['success'] is False
def test_get_statistics_not_connected(self, graphiti_client):
"""测试未连接时获取统计信息."""
result = graphiti_client.get_statistics()
assert result['success'] is False
def test_validate_data_not_connected(self, graphiti_client):
"""测试未连接时验证数据."""
result = graphiti_client.validate_data()
assert result['success'] is False
def test_clean_orphaned_nodes_not_connected(self, graphiti_client):
"""测试未连接时清理孤立节点."""
result = graphiti_client.clean_orphaned_nodes()
assert result['success'] is False
def test_rebuild_indexes_not_connected(self, graphiti_client):
"""测试未连接时重建索引."""
result = graphiti_client.rebuild_indexes()
assert result['success'] is False
