Timmy-time-dashboard/tests/timmy/nexus/test_persistence.py

"""Tests for the Nexus Session Persistence store."""

import pytest

from timmy.nexus.persistence import MAX_MESSAGES, NexusStore


@pytest.fixture
def store(tmp_path):
    """Provide a NexusStore backed by a temp database."""
    db = tmp_path / "test_nexus.db"
    s = NexusStore(db_path=db)
    yield s
    s.close()


class TestNexusStoreBasic:
    def test_append_and_retrieve(self, store):
        store.append("user", "hello")
        store.append("assistant", "hi there")
        history = store.get_history()
        assert len(history) == 2
        assert history[0]["role"] == "user"
        assert history[0]["content"] == "hello"
        assert history[1]["role"] == "assistant"

    def test_message_count(self, store):
        assert store.message_count() == 0
        store.append("user", "a")
        store.append("user", "b")
        assert store.message_count() == 2

    def test_custom_timestamp(self, store):
        store.append("user", "msg", timestamp="12:34:56")
        history = store.get_history()
        assert history[0]["timestamp"] == "12:34:56"

    def test_clear_session(self, store):
        store.append("user", "a")
        store.append("assistant", "b")
        deleted = store.clear()
        assert deleted == 2
        assert store.message_count() == 0

    def test_clear_empty_session(self, store):
        deleted = store.clear()
        assert deleted == 0

    def test_clear_all(self, store):
        store.append("user", "a", session_tag="s1")
        store.append("user", "b", session_tag="s2")
        deleted = store.clear_all()
        assert deleted == 2
        assert store.message_count(session_tag="s1") == 0
        assert store.message_count(session_tag="s2") == 0


class TestNexusStoreOrdering:
    def test_chronological_order(self, store):
        for i in range(5):
            store.append("user", f"msg-{i}")
        history = store.get_history()
        contents = [m["content"] for m in history]
        assert contents == ["msg-0", "msg-1", "msg-2", "msg-3", "msg-4"]

    def test_limit_parameter(self, store):
        for i in range(10):
            store.append("user", f"msg-{i}")
        history = store.get_history(limit=3)
        assert len(history) == 3
        # Should be the 3 most recent
        assert history[0]["content"] == "msg-7"
        assert history[2]["content"] == "msg-9"


class TestNexusStoreSessionTags:
    def test_session_isolation(self, store):
        store.append("user", "nexus-msg", session_tag="nexus")
        store.append("user", "other-msg", session_tag="other")
        nexus_history = store.get_history(session_tag="nexus")
        other_history = store.get_history(session_tag="other")
        assert len(nexus_history) == 1
        assert len(other_history) == 1
        assert nexus_history[0]["content"] == "nexus-msg"

    def test_clear_only_affects_target_session(self, store):
        store.append("user", "a", session_tag="s1")
        store.append("user", "b", session_tag="s2")
        store.clear(session_tag="s1")
        assert store.message_count(session_tag="s1") == 0
        assert store.message_count(session_tag="s2") == 1


class TestNexusStorePruning:
    def test_prune_excess_messages(self, tmp_path):
        """Inserting beyond MAX_MESSAGES should prune oldest."""
        db = tmp_path / "prune_test.db"
        s = NexusStore(db_path=db)
        # Insert MAX_MESSAGES + 5 to trigger pruning
        for i in range(MAX_MESSAGES + 5):
            s.append("user", f"msg-{i}")
        assert s.message_count() == MAX_MESSAGES
        # Get full history — oldest remaining should be msg-5
        history = s.get_history(limit=MAX_MESSAGES)
        assert history[0]["content"] == "msg-5"
        s.close()


class TestNexusStoreReopen:
    def test_data_survives_close_reopen(self, tmp_path):
        """Data persists across store instances (simulates process restart)."""
        db = tmp_path / "reopen.db"

        s1 = NexusStore(db_path=db)
        s1.append("user", "persistent message")
        s1.close()

        s2 = NexusStore(db_path=db)
        history = s2.get_history()
        assert len(history) == 1
        assert history[0]["content"] == "persistent message"
        s2.close()


class TestNexusStoreReturnedId:
    def test_append_returns_row_id(self, store):
        id1 = store.append("user", "first")
        id2 = store.append("user", "second")
        assert isinstance(id1, int)
        assert id2 > id1


class TestNexusStoreClose:
    def test_close_is_idempotent(self, store):
        store.close()
        store.close()  # Should not raise

    def test_operations_after_close_reconnect(self, store):
        """After close, next operation should reconnect automatically."""
        store.append("user", "before close")
        store.close()
        # Should auto-reconnect
        store.append("user", "after close")
        assert store.message_count() == 2