"""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) # ── Minimal embedding stub ────────────────────────────────────────────── 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]] # ── SQLite-based DatabaseManager stub that mirrors the real API ───────── 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() # -- connection helpers ------------------------------------------------ def _conn(self): return sqlite3.connect(self._db_path) # -- schema ------------------------------------------------------------ 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() # -- users ------------------------------------------------------------- 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 # -- conversations & messages ------------------------------------------ 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 # -- OpenWebUI integration --------------------------------------------- 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) ] # -- profile ----------------------------------------------------------- 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() # -- semantic memory --------------------------------------------------- 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] # -- episodic memory (conversation summaries) -------------------------- 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() # -- semantic memory (user facts) -------------------------------------- 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 # -- guest / cleanup (stubs for test) ---------------------------------- def create_guest_user(self, ip: str) -> int: return 0 def cleanup_expired_guests(self, hours: int = 48): pass def close(self): pass # ── Test runner ───────────────────────────────────────────────────────── def run(): print("=== DB smoke-test ===\n") db = SQLiteDatabaseManager(TEST_DB) # 1. Users & auth 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}") # 2. Conversations & messages 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)") # 3. Working memory — get_recent_messages (backward compat) recent = db.get_recent_messages(uid) assert len(recent) == 2 print(f"[OK] recent messages across conversations ({len(recent)} messages)") # 4. Working memory — conversation-scoped turns 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)") # 5. Profile 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}") # 6. Semantic memory (legacy) 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]}" ) # 7. User facts (new semantic memory) 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)") # 7a. Deduplication — same content should reinforce, not duplicate 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)") # 7b. Delete deleted = db.delete_user_fact(uid, "Enjoys long walks.") assert deleted assert len(db.get_user_facts(uid)) == 2 print("[OK] user fact deletion") # 8. Conversation summary (episodic memory) 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]}") # 8a. Upsert — update should not create a duplicate 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)") # 9. Verify context assembly inputs 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") # 10. OpenWebUI integration 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}") # 10a. Same email returns same user 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") # 10b. Conversation mapping 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}") # 10c. Same chat ID returns same conversation 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") # 10d. Messages work in mapped conversation 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()