diff --git a/modules/commands/base_command.py b/modules/commands/base_command.py index d48369d..20cd811 100644 --- a/modules/commands/base_command.py +++ b/modules/commands/base_command.py @@ -954,6 +954,36 @@ class BaseCommand(ABC): message.content_lower = content.lower() return message.content_lower + def split_trigger_and_args(self, content: str) -> tuple[Optional[str], str]: + """Split message content into ``(matched_keyword, args)``. + + Matches against ``self.keywords`` (built-in stems plus config ``aliases``), + preferring the longest keyword so multi-word triggers win. Leading ``!`` + is stripped for execute paths that still see raw command-style text. + + Args: + content: Raw or partially cleaned message text. + + Returns: + ``(keyword, args)`` when a keyword matches as the first token(s); + ``(None, content)`` (after optional ``!`` strip) otherwise. + """ + text = content.strip() + if text.startswith('!'): + text = text[1:].strip() + lower = text.lower() + if not lower or not self.keywords: + return None, text + + # Longest first so "dad joke" wins over a hypothetical shorter stem + for keyword in sorted(self.keywords, key=lambda k: len(k), reverse=True): + kw = keyword.lower() + if lower == kw: + return kw, "" + if lower.startswith(kw + " "): + return kw, text[len(kw):].strip() + return None, text + def matches_keyword(self, message: MeshMessage) -> bool: """Check if this command matches the message content based on keywords. diff --git a/modules/commands/channels_command.py b/modules/commands/channels_command.py index c39b2c5..9734a90 100644 --- a/modules/commands/channels_command.py +++ b/modules/commands/channels_command.py @@ -93,9 +93,11 @@ class ChannelsCommand(BaseCommand): # Don't match if this looks like a subcommand of another command # (e.g., "stats channels" should not match "channels" command) + # First word must be one of our keywords (including config aliases). if ' ' in content_lower: parts = content_lower.split() - if len(parts) > 1 and parts[0] not in ['channels', 'channel']: + keyword_stems = {k.lower() for k in self.keywords} + if len(parts) > 1 and parts[0] not in keyword_stems: return False for keyword in self.keywords: @@ -127,40 +129,36 @@ class ChannelsCommand(BaseCommand): bool: True if execution was successful. """ try: - # Parse the command to check for sub-commands - content = message.content.strip() - if content.startswith('!'): - content = content[1:].strip() + # Remainder after trigger (built-in stem or config alias), e.g. + # "channels seattle", "channel seahawks", "ch list", "channels #bot" + _trigger, args = self.split_trigger_and_args(message.content) - # Check for sub-command (e.g., "channels seattle", "channel seahawks", "channels list", "channels #bot") sub_command = None specific_channel = None - if content.lower().startswith('channels ') or content.lower().startswith('channel '): - parts = content.split(' ', 1) - if len(parts) > 1: - sub_command = parts[1].strip().lower() + if args: + sub_command = args.lower() - # Handle special "list" command to show all categories - if sub_command == 'list': - await self._show_all_categories(message) - return True + # Handle special "list" command to show all categories + if sub_command == 'list': + await self._show_all_categories(message) + return True - # Check if user is asking for a specific channel (starts with #) - if sub_command.startswith('#'): - specific_channel = sub_command - sub_command = None + # Check if user is asking for a specific channel (starts with #) + if sub_command.startswith('#'): + specific_channel = sub_command + sub_command = None + else: + # First check if this is a valid category + if self._is_valid_category(sub_command): + # It's a category, keep it as sub_command + pass else: - # First check if this is a valid category - if self._is_valid_category(sub_command): - # It's a category, keep it as sub_command - pass - else: - # Check if this might be a channel search (not a category) - # Try to find a channel that matches this name across all categories - found_channel = self._find_channel_by_name(sub_command) - if found_channel: - specific_channel = '#' + found_channel - sub_command = None + # Check if this might be a channel search (not a category) + # Try to find a channel that matches this name across all categories + found_channel = self._find_channel_by_name(sub_command) + if found_channel: + specific_channel = '#' + found_channel + sub_command = None # Handle specific channel request if specific_channel: diff --git a/modules/commands/dice_command.py b/modules/commands/dice_command.py index 993159e..f35fe4c 100644 --- a/modules/commands/dice_command.py +++ b/modules/commands/dice_command.py @@ -76,30 +76,6 @@ class DiceCommand(BaseCommand): """ return self.translate('commands.dice.help') - def matches_keyword(self, message: MeshMessage) -> bool: - """Override to handle dice-specific matching. - - Args: - message: The received message. - - Returns: - bool: True if message is a dice command, False otherwise. - """ - content_lower = self.cleanup_message_for_matching(message) - - # Check for exact "dice" match - if content_lower == "dice": - return True - - # Check for dice with parameters (dice d20, dice 20, dice d6, etc.) - # Match any message starting with "dice " - validation happens in execute() - if content_lower.startswith("dice "): - words = content_lower.split() - if len(words) >= 2 and words[0] == "dice": - return True # Match any dice command, validation in execute() - - return False - def parse_dice_notation(self, dice_input: str) -> tuple: """Parse dice notation and return (sides, count, is_decade). @@ -281,22 +257,17 @@ class DiceCommand(BaseCommand): Returns: bool: True if executed successfully, False otherwise. """ - content = message.content.strip() - - # Handle command-style messages - if content.startswith('!'): - content = content[1:].strip() + _trigger, dice_part = self.split_trigger_and_args(message.content) # Default to d6 if no specification - if content.lower() == "dice": + if not dice_part: sides = 6 count = 1 results = self.roll_dice(sides, count) response = self.format_dice_result(sides, count, results) return await self.send_response(message, response) - # Parse dice specification - dice_part = content[5:].strip() # Get everything after "dice " + # Parse dice specification after the trigger keyword / alias # Try parsing as mixed dice first (multiple dice types) mixed_dice = self.parse_mixed_dice(dice_part) diff --git a/modules/commands/hacker_command.py b/modules/commands/hacker_command.py index 1b01ca3..1184964 100644 --- a/modules/commands/hacker_command.py +++ b/modules/commands/hacker_command.py @@ -66,8 +66,16 @@ class HackerCommand(BaseCommand): if content.startswith('!'): content = content[1:].strip() + # Config aliases are extra stems; strip them so "hack sudo ls" routes like "sudo ls". + # Built-in comedy keywords stay in the routed text (get_hacker_error matches on prefix). + trigger, args = self.split_trigger_and_args(message.content) + builtin_stems = {k.lower() for k in type(self).keywords} + routed = content + if trigger is not None and trigger not in builtin_stems: + routed = args + # Get the appropriate error message - error_msg = self.get_hacker_error(content) + error_msg = self.get_hacker_error(routed) # Send the response return await self.send_response(message, error_msg) @@ -510,4 +518,13 @@ class HackerCommand(BaseCommand): if len(content_lower) == len(keyword.lower()) or content_lower[len(keyword.lower())] == ' ': return True + # Config aliases (keywords not already covered by the comedy lists above) + known = {k.lower() for k in exact_match_commands + prefix_match_commands} + for keyword in self.keywords: + kw = keyword.lower() + if kw in known: + continue + if content_lower == kw or content_lower.startswith(kw + ' '): + return True + return False diff --git a/modules/commands/multitest_command.py b/modules/commands/multitest_command.py index 6a1e3fc..e69a965 100644 --- a/modules/commands/multitest_command.py +++ b/modules/commands/multitest_command.py @@ -654,25 +654,6 @@ class MultitestCommand(BaseCommand): def get_help_text(self) -> str: return self.translate('commands.multitest.help', fallback="Listens for 6 seconds and collects all unique paths from incoming messages") - def matches_keyword(self, message: MeshMessage) -> bool: - """Check if message matches multitest keyword""" - content_lower = self.cleanup_message_for_matching(message) - - # Check for exact match or keyword followed by space - for keyword in self.keywords: - if content_lower == keyword or content_lower.startswith(keyword + ' '): - return True - - # Check for variants: "mt long", "mt xlong", "multitest long", "multitest xlong" - if content_lower.startswith('mt ') or content_lower.startswith('multitest '): - parts = content_lower.split() - if len(parts) >= 2 and parts[0] in ['mt', 'multitest']: - variant = parts[1] - if variant in ['long', 'xlong']: - return True - - return False - def extract_path_from_rf_data(self, rf_data: dict) -> Optional[str]: """Extract path in prefix string format from RF data routing_info. Supports 1-, 2-, and 3-byte-per-hop (2, 4, or 6 hex chars per node). @@ -996,27 +977,16 @@ class MultitestCommand(BaseCommand): self.record_execution(user_id) # Determine listening duration based on command variant - content = message.content.strip() - if content.startswith('!'): - content = content[1:].strip() - - content_lower = content.lower() + _trigger, args = self.split_trigger_and_args(message.content) listening_duration = 6.0 # Default - # Check for variants: "mt long", "mt xlong", "multitest long", "multitest xlong" - if content_lower.startswith('mt ') or content_lower.startswith('multitest '): - parts = content_lower.split() - if len(parts) >= 2 and parts[0] in ['mt', 'multitest']: - variant = parts[1] - if variant == 'long': - listening_duration = 10.0 - self.logger.info(f"Multitest command (long) executed by {user_id} - starting 10 second listening window") - elif variant == 'xlong': - listening_duration = 14.0 - self.logger.info(f"Multitest command (xlong) executed by {user_id} - starting 14 second listening window") - else: - self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window") - else: - self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window") + # Variants: " long" / " xlong" (trigger = multitest, mt, or alias) + variant = args.split()[0].lower() if args else "" + if variant == 'long': + listening_duration = 10.0 + self.logger.info(f"Multitest command (long) executed by {user_id} - starting 10 second listening window") + elif variant == 'xlong': + listening_duration = 14.0 + self.logger.info(f"Multitest command (xlong) executed by {user_id} - starting 14 second listening window") else: self.logger.info(f"Multitest command executed by {user_id} - starting 6 second listening window") diff --git a/modules/commands/prefix_command.py b/modules/commands/prefix_command.py index 882ae01..295297a 100644 --- a/modules/commands/prefix_command.py +++ b/modules/commands/prefix_command.py @@ -26,7 +26,7 @@ class PrefixCommand(BaseCommand): # Read-only informational output; safe for scheduled {cmd:...} rendering. render_safe = True name = "prefix" - keywords = ['prefix', 'repeater', 'lookup'] + keywords = ['prefix', 'lookup'] description = "Look up repeaters by prefix (2, 4, or 6 hex chars = 1–3 bytes; longer input truncated)" category = "meshcore_info" requires_dm = False @@ -178,11 +178,6 @@ class PrefixCommand(BaseCommand): return self.translate('commands.prefix.help_no_api', location_note=location_note) return self.translate('commands.prefix.help_api', location_note=location_note) - def matches_keyword(self, message: MeshMessage) -> bool: - """Check if message starts with 'prefix' keyword""" - content_lower = self.cleanup_message_for_matching(message) - return content_lower == 'prefix' or content_lower.startswith('prefix ') - async def _parse_location_to_lat_lon(self, location: str) -> tuple[Optional[float], Optional[float], Optional[str]]: """Parse location string to latitude/longitude coordinates. diff --git a/modules/commands/roll_command.py b/modules/commands/roll_command.py index b01b21a..0356816 100644 --- a/modules/commands/roll_command.py +++ b/modules/commands/roll_command.py @@ -65,31 +65,18 @@ class RollCommand(BaseCommand): return self.translate('commands.roll.help') def matches_keyword(self, message: MeshMessage) -> bool: - """Override to handle roll-specific matching. - - Custom matching logic to support variable maximums (e.g., "roll 50"). - - Args: - message: The message to check for a match. - - Returns: - bool: True if the message matches the roll command syntax, False otherwise. - """ + """Match ``roll`` / aliases; with args, only when the arg is a valid max.""" content_lower = self.cleanup_message_for_matching(message) + if not content_lower: + return False - # Check for exact "roll" match - if content_lower == "roll": - return True - - # Check for roll with parameters (roll 50, roll 1000, etc.) - # Ensure "roll" is the first word and followed by valid number - if content_lower.startswith("roll "): - words = content_lower.split() - if len(words) >= 2 and words[0] == "roll": - roll_part = content_lower[5:].strip() # Get everything after "roll " - # Check if the roll part is valid number notation (not just any word) - max_num = self.parse_roll_notation(roll_part) - return max_num is not None # Only match if it's valid number notation + for keyword in self.keywords: + kw = keyword.lower() + if content_lower == kw: + return True + if content_lower.startswith(kw + " "): + roll_part = content_lower[len(kw):].strip() + return self.parse_roll_notation(roll_part) is not None return False @@ -151,18 +138,12 @@ class RollCommand(BaseCommand): Returns: bool: True if executed successfully, False otherwise. """ - content = message.content.strip() - - # Handle command-style messages - if content.startswith('!'): - content = content[1:].strip() + _trigger, roll_part = self.split_trigger_and_args(message.content) # Default to 1-100 if no specification - if content.lower() == "roll": + if not roll_part: max_num: Optional[int] = 100 else: - # Parse roll specification - roll_part = content[5:].strip() # Get everything after "roll " max_num = self.parse_roll_notation(roll_part) if max_num is None: diff --git a/modules/commands/trace_command.py b/modules/commands/trace_command.py index 061e842..fe94128 100644 --- a/modules/commands/trace_command.py +++ b/modules/commands/trace_command.py @@ -93,12 +93,6 @@ class TraceCommand(BaseCommand): "No path = use your message path (round-trip)." ) - def matches_keyword(self, message: MeshMessage) -> bool: - content_lower = self.cleanup_message_for_matching(message) - if content_lower == "trace" or content_lower == "tracer": - return True - return bool(content_lower.startswith("trace ") or content_lower.startswith("tracer ")) - def _extract_path_from_message(self, message: MeshMessage) -> list[str]: """Extract path node IDs from message.path (supports 1-byte, 2-byte, and 3-byte hashes).""" if not message.path: @@ -133,21 +127,14 @@ class TraceCommand(BaseCommand): return valid def _parse_path_arg(self, content: str) -> Optional[list[str]]: - """Parse path from command content after 'trace ' or 'tracer '. + """Parse path from command content after the matched trigger keyword. Accepts comma-separated hex nodes where each segment is the same length: 2-char = 1-byte (e.g. 01,7a,55), 4-char = 2-byte (e.g. feed,6ddf), 6-char = 3-byte (e.g. feedca,6ddf01). Without commas, treats contiguous hex as 2-char (1-byte) nodes. Returns list of hex node IDs, or None if no path args / invalid. """ - content = content.strip() - if content.startswith("!"): - content = content[1:].strip() - rest = "" - for kw in ["tracer ", "trace "]: - if content.lower().startswith(kw): - rest = content[len(kw) :].strip() - break + _trigger, rest = self.split_trigger_and_args(content) if not rest: return None # Comma-separated: each segment is one node; preserves multibyte groupings @@ -224,11 +211,9 @@ class TraceCommand(BaseCommand): return "\n".join(lines) async def execute(self, message: MeshMessage) -> bool: - content = message.content.strip() - if content.startswith("!"): - content = content[1:].strip() - - is_tracer = content.lower().startswith("tracer") + trigger, _args = self.split_trigger_and_args(message.content) + # Reciprocal round-trip only when the tracer stem is used (not aliases of trace) + is_tracer = trigger == "tracer" path_arg = self._parse_path_arg(message.content) if path_arg is not None: path_nodes = path_arg[: self.maximum_hops] diff --git a/tests/unit/test_disabled_command_alias_fallback.py b/tests/unit/test_disabled_command_alias_fallback.py index 4469948..054414a 100644 --- a/tests/unit/test_disabled_command_alias_fallback.py +++ b/tests/unit/test_disabled_command_alias_fallback.py @@ -1,36 +1,150 @@ #!/usr/bin/env python3 -"""Regression: disabled built-in must not block another command's alias. - -Reproduces the report where [Test_Command] aliases = path, p with -[Path_Command] enabled = false still claimed !path and sent no reply. -""" +"""Regression: config aliases work across commands that had hardcoded matchers.""" from unittest.mock import AsyncMock, MagicMock, Mock import pytest +from modules.commands.channels_command import ChannelsCommand +from modules.commands.dice_command import DiceCommand +from modules.commands.hacker_command import HackerCommand +from modules.commands.multitest_command import MultitestCommand from modules.commands.path_command import PathCommand +from modules.commands.prefix_command import PrefixCommand +from modules.commands.roll_command import RollCommand from modules.commands.test_command import TestCommand as MeshTestCommand +from modules.commands.trace_command import TraceCommand from tests.conftest import mock_message from tests.test_command_manager import make_manager from tests.unit.test_command_path_byte_gating import _base_bot -@pytest.mark.unit -def test_test_command_aliases_match_path_and_p(): - bot = _base_bot() - bot.config.add_section("Test_Command") - bot.config.set("Test_Command", "enabled", "true") - bot.config.set("Test_Command", "aliases", "path, p") +def _with_aliases(bot, section: str, aliases: str): + if not bot.config.has_section(section): + bot.config.add_section(section) + bot.config.set(section, "enabled", "true") + bot.config.set(section, "aliases", aliases) + return bot - cmd = MeshTestCommand(bot) - assert "path" in cmd.keywords - assert "p" in cmd.keywords - assert cmd.matches_keyword(mock_message(content="!path", is_dm=True)) is True - assert cmd.matches_keyword(mock_message(content="path", is_dm=True)) is True - assert cmd.matches_keyword(mock_message(content="p", is_dm=True)) is True - assert cmd.matches_keyword(mock_message(content="test", is_dm=True)) is True - assert cmd.matches_keyword(mock_message(content="ping", is_dm=True)) is False + +@pytest.mark.unit +@pytest.mark.parametrize( + "factory,section,alias,content", + [ + (MeshTestCommand, "Test_Command", "path", "!path"), + (DiceCommand, "Dice_Command", "d", "d d20"), + (DiceCommand, "Dice_Command", "d", "d"), + (RollCommand, "Roll_Command", "r", "r"), + (RollCommand, "Roll_Command", "r", "r 50"), + (TraceCommand, "Trace_Command", "tr", "tr"), + (TraceCommand, "Trace_Command", "tr", "tr 01,7a"), + (PrefixCommand, "Prefix_Command", "pfx", "pfx"), + (PrefixCommand, "Prefix_Command", "pfx", "pfx free"), + (MultitestCommand, "Multitest_Command", "mtest", "mtest"), + (MultitestCommand, "Multitest_Command", "mtest", "mtest long"), + (ChannelsCommand, "Channels_Command", "ch", "ch"), + (PathCommand, "Path_Command", "routehex", "routehex"), + ], +) +def test_config_alias_matches(factory, section, alias, content): + bot = _with_aliases(_base_bot(), section, alias) + cmd = factory(bot) + assert alias in [k.lower() for k in cmd.keywords] + assert cmd.matches_keyword(mock_message(content=content, is_dm=True)) is True + + +@pytest.mark.unit +def test_roll_alias_still_rejects_non_numeric_args(): + bot = _with_aliases(_base_bot(), "Roll_Command", "r") + cmd = RollCommand(bot) + assert cmd.matches_keyword(mock_message(content="r abc", is_dm=True)) is False + + +@pytest.mark.unit +def test_channels_alias_does_not_match_as_subcommand(): + bot = _with_aliases(_base_bot(), "Channels_Command", "ch") + cmd = ChannelsCommand(bot) + # "stats channels" style: first word is not a channels keyword/alias + assert cmd.matches_keyword(mock_message(content="stats ch", is_dm=True)) is False + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_dice_alias_execute_uses_args(): + bot = _with_aliases(_base_bot(), "Dice_Command", "d") + cmd = DiceCommand(bot) + cmd.send_response = AsyncMock(return_value=True) + cmd.roll_dice = Mock(return_value=[4]) + cmd.format_dice_result = Mock(return_value="ok") + + await cmd.execute(mock_message(content="!d d20", is_dm=True)) + + cmd.roll_dice.assert_called() + cmd.send_response.assert_awaited() + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_roll_alias_execute_parses_max(): + bot = _with_aliases(_base_bot(), "Roll_Command", "r") + cmd = RollCommand(bot) + cmd.send_response = AsyncMock(return_value=True) + cmd.roll_number = Mock(return_value=7) + cmd.format_roll_result = Mock(return_value="rolled") + + await cmd.execute(mock_message(content="!r 50", is_dm=True)) + + cmd.roll_number.assert_called_once_with(50) + + +@pytest.mark.unit +def test_trace_parse_path_arg_honors_alias(): + bot = _with_aliases(_base_bot(), "Trace_Command", "tr") + cmd = TraceCommand(bot) + assert cmd._parse_path_arg("!tr 01,7a") == ["01", "7a"] + + +@pytest.mark.unit +def test_hacker_config_alias_matches(): + bot = _with_aliases(_base_bot(), "Hacker_Command", "hack") + # Hacker may use a different enabled key; force on + bot.config.set("Hacker_Command", "enabled", "true") + cmd = HackerCommand(bot) + cmd.enabled = True + assert cmd.matches_keyword(mock_message(content="hack", is_dm=True)) is True + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_channels_alias_execute_honors_list_subcommand(): + bot = _with_aliases(_base_bot(), "Channels_Command", "ch") + cmd = ChannelsCommand(bot) + cmd._show_all_categories = AsyncMock() + cmd._show_specific_channel = AsyncMock() + cmd.send_response = AsyncMock(return_value=True) + + await cmd.execute(mock_message(content="!ch list", is_dm=True)) + + cmd._show_all_categories.assert_awaited_once() + cmd._show_specific_channel.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.unit +async def test_hacker_alias_execute_routes_inner_command(): + bot = _with_aliases(_base_bot(), "Hacker_Command", "hack") + bot.config.set("Hacker_Command", "enabled", "true") + cmd = HackerCommand(bot) + cmd.enabled = True + cmd.get_hacker_error = Mock(return_value="denied") + cmd.send_response = AsyncMock(return_value=True) + + await cmd.execute(mock_message(content="!hack sudo ls", is_dm=True)) + cmd.get_hacker_error.assert_called_once_with("sudo ls") + + cmd.get_hacker_error.reset_mock() + await cmd.execute(mock_message(content="!sudo ls", is_dm=True)) + cmd.get_hacker_error.assert_called_once_with("sudo ls") @pytest.mark.unit @@ -45,7 +159,6 @@ def test_check_keywords_prefers_test_alias_when_path_disabled(): path_cmd = PathCommand(bot) test_cmd = MeshTestCommand(bot) - # path before test — same claim order as the user report manager = make_manager(bot, commands={"path": path_cmd, "test": test_cmd}) matches = manager.check_keywords(mock_message(content="!path", is_dm=True))