gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore: unit test fix for cache mocking

Browse source
This commit is contained in:
hajdul88 2025-10-17 11:06:34 +02:00
parent 30a31889d0
commit ebb5b94265
1 changed files with 23 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
cognee/tests/unit/modules/retrieval/conversation_history_test.py
View file

@ -1,48 +1,41 @@
import pytest import pytest
from unittest.mock import AsyncMock, patch, MagicMock from unittest.mock import AsyncMock, patch, MagicMock
from cognee.context_global_variables import session_user from cognee.context_global_variables import session_user
import importlib
def create_mock_cache_engine(qa_history=None): def create_mock_cache_engine(qa_history=None):
"""Mocking cache engine as it is tested somewhere else"""
mock_cache = AsyncMock() mock_cache = AsyncMock()
if qa_history is None: if qa_history is None:
qa_history = [] qa_history = []
mock_cache.get_latest_qa = AsyncMock(return_value=qa_history) mock_cache.get_latest_qa = AsyncMock(return_value=qa_history)
mock_cache.add_qa = AsyncMock(return_value=None) mock_cache.add_qa = AsyncMock(return_value=None)
return mock_cache return mock_cache
def create_mock_user(): def create_mock_user():
"""Create a mock user without database access"""
mock_user = MagicMock() mock_user = MagicMock()
mock_user.id = "test-user-id-123" mock_user.id = "test-user-id-123"
return mock_user return mock_user
class TestConversationHistoryUtils: class TestConversationHistoryUtils:
"""Test the two utility functions: get_conversation_history and save_to_session_cache"""
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_conversation_history_returns_empty_when_no_history(self): async def test_get_conversation_history_returns_empty_when_no_history(self):
"""Test get_conversation_history returns empty string when no history exists."""
user = create_mock_user() user = create_mock_user()
session_user.set(user) session_user.set(user)
mock_cache = create_mock_cache_engine([]) mock_cache = create_mock_cache_engine([])
with patch( cache_module = importlib.import_module(
"cognee.infrastructure.databases.cache.get_cache_engine.get_cache_engine", "cognee.infrastructure.databases.cache.get_cache_engine"
return_value=mock_cache, )
):
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
from cognee.modules.retrieval.utils.session_cache import get_conversation_history from cognee.modules.retrieval.utils.session_cache import get_conversation_history
result = await get_conversation_history(session_id="test_session") result = await get_conversation_history(session_id="test_session")
assert result == "" assert result == ""
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_get_conversation_history_formats_history_correctly(self): async def test_get_conversation_history_formats_history_correctly(self):
@ -60,14 +53,15 @@ class TestConversationHistoryUtils:
] ]
mock_cache = create_mock_cache_engine(mock_history) mock_cache = create_mock_cache_engine(mock_history)
with patch( # ✅ Import the real module to patch safely
"cognee.infrastructure.databases.cache.get_cache_engine.get_cache_engine", cache_module = importlib.import_module(
return_value=mock_cache, "cognee.infrastructure.databases.cache.get_cache_engine"
): )
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
with patch( with patch(
"cognee.modules.retrieval.utils.session_cache.CacheConfig" "cognee.modules.retrieval.utils.session_cache.CacheConfig"
) as MockCacheConfig: ) as MockCacheConfig:
# Enable caching
mock_config = MagicMock() mock_config = MagicMock()
mock_config.caching = True mock_config.caching = True
MockCacheConfig.return_value = mock_config MockCacheConfig.return_value = mock_config
@ -92,14 +86,14 @@ class TestConversationHistoryUtils:
mock_cache = create_mock_cache_engine([]) mock_cache = create_mock_cache_engine([])
with patch( cache_module = importlib.import_module(
"cognee.infrastructure.databases.cache.get_cache_engine.get_cache_engine", "cognee.infrastructure.databases.cache.get_cache_engine"
return_value=mock_cache, )
):
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
with patch( with patch(
"cognee.modules.retrieval.utils.session_cache.CacheConfig" "cognee.modules.retrieval.utils.session_cache.CacheConfig"
) as MockCacheConfig: ) as MockCacheConfig:
# Enable caching
mock_config = MagicMock() mock_config = MagicMock()
mock_config.caching = True mock_config.caching = True
MockCacheConfig.return_value = mock_config MockCacheConfig.return_value = mock_config
@ -132,14 +126,14 @@ class TestConversationHistoryUtils:
mock_cache = create_mock_cache_engine([]) mock_cache = create_mock_cache_engine([])
with patch( cache_module = importlib.import_module(
"cognee.infrastructure.databases.cache.get_cache_engine.get_cache_engine", "cognee.infrastructure.databases.cache.get_cache_engine"
return_value=mock_cache, )
):
with patch.object(cache_module, "get_cache_engine", return_value=mock_cache):
with patch( with patch(
"cognee.modules.retrieval.utils.session_cache.CacheConfig" "cognee.modules.retrieval.utils.session_cache.CacheConfig"
) as MockCacheConfig: ) as MockCacheConfig:
# Enable caching
mock_config = MagicMock() mock_config = MagicMock()
mock_config.caching = True mock_config.caching = True
MockCacheConfig.return_value = mock_config MockCacheConfig.return_value = mock_config