45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
import tempfile, pathlib
|
|
from agentmsgs.stores.sqlite import SQLiteStore
|
|
|
|
def test_sqlite_persists_across_handles():
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = pathlib.Path(d)/"test.db"
|
|
s1 = SQLiteStore(p)
|
|
try:
|
|
a = s1.get_or_create_agent("Alice"); b = s1.get_or_create_agent("Bob")
|
|
t = s1.create_thread({a,b})
|
|
s1.append_message(t.id, a, "hi")
|
|
# new handle same file
|
|
s2 = SQLiteStore(p)
|
|
try:
|
|
assert s2.get_agent_by_name("Alice").id == a.id
|
|
assert len(s2.list_messages(t.id)) == 1
|
|
assert s2.find_threads({a})[0].id == t.id
|
|
finally:
|
|
s2.close()
|
|
finally:
|
|
s1.close()
|
|
|
|
def test_sqlite_concurrent_append():
|
|
import pathlib, tempfile
|
|
from agentmsgs.stores.sqlite import SQLiteStore
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = pathlib.Path(d)/"c.db"
|
|
s1 = SQLiteStore(p); s2 = SQLiteStore(p)
|
|
try:
|
|
a = s1.get_or_create_agent("A"); b = s1.get_or_create_agent("B")
|
|
# s2 sees same agents via file
|
|
a2 = s2.get_agent_by_name("A"); b2 = s2.get_agent_by_name("B")
|
|
t = s1.create_thread({a,b})
|
|
s1.append_message(t.id, a, "from s1")
|
|
s2.append_message(t.id, b2, "from s2")
|
|
assert len(s1.list_messages(t.id)) == 2
|
|
# verify seq uniqueness and monotonicity
|
|
msgs = s1.list_messages(t.id)
|
|
seqs = [m.seq for m in msgs]
|
|
assert seqs == [1, 2]
|
|
assert len(set(seqs)) == 2
|
|
finally:
|
|
s2.close()
|
|
s1.close()
|