229 lines
8.1 KiB
Python
229 lines
8.1 KiB
Python
"""Unit tests for timmy.briefing — the morning briefing engine."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from timmy.briefing import (
|
|
ApprovalItem,
|
|
Briefing,
|
|
BriefingEngine,
|
|
_gather_swarm_summary,
|
|
_load_latest,
|
|
_save_briefing,
|
|
is_fresh,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ApprovalItem / Briefing dataclass basics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestApprovalItem:
|
|
def test_fields(self):
|
|
now = datetime.now(UTC)
|
|
item = ApprovalItem(
|
|
id="a1",
|
|
title="Deploy v2",
|
|
description="Upgrade prod",
|
|
proposed_action="deploy",
|
|
impact="high",
|
|
created_at=now,
|
|
status="pending",
|
|
)
|
|
assert item.id == "a1"
|
|
assert item.status == "pending"
|
|
|
|
def test_briefing_defaults(self):
|
|
b = Briefing(generated_at=datetime.now(UTC), summary="hello")
|
|
assert b.approval_items == []
|
|
assert b.period_start < b.period_end
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_fresh
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestIsFresh:
|
|
def test_fresh_briefing(self):
|
|
b = Briefing(generated_at=datetime.now(UTC), summary="ok")
|
|
assert is_fresh(b) is True
|
|
|
|
def test_stale_briefing(self):
|
|
old = datetime.now(UTC) - timedelta(hours=2)
|
|
b = Briefing(generated_at=old, summary="old")
|
|
assert is_fresh(b) is False
|
|
|
|
def test_custom_max_age(self):
|
|
recent = datetime.now(UTC) - timedelta(minutes=10)
|
|
b = Briefing(generated_at=recent, summary="recent")
|
|
assert is_fresh(b, max_age_minutes=5) is False
|
|
assert is_fresh(b, max_age_minutes=15) is True
|
|
|
|
def test_naive_datetime_handled(self):
|
|
# briefing.generated_at without tzinfo should still work
|
|
naive = datetime.now(UTC).replace(tzinfo=None)
|
|
b = Briefing(generated_at=naive, summary="naive")
|
|
assert is_fresh(b) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SQLite cache round-trip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSqliteCache:
|
|
def test_save_and_load(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
now = datetime.now(UTC)
|
|
b = Briefing(
|
|
generated_at=now,
|
|
summary="Good morning",
|
|
period_start=now - timedelta(hours=6),
|
|
period_end=now,
|
|
)
|
|
_save_briefing(b, db)
|
|
loaded = _load_latest(db)
|
|
assert loaded is not None
|
|
assert loaded.summary == "Good morning"
|
|
assert loaded.generated_at.isoformat()[:19] == now.isoformat()[:19]
|
|
|
|
def test_load_latest_returns_most_recent(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
old = datetime.now(UTC) - timedelta(hours=12)
|
|
new = datetime.now(UTC)
|
|
_save_briefing(Briefing(generated_at=old, summary="old"), db)
|
|
_save_briefing(Briefing(generated_at=new, summary="new"), db)
|
|
loaded = _load_latest(db)
|
|
assert loaded.summary == "new"
|
|
|
|
def test_load_latest_empty_db(self, tmp_path):
|
|
db = tmp_path / "empty.db"
|
|
assert _load_latest(db) is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _gather_swarm_summary
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGatherSwarmSummary:
|
|
def test_missing_db_file(self):
|
|
with patch("timmy.briefing.Path") as mock_path_cls:
|
|
# Simulate swarm_db.exists() -> False
|
|
mock_instance = MagicMock()
|
|
mock_instance.exists.return_value = False
|
|
|
|
original_path = Path
|
|
|
|
def side_effect(arg):
|
|
if arg == "data/swarm.db":
|
|
return mock_instance
|
|
return original_path(arg)
|
|
|
|
mock_path_cls.side_effect = side_effect
|
|
mock_path_cls.home = original_path.home
|
|
|
|
result = _gather_swarm_summary(datetime.now(UTC))
|
|
assert (
|
|
"No swarm" in result
|
|
or "unavailable" in result.lower()
|
|
or "No swarm activity" in result
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BriefingEngine
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBriefingEngine:
|
|
def test_get_cached_empty(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
engine = BriefingEngine(db_path=db)
|
|
assert engine.get_cached() is None
|
|
|
|
def test_needs_refresh_empty(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
engine = BriefingEngine(db_path=db)
|
|
assert engine.needs_refresh() is True
|
|
|
|
def test_needs_refresh_fresh(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
_save_briefing(Briefing(generated_at=datetime.now(UTC), summary="fresh"), db)
|
|
engine = BriefingEngine(db_path=db)
|
|
assert engine.needs_refresh() is False
|
|
|
|
def test_needs_refresh_stale(self, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
old = datetime.now(UTC) - timedelta(hours=2)
|
|
_save_briefing(Briefing(generated_at=old, summary="stale"), db)
|
|
engine = BriefingEngine(db_path=db)
|
|
assert engine.needs_refresh() is True
|
|
|
|
@patch("timmy.briefing.BriefingEngine._call_agent")
|
|
@patch("timmy.briefing.BriefingEngine._load_pending_items")
|
|
@patch("timmy.briefing._gather_swarm_summary")
|
|
@patch("timmy.briefing._gather_chat_summary")
|
|
@patch("timmy.briefing._gather_task_queue_summary")
|
|
def test_generate(self, mock_task, mock_chat, mock_swarm, mock_pending, mock_agent, tmp_path):
|
|
mock_swarm.return_value = "2 tasks completed"
|
|
mock_chat.return_value = "No conversations"
|
|
mock_task.return_value = "No tasks"
|
|
mock_agent.return_value = "Good morning, Alexander."
|
|
mock_pending.return_value = []
|
|
|
|
db = tmp_path / "test.db"
|
|
engine = BriefingEngine(db_path=db)
|
|
briefing = engine.generate()
|
|
|
|
assert briefing.summary == "Good morning, Alexander."
|
|
mock_agent.assert_called_once()
|
|
# Verify it was cached
|
|
assert _load_latest(db) is not None
|
|
|
|
@patch("timmy.briefing.BriefingEngine._call_agent")
|
|
@patch("timmy.briefing.BriefingEngine._load_pending_items")
|
|
@patch("timmy.briefing._gather_swarm_summary")
|
|
@patch("timmy.briefing._gather_chat_summary")
|
|
@patch("timmy.briefing._gather_task_queue_summary")
|
|
def test_generate_agent_failure(
|
|
self, mock_task, mock_chat, mock_swarm, mock_pending, mock_agent, tmp_path
|
|
):
|
|
mock_swarm.return_value = ""
|
|
mock_chat.return_value = ""
|
|
mock_task.return_value = ""
|
|
mock_agent.side_effect = Exception("LLM offline")
|
|
mock_pending.return_value = []
|
|
|
|
db = tmp_path / "test.db"
|
|
engine = BriefingEngine(db_path=db)
|
|
briefing = engine.generate()
|
|
# Should gracefully degrade
|
|
assert "offline" in briefing.summary.lower()
|
|
|
|
@patch("timmy.briefing.BriefingEngine._load_pending_items")
|
|
def test_get_or_generate_returns_cached(self, mock_pending, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
_save_briefing(Briefing(generated_at=datetime.now(UTC), summary="cached"), db)
|
|
mock_pending.return_value = []
|
|
|
|
engine = BriefingEngine(db_path=db)
|
|
result = engine.get_or_generate()
|
|
assert result.summary == "cached"
|
|
|
|
@patch("timmy.briefing.BriefingEngine.generate")
|
|
def test_get_or_generate_regenerates_when_stale(self, mock_gen, tmp_path):
|
|
db = tmp_path / "test.db"
|
|
old = datetime.now(UTC) - timedelta(hours=2)
|
|
_save_briefing(Briefing(generated_at=old, summary="stale"), db)
|
|
|
|
fresh = Briefing(generated_at=datetime.now(UTC), summary="fresh")
|
|
mock_gen.return_value = fresh
|
|
|
|
engine = BriefingEngine(db_path=db)
|
|
result = engine.get_or_generate()
|
|
assert result.summary == "fresh"
|
|
mock_gen.assert_called_once()
|