import logging from datetime import datetime, timezone from unittest.mock import MagicMock import pytest from mem0.memory.main import AsyncMemory, Memory, _normalize_iso_timestamp_to_utc def _setup_mocks(mocker): """Helper to setup common mocks for both sync and async fixtures""" mock_embedder = mocker.MagicMock() mock_embedder.return_value.embed.return_value = [0.1, 0.2, 0.3] mocker.patch("mem0.utils.factory.EmbedderFactory.create", mock_embedder) mock_vector_store = mocker.MagicMock() mock_vector_store.return_value.search.return_value = [] mocker.patch( "mem0.utils.factory.VectorStoreFactory.create", side_effect=[mock_vector_store.return_value, mocker.MagicMock()] ) mock_llm = mocker.MagicMock() mocker.patch("mem0.utils.factory.LlmFactory.create", mock_llm) mocker.patch("mem0.memory.storage.SQLiteManager", mocker.MagicMock()) return mock_llm, mock_vector_store class TestAddToVectorStoreErrors: @pytest.fixture def mock_memory(self, mocker): """Fixture that returns a Memory instance with mocker-based mocks""" mock_llm, _ = _setup_mocks(mocker) memory = Memory() memory.config = mocker.MagicMock() memory.config.custom_fact_extraction_prompt = None memory.config.custom_update_memory_prompt = None memory.api_version = "v1.1" return memory def test_empty_llm_response_fact_extraction(self, mocker, mock_memory, caplog): """Test empty response from LLM during fact extraction""" # Setup mock_memory.llm.generate_response.return_value = "invalid json" # This will trigger a JSON decode error mock_capture_event = mocker.MagicMock() mocker.patch("mem0.memory.main.capture_event", mock_capture_event) # Execute with caplog.at_level(logging.ERROR): result = mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True ) # Verify assert mock_memory.llm.generate_response.call_count == 1 assert result == [] # Should return empty list when no memories processed # Check for error message in any of the log records assert any("Error in new_retrieved_facts" in record.msg for record in caplog.records), "Expected error message not found in logs" assert mock_capture_event.call_count == 1 def test_empty_llm_response_memory_actions(self, mock_memory, caplog): """Test empty response from LLM during memory actions""" # Setup # First call returns valid JSON, second call returns empty string mock_memory.llm.generate_response.side_effect = ['{"facts": ["test fact"]}', ""] # Execute with caplog.at_level(logging.WARNING): result = mock_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, filters={}, infer=True ) # Verify assert mock_memory.llm.generate_response.call_count == 2 assert result == [] # Should return empty list when no memories processed assert "Empty response from LLM, no memories to extract" in caplog.text @pytest.mark.asyncio class TestAsyncAddToVectorStoreErrors: @pytest.fixture def mock_async_memory(self, mocker): """Fixture for AsyncMemory with mocker-based mocks""" mock_llm, _ = _setup_mocks(mocker) memory = AsyncMemory() memory.config = mocker.MagicMock() memory.config.custom_fact_extraction_prompt = None memory.config.custom_update_memory_prompt = None memory.api_version = "v1.1" return memory @pytest.mark.asyncio async def test_async_empty_llm_response_fact_extraction(self, mock_async_memory, caplog, mocker): """Test empty response in AsyncMemory._add_to_vector_store""" mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock()) mock_async_memory.llm.generate_response.return_value = "invalid json" # This will trigger a JSON decode error mock_capture_event = mocker.MagicMock() mocker.patch("mem0.memory.main.capture_event", mock_capture_event) with caplog.at_level(logging.ERROR): result = await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True ) assert mock_async_memory.llm.generate_response.call_count == 1 assert result == [] # Check for error message in any of the log records assert any("Error in new_retrieved_facts" in record.msg for record in caplog.records), "Expected error message not found in logs" assert mock_capture_event.call_count == 1 @pytest.mark.asyncio async def test_async_empty_llm_response_memory_actions(self, mock_async_memory, caplog, mocker): """Test empty response in AsyncMemory._add_to_vector_store""" mocker.patch("mem0.utils.factory.EmbedderFactory.create", return_value=MagicMock()) mock_async_memory.llm.generate_response.side_effect = ['{"facts": ["test fact"]}', ""] mock_capture_event = mocker.MagicMock() mocker.patch("mem0.memory.main.capture_event", mock_capture_event) with caplog.at_level(logging.WARNING): result = await mock_async_memory._add_to_vector_store( messages=[{"role": "user", "content": "test"}], metadata={}, effective_filters={}, infer=True ) assert result == [] assert "Empty response from LLM, no memories to extract" in caplog.text assert mock_capture_event.call_count == 1 def _build_memory_instance(mocker, memory_cls): _setup_mocks(mocker) mocker.patch("mem0.memory.main.SQLiteManager", mocker.MagicMock()) mocker.patch("mem0.memory.main.MEM0_TELEMETRY", False) memory = memory_cls() memory.config = mocker.MagicMock() memory.config.custom_fact_extraction_prompt = None memory.config.custom_update_memory_prompt = None memory.api_version = "v1.1" memory.vector_store = mocker.MagicMock() memory.db = mocker.MagicMock() return memory def _assert_utc_timestamp(timestamp: str): parsed = datetime.fromisoformat(timestamp) assert parsed.tzinfo == timezone.utc assert parsed.utcoffset().total_seconds() == 0 def test_create_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, Memory) memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] _assert_utc_timestamp(payload["created_at"]) def test_update_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, Memory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old memory", "created_at": "2026-03-17T17:00:00-07:00"} ) memory._update_memory("memory-id", "new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.update.call_args.kwargs["payload"] assert payload["created_at"] == "2026-03-18T00:00:00+00:00" _assert_utc_timestamp(payload["updated_at"]) @pytest.mark.asyncio async def test_async_create_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, AsyncMemory) await memory._create_memory("new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] _assert_utc_timestamp(payload["created_at"]) @pytest.mark.asyncio async def test_async_update_memory_uses_utc_timestamps(mocker): memory = _build_memory_instance(mocker, AsyncMemory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old memory", "created_at": "2026-03-17T17:00:00-07:00"} ) await memory._update_memory("memory-id", "new memory", {"new memory": [0.1, 0.2, 0.3]}, metadata={}) payload = memory.vector_store.update.call_args.kwargs["payload"] assert payload["created_at"] == "2026-03-18T00:00:00+00:00" _assert_utc_timestamp(payload["updated_at"]) class TestMetadataNotMutated: """Tests that metadata dicts passed to memory methods are not mutated in-place (issue #2648).""" def test_create_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) original_metadata = {"user_id": "test_user", "category": "sports"} metadata_copy = original_metadata.copy() memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"_create_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_create_memory_stores_correct_payload(self, mocker): memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "category": "sports"} memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["data"] == "test data" assert payload["user_id"] == "test_user" assert payload["category"] == "sports" assert "hash" in payload assert "created_at" in payload def test_create_memory_with_none_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=None) payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["data"] == "test data" assert "hash" in payload def test_create_memory_shared_metadata_across_calls(self, mocker): """Verify that sharing a metadata dict between multiple _create_memory calls is safe.""" memory = _build_memory_instance(mocker, Memory) shared_metadata = {"user_id": "test_user"} memory._create_memory("first memory", {"first memory": [0.1, 0.2, 0.3]}, metadata=shared_metadata) memory._create_memory("second memory", {"second memory": [0.4, 0.5, 0.6]}, metadata=shared_metadata) assert shared_metadata == {"user_id": "test_user"}, "shared metadata was mutated across calls" # Verify each call got the correct data first_payload = memory.vector_store.insert.call_args_list[0].kwargs["payloads"][0] second_payload = memory.vector_store.insert.call_args_list[1].kwargs["payloads"][0] assert first_payload["data"] == "first memory" assert second_payload["data"] == "second memory" def test_create_memory_preserves_role_and_actor_id_in_history(self, mocker): """Verify that role and actor_id from metadata flow through to add_history after deepcopy.""" memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "role": "assistant", "actor_id": "bot-1"} memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) # Verify the payload stored in vector store has all fields payload = memory.vector_store.insert.call_args.kwargs["payloads"][0] assert payload["role"] == "assistant" assert payload["actor_id"] == "bot-1" assert payload["user_id"] == "test_user" assert payload["data"] == "test data" # Verify add_history received the correct role and actor_id history_call = memory.db.add_history.call_args assert history_call.kwargs["role"] == "assistant" assert history_call.kwargs["actor_id"] == "bot-1" # And the original metadata is still untouched assert metadata == {"user_id": "test_user", "role": "assistant", "actor_id": "bot-1"} def test_create_memory_with_nested_metadata_not_mutated(self, mocker): """Verify deepcopy protects nested structures in metadata.""" memory = _build_memory_instance(mocker, Memory) metadata = {"user_id": "test_user", "tags": ["important", "urgent"], "config": {"key": "val"}} import copy metadata_snapshot = copy.deepcopy(metadata) memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=metadata) assert metadata == metadata_snapshot, "Nested metadata structures were mutated" def test_update_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, Memory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old data", "user_id": "test_user", "created_at": "2026-01-01T00:00:00+00:00"} ) original_metadata = {"category": "updated"} metadata_copy = original_metadata.copy() memory._update_memory("mem-id", "new data", {"new data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"_update_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_add_to_vector_store_no_infer_does_not_mutate_metadata(self, mocker): """Verify _add_to_vector_store with infer=False doesn't leak metadata between messages.""" memory = _build_memory_instance(mocker, Memory) memory.embedding_model.embed.return_value = [0.1, 0.2, 0.3] original_metadata = {"user_id": "test_user"} metadata_copy = original_metadata.copy() messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there", "name": "bot-1"}, ] result = memory._add_to_vector_store(messages, original_metadata, filters={}, infer=False) # Metadata should not be mutated assert original_metadata == metadata_copy, ( f"_add_to_vector_store mutated the caller's metadata: {original_metadata}" ) # Should have created 2 memories assert len(result) == 2 assert result[0]["role"] == "user" assert result[1]["role"] == "assistant" assert result[1]["actor_id"] == "bot-1" # Verify each insert got distinct payloads with correct roles insert_calls = memory.vector_store.insert.call_args_list first_payload = insert_calls[0].kwargs["payloads"][0] second_payload = insert_calls[1].kwargs["payloads"][0] assert first_payload["role"] == "user" assert "actor_id" not in first_payload # user message has no name assert second_payload["role"] == "assistant" assert second_payload["actor_id"] == "bot-1" @pytest.mark.asyncio async def test_async_create_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) original_metadata = {"user_id": "test_user", "category": "sports"} metadata_copy = original_metadata.copy() await memory._create_memory("test data", {"test data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"async _create_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) @pytest.mark.asyncio async def test_async_create_memory_shared_metadata_across_calls(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) shared_metadata = {"user_id": "test_user"} await memory._create_memory("first memory", {"first memory": [0.1, 0.2, 0.3]}, metadata=shared_metadata) await memory._create_memory("second memory", {"second memory": [0.4, 0.5, 0.6]}, metadata=shared_metadata) assert shared_metadata == {"user_id": "test_user"}, "shared metadata was mutated across async calls" @pytest.mark.asyncio async def test_async_add_to_vector_store_no_infer_does_not_mutate_metadata(self, mocker): """Verify async _add_to_vector_store with infer=False doesn't leak metadata between messages.""" memory = _build_memory_instance(mocker, AsyncMemory) memory.embedding_model.embed.return_value = [0.1, 0.2, 0.3] original_metadata = {"user_id": "test_user"} metadata_copy = original_metadata.copy() messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there", "name": "bot-1"}, ] result = await memory._add_to_vector_store(messages, original_metadata, effective_filters={}, infer=False) assert original_metadata == metadata_copy, ( f"async _add_to_vector_store mutated the caller's metadata: {original_metadata}" ) assert len(result) == 2 assert result[0]["role"] == "user" assert result[1]["role"] == "assistant" @pytest.mark.asyncio async def test_async_update_memory_does_not_mutate_metadata(self, mocker): memory = _build_memory_instance(mocker, AsyncMemory) memory.vector_store.get.return_value = MagicMock( payload={"data": "old data", "user_id": "test_user", "created_at": "2026-01-01T00:00:00+00:00"} ) original_metadata = {"category": "updated"} metadata_copy = original_metadata.copy() await memory._update_memory("mem-id", "new data", {"new data": [0.1, 0.2, 0.3]}, metadata=original_metadata) assert original_metadata == metadata_copy, ( f"async _update_memory mutated the caller's metadata dict: {original_metadata} != {metadata_copy}" ) def test_normalize_iso_timestamp_to_utc_preserves_naive_values(): assert _normalize_iso_timestamp_to_utc("2026-03-18T00:00:00") == "2026-03-18T00:00:00" def test_normalize_iso_timestamp_to_utc_converts_pacific(): result = _normalize_iso_timestamp_to_utc("2026-03-17T17:00:00-07:00") assert result == "2026-03-18T00:00:00+00:00" def test_normalize_iso_timestamp_to_utc_handles_none(): assert _normalize_iso_timestamp_to_utc(None) is None def test_normalize_iso_timestamp_to_utc_handles_empty(): assert _normalize_iso_timestamp_to_utc("") == ""