125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from javis.memory.sqlite_store import SessionNotFoundError, SQLiteSessionStore
|
|
|
|
|
|
class SQLiteSessionStoreTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary_directory = tempfile.TemporaryDirectory()
|
|
database = Path(self.temporary_directory.name) / "sessions.sqlite3"
|
|
self.store = SQLiteSessionStore(database)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary_directory.cleanup()
|
|
|
|
def test_create_list_and_load_session(self) -> None:
|
|
created = self.store.create_session("ollama", "test-model")
|
|
loaded = self.store.get_session(created.id)
|
|
|
|
self.assertEqual(loaded.id, created.id)
|
|
self.assertEqual(loaded.provider, "ollama")
|
|
self.assertEqual(loaded.model, "test-model")
|
|
self.assertEqual(loaded.title, "Neue Sitzung")
|
|
self.assertEqual(loaded.last_provider, "ollama")
|
|
self.assertEqual(self.store.list_sessions()[0].id, created.id)
|
|
|
|
def test_messages_keep_exchange_order(self) -> None:
|
|
session = self.store.create_session("ollama", "test-model")
|
|
self.store.append_exchange(session.id, "eins", "zwei")
|
|
self.store.append_exchange(session.id, "drei", "vier")
|
|
|
|
messages = self.store.get_messages(session.id)
|
|
|
|
self.assertEqual(
|
|
[(message.role, message.content) for message in messages],
|
|
[
|
|
("user", "eins"),
|
|
("assistant", "zwei"),
|
|
("user", "drei"),
|
|
("assistant", "vier"),
|
|
],
|
|
)
|
|
|
|
def test_unknown_session_raises_clear_error(self) -> None:
|
|
with self.assertRaises(SessionNotFoundError):
|
|
self.store.get_session("missing")
|
|
|
|
def test_clear_keeps_session_and_removes_messages(self) -> None:
|
|
session = self.store.create_session("ollama", "test-model")
|
|
self.store.append_exchange(session.id, "eins", "zwei")
|
|
|
|
self.store.clear_messages(session.id)
|
|
|
|
self.assertEqual(self.store.get_messages(session.id), [])
|
|
self.assertEqual(self.store.get_session(session.id).message_count, 0)
|
|
|
|
def test_first_exchange_sets_title_and_last_provider_atomically(self) -> None:
|
|
session = self.store.create_session("hybrid", "auto")
|
|
|
|
self.store.append_exchange(
|
|
session.id,
|
|
"Frage",
|
|
"Antwort",
|
|
title_if_first="Kurzer Titel",
|
|
last_provider="gemini",
|
|
)
|
|
self.store.append_exchange(
|
|
session.id,
|
|
"Noch eine Frage",
|
|
"Noch eine Antwort",
|
|
title_if_first="Darf Titel nicht ersetzen",
|
|
last_provider="ollama",
|
|
)
|
|
|
|
loaded = self.store.get_session(session.id)
|
|
self.assertEqual(loaded.title, "Kurzer Titel")
|
|
self.assertEqual(loaded.last_provider, "ollama")
|
|
|
|
def test_old_database_is_migrated_idempotently_without_data_loss(self) -> None:
|
|
database = Path(self.temporary_directory.name) / "legacy.sqlite3"
|
|
with sqlite3.connect(database) as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE sessions (
|
|
id TEXT PRIMARY KEY,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
provider TEXT NOT NULL,
|
|
model TEXT NOT NULL
|
|
);
|
|
CREATE TABLE messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
session_id TEXT NOT NULL,
|
|
position INTEGER NOT NULL,
|
|
role TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
UNIQUE (session_id, position)
|
|
);
|
|
INSERT INTO sessions VALUES
|
|
('legacy', '2026-01-01', '2026-01-02', 'ollama', 'old-model');
|
|
INSERT INTO messages
|
|
(session_id, position, role, content, created_at)
|
|
VALUES ('legacy', 0, 'user', 'Alte Nachricht', '2026-01-02');
|
|
"""
|
|
)
|
|
connection.close()
|
|
|
|
migrated = SQLiteSessionStore(database)
|
|
migrated_again = SQLiteSessionStore(database)
|
|
|
|
session = migrated_again.get_session("legacy")
|
|
self.assertEqual(session.title, "Neue Sitzung")
|
|
self.assertEqual(session.last_provider, "ollama")
|
|
self.assertEqual(
|
|
[message.content for message in migrated.get_messages("legacy")],
|
|
["Alte Nachricht"],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|