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.
This commit is contained in:
Stacy Olivas
2026-03-17 17:43:37 -07:00
parent 25eb7ccf5c
commit 97e5c59ca6
2 changed files with 300 additions and 0 deletions
+110
View File
@@ -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
+190
View File
@@ -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