"""Tests for the briefing aggregator."""
from datetime import date, datetime, timedelta
from unittest.mock import AsyncMock
from zoneinfo import ZoneInfo
import pytest
from daily_briefing.aggregator import BriefingAggregator
from daily_briefing.models import CalendarEvent, EventType, MeetingSummary
class TestBriefingAggregator:
"""Tests for BriefingAggregator."""
@pytest.fixture
def mock_calendar_source(self, tz, today):
"""Create a mock calendar source."""
source = AsyncMock()
source.name = "Test Calendar"
source.enabled = True
# Create sample events
events = [
CalendarEvent(
id="event-1",
title="Morning Standup",
start=datetime.combine(today, datetime.min.time().replace(hour=9)).replace(tzinfo=tz),
end=datetime.combine(today, datetime.min.time().replace(hour=9, minute=30)).replace(tzinfo=tz),
source="Test Calendar",
event_type=EventType.MEETING,
),
CalendarEvent(
id="event-2",
title="Product Review",
start=datetime.combine(today, datetime.min.time().replace(hour=14)).replace(tzinfo=tz),
end=datetime.combine(today, datetime.min.time().replace(hour=15)).replace(tzinfo=tz),
source="Test Calendar",
event_type=EventType.MEETING,
),
]
source.get_events.return_value = events
source.get_meetings.return_value = []
source.get_action_items.return_value = []
source.get_travel_info.return_value = []
source.health_check.return_value = {"status": "healthy", "source": "Test Calendar"}
return source
@pytest.fixture
def mock_fireflies_source(self, tz, today):
"""Create a mock Fireflies source."""
source = AsyncMock()
source.name = "Fireflies"
source.enabled = True
yesterday = today - timedelta(days=1)
meetings = [
MeetingSummary(
id="meeting-1",
title="Sprint Planning",
date=datetime.combine(yesterday, datetime.min.time().replace(hour=10)).replace(tzinfo=tz),
duration_minutes=60,
participants=["alice@example.com"],
overview="Discussed sprint goals.",
action_items=["Review PR", "Update docs"],
),
]
source.get_events.return_value = []
source.get_meetings.return_value = meetings
source.get_action_items.return_value = []
source.get_travel_info.return_value = []
source.health_check.return_value = {"status": "healthy", "source": "Fireflies"}
return source
@pytest.mark.asyncio
async def test_generate_briefing(self, mock_calendar_source, today):
"""Test generating a daily briefing."""
aggregator = BriefingAggregator([mock_calendar_source])
briefing = await aggregator.generate_briefing(today)
assert briefing.date == today
assert len(briefing.events) == 2
assert briefing.total_meetings == 2
@pytest.mark.asyncio
async def test_generate_briefing_with_meetings(
self, mock_calendar_source, mock_fireflies_source, today
):
"""Test generating a briefing with meeting summaries."""
aggregator = BriefingAggregator([mock_calendar_source, mock_fireflies_source])
briefing = await aggregator.generate_briefing(today)
assert len(briefing.recent_meetings) == 1
assert briefing.recent_meetings[0].title == "Sprint Planning"
@pytest.mark.asyncio
async def test_detect_conflicts(self, tz, today):
"""Test conflict detection."""
source = AsyncMock()
source.name = "Calendar"
source.enabled = True
# Create overlapping events
events = [
CalendarEvent(
id="1",
title="Meeting A",
start=datetime.combine(today, datetime.min.time().replace(hour=10)).replace(tzinfo=tz),
end=datetime.combine(today, datetime.min.time().replace(hour=11)).replace(tzinfo=tz),
source="Calendar",
),
CalendarEvent(
id="2",
title="Meeting B",
start=datetime.combine(today, datetime.min.time().replace(hour=10, minute=30)).replace(tzinfo=tz),
end=datetime.combine(today, datetime.min.time().replace(hour=11, minute=30)).replace(tzinfo=tz),
source="Calendar",
),
]
source.get_events.return_value = events
source.get_meetings.return_value = []
source.get_action_items.return_value = []
source.get_travel_info.return_value = []
aggregator = BriefingAggregator([source])
briefing = await aggregator.generate_briefing(today)
assert len(briefing.conflicts) > 0
assert any(c.conflict_type.value == "overlap" for c in briefing.conflicts)
@pytest.mark.asyncio
async def test_find_free_slots(self, tz, today):
"""Test finding free time slots."""
source = AsyncMock()
source.name = "Calendar"
source.enabled = True
# Single morning meeting leaves afternoon free
events = [
CalendarEvent(
id="1",
title="Morning Meeting",
start=datetime.combine(today, datetime.min.time().replace(hour=9)).replace(tzinfo=tz),
end=datetime.combine(today, datetime.min.time().replace(hour=10)).replace(tzinfo=tz),
source="Calendar",
),
]
source.get_events.return_value = events
source.get_meetings.return_value = []
source.get_action_items.return_value = []
source.get_travel_info.return_value = []
aggregator = BriefingAggregator([source])
briefing = await aggregator.generate_briefing(today)
# Should have free time after 10am
assert len(briefing.free_slots) > 0
assert briefing.focus_time_hours > 0
@pytest.mark.asyncio
async def test_source_health(self, mock_calendar_source, mock_fireflies_source):
"""Test getting health status of all sources."""
aggregator = BriefingAggregator([mock_calendar_source, mock_fireflies_source])
health = await aggregator.get_source_health()
assert "Test Calendar" in health
assert "Fireflies" in health
assert health["Test Calendar"]["status"] == "healthy"
