114 lines
3.2 KiB
Python
114 lines
3.2 KiB
Python
"""Quick smoke test for the memory system."""
|
|
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
|
|
from memory import MemoryManager
|
|
from memory.types import MemoryConfig
|
|
from memory.internal import chunk_markdown, hash_text, list_memory_files
|
|
|
|
|
|
def test_internals():
|
|
print("── Internal utilities ──")
|
|
|
|
# Hashing
|
|
h = hash_text("hello world")
|
|
assert len(h) == 64
|
|
print(f"✅ hash_text: {h[:16]}...")
|
|
|
|
# Chunking
|
|
text = "# Title\n\nLine1\nLine2\nLine3\n\n## Section\n\nMore text here"
|
|
chunks = chunk_markdown(text, chunk_tokens=50, chunk_overlap=10)
|
|
assert len(chunks) >= 1
|
|
print(f"✅ chunk_markdown: {len(chunks)} chunks")
|
|
for c in chunks:
|
|
print(f" lines {c.start_line}-{c.end_line}: {c.text[:50]!r}")
|
|
|
|
print()
|
|
|
|
|
|
async def test_manager():
|
|
print("── MemoryManager ──")
|
|
|
|
# Clean slate
|
|
test_dir = "/tmp/aetheel_test_workspace"
|
|
test_db = "/tmp/aetheel_test_memory.db"
|
|
for p in [test_dir, test_db]:
|
|
if os.path.exists(p):
|
|
if os.path.isdir(p):
|
|
shutil.rmtree(p)
|
|
else:
|
|
os.remove(p)
|
|
|
|
config = MemoryConfig(
|
|
workspace_dir=test_dir,
|
|
db_path=test_db,
|
|
)
|
|
|
|
mgr = MemoryManager(config)
|
|
print(f"✅ Created: workspace={mgr._workspace_dir}")
|
|
|
|
# Identity files
|
|
soul = mgr.read_soul()
|
|
assert soul and len(soul) > 0
|
|
print(f"✅ SOUL.md: {len(soul)} chars")
|
|
|
|
user = mgr.read_user()
|
|
assert user and len(user) > 0
|
|
print(f"✅ USER.md: {len(user)} chars")
|
|
|
|
memory = mgr.read_long_term_memory()
|
|
assert memory and len(memory) > 0
|
|
print(f"✅ MEMORY.md: {len(memory)} chars")
|
|
|
|
# Append to memory
|
|
mgr.append_to_memory("Test entry: Python 3.14 works great!")
|
|
memory2 = mgr.read_long_term_memory()
|
|
assert len(memory2) > len(memory)
|
|
print(f"✅ Appended to MEMORY.md: {len(memory2)} chars")
|
|
|
|
# Log a session
|
|
log_path = mgr.log_session(
|
|
"User: Hello!\nAssistant: Hi, how can I help?",
|
|
channel="terminal",
|
|
)
|
|
assert os.path.exists(log_path)
|
|
print(f"✅ Session logged: {log_path}")
|
|
|
|
# Sync
|
|
print("\n⏳ Syncing (loading embedding model on first run)...")
|
|
stats = await mgr.sync()
|
|
print(f"✅ Sync complete:")
|
|
for k, v in stats.items():
|
|
print(f" {k}: {v}")
|
|
|
|
# Search
|
|
print("\n🔍 Searching for 'personality values'...")
|
|
results = await mgr.search("personality values")
|
|
print(f"✅ Found {len(results)} results:")
|
|
for i, r in enumerate(results[:3]):
|
|
print(f" [{i+1}] score={r.score:.3f} path={r.path} lines={r.start_line}-{r.end_line}")
|
|
print(f" {r.snippet[:80]}...")
|
|
|
|
print("\n🔍 Searching for 'preferences'...")
|
|
results2 = await mgr.search("preferences")
|
|
print(f"✅ Found {len(results2)} results:")
|
|
for i, r in enumerate(results2[:3]):
|
|
print(f" [{i+1}] score={r.score:.3f} path={r.path} lines={r.start_line}-{r.end_line}")
|
|
print(f" {r.snippet[:80]}...")
|
|
|
|
# Status
|
|
print("\n📊 Status:")
|
|
status = mgr.status()
|
|
for k, v in status.items():
|
|
print(f" {k}: {v}")
|
|
|
|
mgr.close()
|
|
print("\n✅ All memory system tests passed!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_internals()
|
|
asyncio.run(test_manager())
|