125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
"""Tests for SQLite-backed chat persistence (issue #46)."""
|
|
|
|
from dashboard.store import Message, MessageLog
|
|
import infrastructure.chat_store as _chat_store
|
|
|
|
|
|
def test_persistence_across_instances(tmp_path):
|
|
"""Messages survive creating a new MessageLog pointing at the same DB."""
|
|
db = tmp_path / "chat.db"
|
|
log1 = MessageLog(db_path=db)
|
|
log1.append(role="user", content="hello", timestamp="10:00:00", source="browser")
|
|
log1.append(role="agent", content="hi back", timestamp="10:00:01", source="browser")
|
|
log1.close()
|
|
|
|
# New instance — simulates server restart
|
|
log2 = MessageLog(db_path=db)
|
|
msgs = log2.all()
|
|
assert len(msgs) == 2
|
|
assert msgs[0].role == "user"
|
|
assert msgs[0].content == "hello"
|
|
assert msgs[1].role == "agent"
|
|
assert msgs[1].content == "hi back"
|
|
log2.close()
|
|
|
|
|
|
def test_retention_policy(tmp_path):
|
|
"""Oldest messages are pruned when count exceeds MAX_MESSAGES."""
|
|
original_max = _chat_store.MAX_MESSAGES
|
|
_chat_store.MAX_MESSAGES = 5 # Small limit for testing
|
|
|
|
try:
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
for i in range(8):
|
|
log.append(role="user", content=f"msg-{i}", timestamp=f"10:00:{i:02d}")
|
|
|
|
assert len(log) == 5
|
|
msgs = log.all()
|
|
# Oldest 3 should have been pruned
|
|
assert msgs[0].content == "msg-3"
|
|
assert msgs[-1].content == "msg-7"
|
|
log.close()
|
|
finally:
|
|
_chat_store.MAX_MESSAGES = original_max
|
|
|
|
|
|
def test_clear_removes_all(tmp_path):
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
log.append(role="user", content="data", timestamp="12:00:00")
|
|
assert len(log) == 1
|
|
log.clear()
|
|
assert len(log) == 0
|
|
assert log.all() == []
|
|
log.close()
|
|
|
|
|
|
def test_recent_returns_limited_newest(tmp_path):
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
for i in range(10):
|
|
log.append(role="user", content=f"msg-{i}", timestamp=f"10:00:{i:02d}")
|
|
|
|
recent = log.recent(limit=3)
|
|
assert len(recent) == 3
|
|
# Should be oldest-first within the window
|
|
assert recent[0].content == "msg-7"
|
|
assert recent[1].content == "msg-8"
|
|
assert recent[2].content == "msg-9"
|
|
log.close()
|
|
|
|
|
|
def test_source_field_persisted(tmp_path):
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
log.append(role="user", content="from api", timestamp="10:00:00", source="api")
|
|
log.append(role="user", content="from tg", timestamp="10:00:01", source="telegram")
|
|
log.close()
|
|
|
|
log2 = MessageLog(db_path=db)
|
|
msgs = log2.all()
|
|
assert msgs[0].source == "api"
|
|
assert msgs[1].source == "telegram"
|
|
log2.close()
|
|
|
|
|
|
def test_message_dataclass_defaults():
|
|
m = Message(role="user", content="hi", timestamp="12:00:00")
|
|
assert m.source == "browser"
|
|
|
|
|
|
def test_empty_db_returns_empty(tmp_path):
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
assert log.all() == []
|
|
assert len(log) == 0
|
|
assert log.recent() == []
|
|
log.close()
|
|
|
|
|
|
def test_concurrent_appends(tmp_path):
|
|
"""Multiple threads can append without corrupting data."""
|
|
import threading
|
|
|
|
db = tmp_path / "chat.db"
|
|
log = MessageLog(db_path=db)
|
|
errors = []
|
|
|
|
def writer(thread_id):
|
|
try:
|
|
for i in range(20):
|
|
log.append(role="user", content=f"t{thread_id}-{i}", timestamp="10:00:00")
|
|
except Exception as e:
|
|
errors.append(e)
|
|
|
|
threads = [threading.Thread(target=writer, args=(t,)) for t in range(4)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert not errors
|
|
assert len(log) == 80
|
|
log.close()
|