my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Tests for the Mem0 importer."""
#2
#3	import json
#4	import pytest
#5	from pathlib import Path
#6	from unittest.mock import patch, MagicMock
#7
#8	from mnemosyne.core.importers.mem0 import Mem0Importer, import_from_mem0
#9	from mnemosyne.core.importers.base import ImporterResult
#10
#11
#12	# Sample Mem0 API response data
#13	MEM0_FIXTURE = [
#14	{
#15	"id": "mem_abc123",
#16	"memory": "Abdias prefers dark mode",
#17	"user_id": "abdias",
#18	"agent_id": None,
#19	"app_id": "fluxspeak",
#20	"run_id": "run_001",
#21	"hash": "a1b2c3d4",
#22	"metadata": {"source": "preferences"},
#23	"categories": ["preference"],
#24	"created_at": "2026-04-01T10:00:00Z",
#25	"updated_at": None,
#26	},
#27	{
#28	"id": "mem_def456",
#29	"memory": "CI pipeline fails with parallel tests",
#30	"user_id": None,
#31	"agent_id": "ci-bot",
#32	"app_id": "mnemosyne",
#33	"run_id": "run_002",
#34	"hash": "b2c3d4e5",
#35	"metadata": {"severity": "high"},
#36	"categories": ["bug"],
#37	"created_at": "2026-04-02T14:30:00Z",
#38	"updated_at": "2026-04-02T15:00:00Z",
#39	},
#40	]
#41
#42	MEM0_PAGINATED = {
#43	"count": 2,
#44	"next": None,
#45	"previous": None,
#46	"results": MEM0_FIXTURE,
#47	}
#48
#49
#50	class TestMem0Transform:
#51	"""Test transform() in isolation — no API needed."""
#52
#53	def test_basic_transform(self):
#54	importer = Mem0Importer(api_key="test_key")
#55	memories = importer.transform(MEM0_FIXTURE)
#56
#57	assert len(memories) == 2
#58	assert memories[0]["content"] == "Abdias prefers dark mode"
#59	assert memories[0]["source"] == "mem0_import"
#60	assert memories[0]["importance"] == 0.5
#61	assert memories[0]["_author_id"] == "mem0_user:abdias"
#62	assert memories[0]["_author_type"] == "human"
#63	assert memories[0]["_channel_id"] == "fluxspeak"
#64
#65	def test_agent_memory_maps_to_agent_type(self):
#66	importer = Mem0Importer(api_key="test_key")
#67	memories = importer.transform(MEM0_FIXTURE)
#68
#69	# Second memory has agent_id but no user_id
#70	assert memories[1]["_author_id"] == "mem0_agent:ci-bot"
#71	assert memories[1]["_author_type"] == "agent"
#72
#73	def test_metadata_preserved(self):
#74	importer = Mem0Importer(api_key="test_key")
#75	memories = importer.transform(MEM0_FIXTURE)
#76
#77	assert memories[0]["metadata"]["source"] == "preferences"
#78	assert memories[0]["metadata"]["_mem0_id"] == "mem_abc123"
#79	assert memories[0]["metadata"]["_mem0_categories"] == ["preference"]
#80
#81	def test_updated_at_preserved(self):
#82	importer = Mem0Importer(api_key="test_key")
#83	memories = importer.transform(MEM0_FIXTURE)
#84
#85	# First memory has no updated_at — should not add _updated_at
#86	assert "_updated_at" not in memories[0]["metadata"]
#87
#88	# Second memory has updated_at different from created_at
#89	assert "_updated_at" in memories[1]["metadata"]
#90	assert memories[1]["metadata"]["_updated_at"] == "2026-04-02T15:00:00Z"
#91
#92	def test_empty_content_skipped(self):
#93	"""Memories with empty content should be skipped."""
#94	data = [{"memory": "", "user_id": "test"}]
#95	importer = Mem0Importer(api_key="test_key")
#96	memories = importer.transform(data)
#97	assert len(memories) == 0
#98
#99	def test_metadata_importance_inference(self):
#100	"""If metadata has importance, use it."""
#101	data = [{
#102	"memory": "important fact",
#103	"user_id": "test",
#104	"metadata": {"importance": 0.95},
#105	}]
#106	importer = Mem0Importer(api_key="test_key")
#107	memories = importer.transform(data)
#108	assert memories[0]["importance"] == 0.95
#109	# importance should be popped from metadata
#110	assert "importance" not in memories[0]["metadata"]
#111
#112
#113	class TestMem0Run:
#114	"""Test run() with mocked Mnemosyne."""
#115
#116	def test_run_imports_memories(self, tmp_path):
#117	"""End-to-end run with mock extraction."""
#118	with patch.object(Mem0Importer, "extract", return_value=MEM0_FIXTURE):
#119	from mnemosyne.core.memory import Mnemosyne
#120	mem = Mnemosyne(session_id="test_mem0", db_path=tmp_path / "test.db")
#121
#122	importer = Mem0Importer(api_key="test_key")
#123	result = importer.run(mem)
#124
#125	assert result.provider == "mem0"
#126	assert result.total == 2
#127	assert result.imported == 2
#128	assert result.failed == 0
#129
#130	# Verify memories were actually stored
#131	wm = mem.beam.get_working_stats()
#132	assert wm["total"] >= 2
#133
#134	def test_run_dry_run_does_not_write(self, tmp_path):
#135	"""Dry run should not write any memories."""
#136	with patch.object(Mem0Importer, "extract", return_value=MEM0_FIXTURE):
#137	from mnemosyne.core.memory import Mnemosyne
#138	mem = Mnemosyne(session_id="test_mem0", db_path=tmp_path / "test.db")
#139
#140	importer = Mem0Importer(api_key="test_key")
#141	result = importer.run(mem, dry_run=True)
#142
#143	assert result.total == 2
#144	assert result.imported == 2 # dry-run reports what WOULD import
#145
#146	# Verify nothing was written
#147	wm = mem.beam.get_working_stats()
#148	assert wm["total"] == 0
#149
#150	def test_run_empty_extract(self, tmp_path):
#151	"""Empty extract should return early with error."""
#152	with patch.object(Mem0Importer, "extract", return_value=[]):
#153	from mnemosyne.core.memory import Mnemosyne
#154	mem = Mnemosyne(session_id="test_mem0", db_path=tmp_path / "test.db")
#155
#156	importer = Mem0Importer(api_key="test_key")
#157	result = importer.run(mem)
#158
#159	assert result.total == 0
#160	assert result.imported == 0
#161	assert len(result.errors) >= 1
#162
#163	def test_run_preserves_identity(self, tmp_path):
#164	"""Verify author_id and channel_id are stored on imported memories."""
#165	with patch.object(Mem0Importer, "extract", return_value=MEM0_FIXTURE):
#166	from mnemosyne.core.memory import Mnemosyne
#167	mem = Mnemosyne(session_id="test_mem0", db_path=tmp_path / "test.db")
#168
#169	importer = Mem0Importer(api_key="test_key")
#170	result = importer.run(mem)
#171
#172	# Query via recall to check identity was stored
#173	memories = mem.recall("dark mode", top_k=10)
#174	assert len(memories) >= 1
#175
#176	# At least one should have author_id
#177	author_ids = [m.get("author_id") for m in memories if m.get("author_id")]
#178	assert "mem0_user:abdias" in author_ids
#179
#180
#181	class TestMem0SDKMock:
#182	"""Test the SDK extraction path with mocked extract method."""
#183
#184	def test_extract_via_sdk_paginates(self):
#185	"""Verify pagination loop works."""
#186	page1 = {
#187	"results": [MEM0_FIXTURE[0]],
#188	"next": "?page=2&page_size=200",
#189	}
#190	page2 = {
#191	"results": [MEM0_FIXTURE[1]],
#192	"next": None,
#193	}
#194
#195	mock_client = MagicMock()
#196	mock_client.get_all.side_effect = [page1, page2]
#197
#198	# Patch the actual SDK import inside _extract_via_sdk
#199	import builtins
#200	real_import = builtins.__import__
#201
#202	def mock_import(name, args, *kwargs):
#203	if name == "mem0":
#204	mem0_mod = MagicMock()
#205	mem0_mod.MemoryClient = MagicMock(return_value=mock_client)
#206	return mem0_mod
#207	return real_import(name, args, *kwargs)
#208
#209	with patch("builtins.__import__", side_effect=mock_import):
#210	importer = Mem0Importer(api_key="test_key", user_id="*")
#211	data = importer._extract_via_sdk()
#212
#213	assert len(data) == 2
#214	assert mock_client.get_all.call_count == 2
#215
#216	def test_extract_via_sdk_with_filters(self):
#217	"""Verify filters are passed to get_all."""
#218	page = {
#219	"results": [MEM0_FIXTURE[0]],
#220	"next": None,
#221	}
#222
#223	mock_client = MagicMock()
#224	mock_client.get_all.return_value = page
#225
#226	import builtins
#227	real_import = builtins.__import__
#228
#229	def mock_import(name, args, *kwargs):
#230	if name == "mem0":
#231	mem0_mod = MagicMock()
#232	mem0_mod.MemoryClient = MagicMock(return_value=mock_client)
#233	return mem0_mod
#234	return real_import(name, args, *kwargs)
#235
#236	with patch("builtins.__import__", side_effect=mock_import):
#237	importer = Mem0Importer(
#238	api_key="test_key",
#239	user_id="abdias",
#240	agent_id="bot-1",
#241	)
#242	data = importer._extract_via_sdk()
#243
#244	assert len(data) == 1
#245	call_kwargs = mock_client.get_all.call_args[1]
#246	assert call_kwargs["filters"]["user_id"] == "abdias"
#247	assert call_kwargs["filters"]["agent_id"] == "bot-1"
#248
#249
#250	class TestConvenienceFunction:
#251	def test_import_from_mem0(self, tmp_path):
#252	"""The convenience function works."""
#253	with patch.object(Mem0Importer, "extract", return_value=MEM0_FIXTURE):
#254	from mnemosyne.core.memory import Mnemosyne
#255	mem = Mnemosyne(session_id="test_mem0", db_path=tmp_path / "test.db")
#256
#257	result = import_from_mem0(
#258	api_key="test_key",
#259	mnemosyne=mem,
#260	user_id="abdias",
#261	)
#262
#263	assert isinstance(result, ImporterResult)
#264	assert result.imported >= 1
#265

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public