test_memory.py

"""
Tests for closeclaw.memory — SQLite conversation storage.
"""

import time
import pytest
from pathlib import Path

from closeclaw.memory import ConversationMemory, Conversation, Message


# ── Fixtures ───────────────────────────────────────────────────────────────────

@pytest.fixture
def mem(tmp_path):
    """A fresh ConversationMemory backed by a temp-dir DB."""
    db = tmp_path / "test_memory.db"
    m = ConversationMemory(db)
    yield m
    m.close()


# ── Lifecycle ──────────────────────────────────────────────────────────────────

class TestLifecycle:
    def test_db_file_created(self, tmp_path):
        db = tmp_path / "sub" / "memory.db"
        m = ConversationMemory(db)
        assert db.exists()
        m.close()

    def test_context_manager(self, tmp_path):
        db = tmp_path / "ctx.db"
        with ConversationMemory(db) as m:
            cid = m.new_conversation()
            assert cid > 0
        # connection closed — re-opening should still work
        with ConversationMemory(db) as m2:
            assert m2.get_conversation(cid) is not None

    def test_double_close_safe(self, mem):
        mem.close()
        mem.close()  # Should not raise


# ── new_conversation ───────────────────────────────────────────────────────────

class TestNewConversation:
    def test_returns_integer_id(self, mem):
        cid = mem.new_conversation()
        assert isinstance(cid, int) and cid > 0

    def test_ids_are_unique(self, mem):
        ids = [mem.new_conversation() for _ in range(5)]
        assert len(set(ids)) == 5

    def test_default_title_set(self, mem):
        cid = mem.new_conversation()
        conv = mem.get_conversation(cid)
        assert conv is not None
        assert conv.title  # non-empty

    def test_custom_title_and_model(self, mem):
        cid = mem.new_conversation(title="My Chat", model="llama3.2:3b")
        conv = mem.get_conversation(cid)
        assert conv.title == "My Chat"
        assert conv.model == "llama3.2:3b"

    def test_timestamps_set(self, mem):
        before = time.time()
        cid = mem.new_conversation()
        after = time.time()
        conv = mem.get_conversation(cid)
        assert before <= conv.created_at <= after
        assert before <= conv.updated_at <= after


# ── add_message ────────────────────────────────────────────────────────────────

class TestAddMessage:
    def test_message_persisted(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "Hello!")
        conv = mem.get_conversation(cid)
        assert len(conv.messages) == 1
        assert conv.messages[0].role == "user"
        assert conv.messages[0].content == "Hello!"

    def test_multiple_messages_ordered(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "First")
        mem.add_message(cid, "assistant", "Second")
        mem.add_message(cid, "user", "Third")
        conv = mem.get_conversation(cid)
        assert [m.content for m in conv.messages] == ["First", "Second", "Third"]

    def test_updated_at_advances(self, mem):
        cid = mem.new_conversation()
        conv_before = mem.get_conversation(cid)
        time.sleep(0.01)
        mem.add_message(cid, "user", "hi")
        conv_after = mem.get_conversation(cid)
        assert conv_after.updated_at > conv_before.updated_at

    def test_message_roles_preserved(self, mem):
        cid = mem.new_conversation()
        for role in ("user", "assistant", "tool"):
            mem.add_message(cid, role, f"content from {role}")
        conv = mem.get_conversation(cid)
        assert [m.role for m in conv.messages] == ["user", "assistant", "tool"]

    def test_multiline_content(self, mem):
        cid = mem.new_conversation()
        content = "Line 1\nLine 2\nLine 3"
        mem.add_message(cid, "assistant", content)
        conv = mem.get_conversation(cid)
        assert conv.messages[0].content == content


# ── get_conversation ───────────────────────────────────────────────────────────

class TestGetConversation:
    def test_returns_none_for_missing(self, mem):
        assert mem.get_conversation(9999) is None

    def test_returns_conversation_dataclass(self, mem):
        cid = mem.new_conversation(title="Test")
        conv = mem.get_conversation(cid)
        assert isinstance(conv, Conversation)
        assert conv.id == cid

    def test_messages_are_message_dataclasses(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "hello")
        conv = mem.get_conversation(cid)
        assert all(isinstance(m, Message) for m in conv.messages)

    def test_empty_conversation_has_no_messages(self, mem):
        cid = mem.new_conversation()
        conv = mem.get_conversation(cid)
        assert conv.messages == []


# ── update_title ───────────────────────────────────────────────────────────────

