42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from agentmsgs.core import ops
|
|
from agentmsgs.stores.memory import InMemoryStore
|
|
|
|
|
|
def test_ops_validates_sender_must_be_in_thread():
|
|
s = InMemoryStore()
|
|
a = ops.get_or_create_agent(s, "A")
|
|
b = ops.get_or_create_agent(s, "B")
|
|
c = ops.get_or_create_agent(s, "C")
|
|
t = ops.create_thread(s, {a, b})
|
|
try:
|
|
ops.append_message(s, t.id, c, "oops")
|
|
assert False, "should raise"
|
|
except ValueError as e:
|
|
assert "not in thread" in str(e)
|
|
|
|
|
|
def test_ops_has_unread_exclude_own():
|
|
s = InMemoryStore()
|
|
a = ops.get_or_create_agent(s, "A")
|
|
b = ops.get_or_create_agent(s, "B")
|
|
t = ops.create_thread(s, {a, b})
|
|
ops.append_message(s, t.id, a, "a1")
|
|
# b has unread, a's own message shouldn't count if exclude_own
|
|
assert ops.has_unread(s, t.id, b) is True
|
|
assert ops.has_unread(s, t.id, a, exclude_own=True) is False
|
|
assert ops.has_unread(s, t.id, a, exclude_own=False) is True
|
|
ops.mark_read(s, t.id, b, 1)
|
|
assert ops.has_unread(s, t.id, b) is False
|
|
|
|
|
|
def test_ops_join_and_delete():
|
|
s = InMemoryStore()
|
|
a = ops.get_or_create_agent(s, "A")
|
|
b = ops.get_or_create_agent(s, "B")
|
|
c = ops.get_or_create_agent(s, "C")
|
|
t = ops.create_thread(s, {a, b})
|
|
t2 = ops.join_thread(s, t.id, c)
|
|
assert c in t2.participants
|
|
ops.delete_agent(s, c.id)
|
|
assert s.get_agent_by_id(c.id) is None
|