my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Tests for Multi-Agent Identity Layer (v2.1)
#3
#4	Tests cover:
#5	- Schema migration adds identity columns
#6	- remember() stores identity fields
#7	- recall() filters by author_id, author_type, channel_id
#8	- Cross-session channel recall
#9	- get_stats() with identity filters
#10	- MCP per-connection instances
#11	- MCP env var fallback
#12	- Backward compatibility (no identity = unchanged behavior)
#13	"""
#14
#15	import pytest
#16	import os
#17	import tempfile
#18	from pathlib import Path
#19	from mnemosyne.core.beam import BeamMemory, init_beam
#20	from mnemosyne.core.memory import Mnemosyne
#21	from mnemosyne.mcp_tools import _create_instance
#22
#23
#24	@pytest.fixture
#25	def temp_db():
#26	"""Create a temporary database for testing."""
#27	tmpdir = Path(tempfile.mkdtemp())
#28	db_path = tmpdir / "test.db"
#29	# Override data dir to isolate tests
#30	old_data_dir = os.environ.get("MNEMOSYNE_DATA_DIR")
#31	os.environ["MNEMOSYNE_DATA_DIR"] = str(tmpdir)
#32	yield db_path
#33	if old_data_dir:
#34	os.environ["MNEMOSYNE_DATA_DIR"] = old_data_dir
#35	else:
#36	del os.environ["MNEMOSYNE_DATA_DIR"]
#37
#38
#39	class TestSchemaMigration:
#40	"""Schema migration adds identity columns."""
#41
#42	def test_working_memory_has_identity_columns(self, temp_db):
#43	bm = BeamMemory(session_id="test", db_path=temp_db)
#44	cols = [c[1] for c in bm.conn.execute("PRAGMA table_info(working_memory)").fetchall()]
#45	assert "author_id" in cols
#46	assert "author_type" in cols
#47	assert "channel_id" in cols
#48
#49	def test_episodic_memory_has_identity_columns(self, temp_db):
#50	bm = BeamMemory(session_id="test", db_path=temp_db)
#51	cols = [c[1] for c in bm.conn.execute("PRAGMA table_info(episodic_memory)").fetchall()]
#52	assert "author_id" in cols
#53	assert "author_type" in cols
#54	assert "channel_id" in cols
#55
#56	def test_identity_indexes_exist(self, temp_db):
#57	bm = BeamMemory(session_id="test", db_path=temp_db)
#58	idxs = [r[0] for r in bm.conn.execute("SELECT name FROM sqlite_master WHERE type='index'").fetchall()]
#59	assert "idx_wm_author" in idxs
#60	assert "idx_wm_channel" in idxs
#61	assert "idx_em_author" in idxs
#62	assert "idx_em_channel" in idxs
#63
#64
#65	class TestRememberIdentity:
#66	"""remember() auto-populates identity fields."""
#67
#68	def test_remember_stores_author(self, temp_db):
#69	mem = Mnemosyne(session_id="test", author_id="abdias", author_type="human",
#70	channel_id="fluxspeak-team", db_path=temp_db)
#71	mid = mem.remember("Dark mode preference", importance=0.9)
#72	row = mem.conn.execute(
#73	"SELECT author_id, author_type, channel_id FROM working_memory WHERE id = ?",
#74	(mid,)
#75	).fetchone()
#76	assert row["author_id"] == "abdias"
#77	assert row["author_type"] == "human"
#78	assert row["channel_id"] == "fluxspeak-team"
#79
#80	def test_remember_without_identity_is_null(self, temp_db):
#81	mem = Mnemosyne(session_id="test", db_path=temp_db)
#82	mid = mem.remember("Some memory")
#83	row = mem.conn.execute(
#84	"SELECT author_id, author_type, channel_id FROM working_memory WHERE id = ?",
#85	(mid,)
#86	).fetchone()
#87	assert row["author_id"] is None
#88	assert row["author_type"] is None
#89	# channel_id defaults to session_id
#90	assert row["channel_id"] == "test"
#91
#92	def test_channel_id_defaults_to_session(self, temp_db):
#93	mem = Mnemosyne(session_id="my-channel", db_path=temp_db)
#94	mid = mem.remember("Channel-scoped memory")
#95	row = mem.conn.execute(
#96	"SELECT channel_id FROM working_memory WHERE id = ?", (mid,)
#97	).fetchone()
#98	assert row["channel_id"] == "my-channel"
#99
#100	def test_different_authors_same_channel(self, temp_db):
#101	abdias = Mnemosyne(session_id="a1", author_id="abdias", author_type="human",
#102	channel_id="team", db_path=temp_db)
#103	sarah = Mnemosyne(session_id="a2", author_id="sarah", author_type="human",
#104	channel_id="team", db_path=temp_db)
#105	abdias.remember("Dark mode")
#106	sarah.remember("Launch Friday")
#107
#108	rows = abdias.conn.execute("SELECT author_id FROM working_memory ORDER BY timestamp").fetchall()
#109	authors = [r["author_id"] for r in rows]
#110	assert "abdias" in authors
#111	assert "sarah" in authors
#112
#113
#114	class TestRecallIdentity:
#115	"""recall() filters by identity."""
#116
#117	def test_recall_by_author(self, temp_db):
#118	mem = Mnemosyne(session_id="test", author_id="abdias", db_path=temp_db)
#119	mem.remember("Dark mode preference", importance=0.9)
#120
#121	results = mem.recall("dark", author_id="abdias")
#122	assert len(results) >= 1
#123	assert results[0]["author_id"] == "abdias"
#124
#125	def test_recall_by_author_no_match(self, temp_db):
#126	mem = Mnemosyne(session_id="test", author_id="abdias", db_path=temp_db)
#127	mem.remember("Dark mode")
#128
#129	results = mem.recall("dark", author_id="sarah")
#130	assert len(results) == 0
#131
#132	def test_recall_by_author_type(self, temp_db):
#133	mem = Mnemosyne(session_id="test", author_id="bot-ci", author_type="agent",
#134	db_path=temp_db)
#135	mem.remember("Deploy succeeded")
#136	mem2 = Mnemosyne(session_id="test2", author_id="abdias", author_type="human",
#137	db_path=temp_db)
#138	mem2.remember("Dark mode")
#139
#140	results = mem.recall("deploy", author_type="agent")
#141	assert len(results) >= 1
#142	results2 = mem.recall("dark", author_type="human", author_id="abdias")
#143	assert len(results2) >= 1
#144
#145	def test_cross_session_channel_recall(self, temp_db):
#146	"""Agents in different sessions but same channel can see each other's memories."""
#147	abdias = Mnemosyne(session_id="session-a", author_id="abdias",
#148	channel_id="fluxspeak-team", db_path=temp_db)
#149	sarah = Mnemosyne(session_id="session-b", author_id="sarah",
#150	channel_id="fluxspeak-team", db_path=temp_db)
#151
#152	abdias.remember("Dark mode is preferred")
#153	sarah.remember("Launch is Friday")
#154
#155	# Abdias recalls by channel — should see Sarah's memories too
#156	results = abdias.recall("Launch", channel_id="fluxspeak-team")
#157	assert len(results) >= 1
#158	assert results[0]["channel_id"] == "fluxspeak-team"
#159
#160	# Sarah recalls by channel — should see Abdias's memories
#161	results = sarah.recall("dark", channel_id="fluxspeak-team")
#162	assert len(results) >= 1
#163	assert results[0]["channel_id"] == "fluxspeak-team"
#164
#165	def test_channel_recall_excludes_other_channels(self, temp_db):
#166	a = Mnemosyne(session_id="a1", author_id="abdias", channel_id="team-a", db_path=temp_db)
#167	b = Mnemosyne(session_id="b1", author_id="sarah", channel_id="team-b", db_path=temp_db)
#168
#169	a.remember("Team A secret")
#170	b.remember("Team B secret")
#171
#172	results = a.recall("secret", channel_id="team-a")
#173	assert len(results) == 1
#174	assert results[0]["channel_id"] == "team-a"
#175
#176
#177	class TestStatsIdentity:
#178	"""get_stats() and get_working_stats() with identity filters."""
#179
#180	def test_stats_by_author(self, temp_db):
#181	mem = Mnemosyne(session_id="test", author_id="abdias", db_path=temp_db)
#182	mem.remember("Memory 1")
#183	mem.remember("Memory 2")
#184
#185	stats = mem.beam.get_working_stats(author_id="abdias")
#186	assert stats["total"] == 2
#187
#188	stats_empty = mem.beam.get_working_stats(author_id="sarah")
#189	assert stats_empty["total"] == 0
#190
#191	def test_stats_by_channel(self, temp_db):
#192	a = Mnemosyne(session_id="a1", author_id="abdias", channel_id="team", db_path=temp_db)
#193	b = Mnemosyne(session_id="b1", author_id="sarah", channel_id="team", db_path=temp_db)
#194	a.remember("A")
#195	b.remember("B")
#196
#197	stats = a.beam.get_working_stats(channel_id="team")
#198	assert stats["total"] == 2
#199
#200
#201	class TestMcpIdentity:
#202	"""MCP per-connection instances and env var fallback."""
#203
#204	def test_create_instance_with_args(self):
#205	mem = _create_instance(author_id="codex", author_type="agent",
#206	channel_id="repo-dev", bank="default")
#207	assert mem.author_id == "codex"
#208	assert mem.author_type == "agent"
#209	assert mem.channel_id == "repo-dev"
#210
#211	def test_create_instance_env_fallback(self):
#212	os.environ["MNEMOSYNE_AUTHOR_ID"] = "envuser"
#213	os.environ["MNEMOSYNE_AUTHOR_TYPE"] = "system"
#214	os.environ["MNEMOSYNE_CHANNEL_ID"] = "env-channel"
#215	try:
#216	mem = _create_instance()
#217	assert mem.author_id == "envuser"
#218	assert mem.author_type == "system"
#219	assert mem.channel_id == "env-channel"
#220	finally:
#221	del os.environ["MNEMOSYNE_AUTHOR_ID"]
#222	del os.environ["MNEMOSYNE_AUTHOR_TYPE"]
#223	del os.environ["MNEMOSYNE_CHANNEL_ID"]
#224
#225	def test_create_instance_args_override_env(self):
#226	os.environ["MNEMOSYNE_AUTHOR_ID"] = "envuser"
#227	try:
#228	mem = _create_instance(author_id="override")
#229	assert mem.author_id == "override"
#230	finally:
#231	del os.environ["MNEMOSYNE_AUTHOR_ID"]
#232
#233
#234	class TestBackwardCompatibility:
#235	"""Without identity params, behavior is unchanged."""
#236
#237	def test_recall_without_identity_filters(self, temp_db):
#238	mem = Mnemosyne(session_id="test", db_path=temp_db)
#239	mid = mem.remember("Dark mode")
#240	results = mem.recall("dark")
#241	assert len(results) >= 1
#242	assert results[0]["id"] == mid
#243
#244	def test_old_constructor_still_works(self, temp_db):
#245	"""The old Mnemosyne(session_id=..., db_path=...) still works."""
#246	mem = Mnemosyne(session_id="test", db_path=temp_db)
#247	mid = mem.remember("Works fine")
#248	assert mid is not None
#249	results = mem.recall("fine")
#250	assert len(results) == 1
#251

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public