my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import json
#2	import sqlite3
#3	from types import SimpleNamespace
#4
#5	from hermes_memory_provider.cli import mnemosyne_command
#6	from mnemosyne.core.importers.base import ImporterResult
#7	from mnemosyne.core.memory import Mnemosyne as RealMnemosyne
#8
#9
#10	def _sample_items():
#11	return [
#12	{
#13	"id": "hs-world-1",
#14	"text": "Hermes CLI Hindsight import must preserve timestamps.",
#15	"fact_type": "world",
#16	"mentioned_at": "2026-04-29T01:36:00+00:00",
#17	"date": "2026-04-29",
#18	"proof_count": 2,
#19	"tags": ["session:cli-import"],
#20	}
#21	]
#22
#23
#24	def _args(**overrides):
#25	defaults = {
#26	"mnemosyne_cmd": "import",
#27	"list_providers": False,
#28	"generate_script": False,
#29	"agentic": False,
#30	"from_provider": None,
#31	"output_script": None,
#32	"input": None,
#33	"file": None,
#34	"force": False,
#35	"api_key": None,
#36	"user_id": None,
#37	"agent_id": None,
#38	"base_url": None,
#39	"bank": None,
#40	"dry_run": False,
#41	"session_id": None,
#42	"channel_id": None,
#43	}
#44	defaults.update(overrides)
#45	return SimpleNamespace(**defaults)
#46
#47
#48	def _patch_cli_memory(monkeypatch, db_path):
#49	class DummyBeamMemory:
#50	def __init__(self, args, *kwargs):
#51	pass
#52
#53	def make_memory(session_id="default", channel_id=None, **kwargs):
#54	return RealMnemosyne(session_id=session_id, channel_id=channel_id, db_path=db_path)
#55
#56	# mnemosyne_command imports these symbols inside the function body, so
#57	# patching their definition modules before invocation keeps tests isolated.
#58	monkeypatch.setattr("mnemosyne.core.beam.BeamMemory", DummyBeamMemory)
#59	monkeypatch.setattr("mnemosyne.core.memory.Mnemosyne", make_memory)
#60
#61
#62	def _write_export(tmp_path, items=None):
#63	export = tmp_path / "hindsight-export.json"
#64	export.write_text(json.dumps({"items": items or _sample_items()}), encoding="utf-8")
#65	return export
#66
#67
#68	def _db_counts_and_row(db_path):
#69	with sqlite3.connect(db_path) as conn:
#70	conn.row_factory = sqlite3.Row
#71	try:
#72	row = conn.execute(
#73	"SELECT timestamp, metadata_json FROM episodic_memory ORDER BY timestamp LIMIT 1"
#74	).fetchone()
#75	working_count = conn.execute("SELECT COUNT(*) FROM working_memory").fetchone()[0]
#76	episodic_count = conn.execute("SELECT COUNT(*) FROM episodic_memory").fetchone()[0]
#77	except sqlite3.OperationalError as exc:
#78	if "no such table" not in str(exc):
#79	raise
#80	return 0, 0, None
#81	return working_count, episodic_count, row
#82
#83
#84	def test_hermes_cli_hindsight_file_import_preserves_timestamp(tmp_path, monkeypatch):
#85	db_path = tmp_path / "mnemosyne.db"
#86	_patch_cli_memory(monkeypatch, db_path)
#87	export = _write_export(tmp_path)
#88
#89	rc = mnemosyne_command(
#90	_args(from_provider="hindsight", file=str(export), bank="hermes")
#91	)
#92
#93	assert rc == 0
#94	working_count, _, row = _db_counts_and_row(db_path)
#95	assert row["timestamp"] == "2026-04-29T01:36:00+00:00"
#96	assert working_count == 0
#97
#98
#99	def test_hermes_cli_hindsight_accepts_input_alias(tmp_path, monkeypatch):
#100	db_path = tmp_path / "mnemosyne.db"
#101	_patch_cli_memory(monkeypatch, db_path)
#102	export = _write_export(tmp_path)
#103
#104	rc = mnemosyne_command(
#105	_args(from_provider="hindsight", input=str(export), bank="hermes")
#106	)
#107
#108	assert rc == 0
#109	working_count, _, row = _db_counts_and_row(db_path)
#110	assert row["timestamp"] == "2026-04-29T01:36:00+00:00"
#111	assert working_count == 0
#112
#113
#114	def test_hermes_cli_hindsight_base_url_does_not_require_api_key(tmp_path, monkeypatch):
#115	db_path = tmp_path / "mnemosyne.db"
#116	_patch_cli_memory(monkeypatch, db_path)
#117	captured = {}
#118
#119	def fake_import_from_provider(provider, mem, **kwargs):
#120	captured["provider"] = provider
#121	captured["kwargs"] = kwargs
#122	return ImporterResult(provider="hindsight", total=1, imported=1)
#123
#124	monkeypatch.setattr(
#125	"mnemosyne.core.importers.import_from_provider", fake_import_from_provider
#126	)
#127
#128	rc = mnemosyne_command(
#129	_args(
#130	from_provider="hindsight",
#131	base_url="http://127.0.0.1:8888",
#132	bank="personal",
#133	)
#134	)
#135
#136	assert rc == 0
#137	assert captured["provider"] == "hindsight"
#138	assert captured["kwargs"]["base_url"] == "http://127.0.0.1:8888"
#139	assert captured["kwargs"]["file_path"] is None
#140	assert captured["kwargs"]["bank"] == "personal"
#141	assert "api_key" not in captured["kwargs"]
#142
#143
#144	def test_hermes_cli_hindsight_forwards_non_default_bank(tmp_path, monkeypatch):
#145	db_path = tmp_path / "mnemosyne.db"
#146	_patch_cli_memory(monkeypatch, db_path)
#147	export = _write_export(tmp_path)
#148
#149	rc = mnemosyne_command(
#150	_args(from_provider="hindsight", file=str(export), bank="personal")
#151	)
#152
#153	assert rc == 0
#154	_, _, row = _db_counts_and_row(db_path)
#155	metadata = json.loads(row["metadata_json"])
#156	assert "hindsight_bank" in metadata
#157	assert metadata["hindsight_bank"] == "personal"
#158
#159
#160	def test_hermes_cli_hindsight_dry_run_writes_no_rows(tmp_path, monkeypatch):
#161	db_path = tmp_path / "mnemosyne.db"
#162	_patch_cli_memory(monkeypatch, db_path)
#163	export = _write_export(tmp_path)
#164
#165	rc = mnemosyne_command(
#166	_args(from_provider="hindsight", file=str(export), dry_run=True)
#167	)
#168
#169	assert rc == 0
#170	working_count, episodic_count, row = _db_counts_and_row(db_path)
#171	assert working_count == 0
#172	assert episodic_count == 0
#173	assert row is None
#174
#175
#176	def test_hermes_cli_hindsight_requires_file_or_base_url(tmp_path, monkeypatch, capsys):
#177	db_path = tmp_path / "mnemosyne.db"
#178	_patch_cli_memory(monkeypatch, db_path)
#179
#180	rc = mnemosyne_command(_args(from_provider="hindsight"))
#181
#182	assert rc == 1
#183	output = capsys.readouterr().out
#184	assert "Hindsight import requires --file/--input or --base-url" in output
#185
#186
#187	def test_hermes_cli_other_provider_still_requires_api_key(tmp_path, monkeypatch, capsys):
#188	db_path = tmp_path / "mnemosyne.db"
#189	_patch_cli_memory(monkeypatch, db_path)
#190	monkeypatch.delenv("MEM0_API_KEY", raising=False)
#191
#192	rc = mnemosyne_command(_args(from_provider="mem0"))
#193
#194	assert rc == 1
#195	output = capsys.readouterr().out
#196	assert "--api-key required for mem0 import" in output
#197

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public