"""GraphitiClient 按时间范围查询测试."""
import pytest
from unittest.mock import Mock, MagicMock
from datetime import datetime
from src.graphiti_client import GraphitiClient
from src.config_manager import ConfigManager
class TestGraphitiClientQueryTimeRange:
"""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_query_by_time_range_with_entity_type_and_start_time(self, mock_client):
"""测试按时间范围查询（有实体类型和开始时间）."""
client, mock_session = mock_client
# Mock 查询结果
mock_result = Mock()
mock_record = Mock()
mock_record.__getitem__ = Mock(side_effect=lambda key: {"e": Mock()} if key == "e" else None)
mock_result.__iter__ = Mock(return_value=iter([mock_record]))
mock_session.run.return_value = mock_result
result = client.query_by_time_range(
start_time="2023-01-01T00:00:00",
end_time="2023-12-31T23:59:59",
entity_type="Technology",
limit=10
)
assert result is not None
assert "success" in result
mock_session.run.assert_called_once()
def test_query_by_time_range_with_entity_type_no_start_time(self, mock_client):
"""测试按时间范围查询（有实体类型，无开始时间）."""
client, mock_session = mock_client
# Mock 查询结果
mock_result = Mock()
mock_record = Mock()
mock_record.__getitem__ = Mock(side_effect=lambda key: {"e": Mock()} if key == "e" else None)
mock_result.__iter__ = Mock(return_value=iter([mock_record]))
mock_session.run.return_value = mock_result
result = client.query_by_time_range(
end_time="2023-12-31T23:59:59",
entity_type="Technology",
limit=10
)
assert result is not None
assert "success" in result
mock_session.run.assert_called_once()
def test_query_by_time_range_no_entity_type_with_start_time(self, mock_client):
"""测试按时间范围查询（无实体类型，有开始时间）."""
client, mock_session = mock_client
# Mock 查询结果
mock_result = Mock()
mock_record = Mock()
mock_record.__getitem__ = Mock(side_effect=lambda key: {"e": Mock()} if key == "e" else None)
mock_result.__iter__ = Mock(return_value=iter([mock_record]))
mock_session.run.return_value = mock_result
result = client.query_by_time_range(
start_time="2023-01-01T00:00:00",
end_time="2023-12-31T23:59:59",
limit=10
)
assert result is not None
assert "success" in result
mock_session.run.assert_called_once()
def test_query_by_time_range_no_entity_type_no_start_time(self, mock_client):
"""测试按时间范围查询（无实体类型，无开始时间）."""
client, mock_session = mock_client
# Mock 查询结果
mock_result = Mock()
mock_record = Mock()
mock_record.__getitem__ = Mock(side_effect=lambda key: {"e": Mock()} if key == "e" else None)
mock_result.__iter__ = Mock(return_value=iter([mock_record]))
mock_session.run.return_value = mock_result
result = client.query_by_time_range(
end_time="2023-12-31T23:59:59",
limit=10
)
assert result is not None
assert "success" in result
mock_session.run.assert_called_once()
def test_query_by_time_range_no_end_time(self, mock_client):
"""测试按时间范围查询（无结束时间，使用当前时间）."""
client, mock_session = mock_client
# Mock 查询结果
mock_result = Mock()
mock_record = Mock()
mock_record.__getitem__ = Mock(side_effect=lambda key: {"e": Mock()} if key == "e" else None)
mock_result.__iter__ = Mock(return_value=iter([mock_record]))
mock_session.run.return_value = mock_result
result = client.query_by_time_range(
start_time="2023-01-01T00:00:00",
limit=10
)
assert result is not None
assert "success" in result
mock_session.run.assert_called_once()
