forked from Rockachopa/Timmy-time-dashboard
6d5eac6049
Automated salvage commit — agent session ended (exit 124). Work in progress, may need continuation.
218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
"""Unit tests for the Dreaming mode engine."""
|
|
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from timmy.dreaming import DreamingEngine, DreamRecord, _SESSION_GAP_SECONDS
|
|
|
|
|
|
# ── 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 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
|