adds full coverage to conv history save
This commit is contained in:
hajdul88
parent
5f6560d6a7
commit
ca593bba15
1 changed files with 350 additions and 0 deletions
|
|
@ -152,3 +152,353 @@ class TestConversationHistoryUtils:
|
|||
assert result is True
|
||||
call_kwargs = mock_cache.add_qa.call_args.kwargs
|
||||
assert call_kwargs["session_id"] == "default_session"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_conversation_history_no_user_id(self):
|
||||
"""Test save_conversation_history returns False when user_id is None."""
|
||||
session_user.set(None)
|
||||
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
save_conversation_history,
|
||||
)
|
||||
|
||||
result = await save_conversation_history(
|
||||
query="Test question",
|
||||
context_summary="Test context",
|
||||
answer="Test answer",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_conversation_history_caching_disabled(self):
|
||||
"""Test save_conversation_history returns False when caching is disabled."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = False
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
save_conversation_history,
|
||||
)
|
||||
|
||||
result = await save_conversation_history(
|
||||
query="Test question",
|
||||
context_summary="Test context",
|
||||
answer="Test answer",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_conversation_history_cache_engine_none(self):
|
||||
"""Test save_conversation_history returns False when cache_engine is None."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=None):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
save_conversation_history,
|
||||
)
|
||||
|
||||
result = await save_conversation_history(
|
||||
query="Test question",
|
||||
context_summary="Test context",
|
||||
answer="Test answer",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_conversation_history_cache_connection_error(self):
|
||||
"""Test save_conversation_history handles CacheConnectionError gracefully."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
from cognee.infrastructure.databases.exceptions import CacheConnectionError
|
||||
|
||||
mock_cache = create_mock_cache_engine([])
|
||||
mock_cache.add_qa = AsyncMock(side_effect=CacheConnectionError("Connection failed"))
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
save_conversation_history,
|
||||
)
|
||||
|
||||
result = await save_conversation_history(
|
||||
query="Test question",
|
||||
context_summary="Test context",
|
||||
answer="Test answer",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_save_conversation_history_generic_exception(self):
|
||||
"""Test save_conversation_history handles generic exceptions gracefully."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
mock_cache = create_mock_cache_engine([])
|
||||
mock_cache.add_qa = AsyncMock(side_effect=ValueError("Unexpected error"))
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
save_conversation_history,
|
||||
)
|
||||
|
||||
result = await save_conversation_history(
|
||||
query="Test question",
|
||||
context_summary="Test context",
|
||||
answer="Test answer",
|
||||
)
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_no_user_id(self):
|
||||
"""Test get_conversation_history returns empty string when user_id is None."""
|
||||
session_user.set(None)
|
||||
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert result == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_caching_disabled(self):
|
||||
"""Test get_conversation_history returns empty string when caching is disabled."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = False
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert result == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_default_session(self):
|
||||
"""Test get_conversation_history uses 'default_session' when session_id is None."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
mock_cache = create_mock_cache_engine([])
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
await get_conversation_history(session_id=None)
|
||||
|
||||
mock_cache.get_latest_qa.assert_called_once_with(
|
||||
str(user.id), "default_session"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_cache_engine_none(self):
|
||||
"""Test get_conversation_history returns empty string when cache_engine is None."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=None):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert result == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_cache_connection_error(self):
|
||||
"""Test get_conversation_history handles CacheConnectionError gracefully."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
from cognee.infrastructure.databases.exceptions import CacheConnectionError
|
||||
|
||||
mock_cache = create_mock_cache_engine([])
|
||||
mock_cache.get_latest_qa = AsyncMock(
|
||||
side_effect=CacheConnectionError("Connection failed")
|
||||
)
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert result == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_generic_exception(self):
|
||||
"""Test get_conversation_history handles generic exceptions gracefully."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
mock_cache = create_mock_cache_engine([])
|
||||
mock_cache.get_latest_qa = AsyncMock(side_effect=ValueError("Unexpected error"))
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert result == ""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_conversation_history_missing_keys(self):
|
||||
"""Test get_conversation_history handles missing keys in history entries."""
|
||||
user = create_mock_user()
|
||||
session_user.set(user)
|
||||
|
||||
mock_history = [
|
||||
{
|
||||
"time": "2024-01-15 10:30:45",
|
||||
"question": "What is AI?",
|
||||
},
|
||||
{
|
||||
"context": "AI is artificial intelligence",
|
||||
"answer": "AI stands for Artificial Intelligence",
|
||||
},
|
||||
{},
|
||||
]
|
||||
mock_cache = create_mock_cache_engine(mock_history)
|
||||
|
||||
cache_module = importlib.import_module(
|
||||
"cognee.infrastructure.databases.cache.get_cache_engine"
|
||||
)
|
||||
|
||||
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
|
||||
with patch(
|
||||
"cognee.modules.retrieval.utils.session_cache.CacheConfig"
|
||||
) as MockCacheConfig:
|
||||
mock_config = MagicMock()
|
||||
mock_config.caching = True
|
||||
MockCacheConfig.return_value = mock_config
|
||||
|
||||
from cognee.modules.retrieval.utils.session_cache import (
|
||||
get_conversation_history,
|
||||
)
|
||||
|
||||
result = await get_conversation_history(session_id="test_session")
|
||||
|
||||
assert "Previous conversation:" in result
|
||||
assert "[2024-01-15 10:30:45]" in result
|
||||
assert "QUESTION: What is AI?" in result
|
||||
assert "Unknown time" in result
|
||||
assert "CONTEXT: AI is artificial intelligence" in result
|
||||
assert "ANSWER: AI stands for Artificial Intelligence" in result
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue