From 4eb96072c1ee3497d4af1461c049660e6a5ad86b Mon Sep 17 00:00:00 2001 From: hermes Date: Thu, 19 Mar 2026 10:44:59 -0400 Subject: [PATCH] test: add unit tests for briefing.py (#422) --- tests/timmy/test_briefing.py | 228 +++++++++++++++++++++++++++++++++++ 1 file changed, 228 insertions(+) create mode 100644 tests/timmy/test_briefing.py diff --git a/tests/timmy/test_briefing.py b/tests/timmy/test_briefing.py new file mode 100644 index 00000000..a8466e3e --- /dev/null +++ b/tests/timmy/test_briefing.py @@ -0,0 +1,228 @@ +"""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()