Bryan Van Deusen
25c767ddf2
PR 1.5 (commit 4168167) changed run_slot_injection's signature from
returning str to returning tuple[str, dict] so the scheduler can get
at the agentic message list for receipt persistence. The
test_run_slot_morning_runs_on_work_day mock was still returning the
old plain-string value, which unpacked to a ValueError at runtime.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
115 lines
5.1 KiB
Python
115 lines
5.1 KiB
Python
"""Tests for briefing_scheduler slot gating and work-day gate."""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
def _make_scheduler():
|
|
"""Return a mock APScheduler with tracked job IDs."""
|
|
sched = MagicMock()
|
|
sched.get_job.return_value = None
|
|
return sched
|
|
|
|
|
|
def test_add_user_jobs_skips_disabled_slots():
|
|
"""_add_user_jobs should not register jobs for slots set to False in config."""
|
|
from fabledassistant.services import briefing_scheduler as bs
|
|
|
|
mock_sched = _make_scheduler()
|
|
bs._scheduler = mock_sched
|
|
bs._loop = MagicMock()
|
|
|
|
config = {"slots": {"compilation": True, "morning": False, "midday": False, "afternoon": False}}
|
|
bs._add_user_jobs(user_id=1, tz="UTC", config=config)
|
|
|
|
added_ids = [call.kwargs.get("id") or call.args[3] for call in mock_sched.add_job.call_args_list]
|
|
# Only compilation should be scheduled
|
|
assert any("compilation" in jid for jid in added_ids)
|
|
assert not any("morning" in jid for jid in added_ids)
|
|
assert not any("midday" in jid for jid in added_ids)
|
|
assert not any("afternoon" in jid for jid in added_ids)
|
|
|
|
|
|
def test_add_user_jobs_schedules_all_when_config_missing():
|
|
"""_add_user_jobs with no config should schedule all 4 slots (safe default)."""
|
|
from fabledassistant.services import briefing_scheduler as bs
|
|
|
|
mock_sched = _make_scheduler()
|
|
bs._scheduler = mock_sched
|
|
bs._loop = MagicMock()
|
|
|
|
bs._add_user_jobs(user_id=1, tz="UTC", config=None)
|
|
|
|
added_ids = [
|
|
call.kwargs.get("id") or call.args[3]
|
|
for call in mock_sched.add_job.call_args_list
|
|
]
|
|
assert any("compilation" in jid for jid in added_ids)
|
|
assert any("morning" in jid for jid in added_ids)
|
|
assert any("midday" in jid for jid in added_ids)
|
|
assert any("afternoon" in jid for jid in added_ids)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slot_skips_morning_on_non_work_day():
|
|
"""_run_slot_for_user should return early for morning slot when today is not a work day."""
|
|
from fabledassistant.services import briefing_scheduler as bs
|
|
|
|
fake_profile = MagicMock()
|
|
fake_profile.work_schedule = {"days": ["Mon", "Tue", "Wed", "Thu", "Fri"]}
|
|
|
|
# Patch datetime at module level (it's imported at module level in briefing_scheduler)
|
|
# Patch get_setting and get_profile at their source modules (they're lazily imported)
|
|
with patch("fabledassistant.services.briefing_scheduler.datetime") as mock_dt, \
|
|
patch("fabledassistant.services.user_profile.get_profile", new_callable=AsyncMock, return_value=fake_profile), \
|
|
patch("fabledassistant.services.settings.get_setting", new_callable=AsyncMock, return_value="UTC"), \
|
|
patch("fabledassistant.services.briefing_pipeline.run_slot_injection", new_callable=AsyncMock) as mock_inject:
|
|
|
|
mock_now = MagicMock()
|
|
mock_now.strftime.return_value = "Sat"
|
|
mock_dt.now.return_value = mock_now
|
|
|
|
await bs._run_slot_for_user(user_id=1, slot="morning")
|
|
mock_inject.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_slot_morning_runs_on_work_day():
|
|
"""_run_slot_for_user should proceed for morning slot when today is a work day."""
|
|
from fabledassistant.services import briefing_scheduler as bs
|
|
|
|
fake_profile = MagicMock()
|
|
fake_profile.work_schedule = {"days": ["Mon", "Tue", "Wed", "Thu", "Fri"]}
|
|
|
|
with patch("fabledassistant.services.briefing_scheduler.datetime") as mock_dt, \
|
|
patch("fabledassistant.services.user_profile.get_profile", new_callable=AsyncMock, return_value=fake_profile), \
|
|
patch("fabledassistant.services.settings.get_setting", new_callable=AsyncMock, return_value="UTC"), \
|
|
patch("fabledassistant.services.briefing_conversations.get_or_create_today_conversation", new_callable=AsyncMock) as mock_conv, \
|
|
patch("fabledassistant.services.briefing_pipeline.run_slot_injection", new_callable=AsyncMock, return_value=("", {})) as mock_inject, \
|
|
patch("fabledassistant.services.briefing_conversations.post_message", new_callable=AsyncMock), \
|
|
patch("fabledassistant.services.push.send_push_notification", new_callable=AsyncMock):
|
|
|
|
mock_now = MagicMock()
|
|
mock_now.strftime.return_value = "Mon"
|
|
mock_dt.now.return_value = mock_now
|
|
mock_conv.return_value = MagicMock(id=1)
|
|
|
|
await bs._run_slot_for_user(user_id=1, slot="morning")
|
|
mock_inject.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_settings_put_timezone_updates_scheduler():
|
|
"""Saving user_timezone should call update_user_schedule when briefing is enabled."""
|
|
import json
|
|
from fabledassistant.services.briefing_scheduler import update_user_schedule
|
|
|
|
enabled_config = json.dumps({"enabled": True, "slots": {}, "locations": {}})
|
|
config = json.loads(enabled_config)
|
|
|
|
with patch("fabledassistant.services.briefing_scheduler._add_user_jobs") as mock_add:
|
|
update_user_schedule(1, config, tz_override="America/New_York")
|
|
mock_add.assert_called_once()
|
|
call_args = mock_add.call_args
|
|
assert call_args.args[0] == 1 # user_id
|
|
assert call_args.args[2] == config # config passed through
|