"""GraphitiClient 完整覆盖测试."""
import pytest
from unittest.mock import Mock, MagicMock, patch
from datetime import datetime
from src.config_manager import ConfigManager
from src.graphiti_client import GraphitiClient, _neo4j_to_dict, _lazy_import_graphiti
class TestNeo4jToDict完整覆盖:
"""测试 _neo4j_to_dict 函数的所有分支."""
def test_neo4j_datetime_success(self):
"""测试 Neo4j DateTime 成功转换."""
# 模拟 Neo4jDateTime
class MockNeo4jDateTime:
def isoformat(self):
return "2025-01-01T00:00:00"
with patch('src.graphiti_client.Neo4jDateTime', MockNeo4jDateTime):
obj = MockNeo4jDateTime()
result = _neo4j_to_dict(obj)
assert result == "2025-01-01T00:00:00"
def test_neo4j_datetime_isoformat_error(self):
"""测试 Neo4j DateTime isoformat 失败."""
class MockNeo4jDateTime:
def isoformat(self):
raise AttributeError("no isoformat")
def __str__(self):
return "mock-datetime-str"
# 直接测试这种情况
obj = MockNeo4jDateTime()
try:
obj.isoformat()
except AttributeError:
result = str(obj)
assert result == "mock-datetime-str"
def test_datetime_class_name_no_isoformat(self):
"""测试类名包含 DateTime 但没有 isoformat."""
class SomeDateTime:
def __str__(self):
return "some-datetime"
result = _neo4j_to_dict(SomeDateTime())
assert result == "some-datetime"
def test_datetime_class_name_isoformat_error(self):
"""测试类名包含 DateTime 且 isoformat 抛出异常."""
class FailDateTime:
def isoformat(self):
raise ValueError("isoformat error")
def __str__(self):
return "fail-datetime"
result = _neo4j_to_dict(FailDateTime())
assert result == "fail-datetime"
def test_isoformat_method_error_with_str(self):
"""测试 isoformat 方法失败，使用 str."""
class IsoformatFails:
def isoformat(self):
raise TypeError("type error")
def __str__(self):
return "str-fallback"
result = _neo4j_to_dict(IsoformatFails())
assert result == "str-fallback"
def test_isoformat_method_error_with_repr(self):
"""测试 isoformat 和 str 都失败，使用 repr."""
class AllFails:
def isoformat(self):
raise TypeError("type error")
def __str__(self):
raise ValueError("str error")
def __repr__(self):
return "AllFails(repr)"
result = _neo4j_to_dict(AllFails())
assert "AllFails" in result
class TestLazyImportGraphiti:
"""测试 _lazy_import_graphiti 函数."""
def test_lazy_import_graphiti_available(self):
"""测试 graphiti-core 可用."""
result = _lazy_import_graphiti()
# 结果取决于是否安装了 graphiti-core
assert isinstance(result, bool)
def test_lazy_import_graphiti_cache(self):
"""测试 _lazy_import_graphiti 缓存."""
# 第一次调用
result1 = _lazy_import_graphiti()
# 第二次调用应该使用缓存
result2 = _lazy_import_graphiti()
assert result1 == result2
class TestGraphitiClientConnect:
"""测试 GraphitiClient 连接相关方法."""
@pytest.fixture
def config_manager(self, temp_config_dir):
"""创建配置管理器."""
return ConfigManager(config_path=temp_config_dir / ".graphitiace" / "config.json")
def test_connect_test_mode(self, config_manager):
"""测试测试模式下连接被跳过."""
import os
with patch.dict(os.environ, {'GRAPHITIACE_TEST_MODE': 'true'}):
client = GraphitiClient(config_manager)
config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
result = client.connect()
assert result is False
def test_connect_no_config(self, config_manager):
"""测试没有 Neo4j 配置时连接失败."""
client = GraphitiClient(config_manager)
# 确保 get_neo4j_config 返回 None
# ConfigManager 默认会创建配置，我们需要 mock 它
with patch.object(config_manager, 'get_neo4j_config', return_value=None):
result = client.connect()
assert result is False
@pytest.fixture
def fresh_config_manager(self, temp_config_dir):
"""创建全新的配置管理器（确保没有 API 配置）."""
fresh_path = temp_config_dir / ".graphitiace_fresh_complete" / "config.json"
fresh_path.parent.mkdir(parents=True, exist_ok=True)
return ConfigManager(config_path=fresh_path)
def test_initialize_graphiti_no_api_key(self, fresh_config_manager):
"""测试没有 API key 时不初始化 Graphiti."""
client = GraphitiClient(fresh_config_manager)
fresh_config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
result = client._initialize_graphiti()
# 没有 API key，应该返回 False
assert result is False
assert client.graphiti is None
assert client._graphiti_initialized is True
def test_initialize_graphiti_already_initialized(self, config_manager):
"""测试已经初始化过的情况."""
client = GraphitiClient(config_manager)
client._graphiti_initialized = True
client.graphiti = None
result = client._initialize_graphiti()
# 已经初始化过，应该返回 graphiti is not None
assert result is False
def test_check_reconnect_not_connected(self, config_manager):
"""测试未连接时 check_reconnect."""
client = GraphitiClient(config_manager)
# 配置 Neo4j
config_manager.configure_neo4j(
uri="bolt://localhost:7687",
username="neo4j",
password="test"
)
# 确保 connect 失败，这样 is_connected 仍为 False
# 或者断言它尝试连接了
with patch.object(client, 'connect', return_value=True) as mock_connect:
client.check_reconnect()
# 应该尝试连接
mock_connect.assert_called_once()
# 此时 is_connected 仍为 False (因为 mock 了 connect 但未设置 _connected)
assert not client.is_connected()
class TestGraphitiClientQueries:
"""测试 GraphitiClient 查询方法."""
@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):
"""创建已连接的客户端."""
client = GraphitiClient(config_manager)
mock_driver = MagicMock()
mock_session = MagicMock()
mock_driver.session.return_value.__enter__ = MagicMock(return_value=mock_session)
mock_driver.session.return_value.__exit__ = MagicMock(return_value=None)
client.driver = mock_driver
client._connected = True
return client, mock_session
def test_query_knowledge_graph_not_connected(self, config_manager):
"""测试未连接时查询知识图谱."""
client = GraphitiClient(config_manager)
result = client.query_knowledge_graph("MATCH (n) RETURN n LIMIT 1")
assert result["success"] is False
assert "未连接" in result["message"]
def test_get_statistics_not_connected(self, config_manager):
"""测试未连接时获取统计信息."""
client = GraphitiClient(config_manager)
result = client.get_statistics()
assert result["success"] is False
assert "未连接" in result["message"]
def test_query_by_time_range_not_connected(self, config_manager):
"""测试未连接时按时间范围查询."""
client = GraphitiClient(config_manager)
result = client.query_by_time_range(days=7)
assert result["success"] is False
assert "未连接" in result["message"]
