my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	import json
#2	import sqlite3
#3
#4	from mnemosyne.core.importers import HindsightImporter, import_from_provider
#5	from mnemosyne.core.memory import Mnemosyne
#6
#7
#8	def _sample_items():
#9	return [
#10	{
#11	"id": "hs-world-1",
#12	"text": "Phin prefers full subject names instead of subject codes.",
#13	"fact_type": "world",
#14	"mentioned_at": "2026-04-29T01:36:00+00:00",
#15	"date": "2026-04-29",
#16	"proof_count": 2,
#17	"tags": ["session:school-preferences"],
#18	"entities": ["Phin"],
#19	"context": "User preference",
#20	},
#21	{
#22	"id": "hs-exp-1",
#23	"text": "Hindsight to Mnemosyne migration must preserve timestamps.",
#24	"fact_type": "experience",
#25	"mentioned_at": "2026-05-07T00:57:24.052845+00:00",
#26	"chunk_id": "chunk-abc",
#27	"proof_count": 1,
#28	},
#29	]
#30
#31
#32	def test_hindsight_importer_preserves_timestamps_and_uses_episodic_memory(tmp_path):
#33	export = tmp_path / "hindsight-export.json"
#34	export.write_text(json.dumps({"items": _sample_items()}), encoding="utf-8")
#35
#36	db_path = tmp_path / "mnemosyne.db"
#37	mem = Mnemosyne(session_id="default", db_path=db_path)
#38	result = HindsightImporter(file_path=str(export), bank="hermes").run(mem)
#39
#40	assert result.failed == 0
#41	assert result.imported == 2
#42	assert result.skipped == 0
#43
#44	conn = sqlite3.connect(db_path)
#45	conn.row_factory = sqlite3.Row
#46	rows = conn.execute(
#47	"SELECT id, content, source, timestamp, session_id, metadata_json, veracity, scope, channel_id "
#48	"FROM episodic_memory ORDER BY timestamp"
#49	).fetchall()
#50	assert len(rows) == 2
#51	assert conn.execute("SELECT COUNT(*) FROM working_memory").fetchone()[0] == 0
#52
#53	first = rows[0]
#54	assert first["content"] == "Phin prefers full subject names instead of subject codes."
#55	assert first["source"] == "hindsight:world"
#56	assert first["timestamp"] == "2026-04-29T01:36:00+00:00"
#57	assert first["session_id"] == "session_school-preferences"
#58	assert first["veracity"] == "imported"
#59	assert first["scope"] == "global"
#60	assert first["channel_id"] == "hindsight"
#61	metadata = json.loads(first["metadata_json"])
#62	assert metadata["migration_source"] == "hindsight"
#63	assert metadata["hindsight_bank"] == "hermes"
#64	assert metadata["hindsight_id"] == "hs-world-1"
#65	assert metadata["hindsight_fact_type"] == "world"
#66
#67	fts_hits = conn.execute(
#68	"SELECT COUNT(*) FROM fts_episodes WHERE fts_episodes MATCH ?",
#69	("timestamps",),
#70	).fetchone()[0]
#71	assert fts_hits == 1
#72
#73
#74	def test_hindsight_importer_skips_duplicates_with_stable_ids(tmp_path):
#75	export = tmp_path / "hindsight-export.json"
#76	export.write_text(json.dumps(_sample_items()), encoding="utf-8")
#77
#78	db_path = tmp_path / "mnemosyne.db"
#79	mem = Mnemosyne(session_id="default", db_path=db_path)
#80	importer = HindsightImporter(file_path=str(export), bank="hermes")
#81
#82	first = importer.run(mem)
#83	second = importer.run(mem)
#84
#85	assert first.imported == 2
#86	assert second.imported == 0
#87	assert second.skipped == 2
#88
#89	conn = sqlite3.connect(db_path)
#90	assert conn.execute("SELECT COUNT(*) FROM episodic_memory").fetchone()[0] == 2
#91
#92
#93	def test_hindsight_provider_registry_import(tmp_path):
#94	export = tmp_path / "hindsight-export.json"
#95	export.write_text(json.dumps({"items": _sample_items()[:1]}), encoding="utf-8")
#96
#97	db_path = tmp_path / "mnemosyne.db"
#98	mem = Mnemosyne(session_id="default", db_path=db_path)
#99	result = import_from_provider("hindsight", mem, file_path=str(export), bank="hermes")
#100
#101	assert result.provider == "hindsight"
#102	assert result.imported == 1
#103

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public