my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Tests for Mnemosyne streaming memory + delta sync.
#3	"""
#4
#5	import pytest
#6	import json
#7	import time
#8	import threading
#9	from datetime import datetime
#10	from pathlib import Path
#11
#12	from mnemosyne.core.streaming import (
#13	MemoryStream, MemoryEvent, EventType, SyncCheckpoint,
#14	DeltaSync, _StreamIterator
#15	)
#16	from mnemosyne.core.memory import Mnemosyne
#17
#18
#19	# ─── Fixtures ─────────────────────────────────────────────────────────
#20
#21	@pytest.fixture
#22	def stream():
#23	return MemoryStream(max_buffer=100)
#24
#25
#26	@pytest.fixture
#27	def sample_event():
#28	return MemoryEvent(
#29	event_type=EventType.MEMORY_ADDED,
#30	memory_id="mem_123",
#31	session_id="sess_456",
#32	content="Test memory",
#33	source="conversation",
#34	importance=0.7
#35	)
#36
#37
#38	@pytest.fixture
#39	def mnemosyne(tmp_path):
#40	db_path = tmp_path / "test_streaming.db"
#41	return Mnemosyne(session_id="test_session", db_path=db_path)
#42
#43
#44	# ─── MemoryEvent ────────────────────────────────────────────────────
#45
#46	class TestMemoryEvent:
#47	def test_event_creation(self, sample_event):
#48	assert sample_event.event_type == EventType.MEMORY_ADDED
#49	assert sample_event.memory_id == "mem_123"
#50	assert sample_event.content == "Test memory"
#51
#52	def test_event_to_dict(self, sample_event):
#53	d = sample_event.to_dict()
#54	assert d["event_type"] == "MEMORY_ADDED"
#55	assert d["memory_id"] == "mem_123"
#56	assert "timestamp" in d
#57
#58	def test_event_to_json(self, sample_event):
#59	j = sample_event.to_json()
#60	data = json.loads(j)
#61	assert data["event_type"] == "MEMORY_ADDED"
#62	assert data["memory_id"] == "mem_123"
#63
#64	def test_event_from_dict(self):
#65	d = {
#66	"event_type": "MEMORY_RECALLED",
#67	"memory_id": "mem_456",
#68	"timestamp": "2026-01-01T00:00:00",
#69	"session_id": "sess_789",
#70	"content": "Recalled memory",
#71	}
#72	event = MemoryEvent.from_dict(d)
#73	assert event.event_type == EventType.MEMORY_RECALLED
#74	assert event.memory_id == "mem_456"
#75
#76	def test_event_delta_field(self):
#77	event = MemoryEvent(
#78	event_type=EventType.MEMORY_UPDATED,
#79	memory_id="mem_789",
#80	delta={"importance": {"old": 0.5, "new": 0.9}}
#81	)
#82	assert event.delta["importance"]["new"] == 0.9
#83
#84
#85	# ─── MemoryStream ───────────────────────────────────────────────────
#86
#87	class TestMemoryStream:
#88	def test_callback_registration(self, stream, sample_event):
#89	called = []
#90	def handler(event):
#91	called.append(event)
#92	stream.on(EventType.MEMORY_ADDED, handler)
#93	stream.emit(sample_event)
#94	assert len(called) == 1
#95	assert called[0].memory_id == "mem_123"
#96
#97	def test_any_callback(self, stream, sample_event):
#98	called = []
#99	def handler(event):
#100	called.append(event)
#101	stream.on_any(handler)
#102	stream.emit(sample_event)
#103	assert len(called) == 1
#104
#105	def test_callback_removal(self, stream, sample_event):
#106	called = []
#107	def handler(event):
#108	called.append(event)
#109	stream.on(EventType.MEMORY_ADDED, handler)
#110	stream.off(EventType.MEMORY_ADDED, handler)
#111	stream.emit(sample_event)
#112	assert len(called) == 0
#113
#114	def test_buffering(self, stream):
#115	for i in range(5):
#116	stream.emit(MemoryEvent(
#117	event_type=EventType.MEMORY_ADDED,
#118	memory_id=f"mem_{i}"
#119	))
#120	buffer = stream.get_buffer()
#121	assert len(buffer) == 5
#122	assert buffer[0].memory_id == "mem_0"
#123
#124	def test_buffer_filter_by_type(self, stream):
#125	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="a"))
#126	stream.emit(MemoryEvent(event_type=EventType.MEMORY_RECALLED, memory_id="b"))
#127	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="c"))
#128	buffer = stream.get_buffer([EventType.MEMORY_ADDED])
#129	assert len(buffer) == 2
#130	assert all(e.event_type == EventType.MEMORY_ADDED for e in buffer)
#131
#132	def test_buffer_filter_since(self, stream):
#133	now = datetime.now().isoformat()
#134	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="old"))
#135	time.sleep(0.01)
#136	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="new"))
#137	buffer = stream.get_buffer(since=now)
#138	assert len(buffer) == 2 # Both should be after "now" since we just created it
#139
#140	def test_buffer_max_size(self):
#141	s = MemoryStream(max_buffer=3)
#142	for i in range(5):
#143	s.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id=f"mem_{i}"))
#144	buffer = s.get_buffer()
#145	assert len(buffer) == 3
#146	assert buffer[0].memory_id == "mem_2" # Oldest kept
#147
#148	def test_clear_buffer(self, stream):
#149	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="a"))
#150	stream.clear_buffer()
#151	assert len(stream.get_buffer()) == 0
#152
#153	def test_iterator_basic(self, stream):
#154	events = []
#155	def collect():
#156	for event in stream.listen([EventType.MEMORY_ADDED]):
#157	events.append(event)
#158	if len(events) >= 2:
#159	break
#160	t = threading.Thread(target=collect)
#161	t.start()
#162	time.sleep(0.05)
#163	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="a"))
#164	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="b"))
#165	t.join(timeout=2)
#166	assert len(events) == 2
#167	assert events[0].memory_id == "a"
#168	assert events[1].memory_id == "b"
#169
#170	def test_iterator_filter(self, stream):
#171	events = []
#172	def collect():
#173	for event in stream.listen([EventType.MEMORY_RECALLED]):
#174	events.append(event)
#175	if len(events) >= 1:
#176	break
#177	t = threading.Thread(target=collect)
#178	t.start()
#179	time.sleep(0.05)
#180	stream.emit(MemoryEvent(event_type=EventType.MEMORY_ADDED, memory_id="a"))
#181	stream.emit(MemoryEvent(event_type=EventType.MEMORY_RECALLED, memory_id="b"))
#182	t.join(timeout=2)
#183	assert len(events) == 1
#184	assert events[0].memory_id == "b"
#185
#186	def test_multiple_callbacks_same_type(self, stream, sample_event):
#187	calls1, calls2 = [], []
#188	stream.on(EventType.MEMORY_ADDED, lambda e: calls1.append(e))
#189	stream.on(EventType.MEMORY_ADDED, lambda e: calls2.append(e))
#190	stream.emit(sample_event)
#191	assert len(calls1) == 1
#192	assert len(calls2) == 1
#193
#194	def test_callback_exception_isolation(self, stream, sample_event):
#195	called = []
#196	def bad_handler(event):
#197	raise RuntimeError("boom")
#198	def good_handler(event):
#199	called.append(event)
#200	stream.on(EventType.MEMORY_ADDED, bad_handler)
#201	stream.on(EventType.MEMORY_ADDED, good_handler)
#202	stream.emit(sample_event) # Should not raise
#203	assert len(called) == 1
#204
#205	def test_all_event_types(self, stream):
#206	for et in EventType:
#207	stream.emit(MemoryEvent(event_type=et, memory_id=f"mem_{et.name}"))
#208	buffer = stream.get_buffer()
#209	assert len(buffer) == len(EventType)
#210
#211
#212	# ─── DeltaSync ──────────────────────────────────────────────────────
#213
#214	class TestDeltaSync:
#215	def test_init(self, mnemosyne):
#216	ds = DeltaSync(mnemosyne)
#217	assert ds.mnemosyne is mnemosyne
#218	assert ds.checkpoint_dir.exists()
#219
#220	def test_compute_delta_first_sync(self, mnemosyne):
#221	ds = DeltaSync(mnemosyne)
#222	# Store some memories
#223	mnemosyne.remember("Memory 1", source="test")
#224	mnemosyne.remember("Memory 2", source="test")
#225	delta = ds.compute_delta("peer_a", "working_memory")
#226	assert len(delta) >= 2
#227
#228	def test_compute_delta_incremental(self, mnemosyne):
#229	ds = DeltaSync(mnemosyne)
#230	mnemosyne.remember("Memory 1", source="test")
#231	# First sync
#232	delta1 = ds.compute_delta("peer_b", "working_memory")
#233	ds.apply_delta("peer_b", delta1, "working_memory")
#234	# Add more
#235	mnemosyne.remember("Memory 2", source="test")
#236	delta2 = ds.compute_delta("peer_b", "working_memory")
#237	assert len(delta2) >= 1
#238	# Should only have the new memory
#239	contents = [d.get("content", "") for d in delta2]
#240	assert "Memory 2" in contents
#241
#242	def test_apply_delta_insert(self, mnemosyne):
#243	ds = DeltaSync(mnemosyne)
#244	delta = [{"id": "test_1", "content": "Imported memory", "source": "import"}]
#245	stats = ds.apply_delta("peer_c", delta, "working_memory")
#246	assert stats["inserted"] == 1
#247	assert stats["updated"] == 0
#248
#249	def test_apply_delta_update(self, mnemosyne):
#250	ds = DeltaSync(mnemosyne)
#251	mid = mnemosyne.remember("Original", source="test")
#252	delta = [{"id": mid, "content": "Updated", "source": "test", "importance": 0.9}]
#253	stats = ds.apply_delta("peer_d", delta, "working_memory")
#254	assert stats["updated"] == 1
#255
#256	def test_apply_delta_skip_no_id(self, mnemosyne):
#257	ds = DeltaSync(mnemosyne)
#258	delta = [{"content": "No ID", "source": "test"}]
#259	stats = ds.apply_delta("peer_e", delta, "working_memory")
#260	assert stats["skipped"] == 1
#261
#262	def test_checkpoint_persistence(self, mnemosyne, tmp_path):
#263	ds = DeltaSync(mnemosyne, checkpoint_dir=tmp_path)
#264	mnemosyne.remember("Memory", source="test")
#265	delta = ds.compute_delta("peer_f", "working_memory")
#266	# Apply delta to trigger checkpoint save
#267	ds.apply_delta("peer_f", delta, "working_memory")
#268	# Check checkpoint saved
#269	cp = ds.get_checkpoint("peer_f")
#270	assert cp is not None
#271	assert cp.peer_id == "peer_f"
#272
#273	def test_checkpoint_reload(self, mnemosyne, tmp_path):
#274	ds1 = DeltaSync(mnemosyne, checkpoint_dir=tmp_path)
#275	mnemosyne.remember("Memory", source="test")
#276	delta = ds1.compute_delta("peer_g", "working_memory")
#277	ds1.apply_delta("peer_g", delta, "working_memory")
#278	# Create new instance pointing to same dir
#279	ds2 = DeltaSync(mnemosyne, checkpoint_dir=tmp_path)
#280	cp = ds2.get_checkpoint("peer_g")
#281	assert cp is not None
#282	assert cp.peer_id == "peer_g"
#283
#284	def test_sync_to_returns_delta(self, mnemosyne):
#285	ds = DeltaSync(mnemosyne)
#286	mnemosyne.remember("Sync test", source="test")
#287	result = ds.sync_to("peer_h", "working_memory")
#288	assert "delta" in result
#289	assert "count" in result
#290	assert result["count"] >= 1
#291
#292	def test_sync_from_applies_delta(self, mnemosyne):
#293	ds = DeltaSync(mnemosyne)
#294	delta = [{"id": "remote_1", "content": "Remote memory", "source": "remote"}]
#295	result = ds.sync_from("peer_i", delta, "working_memory")
#296	assert result["stats"]["inserted"] == 1
#297
#298	def test_invalid_mnemosyne_type(self):
#299	with pytest.raises(TypeError):
#300	DeltaSync("not a mnemosyne")
#301
#302
#303	# ─── SyncCheckpoint ─────────────────────────────────────────────────
#304
#305	class TestSyncCheckpoint:
#306	def test_checkpoint_creation(self):
#307	cp = SyncCheckpoint(peer_id="p1", last_sync_at="2026-01-01T00:00:00", last_rowid=42)
#308	assert cp.peer_id == "p1"
#309	assert cp.last_rowid == 42
#310
#311	def test_checkpoint_serialization(self):
#312	cp = SyncCheckpoint(peer_id="p1", last_sync_at="2026-01-01T00:00:00", last_rowid=42)
#313	j = cp.to_json()
#314	data = json.loads(j)
#315	assert data["peer_id"] == "p1"
#316	assert data["last_rowid"] == 42
#317

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public