34 lines
1.1 KiB
Python
34 lines
1.1 KiB
Python
from agentmsgs.stores.memory import InMemoryStore
|
|
|
|
def test_memory_create_and_find_thread():
|
|
s = InMemoryStore()
|
|
alice = s.get_or_create_agent("Alice")
|
|
bob = s.get_or_create_agent("Bob")
|
|
t = s.create_thread({alice, bob})
|
|
assert s.get_thread(t.id) == t
|
|
assert t in s.find_threads({alice})
|
|
assert t in s.find_threads({alice, bob})
|
|
assert s.find_threads({alice, bob, s.get_or_create_agent("Charlie")}) == []
|
|
|
|
def test_memory_soft_unique_name():
|
|
s = InMemoryStore()
|
|
a1 = s.get_or_create_agent("Alice")
|
|
a2 = s.get_or_create_agent("Alice")
|
|
assert a1.id == a2.id
|
|
|
|
def test_memory_delete_account():
|
|
s = InMemoryStore()
|
|
a = s.get_or_create_agent("Alice")
|
|
s.delete_agent(a.id)
|
|
assert s.get_agent_by_id(a.id) is None
|
|
assert s.get_agent_by_name("Alice") is None
|
|
|
|
def test_memory_append_and_list():
|
|
s = InMemoryStore()
|
|
a = s.get_or_create_agent("A"); b = s.get_or_create_agent("B")
|
|
t = s.create_thread({a,b})
|
|
m1 = s.append_message(t.id, a, "hi")
|
|
m2 = s.append_message(t.id, b, "yo")
|
|
assert m1.seq == 1 and m2.seq == 2
|
|
assert s.list_messages(t.id, after_seq=1) == [m2]
|