1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
"""
Tests for the SQLite-backed chat message store.
"""
import pytest
import pytest_asyncio
from meshbay_node.chat.store import ChatStore
@pytest_asyncio.fixture
async def store(tmp_path):
s = ChatStore(db_path=tmp_path / "test_chat.db")
await s.open()
yield s
await s.close()
@pytest.mark.asyncio
async def test_save_and_retrieve(store):
row_id = await store.save_message(
sender_id="alice", iteration=0, payload=b"hello",
)
assert row_id == 1
msgs = await store.get_messages()
assert len(msgs) == 1
assert msgs[0].sender_id == "alice"
assert msgs[0].iteration == 0
assert msgs[0].payload == b"hello"
@pytest.mark.asyncio
async def test_message_count(store):
assert await store.message_count() == 0
await store.save_message("alice", 0, b"msg1")
await store.save_message("bob", 1, b"msg2")
assert await store.message_count() == 2
@pytest.mark.asyncio
async def test_get_messages_since(store):
await store.save_message("alice", 0, b"old")
all_msgs = await store.get_messages()
cutoff = all_msgs[0].timestamp
await store.save_message("bob", 1, b"new")
msgs = await store.get_messages(since=cutoff)
assert len(msgs) == 1
assert msgs[0].sender_id == "bob"
@pytest.mark.asyncio
async def test_the_thread_column_stays_for_an_older_node_and_its_index_goes(tmp_path):
"""Chat has no threads; a node downgraded to a version that still names
`thread_id` in every query must keep reading its rows."""
import sqlite3
path = tmp_path / "chat.db"
with sqlite3.connect(path) as c: # as an older node left it
c.executescript("""
CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT,
sender_id TEXT NOT NULL, iteration INTEGER NOT NULL,
payload BLOB NOT NULL, timestamp REAL NOT NULL,
thread_id TEXT DEFAULT NULL, sender_name TEXT DEFAULT '');
CREATE INDEX idx_messages_thread ON messages(thread_id);
INSERT INTO messages (sender_id, iteration, payload, timestamp, thread_id)
VALUES ('alice', 0, X'6869', 1.0, 't1');""")
s = ChatStore(path)
await s.open()
await s.save_message("bob", 1, b"now")
assert [m.payload for m in await s.get_recent()] == [b"hi", b"now"]
await s.close()
with sqlite3.connect(path) as c:
assert "thread_id" in [r[1] for r in c.execute("PRAGMA table_info(messages)")]
assert not c.execute("SELECT 1 FROM sqlite_master WHERE name = "
"'idx_messages_thread'").fetchone()
# What an older node selects still answers.
assert c.execute("SELECT thread_id FROM messages ORDER BY id").fetchall() == [
("t1",), (None,)]
@pytest.mark.asyncio
async def test_message_ordering(store):
for i in range(5):
await store.save_message(f"user-{i}", i, f"msg-{i}".encode())
msgs = await store.get_messages()
assert len(msgs) == 5
for i, m in enumerate(msgs):
assert m.sender_id == f"user-{i}"
@pytest.mark.asyncio
async def test_limit(store):
for i in range(10):
await store.save_message("alice", i, f"msg-{i}".encode())
msgs = await store.get_messages(limit=3)
assert len(msgs) == 3
@pytest.mark.asyncio
async def test_context_manager(tmp_path):
async with ChatStore(db_path=tmp_path / "ctx_test.db") as store:
await store.save_message("alice", 0, b"test")
assert await store.message_count() == 1
|