"""GraphitiClient 验证重复节点覆盖测试."""
import pytest
from unittest.mock import Mock, MagicMock
from src.graphiti_client import GraphitiClient
from src.config_manager import ConfigManager
class TestGraphitiClientValidateDuplicatesCoverage:
"""GraphitiClient 验证重复节点覆盖测试类."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
@pytest.fixture
def mock_client(self, config_manager):
"""创建 Mock 客户端."""
client = GraphitiClient(config_manager)
# Mock driver 和 session
mock_driver = Mock()
mock_session = Mock()
mock_driver.session.return_value.__enter__ = Mock(return_value=mock_session)
mock_driver.session.return_value.__exit__ = Mock(return_value=None)
client.driver = mock_driver
client._connected = True
return client, mock_session
def test_validate_data_duplicates_coverage(self, mock_client):
"""测试验证数据重复节点覆盖（行1229-1233）."""
client, mock_session = mock_client
# Mock 孤立节点查询结果（空结果）
mock_orphaned_result = Mock()
mock_orphaned_result.__iter__ = Mock(return_value=iter([]))
# Mock 重复节点查询结果（有重复节点）
mock_record1 = Mock()
def getitem1(key):
if key == "label":
return "Technology"
elif key == "name":
return "TypeScript"
elif key == "count":
return 3
return None
mock_record1.__getitem__ = Mock(side_effect=getitem1)
mock_record2 = Mock()
def getitem2(key):
if key == "label":
return "Framework"
elif key == "name":
return "React"
elif key == "count":
return 2
return None
mock_record2.__getitem__ = Mock(side_effect=getitem2)
mock_duplicates_result = Mock()
mock_duplicates_result.__iter__ = Mock(return_value=iter([mock_record1, mock_record2]))
# Mock 完整性查询结果（空结果）
mock_integrity_result = Mock()
mock_integrity_record = Mock()
mock_integrity_record.__getitem__ = Mock(side_effect=lambda key: 0 if key == "count" else None)
mock_integrity_result.single.return_value = mock_integrity_record
# 设置 session.run 的返回值（按调用顺序）
mock_session.run.side_effect = [
mock_orphaned_result, # 孤立节点查询
mock_duplicates_result, # 重复节点查询
mock_integrity_result # 完整性查询
]
result = client.validate_data(
check_orphaned=True,
check_duplicates=True,
check_integrity=True
)
assert result is not None
assert result['success'] is True
assert len(result['issues']) >= 2
assert any("重复" in issue for issue in result['issues'])