class TestUpdateTitle:
    def test_title_updated(self, mem):
        cid = mem.new_conversation(title="Old")
        mem.update_title(cid, "New Title")
        conv = mem.get_conversation(cid)
        assert conv.title == "New Title"

    def test_update_nonexistent_is_silent(self, mem):
        # Should not raise, just affects zero rows
        mem.update_title(9999, "Ghost")


# ── delete_conversation ────────────────────────────────────────────────────────

class TestDeleteConversation:
    def test_delete_existing(self, mem):
        cid = mem.new_conversation()
        assert mem.delete_conversation(cid) is True
        assert mem.get_conversation(cid) is None

    def test_delete_missing_returns_false(self, mem):
        assert mem.delete_conversation(9999) is False

    def test_messages_deleted_with_conversation(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "hi")
        mem.delete_conversation(cid)
        conv = mem.get_conversation(cid)
        assert conv is None

    def test_only_target_deleted(self, mem):
        cid1 = mem.new_conversation(title="Keep")
        cid2 = mem.new_conversation(title="Delete")
        mem.delete_conversation(cid2)
        assert mem.get_conversation(cid1) is not None
        assert mem.get_conversation(cid2) is None


# ── list_conversations ─────────────────────────────────────────────────────────

class TestListConversations:
    def test_empty_db_returns_empty(self, mem):
        assert mem.list_conversations() == []

    def test_returns_dicts(self, mem):
        mem.new_conversation()
        rows = mem.list_conversations()
        assert len(rows) == 1
        assert isinstance(rows[0], dict)

    def test_ordered_by_updated_desc(self, mem):
        c1 = mem.new_conversation(title="First")
        time.sleep(0.01)
        c2 = mem.new_conversation(title="Second")
        rows = mem.list_conversations()
        assert rows[0]["id"] == c2
        assert rows[1]["id"] == c1

    def test_limit_and_offset(self, mem):
        for i in range(10):
            mem.new_conversation(title=f"Conv {i}")
        assert len(mem.list_conversations(limit=3)) == 3
        assert len(mem.list_conversations(limit=5, offset=7)) == 3  # 10 - 7 = 3

    def test_message_count_field(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "a")
        mem.add_message(cid, "assistant", "b")
        rows = mem.list_conversations()
        assert rows[0]["message_count"] == 2

    def test_required_fields_present(self, mem):
        mem.new_conversation(title="t", model="m")
        row = mem.list_conversations()[0]
        for field in ("id", "title", "model", "created_at", "updated_at", "message_count"):
            assert field in row


# ── search ─────────────────────────────────────────────────────────────────────

class TestSearch:
    def test_empty_query_returns_empty(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "hello world")
        assert mem.search("") == []

    def test_basic_keyword_match(self, mem):
        cid = mem.new_conversation(title="Chat 1")
        mem.add_message(cid, "user", "Tell me about Python")
        results = mem.search("Python")
        assert len(results) == 1
        assert results[0]["id"] == cid

    def test_case_insensitive(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "HELLO WORLD")
        assert len(mem.search("hello")) == 1
        assert len(mem.search("HELLO")) == 1
        assert len(mem.search("Hello")) == 1

    def test_no_match_returns_empty(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "apples and oranges")
        assert mem.search("banana") == []

    def test_matches_across_multiple_conversations(self, mem):
        for i in range(3):
            cid = mem.new_conversation()
            mem.add_message(cid, "user", f"question {i} about cats")
        results = mem.search("cats")
        assert len(results) == 3

    def test_only_matching_conversation_returned(self, mem):
        c1 = mem.new_conversation()
        mem.add_message(c1, "user", "I love cats")
        c2 = mem.new_conversation()
        mem.add_message(c2, "user", "I love dogs")
        results = mem.search("cats")
        assert len(results) == 1
        assert results[0]["id"] == c1

    def test_limit_respected(self, mem):
        for i in range(10):
            cid = mem.new_conversation()
            mem.add_message(cid, "user", f"keyword match {i}")
        results = mem.search("keyword", limit=3)
        assert len(results) == 3

    def test_special_like_chars_escaped(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "100% complete")
        # Should match literal "%" without SQL injection
        results = mem.search("100%")
        assert len(results) == 1

    def test_matched_snippet_field(self, mem):
        cid = mem.new_conversation()
        mem.add_message(cid, "user", "testing snippet extraction")
        results = mem.search("snippet")
        assert "matched_snippet" in results[0]
        assert "snippet" in results[0]["matched_snippet"].lower()

    def test_search_result_fields(self, mem):
        cid = mem.new_conversation(title="t", model="llama")
        mem.add_message(cid, "user", "hello")
        row = mem.search("hello")[0]
        for field in ("id", "title", "model", "created_at", "updated_at", "matched_snippet"):
            assert field in row