my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Integration tests for Mnemosyne Structured Fact Extraction (Phase 2)
#3	Tests end-to-end: remember with extract -> recall finds facts
#4	"""
#5
#6	import os
#7	import sys
#8	import json
#9	import tempfile
#10	from pathlib import Path
#11	from datetime import datetime
#12
#13	sys.path.insert(0, str(Path(__file__).parent.parent))
#14
#15	from mnemosyne.core.memory import Mnemosyne
#16	from mnemosyne.core.triples import TripleStore, init_triples
#17	from mnemosyne.core.beam import BeamMemory
#18
#19
#20	class MockLLMExtractor:
#21	"""Mock LLM for fact extraction that returns predictable facts."""
#22	def __init__(self, facts=None):
#23	self.facts = facts or [
#24	"The user loves coffee",
#25	"The user hates mornings",
#26	"The user prefers dark roast"
#27	]
#28
#29	def __call__(self, prompt, **kwargs):
#30	return "\n".join(self.facts)
#31
#32
#33	def test_end_to_end_extract_recall():
#34	"""
#35	Test: remember with extract=True -> facts stored -> recall finds them
#36	"""
#37	with tempfile.TemporaryDirectory() as tmpdir:
#38	db_path = Path(tmpdir) / "test.db"
#39
#40	# Initialize
#41	init_triples(db_path)
#42
#43	# Create Mnemosyne instance
#44	mem = Mnemosyne(session_id="test_session", db_path=db_path)
#45
#46	# Store a memory WITH fact extraction (mocked)
#47	content = "I absolutely love coffee, especially dark roast. I hate mornings though."
#48
#49	# Manually inject facts (simulating LLM extraction)
#50	memory_id = mem.remember(content, source="test", extract=False)
#51
#52	# Manually add facts to triples
#53	triples = TripleStore(db_path=db_path)
#54	triples.add_facts(memory_id, [
#55	"The user loves coffee",
#56	"The user hates mornings",
#57	"The user prefers dark roast"
#58	], source="test", confidence=0.7)
#59
#60	# Now recall with a query that matches the facts
#61	results = mem.recall("what does the user like", top_k=5)
#62
#63	# Should find the memory via fact matching
#64	assert len(results) > 0, "Recall should find results"
#65
#66	# Check if fact_match is present in any result
#67	fact_matches = [r for r in results if r.get("fact_match")]
#68	assert len(fact_matches) > 0, "At least one result should have fact_match=True"
#69
#70	# The memory should be in results
#71	memory_ids = [r["id"] for r in results]
#72	assert memory_id in memory_ids, f"Memory {memory_id} should be in recall results"
#73
#74	print("PASS: test_end_to_end_extract_recall")
#75
#76
#77	def test_fact_recall_keyword_matching():
#78	"""
#79	Test: Fact recall uses keyword matching against stored facts
#80	"""
#81	with tempfile.TemporaryDirectory() as tmpdir:
#82	db_path = Path(tmpdir) / "test.db"
#83	init_triples(db_path)
#84
#85	mem = Mnemosyne(session_id="test_session_2", db_path=db_path)
#86
#87	# Store two memories
#88	id1 = mem.remember("I love hiking in the mountains", source="test", extract=False)
#89	id2 = mem.remember("I enjoy swimming at the beach", source="test", extract=False)
#90
#91	# Add facts manually
#92	triples = TripleStore(db_path=db_path)
#93	triples.add_facts(id1, ["The user loves hiking", "The user enjoys mountains"], source="test")
#94	triples.add_facts(id2, ["The user enjoys swimming", "The user likes the beach"], source="test")
#95
#96	# Recall for "hiking" should find id1 via facts
#97	results = mem.recall("hiking", top_k=5)
#98	result_ids = [r["id"] for r in results]
#99	assert id1 in result_ids, "Should find memory about hiking via fact match"
#100
#101	# Recall for "swimming" should find id2 via facts
#102	results = mem.recall("swimming", top_k=5)
#103	result_ids = [r["id"] for r in results]
#104	assert id2 in result_ids, "Should find memory about swimming via fact match"
#105
#106	print("PASS: test_fact_recall_keyword_matching")
#107
#108
#109	def test_fact_and_entity_extraction_together():
#110	"""
#111	Test: Both extract_entities=True and extract=True work together
#112	"""
#113	with tempfile.TemporaryDirectory() as tmpdir:
#114	db_path = Path(tmpdir) / "test.db"
#115	init_triples(db_path)
#116
#117	mem = Mnemosyne(session_id="test_session_3", db_path=db_path)
#118
#119	content = "I met Abdias in New York. He loves coffee."
#120
#121	# Store with both extraction flags (entities will extract, facts won't because no LLM)
#122	memory_id = mem.remember(content, source="test", extract_entities=True, extract=False)
#123
#124	# Manually add facts
#125	triples = TripleStore(db_path=db_path)
#126	triples.add_facts(memory_id, ["The user met Abdias", "Abdias loves coffee"], source="test")
#127
#128	# Recall for "Abdias" should find via entity
#129	results = mem.recall("Abdias", top_k=5)
#130	result_ids = [r["id"] for r in results]
#131	assert memory_id in result_ids, "Should find memory via entity match"
#132
#133	# Recall for "coffee" should find via fact
#134	results = mem.recall("coffee", top_k=5)
#135	result_ids = [r["id"] for r in results]
#136	assert memory_id in result_ids, "Should find memory via fact match"
#137
#138	print("PASS: test_fact_and_entity_extraction_together")
#139
#140
#141	def test_backward_compatibility():
#142	"""
#143	Test: remember() without extract works exactly as before
#144	"""
#145	with tempfile.TemporaryDirectory() as tmpdir:
#146	db_path = Path(tmpdir) / "test.db"
#147	init_triples(db_path)
#148
#149	mem = Mnemosyne(session_id="test_session_4", db_path=db_path)
#150
#151	# Store without any extraction
#152	memory_id = mem.remember("Simple memory without extraction", source="test")
#153
#154	# Should be retrievable
#155	results = mem.recall("simple memory", top_k=5)
#156	result_ids = [r["id"] for r in results]
#157	assert memory_id in result_ids, "Should find memory without extraction"
#158
#159	# Triples should be empty
#160	triples = TripleStore(db_path=db_path)
#161	all_facts = triples.query_by_predicate("fact")
#162	assert len(all_facts) == 0, "No facts should be stored without extract=True"
#163
#164	print("PASS: test_backward_compatibility")
#165
#166
#167	def test_graceful_fallback_no_llm():
#168	"""
#169	Test: When LLM unavailable, memory still stores, no error
#170	"""
#171	from unittest.mock import patch
#172
#173	with tempfile.TemporaryDirectory() as tmpdir:
#174	db_path = Path(tmpdir) / "test.db"
#175	init_triples(db_path)
#176
#177	mem = Mnemosyne(session_id="test_session_5", db_path=db_path)
#178
#179	# Patch llm_available so extraction is skipped regardless of env state.
#180	# extract_facts() calls local_llm.llm_available() through the live module
#181	# reference, not through extraction's at-import-time binding.
#182	with patch("mnemosyne.core.local_llm.llm_available", return_value=False):
#183	# This should NOT raise even though extract=True and no LLM
#184	memory_id = mem.remember(
#185	"I love coffee",
#186	source="test",
#187	extract=True # LLM unavailable, but should not fail
#188	)
#189
#190	assert memory_id is not None, "Memory ID should be returned"
#191
#192	# No facts should be stored
#193	triples = TripleStore(db_path=db_path)
#194	all_facts = triples.query_by_predicate("fact")
#195	assert len(all_facts) == 0, "No facts when LLM unavailable"
#196
#197	print("PASS: test_graceful_fallback_no_llm")
#198
#199
#200	def test_fact_aware_recall_boosts_scores():
#201	"""
#202	Test: Fact matches get score boost (1.2x)
#203	"""
#204	with tempfile.TemporaryDirectory() as tmpdir:
#205	db_path = Path(tmpdir) / "test.db"
#206	init_triples(db_path)
#207
#208	mem = Mnemosyne(session_id="test_session_6", db_path=db_path)
#209
#210	# Store two similar memories (different content to avoid dedup)
#211	id1 = mem.remember("I love coffee and tea in the morning", source="test", importance=0.5)
#212	id2 = mem.remember("I love coffee and tea in the evening", source="test", importance=0.5)
#213
#214	# Add fact only to id1
#215	triples = TripleStore(db_path=db_path)
#216	triples.add_facts(id1, ["The user loves coffee"], source="test")
#217
#218	# Recall for "coffee" - id1 should have fact_match
#219	results = mem.recall("coffee", top_k=5)
#220
#221	# Find results for id1 and id2
#222	r1 = [r for r in results if r["id"] == id1]
#223	r2 = [r for r in results if r["id"] == id2]
#224
#225	if r1 and r2:
#226	# id1 should have fact_match and boosted score
#227	assert r1[0].get("fact_match") == True, "id1 should have fact_match"
#228	assert r1[0]["score"] > r2[0]["score"], "Fact match should boost score"
#229
#230	print("PASS: test_fact_aware_recall_boosts_scores")
#231
#232
#233	def run_all_tests():
#234	"""Run all integration tests."""
#235	print("=" * 60)
#236	print("Phase 2: Structured Fact Extraction — Integration Tests")
#237	print("=" * 60)
#238
#239	tests = [
#240	test_end_to_end_extract_recall,
#241	test_fact_recall_keyword_matching,
#242	test_fact_and_entity_extraction_together,
#243	test_backward_compatibility,
#244	test_graceful_fallback_no_llm,
#245	test_fact_aware_recall_boosts_scores,
#246	]
#247
#248	passed = 0
#249	failed = 0
#250
#251	for test in tests:
#252	try:
#253	test()
#254	passed += 1
#255	except Exception as e:
#256	failed += 1
#257	print(f"FAIL: {test.__name__}: {e}")
#258	import traceback
#259	traceback.print_exc()
#260
#261	print("=" * 60)
#262	print(f"Results: {passed} passed, {failed} failed, {len(tests)} total")
#263	print("=" * 60)
#264
#265	return failed == 0
#266
#267
#268	if __name__ == "__main__":
#269	success = run_all_tests()
#270	sys.exit(0 if success else 1)
#271

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public