From ca593bba152a209f9b34ac3f554814f8e42d7ca2 Mon Sep 17 00:00:00 2001 From: hajdul88 <52442977+hajdul88@users.noreply.github.com> Date: Thu, 11 Dec 2025 13:13:01 +0100 Subject: [PATCH] adds full coverage to conv history save --- .../retrieval/conversation_history_test.py | 350 ++++++++++++++++++ 1 file changed, 350 insertions(+) diff --git a/cognee/tests/unit/modules/retrieval/conversation_history_test.py b/cognee/tests/unit/modules/retrieval/conversation_history_test.py index d464a99d8..1ca5cccfb 100644 --- a/cognee/tests/unit/modules/retrieval/conversation_history_test.py +++ b/cognee/tests/unit/modules/retrieval/conversation_history_test.py @@ -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