my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import logging
#2	from unittest.mock import MagicMock
#3
#4	import pytest
#5
#6	from mem0.memory.main import AsyncMemory, Memory
#7
#8
#9	def _setup_mocks(mocker):
#10	"""Helper to setup common mocks for both sync and async fixtures"""
#11	mock_embedder = mocker.MagicMock()
#12	mock_embedder.return_value.embed.return_value = [0.1, 0.2, 0.3]
#13	mocker.patch("mem0.utils.factory.EmbedderFactory.create", mock_embedder)
#14
#15	mock_vector_store = mocker.MagicMock()
#16	mock_vector_store.return_value.search.return_value = []
#17	mocker.patch(
#18	"mem0.utils.factory.VectorStoreFactory.create", side_effect=[mock_vector_store.return_value, mocker.MagicMock()]
#19	)
#20
#21	mock_llm = mocker.MagicMock()
#22	mocker.patch("mem0.utils.factory.LlmFactory.create", mock_llm)
#23
#24	mocker.patch("mem0.memory.storage.SQLiteManager", mocker.MagicMock())
#25
#26	return mock_llm, mock_vector_store
#27
#28
#29	class TestAddToVectorStoreErrors:
#30	@pytest.fixture
#31	def mock_memory(self, mocker):
#32	"""Fixture that returns a Memory instance with mocker-based mocks"""
#33	mock_llm, _ = _setup_mocks(mocker)
#34
#35	memory = Memory()
#36	memory.config = mocker.MagicMock()
#37	memory.config.custom_fact_extraction_prompt = None
#38	memory.config.custom_update_memory_prompt = None
#39	memory.api_version = "v1.1"
#40
#41	return memory
#42
#43	def test_empty_llm_response_fact_extraction(self, mocker, mock_memory, caplog):
#44	"""Test empty response from LLM during fact extraction"""
#45	# Setup
#46	mock_memory.llm.generate_response.return_value = "invalid json" # This will trigger a JSON decode error
#47	mock_capture_event = mocker.MagicMock()
#48	mocker.patch("mem0.memory.main.capture_event", mock_capture_event)
#49
#50	# Execute
#51	with caplog.at_level(logging.ERROR):
#52	result = mock_memory._add_to_vector_store(
#53	messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True
#54	)
#55
#56	# Verify
#57	assert mock_memory.llm.generate_response.call_count == 1
#58	assert result == [] # Should return empty list when no memories processed
#59	# Check for error message in any of the log records
#60	assert any("Error in new_retrieved_facts" in record.msg for record in caplog.records), "Expected error message not found in logs"
#61	assert mock_capture_event.call_count == 1
#62
#63	def test_empty_llm_response_memory_actions(self, mock_memory, caplog):
#64	"""Test empty response from LLM during memory actions"""
#65	# Setup
#66	# First call returns valid JSON, second call returns empty string
#67	mock_memory.llm.generate_response.side_effect = ['{"facts": ["test fact"]}', ""]
#68
#69	# Execute
#70	with caplog.at_level(logging.WARNING):
#71	result = mock_memory._add_to_vector_store(
#72	messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True
#73	)
#74
#75	# Verify
#76	assert mock_memory.llm.generate_response.call_count == 2
#77	assert result == [] # Should return empty list when no memories processed
#78	assert "Empty response from LLM, no memories to extract" in caplog.text
#79
#80
#81	@pytest.mark.asyncio
#82	class TestAsyncAddToVectorStoreErrors:
#83	@pytest.fixture
#84	def mock_async_memory(self, mocker):
#85	"""Fixture for AsyncMemory with mocker-based mocks"""
#86	mock_llm, _ = _setup_mocks(mocker)
#87
#88	memory = AsyncMemory()
#89	memory.config = mocker.MagicMock()
#90	memory.config.custom_fact_extraction_prompt = None
#91	memory.config.custom_update_memory_prompt = None
#92	memory.api_version = "v1.1"
#93
#94	return memory
#95
#96	@pytest.mark.asyncio
#97	async def test_async_empty_llm_response_fact_extraction(self, mock_async_memory, caplog, mocker):
#98	"""Test empty response in AsyncMemory._add_to_vector_store"""
#99	mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock())
#100	mock_async_memory.llm.generate_response.return_value = "invalid json" # This will trigger a JSON decode error
#101	mock_capture_event = mocker.MagicMock()
#102	mocker.patch("mem0.memory.main.capture_event", mock_capture_event)
#103
#104	with caplog.at_level(logging.ERROR):
#105	result = await mock_async_memory._add_to_vector_store(
#106	messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True
#107	)
#108	assert mock_async_memory.llm.generate_response.call_count == 1
#109	assert result == []
#110	# Check for error message in any of the log records
#111	assert any("Error in new_retrieved_facts" in record.msg for record in caplog.records), "Expected error message not found in logs"
#112	assert mock_capture_event.call_count == 1
#113
#114	@pytest.mark.asyncio
#115	async def test_async_empty_llm_response_memory_actions(self, mock_async_memory, caplog, mocker):
#116	"""Test empty response in AsyncMemory._add_to_vector_store"""
#117	mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock())
#118	mock_async_memory.llm.generate_response.side_effect = ['{"facts": ["test fact"]}', ""]
#119	mock_capture_event = mocker.MagicMock()
#120	mocker.patch("mem0.memory.main.capture_event", mock_capture_event)
#121
#122	with caplog.at_level(logging.WARNING):
#123	result = await mock_async_memory._add_to_vector_store(
#124	messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True
#125	)
#126
#127	assert result == []
#128	assert "Empty response from LLM, no memories to extract" in caplog.text
#129	assert mock_capture_event.call_count == 1
#130

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public