89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
import uuid
|
|
|
|
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]
|
|
|
|
|
|
def test_memory_thread_info_and_read_state():
|
|
s = InMemoryStore()
|
|
a = s.get_or_create_agent("A")
|
|
b = s.get_or_create_agent("B")
|
|
t = s.create_thread({a, b})
|
|
s.append_message(t.id, a, "one")
|
|
s.append_message(t.id, b, "two")
|
|
m3 = s.append_message(t.id, a, "three")
|
|
|
|
info = s.get_thread_info(t.id)
|
|
assert info is not None
|
|
assert info.thread == t
|
|
assert info.message_count == 3
|
|
assert info.latest_message == m3
|
|
|
|
state = s.get_thread_read_state(t.id, a)
|
|
assert state.agent == a
|
|
assert state.cursor_seq == 0
|
|
assert state.latest_seq == 3
|
|
assert state.unread_count == 3
|
|
assert state.unread_from_others_count == 1
|
|
|
|
s.set_cursor(t.id, a, 2)
|
|
state = s.get_thread_read_state(t.id, a)
|
|
assert state.cursor_seq == 2
|
|
assert state.unread_count == 1
|
|
assert state.unread_from_others_count == 0
|
|
|
|
|
|
def test_memory_thread_info_for_empty_thread():
|
|
s = InMemoryStore()
|
|
a = s.get_or_create_agent("A")
|
|
b = s.get_or_create_agent("B")
|
|
t = s.create_thread({a, b})
|
|
|
|
info = s.get_thread_info(t.id)
|
|
state = s.get_thread_read_state(t.id, a)
|
|
|
|
assert info is not None
|
|
assert info.message_count == 0
|
|
assert info.latest_message is None
|
|
assert state.latest_seq == 0
|
|
assert state.unread_count == 0
|
|
assert state.unread_from_others_count == 0
|
|
assert s.get_thread_info(uuid.uuid4()) is None
|
|
assert s.get_thread_read_state(uuid.uuid4(), a) is None
|