Timmy-time-dashboard/tests/unit/test_dreaming.py

"""Unit tests for the Dreaming mode engine."""

import sqlite3
from contextlib import closing
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, patch

import pytest

from timmy.dreaming import _SESSION_GAP_SECONDS, DreamingEngine, DreamRecord

pytestmark = pytest.mark.unit

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


@pytest.fixture()
def tmp_dreams_db(tmp_path):
    """Return a temporary path for the dreams database."""
    return tmp_path / "dreams.db"


@pytest.fixture()
def engine(tmp_dreams_db):
    """DreamingEngine backed by a temp database."""
    return DreamingEngine(db_path=tmp_dreams_db)


@pytest.fixture()
def chat_db(tmp_path):
    """Create a minimal chat database with some messages."""
    db_path = tmp_path / "chat.db"
    with closing(sqlite3.connect(str(db_path))) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS chat_messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                timestamp TEXT NOT NULL,
                source TEXT NOT NULL DEFAULT 'browser'
            )
        """)
        now = datetime.now(UTC)
        messages = [
            ("user",  "Hello, can you help me?",          (now - timedelta(hours=2)).isoformat()),
            ("agent", "Of course! What do you need?",     (now - timedelta(hours=2, seconds=-5)).isoformat()),
            ("user",  "How does Python handle errors?",   (now - timedelta(hours=2, seconds=-60)).isoformat()),
            ("agent", "Python uses try/except blocks.",   (now - timedelta(hours=2, seconds=-120)).isoformat()),
            ("user",  "Thanks!",                          (now - timedelta(hours=2, seconds=-180)).isoformat()),
        ]
        conn.executemany(
            "INSERT INTO chat_messages (role, content, timestamp) VALUES (?, ?, ?)",
            messages,
        )
        conn.commit()
    return db_path


# ── Idle detection ─────────────────────────────────────────────────────────────


class TestIdleDetection:
    def test_not_idle_immediately(self, engine):
        assert engine.is_idle() is False

    def test_idle_after_threshold(self, engine):
        engine._last_activity_time = datetime.now(UTC) - timedelta(minutes=20)
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_idle_threshold_minutes = 10
            assert engine.is_idle() is True

    def test_not_idle_when_threshold_zero(self, engine):
        engine._last_activity_time = datetime.now(UTC) - timedelta(hours=99)
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_idle_threshold_minutes = 0
            assert engine.is_idle() is False

    def test_record_activity_resets_timer(self, engine):
        engine._last_activity_time = datetime.now(UTC) - timedelta(minutes=30)
        engine.record_activity()
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_idle_threshold_minutes = 10
            assert engine.is_idle() is False


# ── Status dict ───────────────────────────────────────────────────────────────


class TestGetStatus:
    def test_status_shape(self, engine):
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_enabled = True
            mock_settings.dreaming_idle_threshold_minutes = 10
            status = engine.get_status()
        assert "enabled" in status
        assert "dreaming" in status
        assert "idle" in status
        assert "dream_count" in status
        assert "idle_minutes" in status

    def test_dream_count_starts_at_zero(self, engine):
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_enabled = True
            mock_settings.dreaming_idle_threshold_minutes = 10
            assert engine.get_status()["dream_count"] == 0


# ── Session grouping ──────────────────────────────────────────────────────────


class TestGroupIntoSessions:
    def test_single_session(self, engine):
        now = datetime.now(UTC)
        rows = [
            {"role": "user",  "content": "hi",   "timestamp": now.isoformat()},
            {"role": "agent", "content": "hello", "timestamp": (now + timedelta(seconds=10)).isoformat()},
        ]
        sessions = engine._group_into_sessions(rows)
        assert len(sessions) == 1
        assert len(sessions[0]) == 2

    def test_splits_on_large_gap(self, engine):
        now = datetime.now(UTC)
        gap = _SESSION_GAP_SECONDS + 100
        rows = [
            {"role": "user",  "content": "hi",    "timestamp": now.isoformat()},
            {"role": "agent", "content": "hello",  "timestamp": (now + timedelta(seconds=gap)).isoformat()},
        ]
        sessions = engine._group_into_sessions(rows)
        assert len(sessions) == 2

    def test_empty_input(self, engine):
        assert engine._group_into_sessions([]) == []


# ── Dream storage ─────────────────────────────────────────────────────────────


class TestDreamStorage:
    def test_store_and_retrieve(self, engine):
        dream = engine._store_dream(
            session_excerpt="User asked about Python.",
            decision_point="Python uses try/except blocks.",
            simulation="I could have given a code example.",
            proposed_rule="When explaining errors, include a code snippet.",
        )
        assert dream.id
        assert dream.proposed_rule == "When explaining errors, include a code snippet."

        retrieved = engine.get_recent_dreams(limit=1)
        assert len(retrieved) == 1
        assert retrieved[0].id == dream.id

    def test_count_increments(self, engine):
        assert engine.count_dreams() == 0
        engine._store_dream(
            session_excerpt="test", decision_point="test", simulation="test", proposed_rule="test"
        )
        assert engine.count_dreams() == 1


# ── dream_once integration ─────────────────────────────────────────────────────


class TestDreamOnce:
    @pytest.mark.asyncio
    async def test_skips_when_disabled(self, engine):
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_enabled = False
            result = await engine.dream_once()
        assert result is None

    @pytest.mark.asyncio
    async def test_skips_when_not_idle(self, engine):
        engine._last_activity_time = datetime.now(UTC)
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_enabled = True
            mock_settings.dreaming_idle_threshold_minutes = 60
            result = await engine.dream_once()
        assert result is None

    @pytest.mark.asyncio
    async def test_skips_when_already_dreaming(self, engine):
        engine._is_dreaming = True
        with patch("timmy.dreaming.settings") as mock_settings:
            mock_settings.dreaming_enabled = True
            mock_settings.dreaming_idle_threshold_minutes = 0
            result = await engine.dream_once()
        # Reset for cleanliness
        engine._is_dreaming = False
        assert result is None

    @pytest.mark.asyncio
    async def test_dream_produces_record_when_idle(self, engine, chat_db):
        """Full cycle: idle + chat data + mocked LLM → produces DreamRecord."""
        engine._last_activity_time = datetime.now(UTC) - timedelta(hours=1)

        with (
            patch("timmy.dreaming.settings") as mock_settings,
            patch("timmy.dreaming.DreamingEngine._call_agent", new_callable=AsyncMock) as mock_agent,
            patch("infrastructure.chat_store.DB_PATH", chat_db),
        ):
            mock_settings.dreaming_enabled = True
            mock_settings.dreaming_idle_threshold_minutes = 10
            mock_settings.dreaming_timeout_seconds = 30
            mock_agent.side_effect = [
                "I could have provided a concrete try/except example.",  # simulation
                "When explaining errors, always include a runnable code snippet.",  # rule
            ]

            result = await engine.dream_once()

        assert result is not None
        assert isinstance(result, DreamRecord)
        assert result.simulation
        assert result.proposed_rule
        assert engine.count_dreams() == 1