my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""
#2	Mnemosyne Integration Tests
#3	===========================
#4	End-to-end tests for all 5 phases:
#5	1. Typed Memory Schema
#6	2. Binary Vectors
#7	3. Episodic Graph
#8	4. Veracity Consolidation
#9	5. Polyphonic Recall
#10
#11	Also tests BEAM benchmark integration.
#12	"""
#13
#14	import unittest
#15	import tempfile
#16	import numpy as np
#17	from pathlib import Path
#18
#19	from mnemosyne.core.typed_memory import classify_memory, MemoryType, get_type_priority
#20	from mnemosyne.core.binary_vectors import BinaryVectorStore
#21	from mnemosyne.core.episodic_graph import EpisodicGraph
#22	from mnemosyne.core.veracity_consolidation import VeracityConsolidator
#23	from mnemosyne.core.polyphonic_recall import PolyphonicRecallEngine
#24
#25
#26	class TestTypedMemory(unittest.TestCase):
#27	"""Test Phase 1: Typed Memory Schema."""
#28
#29	def test_fact_classification(self):
#30	result = classify_memory("The API is at https://example.com")
#31	self.assertEqual(result.memory_type, MemoryType.FACT)
#32	self.assertGreater(result.confidence, 0.5)
#33
#34	def test_preference_classification(self):
#35	result = classify_memory("I prefer dark mode")
#36	self.assertEqual(result.memory_type, MemoryType.PREFERENCE)
#37
#38	def test_commitment_classification(self):
#39	result = classify_memory("I will deliver by Friday")
#40	self.assertEqual(result.memory_type, MemoryType.COMMITMENT)
#41
#42	def test_priority_ranking(self):
#43	self.assertGreater(get_type_priority(MemoryType.INSTRUCTION), get_type_priority(MemoryType.EVENT))
#44
#45	def test_decay_rates(self):
#46	from mnemosyne.core.typed_memory import get_decay_rate
#47	self.assertGreater(get_decay_rate(MemoryType.CONTEXT), get_decay_rate(MemoryType.FACT))
#48
#49
#50	class TestBinaryVectors(unittest.TestCase):
#51	"""Test Phase 2: Binary Vectors."""
#52
#53	def setUp(self):
#54	self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
#55	self.store = BinaryVectorStore(db_path=Path(self.temp_file.name))
#56	self.embedding = np.random.randn(384).astype(np.float32)
#57
#58	def tearDown(self):
#59	self.store.close()
#60	self.temp_file.close()
#61
#62	def test_binarization(self):
#63	binary = self.store.maximally_informative_binarization(self.embedding)
#64	self.assertEqual(len(binary), 48) # 384 bits / 8 = 48 bytes
#65
#66	def test_hamming_distance(self):
#67	binary_a = self.store.maximally_informative_binarization(self.embedding)
#68	binary_b = self.store.maximally_informative_binarization(self.embedding)
#69	distance = self.store.hamming_distance(binary_a, binary_b)
#70	self.assertEqual(distance, 0) # Same embedding = 0 distance
#71
#72	def test_store_and_search(self):
#73	self.store.store_vector("test_1", self.embedding)
#74	results = self.store.search(self.embedding, top_k=1)
#75	self.assertEqual(len(results), 1)
#76	self.assertEqual(results[0]["memory_id"], "test_1")
#77	self.assertEqual(results[0]["distance"], 0)
#78
#79	def test_compression_ratio(self):
#80	stats = self.store.get_stats()
#81	self.assertLess(stats["compression_ratio"], 0.1) # Should be ~3%
#82
#83
#84	class TestEpisodicGraph(unittest.TestCase):
#85	"""Test Phase 3: Episodic Graph."""
#86
#87	def setUp(self):
#88	self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
#89	self.graph = EpisodicGraph(db_path=Path(self.temp_file.name))
#90
#91	def tearDown(self):
#92	self.graph.close()
#93	self.temp_file.close()
#94
#95	def test_gist_extraction(self):
#96	content = "Alice met Bob at the office yesterday. She was happy."
#97	gist = self.graph.extract_gist(content, "mem_001")
#98	self.assertIn("Alice", gist.participants)
#99	self.assertEqual(gist.emotion, "positive")
#100	self.assertEqual(gist.location, "the office")
#101
#102	def test_fact_extraction(self):
#103	content = "Alice is a developer. She uses Python."
#104	facts = self.graph.extract_facts(content, "mem_001")
#105	self.assertGreater(len(facts), 0)
#106	self.assertEqual(facts[0].subject, "Alice")
#107
#108	def test_graph_storage(self):
#109	gist = self.graph.extract_gist("Test content", "mem_001")
#110	self.graph.store_gist(gist, "mem_001")
#111
#112	facts = self.graph.extract_facts("Alice is a developer", "mem_001")
#113	for fact in facts:
#114	self.graph.store_fact(fact, "mem_001")
#115
#116	stats = self.graph.get_stats()
#117	self.assertGreaterEqual(stats["gists"], 1)
#118	self.assertGreaterEqual(stats["facts"], 1)
#119
#120	def test_graph_traversal(self):
#121	from mnemosyne.core.episodic_graph import GraphEdge
#122	from datetime import datetime
#123
#124	self.graph.add_edge(GraphEdge("mem_001", "mem_002", "rel", 0.8, datetime.now().isoformat()))
#125	related = self.graph.find_related_memories("mem_001", depth=1)
#126	self.assertIn("mem_002", related)
#127
#128
#129	class TestVeracityConsolidation(unittest.TestCase):
#130	"""Test Phase 4: Veracity Consolidation."""
#131
#132	def setUp(self):
#133	self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
#134	self.cons = VeracityConsolidator(db_path=Path(self.temp_file.name))
#135
#136	def tearDown(self):
#137	self.cons.close()
#138	self.temp_file.close()
#139
#140	def test_basic_consolidation(self):
#141	fact = self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#142	self.assertEqual(fact.subject, "Alice")
#143	self.assertGreater(fact.confidence, 0.0)
#144
#145	def test_bayesian_update(self):
#146	fact1 = self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#147	fact2 = self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#148	self.assertGreater(fact2.confidence, fact1.confidence)
#149	self.assertEqual(fact2.mention_count, 2)
#150
#151	def test_conflict_detection(self):
#152	self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#153	self.cons.consolidate_fact("Alice", "is", "manager", "inferred")
#154	conflicts = self.cons.get_conflicts()
#155	self.assertGreater(len(conflicts), 0)
#156
#157	def test_conflict_resolution(self):
#158	self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#159	self.cons.consolidate_fact("Alice", "is", "manager", "inferred")
#160	conflicts = self.cons.get_conflicts()
#161
#162	if conflicts:
#163	self.cons.resolve_conflict(conflicts[0]["id"], "cf_Alice_is_developer")
#164	resolved = self.cons.get_conflicts()
#165	self.assertEqual(len(resolved), 0)
#166
#167	def test_high_confidence_summary(self):
#168	self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#169	self.cons.consolidate_fact("Alice", "is", "developer", "stated")
#170	summary = self.cons.get_high_confidence_summary("Alice", threshold=0.5)
#171	self.assertIn("Alice", summary)
#172
#173
#174	class TestPolyphonicRecall(unittest.TestCase):
#175	"""Test Phase 5: Polyphonic Recall."""
#176
#177	def setUp(self):
#178	self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
#179	self.engine = PolyphonicRecallEngine(db_path=Path(self.temp_file.name))
#180
#181	def tearDown(self):
#182	self.engine.close()
#183	self.temp_file.close()
#184
#185	def test_empty_recall(self):
#186	results = self.engine.recall("test query")
#187	self.assertIsInstance(results, list)
#188
#189	def test_voice_weights(self):
#190	weights = self.engine.voice_weights
#191	self.assertEqual(sum(weights.values()), 1.0)
#192	self.assertIn("vector", weights)
#193	self.assertIn("graph", weights)
#194	self.assertIn("fact", weights)
#195	self.assertIn("temporal", weights)
#196
#197	def test_stats(self):
#198	stats = self.engine.get_stats()
#199	self.assertIn("voice_weights", stats)
#200	self.assertIn("vector_stats", stats)
#201
#202
#203	class TestIntegration(unittest.TestCase):
#204	"""Test full pipeline integration."""
#205
#206	def setUp(self):
#207	self.temp_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
#208	self.db_path = Path(self.temp_file.name)
#209
#210	self.store = BinaryVectorStore(db_path=self.db_path)
#211	self.graph = EpisodicGraph(db_path=self.db_path)
#212	self.cons = VeracityConsolidator(db_path=self.db_path)
#213	self.engine = PolyphonicRecallEngine(db_path=self.db_path)
#214
#215	def tearDown(self):
#216	self.engine.close()
#217	self.cons.close()
#218	self.graph.close()
#219	self.store.close()
#220	self.temp_file.close()
#221
#222	def test_full_pipeline(self):
#223	# 1. Classify memory
#224	content = "Alice decided to use PostgreSQL for the new project."
#225	result = classify_memory(content)
#226	self.assertEqual(result.memory_type, MemoryType.DECISION)
#227
#228	# 2. Store binary vector
#229	embedding = np.random.randn(384).astype(np.float32)
#230	self.store.store_vector("mem_001", embedding)
#231
#232	# 3. Extract gist and facts
#233	gist = self.graph.extract_gist(content, "mem_001")
#234	self.graph.store_gist(gist, "mem_001")
#235
#236	facts = self.graph.extract_facts(content, "mem_001")
#237	for fact in facts:
#238	self.graph.store_fact(fact, "mem_001")
#239
#240	# 4. Consolidate facts
#241	for fact in facts:
#242	self.cons.consolidate_fact(
#243	fact.subject, fact.predicate, fact.object,
#244	veracity="stated", source="mem_001"
#245	)
#246
#247	# 5. Recall
#248	results = self.engine.recall("What database did Alice choose?", embedding)
#249	self.assertIsInstance(results, list)
#250
#251	# Verify stats
#252	stats = self.engine.get_stats()
#253	self.assertIn("vector_stats", stats)
#254	self.assertIn("graph_stats", stats)
#255	self.assertIn("consolidation_stats", stats)
#256
#257
#258	if __name__ == "__main__":
#259	unittest.main(verbosity=2)
#260

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public