my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Tests for the BaseImporter and ImporterResult classes."""
#2
#3	import pytest
#4	import json
#5	from pathlib import Path
#6	from mnemosyne.core.importers.base import (
#7	BaseImporter, ImporterResult, import_from_file
#8	)
#9
#10
#11	class TestImporterResult:
#12	def test_default_values(self):
#13	result = ImporterResult(provider="test")
#14	assert result.provider == "test"
#15	assert result.total == 0
#16	assert result.imported == 0
#17	assert result.skipped == 0
#18	assert result.failed == 0
#19	assert result.errors == []
#20	assert result.memory_ids == []
#21
#22	def test_to_dict_includes_counts(self):
#23	result = ImporterResult(
#24	provider="mem0",
#25	total=10,
#26	imported=8,
#27	failed=2,
#28	errors=["e1", "e2"],
#29	memory_ids=["id1", "id2"],
#30	)
#31	d = result.to_dict()
#32	assert d["provider"] == "mem0"
#33	assert d["total"] == 10
#34	assert d["imported"] == 8
#35	assert d["failed"] == 2
#36	assert "e1" in d["errors"]
#37
#38	def test_to_dict_caps_errors_and_ids(self):
#39	result = ImporterResult(
#40	provider="test",
#41	total=100,
#42	imported=100,
#43	errors=[f"err_{i}" for i in range(30)],
#44	memory_ids=[f"id_{i}" for i in range(100)],
#45	)
#46	d = result.to_dict()
#47	assert len(d["errors"]) == 20 # capped at 20
#48	assert len(d["memory_ids"]) == 50 # capped at 50
#49
#50	def test_to_json_serializable(self):
#51	result = ImporterResult(provider="test", total=5)
#52	j = result.to_json()
#53	parsed = json.loads(j)
#54	assert parsed["provider"] == "test"
#55
#56
#57	class TestBaseImporter:
#58	def test_abstract_methods(self):
#59	"""BaseImporter can't be instantiated directly (abstract)."""
#60	with pytest.raises(TypeError):
#61	BaseImporter()
#62
#63	def test_concrete_subclass(self):
#64	"""A minimal concrete subclass works."""
#65
#66	class TestImporter(BaseImporter):
#67	provider_name = "test"
#68
#69	def extract(self):
#70	return [{"content": "hello"}]
#71
#72	def transform(self, raw_data):
#73	return [
#74	{
#75	"content": item.get("content", ""),
#76	"source": "test",
#77	"importance": 0.5,
#78	"metadata": {},
#79	"valid_until": None,
#80	"scope": "session",
#81	}
#82	for item in raw_data
#83	]
#84
#85	importer = TestImporter()
#86	assert importer.provider_name == "test"
#87
#88	def test_validate_empty(self):
#89	class EmptyImporter(BaseImporter):
#90	provider_name = "empty"
#91
#92	def extract(self):
#93	return []
#94
#95	def transform(self, raw_data):
#96	return []
#97
#98	importer = EmptyImporter()
#99	assert importer.validate([]) is False # empty should fail
#100
#101	def test_validate_non_list(self):
#102	class BadImporter(BaseImporter):
#103	provider_name = "bad"
#104
#105	def extract(self):
#106	return "not a list"
#107
#108	def transform(self, raw_data):
#109	return []
#110
#111	importer = BadImporter()
#112	assert importer.validate("not a list") is False
#113
#114	def test_content_hash_deterministic(self):
#115	h1 = BaseImporter._content_hash("hello")
#116	h2 = BaseImporter._content_hash("hello")
#117	assert h1 == h2
#118	assert len(h1) == 16
#119
#120
#121	class TestImportFromFile:
#122	def test_import_json_array(self, tmp_path):
#123	"""Import a simple JSON array of memories."""
#124	data = [
#125	{"content": "memory one", "importance": 0.8},
#126	{"content": "memory two", "source": "test"},
#127	]
#128	f = tmp_path / "export.json"
#129	f.write_text(json.dumps(data))
#130
#131	from mnemosyne.core.memory import Mnemosyne
#132	mem = Mnemosyne(session_id="test_import", db_path=tmp_path / "test.db")
#133
#134	result = import_from_file(str(f), mem)
#135	assert result.provider == "file"
#136	assert result.total == 2
#137	assert result.imported == 2
#138	assert result.failed == 0
#139
#140	def test_import_wrapped_response(self, tmp_path):
#141	"""Import Mem0-style wrapped response: {"results": [...]}."""
#142	data = {"results": [{"memory": "wrapped memory", "user_id": "alice"}]}
#143	f = tmp_path / "wrapped.json"
#144	f.write_text(json.dumps(data))
#145
#146	from mnemosyne.core.memory import Mnemosyne
#147	mem = Mnemosyne(session_id="test_import", db_path=tmp_path / "test.db")
#148
#149	result = import_from_file(str(f), mem)
#150	assert result.total == 1
#151	assert result.imported == 1
#152
#153	def test_dry_run(self, tmp_path):
#154	"""Dry run should validate but not write."""
#155	data = [{"content": "test dry run"}]
#156	f = tmp_path / "export.json"
#157	f.write_text(json.dumps(data))
#158
#159	from mnemosyne.core.memory import Mnemosyne
#160	mem = Mnemosyne(session_id="test_import", db_path=tmp_path / "test.db")
#161
#162	result = import_from_file(str(f), mem, dry_run=True)
#163	assert result.total == 1
#164	assert result.imported == 1 # dry-run shows what WOULD be imported
#165
#166	# Verify nothing was actually written
#167	wm = mem.beam.get_working_stats()
#168	assert wm["total"] == 0
#169

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public