Blame for test_db.py

"""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()