77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from traderai.memory import MemoryStore
|
|
|
|
|
|
def test_memory_store_recalls_saved_fact(tmp_path):
|
|
store = MemoryStore(str(tmp_path / "memory.sqlite3"))
|
|
store.remember("preference", "The user prefers Polaris Bits searches to include barter listings.", importance=5)
|
|
|
|
results = store.recall("Polaris barter")
|
|
|
|
assert results
|
|
assert "Polaris Bits" in results[0]["content"]
|
|
|
|
|
|
def test_memory_store_clear_selected_sections(tmp_path):
|
|
store = MemoryStore(str(tmp_path / "memory.sqlite3"))
|
|
store.remember("note", "Forgettable note")
|
|
store.add_conversation("user", "hello")
|
|
store.set_profile("configured_name", "Hudson")
|
|
|
|
deleted = store.clear(include_profile=False)
|
|
snapshot = store.inspect()
|
|
|
|
assert deleted["memories"] == 1
|
|
assert deleted["conversations"] == 1
|
|
assert snapshot["memories"] == []
|
|
assert snapshot["conversations"] == []
|
|
assert snapshot["profile"][0]["key"] == "configured_name"
|
|
|
|
|
|
def test_memory_store_separates_chat_threads_but_keeps_shared_memories(tmp_path):
|
|
store = MemoryStore(str(tmp_path / "memory.sqlite3"))
|
|
first = store.create_thread("First")
|
|
second = store.create_thread("Second")
|
|
store.add_conversation("user", "first thread message", first["id"])
|
|
store.add_conversation("user", "second thread message", second["id"])
|
|
store.remember("preference", "Shared trading preference", importance=5)
|
|
|
|
first_rows = store.recent_conversation(thread_id=first["id"])
|
|
second_rows = store.recent_conversation(thread_id=second["id"])
|
|
|
|
assert [row["content"] for row in first_rows] == ["first thread message"]
|
|
assert [row["content"] for row in second_rows] == ["second thread message"]
|
|
assert store.recall("trading preference")[0]["content"] == "Shared trading preference"
|
|
|
|
|
|
def test_memory_store_renames_threads_and_deletes_outbox_items(tmp_path):
|
|
store = MemoryStore(str(tmp_path / "memory.sqlite3"))
|
|
thread = store.create_thread("New chat")
|
|
store.add_outbox("Wake job result")
|
|
inbox_id = store.list_outbox()[0]["id"]
|
|
|
|
renamed = store.rename_thread(thread["id"], " Market Check ")
|
|
deleted = store.delete_outbox(inbox_id)
|
|
|
|
assert renamed["title"] == "Market Check"
|
|
assert deleted is True
|
|
assert store.list_outbox() == []
|
|
|
|
|
|
def test_memory_store_uses_absolute_path_across_working_directory_changes(tmp_path, monkeypatch):
|
|
original_cwd = tmp_path / "start"
|
|
original_cwd.mkdir()
|
|
monkeypatch.chdir(original_cwd)
|
|
|
|
store = MemoryStore("data/memory.sqlite3")
|
|
|
|
moved_cwd = tmp_path / "moved"
|
|
moved_cwd.mkdir()
|
|
monkeypatch.chdir(moved_cwd)
|
|
|
|
store.add_outbox("Notification survived cwd change")
|
|
|
|
snapshot = store.inspect()
|
|
|
|
assert store.path.is_absolute()
|
|
assert snapshot["outbox"][0]["content"] == "Notification survived cwd change"
|