| """Smoke tests for database operations and the new memory-layer methods. |
| |
| Uses a local SQLite database so no PostgreSQL or sentence_transformers |
| dependencies are required. The test exercises the same DatabaseManager |
| code paths that production uses, adapted here via a thin SQLite wrapper |
| that mirrors its public API. |
| """ |
|
|
| import json |
| import os |
| import sqlite3 |
| import sys |
|
|
| sys.path.insert(0, os.path.dirname(__file__)) |
|
|
| TEST_DB = "data/test_montaigne.db" |
| os.makedirs(os.path.dirname(TEST_DB), exist_ok=True) |
|
|
| if os.path.exists(TEST_DB): |
| os.remove(TEST_DB) |
|
|
|
|
| |
| def _fake_embed(text: str) -> list[float]: |
| """Deterministic 8-dim vector so cosine similarity still works.""" |
| import hashlib |
|
|
| h = hashlib.sha256(text.encode()).digest() |
| return [b / 255.0 for b in h[:8]] |
|
|
|
|
| |
| class SQLiteDatabaseManager: |
| """Thin SQLite wrapper exposing the same public methods the test uses.""" |
|
|
| def __init__(self, db_path: str): |
| self._db_path = db_path |
| self._init_schema() |
|
|
| |
| def _conn(self): |
| return sqlite3.connect(self._db_path) |
|
|
| |
| def _init_schema(self): |
| conn = self._conn() |
| c = conn.cursor() |
| c.executescript(""" |
| CREATE TABLE IF NOT EXISTS users ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| username TEXT UNIQUE NOT NULL, |
| password_hash TEXT NOT NULL, |
| is_guest INTEGER DEFAULT 0, |
| guest_ip TEXT, |
| is_admin INTEGER DEFAULT 0, |
| email TEXT, |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE TABLE IF NOT EXISTS conversations ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| user_id INTEGER NOT NULL REFERENCES users(id), |
| title TEXT, |
| openwebui_chat_id TEXT, |
| started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE TABLE IF NOT EXISTS messages ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| conversation_id INTEGER NOT NULL REFERENCES conversations(id), |
| role TEXT NOT NULL CHECK(role IN ('user','assistant')), |
| content TEXT NOT NULL, |
| timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE TABLE IF NOT EXISTS user_profiles ( |
| user_id INTEGER PRIMARY KEY REFERENCES users(id), |
| summary_text TEXT DEFAULT '', |
| last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE TABLE IF NOT EXISTS memories ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| user_id INTEGER NOT NULL REFERENCES users(id), |
| content TEXT NOT NULL, |
| embedding TEXT NOT NULL, |
| metadata TEXT DEFAULT '{}', |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE TABLE IF NOT EXISTS conversation_summaries ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| conversation_id INTEGER NOT NULL REFERENCES conversations(id), |
| user_id INTEGER NOT NULL REFERENCES users(id), |
| summary_text TEXT NOT NULL, |
| message_count INTEGER DEFAULT 0, |
| updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP |
| ); |
| CREATE UNIQUE INDEX IF NOT EXISTS idx_conv_summaries_unique_conv |
| ON conversation_summaries(conversation_id); |
| CREATE TABLE IF NOT EXISTS user_facts ( |
| id INTEGER PRIMARY KEY AUTOINCREMENT, |
| user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, |
| category TEXT NOT NULL DEFAULT 'general', |
| content TEXT NOT NULL, |
| importance REAL DEFAULT 0.5, |
| confidence REAL DEFAULT 0.8, |
| source_conv_id INTEGER REFERENCES conversations(id), |
| created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| last_reinforced TIMESTAMP DEFAULT CURRENT_TIMESTAMP, |
| is_pinned INTEGER DEFAULT 0 |
| ); |
| CREATE UNIQUE INDEX IF NOT EXISTS idx_user_facts_unique_content |
| ON user_facts(user_id, content); |
| """) |
| conn.commit() |
| conn.close() |
|
|
| |
| def create_user(self, username: str, password_hash: str) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "INSERT INTO users (username, password_hash) VALUES (?, ?) RETURNING id", |
| (username, password_hash), |
| ) |
| uid = c.fetchone()[0] |
| c.execute( |
| "INSERT INTO user_profiles (user_id, summary_text) VALUES (?, '')", (uid,) |
| ) |
| conn.commit() |
| conn.close() |
| return uid |
|
|
| def get_user_by_username(self, username: str): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT id, username, password_hash FROM users WHERE username = ?", |
| (username,), |
| ) |
| row = c.fetchone() |
| conn.close() |
| return ( |
| {"id": row[0], "username": row[1], "password_hash": row[2]} if row else None |
| ) |
|
|
| def verify_password(self, username: str, password: str): |
| from backend.auth import verify_password as _verify |
|
|
| user = self.get_user_by_username(username) |
| if user and _verify(password, user["password_hash"]): |
| return user["id"] |
| return None |
|
|
| |
| def create_conversation(self, user_id: int) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "INSERT INTO conversations (user_id) VALUES (?) RETURNING id", (user_id,) |
| ) |
| cid = c.fetchone()[0] |
| conn.commit() |
| conn.close() |
| return cid |
|
|
| |
| def get_user_by_email(self, email: str): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT id, username, password_hash, is_guest, email FROM users WHERE email = ?", |
| (email,), |
| ) |
| row = c.fetchone() |
| conn.close() |
| if not row: |
| return None |
| return { |
| "id": row[0], |
| "username": row[1], |
| "password_hash": row[2], |
| "is_guest": bool(row[3]), |
| "email": row[4], |
| } |
|
|
| def create_oauth_user(self, username: str, email: str = None) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "INSERT INTO users (username, password_hash, email) VALUES (?, '', ?) RETURNING id", |
| (username, email), |
| ) |
| uid = c.fetchone()[0] |
| c.execute( |
| "INSERT INTO user_profiles (user_id, summary_text) VALUES (?, '')", (uid,) |
| ) |
| conn.commit() |
| conn.close() |
| return uid |
|
|
| def get_or_create_openwebui_user(self, email: str) -> int: |
| existing = self.get_user_by_email(email) |
| if existing and not existing.get("is_guest"): |
| return existing["id"] |
| username = email.split("@")[0] |
| if self.get_user_by_username(username): |
| import secrets |
|
|
| username = f"{username}_{secrets.token_hex(4)}" |
| return self.create_oauth_user(username, email) |
|
|
| def get_or_create_conversation_by_openwebui_chat( |
| self, user_id: int, openwebui_chat_id: str |
| ) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT id FROM conversations WHERE openwebui_chat_id = ?", |
| (openwebui_chat_id,), |
| ) |
| row = c.fetchone() |
| if row: |
| conv_id = row[0] |
| else: |
| c.execute( |
| "INSERT INTO conversations (user_id, openwebui_chat_id) VALUES (?, ?) RETURNING id", |
| (user_id, openwebui_chat_id), |
| ) |
| conv_id = c.fetchone()[0] |
| c.execute( |
| "INSERT INTO user_profiles (user_id, summary_text) VALUES (?, '') ON CONFLICT (user_id) DO NOTHING", |
| (user_id,), |
| ) |
| conn.commit() |
| conn.close() |
| return conv_id |
|
|
| def add_message(self, conversation_id: int, role: str, content: str) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "INSERT INTO messages (conversation_id, role, content) VALUES (?, ?, ?) RETURNING id", |
| (conversation_id, role, content), |
| ) |
| mid = c.fetchone()[0] |
| conn.commit() |
| conn.close() |
| return mid |
|
|
| def get_conversation_messages(self, conversation_id: int, limit: int = 50): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT role, content, timestamp FROM messages WHERE conversation_id = ? ORDER BY timestamp ASC LIMIT ?", |
| (conversation_id, limit), |
| ) |
| rows = c.fetchall() |
| conn.close() |
| return [{"role": r[0], "content": r[1], "timestamp": str(r[2])} for r in rows] |
|
|
| def get_recent_messages(self, user_id: int, limit: int = 10): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| """SELECT m.role, m.content, m.timestamp |
| FROM messages m JOIN conversations c ON m.conversation_id = c.id |
| WHERE c.user_id = ? ORDER BY m.timestamp DESC LIMIT ?""", |
| (user_id, limit), |
| ) |
| rows = c.fetchall() |
| conn.close() |
| return [ |
| {"role": r[0], "content": r[1], "timestamp": str(r[2])} |
| for r in reversed(rows) |
| ] |
|
|
| def count_conversation_messages(self, conversation_id: int) -> int: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT COUNT(*) FROM messages WHERE conversation_id = ?", |
| (conversation_id,), |
| ) |
| count = c.fetchone()[0] |
| conn.close() |
| return count |
|
|
| def get_conversation_turns( |
| self, conversation_id: int, limit: int = 20, offset_from_end: int = 0 |
| ): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| """SELECT role, content, timestamp, id FROM messages |
| WHERE conversation_id = ? ORDER BY id DESC LIMIT ? OFFSET ?""", |
| (conversation_id, limit, offset_from_end), |
| ) |
| rows = c.fetchall() |
| conn.close() |
| return [ |
| {"role": r[0], "content": r[1], "timestamp": str(r[2]), "id": r[3]} |
| for r in reversed(rows) |
| ] |
|
|
| |
| def get_user_profile(self, user_id: int) -> str: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT summary_text FROM user_profiles WHERE user_id = ?", (user_id,) |
| ) |
| row = c.fetchone() |
| conn.close() |
| return row[0] if row else "" |
|
|
| def update_user_profile(self, user_id: int, summary_text: str): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "UPDATE user_profiles SET summary_text = ?, last_updated = CURRENT_TIMESTAMP WHERE user_id = ?", |
| (summary_text, user_id), |
| ) |
| conn.commit() |
| conn.close() |
|
|
| |
| def add_memory(self, user_id: int, text: str, metadata=None): |
| emb = json.dumps(_fake_embed(text)) |
| meta = json.dumps(metadata or {}) |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "INSERT INTO memories (user_id, content, embedding, metadata) VALUES (?, ?, ?, ?)", |
| (user_id, text, emb, meta), |
| ) |
| conn.commit() |
| conn.close() |
|
|
| def search_memories(self, user_id: int, query: str, n_results: int = 5): |
| import numpy as np |
|
|
| query_vec = np.array(_fake_embed(query)) |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT content, embedding, metadata FROM memories WHERE user_id = ?", |
| (user_id,), |
| ) |
| rows = c.fetchall() |
| conn.close() |
|
|
| scored = [] |
| for content, emb_json, meta_json in rows: |
| vec = np.array(json.loads(emb_json)) |
| sim = float( |
| np.dot(query_vec, vec) |
| / (np.linalg.norm(query_vec) * np.linalg.norm(vec)) |
| ) |
| scored.append( |
| { |
| "content": content, |
| "metadata": json.loads(meta_json), |
| "similarity": sim, |
| } |
| ) |
| scored.sort(key=lambda x: x["similarity"], reverse=True) |
| return [m for m in scored[:n_results] if m["similarity"] > 0.3] |
|
|
| |
| def get_conversation_summary(self, conversation_id: int): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "SELECT summary_text FROM conversation_summaries WHERE conversation_id = ? ORDER BY updated_at DESC LIMIT 1", |
| (conversation_id,), |
| ) |
| row = c.fetchone() |
| conn.close() |
| return row[0] if row else None |
|
|
| def save_conversation_summary( |
| self, |
| conversation_id: int, |
| user_id: int, |
| summary_text: str, |
| message_count: int = 0, |
| ): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| """INSERT INTO conversation_summaries (conversation_id, user_id, summary_text, message_count) |
| VALUES (?, ?, ?, ?) |
| ON CONFLICT (conversation_id) DO UPDATE SET |
| summary_text = excluded.summary_text, |
| message_count = excluded.message_count, |
| updated_at = CURRENT_TIMESTAMP""", |
| (conversation_id, user_id, summary_text, message_count), |
| ) |
| conn.commit() |
| conn.close() |
|
|
| |
| def add_user_fact( |
| self, |
| user_id: int, |
| content: str, |
| category: str = "general", |
| importance: float = 0.5, |
| source_conv_id=None, |
| ): |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| """INSERT INTO user_facts (user_id, category, content, importance, source_conv_id) |
| VALUES (?, ?, ?, ?, ?) |
| ON CONFLICT (user_id, content) DO UPDATE SET |
| category = excluded.category, |
| importance = MAX(user_facts.importance, excluded.importance), |
| last_reinforced = CURRENT_TIMESTAMP""", |
| (user_id, category, content, importance, source_conv_id), |
| ) |
| conn.commit() |
| conn.close() |
|
|
| def get_user_facts(self, user_id: int, limit: int = 8, category=None): |
| conn = self._conn() |
| c = conn.cursor() |
| if category: |
| c.execute( |
| "SELECT category, content, importance, is_pinned FROM user_facts WHERE user_id = ? AND category = ? " |
| "ORDER BY is_pinned DESC, importance DESC, last_reinforced DESC LIMIT ?", |
| (user_id, category, limit), |
| ) |
| else: |
| c.execute( |
| "SELECT category, content, importance, is_pinned FROM user_facts WHERE user_id = ? " |
| "ORDER BY is_pinned DESC, importance DESC, last_reinforced DESC LIMIT ?", |
| (user_id, limit), |
| ) |
| rows = c.fetchall() |
| conn.close() |
| return [ |
| {"category": r[0], "content": r[1], "importance": r[2], "is_pinned": r[3]} |
| for r in rows |
| ] |
|
|
| def delete_user_fact(self, user_id: int, content: str) -> bool: |
| conn = self._conn() |
| c = conn.cursor() |
| c.execute( |
| "DELETE FROM user_facts WHERE user_id = ? AND content = ?", |
| (user_id, content), |
| ) |
| deleted = c.rowcount > 0 |
| conn.commit() |
| conn.close() |
| return deleted |
|
|
| |
| def create_guest_user(self, ip: str) -> int: |
| return 0 |
|
|
| def cleanup_expired_guests(self, hours: int = 48): |
| pass |
|
|
| def close(self): |
| pass |
|
|
|
|
| |
| def run(): |
| print("=== DB smoke-test ===\n") |
| db = SQLiteDatabaseManager(TEST_DB) |
|
|
| |
| from backend.auth import get_password_hash, verify_password as auth_verify |
|
|
| pw_hash = get_password_hash("hello123") |
| uid = db.create_user("testuser", pw_hash) |
| assert uid > 0 |
| print(f"[OK] created user id={uid}") |
|
|
| assert auth_verify("hello123", pw_hash), "password verify failed" |
| assert not auth_verify("wrong", pw_hash), "wrong password should fail" |
| print("[OK] password hashing + verification") |
|
|
| verified_id = db.verify_password("testuser", "hello123") |
| assert verified_id == uid |
| print(f"[OK] db.verify_password -> {verified_id}") |
|
|
| |
| cid = db.create_conversation(uid) |
| assert cid > 0 |
| print(f"[OK] created conversation id={cid}") |
|
|
| db.add_message(cid, "user", "Hello Montaigne!") |
| db.add_message(cid, "assistant", "Ah, welcome, friend.") |
| msgs = db.get_conversation_messages(cid) |
| assert len(msgs) == 2 |
| print(f"[OK] messages stored/retrieved ({len(msgs)} messages)") |
|
|
| |
| recent = db.get_recent_messages(uid) |
| assert len(recent) == 2 |
| print(f"[OK] recent messages across conversations ({len(recent)} messages)") |
|
|
| |
| turns = db.get_conversation_turns(cid) |
| assert len(turns) == 2 |
| assert turns[0]["content"] == "Hello Montaigne!" |
| print(f"[OK] conversation-scoped turns ({len(turns)} turns)") |
|
|
| |
| db.update_user_profile(uid, "Interested in Stoicism and friendship.") |
| profile = db.get_user_profile(uid) |
| assert "Stoicism" in profile |
| print(f"[OK] user profile: {profile!r}") |
|
|
| |
| db.add_memory(uid, "The user finds joy in long walks and reading Plutarch.") |
| db.add_memory(uid, "The user dislikes crowded places and noise.") |
| results = db.search_memories(uid, "Does the user enjoy nature?") |
| assert len(results) > 0, "expected at least one memory hit" |
| print(f"[OK] semantic search returned {len(results)} result(s)") |
| print( |
| f" top hit (sim={results[0]['similarity']:.3f}): {results[0]['content'][:60]}" |
| ) |
|
|
| |
| db.add_user_fact( |
| uid, |
| "Likes Stoicism.", |
| category="preference", |
| importance=0.8, |
| source_conv_id=cid, |
| ) |
| db.add_user_fact( |
| uid, "Name is Marcus.", category="name", importance=0.9, source_conv_id=cid |
| ) |
| db.add_user_fact( |
| uid, "Enjoys long walks.", category="habit", importance=0.5, source_conv_id=cid |
| ) |
| facts = db.get_user_facts(uid) |
| assert len(facts) == 3 |
| assert facts[0]["category"] == "name", "facts should be sorted by importance DESC" |
| assert facts[1]["category"] == "preference" |
| print(f"[OK] user facts ({len(facts)} stored, sorted by importance)") |
|
|
| |
| db.add_user_fact( |
| uid, |
| "Likes Stoicism.", |
| category="preference", |
| importance=0.9, |
| source_conv_id=cid, |
| ) |
| facts_after = db.get_user_facts(uid) |
| assert len(facts_after) == 3, "duplicate fact should not create a new row" |
| stoicism = next(f for f in facts_after if f["content"] == "Likes Stoicism.") |
| assert stoicism["importance"] == 0.9, "reinforcement should use MAX(importance)" |
| print("[OK] user facts deduplication (ON CONFLICT update)") |
|
|
| |
| deleted = db.delete_user_fact(uid, "Enjoys long walks.") |
| assert deleted |
| assert len(db.get_user_facts(uid)) == 2 |
| print("[OK] user fact deletion") |
|
|
| |
| assert db.get_conversation_summary(cid) is None |
| db.save_conversation_summary( |
| cid, uid, "Discussing friendship and doubts about identity.", message_count=4 |
| ) |
| summary = db.get_conversation_summary(cid) |
| assert summary is not None |
| assert "friendship" in summary |
| print(f"[OK] conversation summary: {summary[:60]}") |
|
|
| |
| db.save_conversation_summary( |
| cid, uid, "Extended to also cover the nature of death.", message_count=8 |
| ) |
| summary2 = db.get_conversation_summary(cid) |
| assert "death" in summary2 |
| print("[OK] conversation summary upsert (ON CONFLICT update)") |
|
|
| |
| turns = db.get_conversation_turns(cid) |
| facts = db.get_user_facts(uid, limit=3) |
| summary = db.get_conversation_summary(cid) |
| profile = db.get_user_profile(uid) |
| assert len(turns) == 2 |
| assert len(facts) == 2 |
| assert summary is not None |
| assert "Stoicism" in profile |
| print("[OK] all memory layers return data for context assembly") |
|
|
| |
| ow_uid = db.get_or_create_openwebui_user("testuser@example.com") |
| assert ow_uid > 0 |
| print(f"[OK] OpenWebUI user auto-provisioned id={ow_uid}") |
|
|
| |
| ow_uid2 = db.get_or_create_openwebui_user("testuser@example.com") |
| assert ow_uid2 == ow_uid, "same email should return same user" |
| print("[OK] OpenWebUI user deduplication by email") |
|
|
| |
| ow_cid = db.get_or_create_conversation_by_openwebui_chat(ow_uid, "chat-abc-123") |
| assert ow_cid > 0 |
| print(f"[OK] OpenWebUI conversation mapped id={ow_cid}") |
|
|
| |
| ow_cid2 = db.get_or_create_conversation_by_openwebui_chat(ow_uid, "chat-abc-123") |
| assert ow_cid2 == ow_cid, "same chat ID should return same conversation" |
| print("[OK] OpenWebUI conversation deduplication by chat ID") |
|
|
| |
| db.add_message(ow_cid, "user", "What is friendship?") |
| db.add_message(ow_cid, "assistant", "A single soul dwelling in two bodies.") |
| ow_turns = db.get_conversation_turns(ow_cid) |
| assert len(ow_turns) == 2 |
| print(f"[OK] messages in OpenWebUI-mapped conversation ({len(ow_turns)} turns)") |
|
|
| db.close() |
| print("\n=== All tests passed ===") |
|
|
|
|
| if __name__ == "__main__": |
| run() |
|
|
