From 97e5c59ca69fcc4e9101bd43c2eaf2c1a22c3b46 Mon Sep 17 00:00:00 2001 From: Stacy Olivas Date: Tue, 17 Mar 2026 17:43:37 -0700 Subject: [PATCH] feat: !schedule command listing scheduled messages and advert interval Add ScheduleCommand (DM-only by default). Displays configured scheduled message times, target channels, message previews, and the current advertisement interval. Read-only; does not modify schedule state. --- modules/commands/schedule_command.py | 110 ++++++++++++++++ tests/test_schedule_command.py | 190 +++++++++++++++++++++++++++ 2 files changed, 300 insertions(+) create mode 100644 modules/commands/schedule_command.py create mode 100644 tests/test_schedule_command.py diff --git a/modules/commands/schedule_command.py b/modules/commands/schedule_command.py new file mode 100644 index 0000000..250867d --- /dev/null +++ b/modules/commands/schedule_command.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +""" +Schedule command for the MeshCore Bot +Lists upcoming scheduled messages and interval advertising settings. +""" + +from typing import Any, Optional + +from ..models import MeshMessage +from .base_command import BaseCommand + + +class ScheduleCommand(BaseCommand): + """Show scheduled messages and advertising interval configured for the bot. + + Responds to ``schedule`` or ``schedule list``. The output is kept compact + so it fits within typical MeshCore message limits (~200 chars). + + Config section ``[Schedule_Command]``: + enabled = true # enable/disable this command + dm_only = true # restrict to DMs (default true — exposes config) + """ + + name = "schedule" + keywords = ["schedule"] + description = "List upcoming scheduled messages and advertising interval." + category = "admin" + + short_description = "Show scheduled messages and advertising interval" + usage = "schedule [list]" + examples = ["schedule", "schedule list"] + + def __init__(self, bot: Any) -> None: + super().__init__(bot) + self._enabled = self.get_config_value( + "Schedule_Command", "enabled", fallback=True, value_type="bool" + ) + self._dm_only = self.get_config_value( + "Schedule_Command", "dm_only", fallback=True, value_type="bool" + ) + + # ------------------------------------------------------------------ + # BaseCommand interface + # ------------------------------------------------------------------ + + def can_execute(self, message: MeshMessage) -> bool: + if not self._enabled: + return False + if self._dm_only and not message.is_dm: + return False + return super().can_execute(message) + + def get_help_text(self) -> str: + return "schedule [list] — show scheduled messages and advert interval" + + async def execute(self, message: MeshMessage) -> bool: + response = self._build_response() + return await self.send_response(message, response) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _build_response(self) -> str: + lines: list[str] = [] + + # --- Scheduled messages --- + scheduled = self._get_scheduled_messages() + if scheduled: + lines.append(f"Scheduled ({len(scheduled)}):") + for time_str, channel, preview in scheduled: + # Format HHMM → HH:MM + hhmm = f"{time_str[:2]}:{time_str[2:]}" + lines.append(f" {hhmm} #{channel}: {preview}") + else: + lines.append("No scheduled messages configured.") + + # --- Interval advertising --- + advert_info = self._get_advert_info() + if advert_info: + lines.append(advert_info) + + return "\n".join(lines) + + def _get_scheduled_messages(self) -> list[tuple]: + """Return sorted list of (time_str, channel, preview) tuples.""" + scheduler = getattr(self.bot, "scheduler", None) + if scheduler is None: + return [] + + scheduled = getattr(scheduler, "scheduled_messages", {}) + results = [] + for time_str, payload in sorted(scheduled.items()): + channel, message = payload + # Truncate long messages so response stays compact + preview = message if len(message) <= 40 else message[:37] + "..." + results.append((time_str, channel, preview)) + return results + + def _get_advert_info(self) -> Optional[str]: + """Return a one-line advert interval summary, or None if disabled.""" + try: + interval_hours = self.bot.config.getint( + "Bot", "advert_interval_hours", fallback=0 + ) + if interval_hours > 0: + return f"Advert interval: every {interval_hours}h" + except Exception: + pass + return None diff --git a/tests/test_schedule_command.py b/tests/test_schedule_command.py new file mode 100644 index 0000000..0d48a62 --- /dev/null +++ b/tests/test_schedule_command.py @@ -0,0 +1,190 @@ +"""Tests for ScheduleCommand.""" + +from configparser import ConfigParser +from unittest.mock import AsyncMock, Mock + +import pytest + +from modules.commands.schedule_command import ScheduleCommand +from tests.conftest import mock_message + + +@pytest.fixture +def bot(mock_logger): + b = Mock() + b.logger = mock_logger + b.config = ConfigParser() + b.config.add_section("Bot") + b.config.set("Bot", "bot_name", "TestBot") + b.translator = Mock() + b.translator.translate = Mock(side_effect=lambda k, **kw: k) + b.translator.get_value = Mock(return_value=None) + b.command_manager = Mock() + b.command_manager.send_response = AsyncMock(return_value=True) + b.command_manager.monitor_channels = ["general"] + b.scheduler = Mock() + b.scheduler.scheduled_messages = {} + return b + + +def make_cmd(bot): + return ScheduleCommand(bot) + + +# --------------------------------------------------------------------------- +# TestCanExecute +# --------------------------------------------------------------------------- + + +class TestCanExecute: + def test_dm_allowed_by_default(self, bot): + cmd = make_cmd(bot) + msg = mock_message(is_dm=True) + assert cmd.can_execute(msg) is True + + def test_channel_blocked_when_dm_only(self, bot): + cmd = make_cmd(bot) + msg = mock_message(channel="general", is_dm=False) + assert cmd.can_execute(msg) is False + + def test_channel_allowed_when_dm_only_false(self, bot): + bot.config.add_section("Schedule_Command") + bot.config.set("Schedule_Command", "dm_only", "false") + cmd = make_cmd(bot) + msg = mock_message(channel="general", is_dm=False) + assert cmd.can_execute(msg) is True + + def test_disabled_blocks_all(self, bot): + bot.config.add_section("Schedule_Command") + bot.config.set("Schedule_Command", "enabled", "false") + cmd = make_cmd(bot) + msg = mock_message(is_dm=True) + assert cmd.can_execute(msg) is False + + +# --------------------------------------------------------------------------- +# TestBuildResponse — no scheduled messages +# --------------------------------------------------------------------------- + + +class TestBuildResponseEmpty: + def test_no_schedules_says_none_configured(self, bot): + cmd = make_cmd(bot) + response = cmd._build_response() + assert "No scheduled messages configured" in response + + def test_no_advert_interval_omits_advert_line(self, bot): + cmd = make_cmd(bot) + response = cmd._build_response() + assert "Advert" not in response + + +# --------------------------------------------------------------------------- +# TestBuildResponse — with scheduled messages +# --------------------------------------------------------------------------- + + +class TestBuildResponseWithSchedules: + def test_shows_scheduled_count(self, bot): + bot.scheduler.scheduled_messages = { + "0900": ("general", "Good morning!"), + "1800": ("general", "Good evening!"), + } + cmd = make_cmd(bot) + response = cmd._build_response() + assert "Scheduled (2)" in response + + def test_formats_time_as_hhmm(self, bot): + bot.scheduler.scheduled_messages = {"0930": ("general", "Hello")} + cmd = make_cmd(bot) + response = cmd._build_response() + assert "09:30" in response + + def test_shows_channel_and_message(self, bot): + bot.scheduler.scheduled_messages = {"1200": ("alerts", "Noon check-in")} + cmd = make_cmd(bot) + response = cmd._build_response() + assert "#alerts" in response + assert "Noon check-in" in response + + def test_long_message_is_truncated(self, bot): + long_msg = "A" * 100 + bot.scheduler.scheduled_messages = {"0800": ("general", long_msg)} + cmd = make_cmd(bot) + response = cmd._build_response() + # Preview should be ≤43 chars (40 + "...") + for line in response.splitlines(): + if "08:00" in line: + # Extract the message part after the channel + parts = line.split(": ", 2) + if len(parts) == 3: + assert len(parts[2]) <= 43 + + def test_entries_sorted_by_time(self, bot): + bot.scheduler.scheduled_messages = { + "1800": ("general", "Evening"), + "0600": ("general", "Morning"), + "1200": ("general", "Noon"), + } + cmd = make_cmd(bot) + response = cmd._build_response() + lines = [l for l in response.splitlines() if ":" in l and "#" in l] + times = [l.strip().split(" ")[0] for l in lines] + assert times == sorted(times) + + +# --------------------------------------------------------------------------- +# TestAdvertInfo +# --------------------------------------------------------------------------- + + +class TestAdvertInfo: + def test_advert_interval_shown(self, bot): + bot.config.set("Bot", "advert_interval_hours", "4") + cmd = make_cmd(bot) + response = cmd._build_response() + assert "Advert interval: every 4h" in response + + def test_zero_interval_omitted(self, bot): + bot.config.set("Bot", "advert_interval_hours", "0") + cmd = make_cmd(bot) + response = cmd._build_response() + assert "Advert" not in response + + def test_missing_advert_interval_omitted(self, bot): + # advert_interval_hours not in config → fallback 0 → omitted + cmd = make_cmd(bot) + response = cmd._build_response() + assert "Advert" not in response + + +# --------------------------------------------------------------------------- +# TestNoScheduler +# --------------------------------------------------------------------------- + + +class TestNoScheduler: + def test_no_scheduler_attr_returns_gracefully(self, bot): + del bot.scheduler + cmd = make_cmd(bot) + response = cmd._build_response() + assert "No scheduled messages configured" in response + + +# --------------------------------------------------------------------------- +# TestExecute +# --------------------------------------------------------------------------- + + +class TestExecute: + @pytest.mark.asyncio + async def test_execute_calls_send_response(self, bot): + cmd = make_cmd(bot) + msg = mock_message(is_dm=True) + result = await cmd.execute(msg) + assert result is True + bot.command_manager.send_response.assert_called_once() + call_args = bot.command_manager.send_response.call_args + response_text = call_args[0][1] # second positional arg is content + assert isinstance(response_text, str) + assert len(response_text) > 0