diff --git a/CHANGELOG.md b/CHANGELOG.md index 753fe5e..0bd8ac9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,41 @@ semantic versioning. read-only beside the default, since they are the reason a channel can ignore it. +- `[Hello_Command] include_sender` (default off) names the user the hello reply is + answering, so a busy channel can tell whose greeting the bot picked up (#292). The + mention takes the place of the random human descriptor, keeping the translated + sentence intact, and is dropped when it would push the reply past the channel body + budget. DMs are unaffected. +- `generate_website.py` accepts `--link-css URL` and `--embed-css FILE` to layer + custom CSS on top of the built-in style chosen with `--style` (#288). Embedded + CSS is appended to the page's `{external_css_link} +""" + + # Continue with body + html_content += f""" + +
+
+ +
+ + + +
+ {nav_html} + +
+
+
+
+

{escape_html(bot_name)}

+
+
+ {escape_html(introduction).replace(chr(10), '
')}{channels_suffix} +
+
+
+ +
+ {commands_html} + {channels_html} +
+ + +
+
+ + + +""" + + return html_content + + +def generate_builtin_css(style: str) -> str: + """Generate the built-in CSS for a given style""" + return f""" :root {{ {STYLES[style]['css_vars']} }} @@ -2368,136 +2581,7 @@ def generate_html(bot_name: str, title: str, introduction: str, commands: list[t }} /* Style-specific CSS overrides */ - {STYLES[style].get('css_overrides', '')} - - - -
-
- -
- - - -
- {nav_html} - -
-
-
-
-

{escape_html(bot_name)}

-
-
- {escape_html(introduction).replace(chr(10), '
')}{channels_suffix} -
-
-
- -
- {commands_html} - {channels_html} -
- - -
-
- - - -""" - - return html_content + {STYLES[style].get('css_overrides', '')}""" def list_styles(): @@ -2513,7 +2597,7 @@ def list_styles(): print(f" {' ' * max_len} {style_info['description']}\n") -def generate_samples(config_file): +def generate_samples(config_file, link_css: Optional[str] = None, custom_css: Optional[str] = None): """Generate sample HTML files for all styles with an index page.""" logger = logging.getLogger(__name__) @@ -2539,7 +2623,7 @@ def generate_samples(config_file): channels_data = load_channels_from_config(config) # Setup minimal bot for plugin loading - minimal_bot = MinimalBot(config, logger) + minimal_bot = MinimalBot(config, logger, bot_root=bot_root) # Initialize database manager if database exists db_path = get_database_path(config, bot_root) @@ -2552,15 +2636,14 @@ def generate_samples(config_file): else: minimal_bot.db_manager = None + local_commands_dir = resolve_local_commands_dir(config, bot_root, logger) + # Load plugins - plugin_loader = PluginLoader(minimal_bot) + plugin_loader = PluginLoader(minimal_bot, local_commands_dir=local_commands_dir) commands = plugin_loader.load_all_plugins() - # Filter out admin and hidden commands - public_commands = { - name: cmd for name, cmd in commands.items() - if name not in admin_commands and not getattr(cmd, 'hidden', False) - } + # Filter out admin, hidden, and disabled commands + public_commands = filter_commands(commands, admin_commands, config) public_commands.update(get_randomline_commands(config)) # Sort commands @@ -2582,7 +2665,9 @@ def generate_samples(config_file): commands=sorted_commands, monitor_channels=monitor_channels, channels_data=channels_data, - style=style_name + style=style_name, + link_css=link_css, + custom_css=custom_css ) output_path = os.path.join(output_dir, f'{style_name}.html') @@ -2730,6 +2815,18 @@ def main(): help='Generate sample pages for all styles with an index.html' ) + parser.add_argument( + '--link-css', + metavar='URL', + help='Link a stylesheet after the built-in style so its rules override it' + ) + + parser.add_argument( + '--embed-css', + metavar='FILE', + help='Inline CSS from a local file after the built-in style so its rules override it' + ) + args = parser.parse_args() # Handle --list-styles flag @@ -2737,9 +2834,19 @@ def main(): list_styles() sys.exit(0) + # Read --embed-css up front so a bad path fails before plugins load + custom_css = None + if args.embed_css: + try: + with open(args.embed_css, encoding='utf-8') as f: + custom_css = f.read() + except (OSError, UnicodeDecodeError) as e: + logger.error(f"Could not read --embed-css file {args.embed_css}: {e}") + sys.exit(1) + # Handle --sample flag if args.sample: - generate_samples(args.config) + generate_samples(args.config, link_css=args.link_css, custom_css=custom_css) sys.exit(0) config_file = args.config @@ -2774,7 +2881,7 @@ def main(): logger.info(f"Monitor channels: {monitor_channels}") # Setup minimal bot for plugin loading - minimal_bot = MinimalBot(config, logger) + minimal_bot = MinimalBot(config, logger, bot_root=bot_root) # Initialize database manager if database exists db_path = get_database_path(config, bot_root) @@ -2789,14 +2896,16 @@ def main(): logger.info("No database found, using default command ordering") minimal_bot.db_manager = None + local_commands_dir = resolve_local_commands_dir(config, bot_root, logger) + # Load plugins logger.info("Loading command plugins...") - plugin_loader = PluginLoader(minimal_bot) + plugin_loader = PluginLoader(minimal_bot, local_commands_dir=local_commands_dir) commands = plugin_loader.load_all_plugins() logger.info(f"Loaded {len(commands)} commands") - # Filter out admin and hidden commands - filtered_commands = filter_commands(commands, admin_commands) + # Filter out admin, hidden, and disabled commands + filtered_commands = filter_commands(commands, admin_commands, config) filtered_commands.update(get_randomline_commands(config)) logger.info(f"Filtered to {len(filtered_commands)} public commands") @@ -2822,7 +2931,7 @@ def main(): # Generate HTML logger.info("Generating HTML...") - html_content = generate_html(bot_name, title, introduction, sorted_commands, monitor_channels, channels_data, style) + html_content = generate_html(bot_name, title, introduction, sorted_commands, monitor_channels, channels_data, style, args.link_css, custom_css) # Create website directory website_dir = os.path.join(bot_root, "website") diff --git a/mkdocs.yml b/mkdocs.yml index 0e0561a..2d8215b 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -67,6 +67,7 @@ nav: - Overview: command-reference.md - Repeater Commands: repeater-commands.md - Custom Command Website: command-reference-website.md + - Developing Commands: develop-command-scripts.md - Service Plugins: - Overview: service-plugins.md - Discord Bridge: discord-bridge.md diff --git a/modules/channel_manager.py b/modules/channel_manager.py index b065783..6d72363 100644 --- a/modules/channel_manager.py +++ b/modules/channel_manager.py @@ -49,15 +49,26 @@ class ChannelManager: """Lowercase channel name without a leading # for cache lookups.""" return name.removeprefix("#").lower() - async def fetch_channels(self): - """Fetch channels from the MeshCore node using enhanced concurrent fetching""" + async def fetch_channels(self, max_attempts: int = 3, retry_delay: float = 2.0) -> bool: + """Fetch configured channels, retrying when the node returns no usable data.""" self.logger.info("Fetching channels from MeshCore node using enhanced concurrent method...") - try: - # Wait a moment for the device to be ready - await asyncio.sleep(2) + if max_attempts < 1: + raise ValueError("max_attempts must be at least 1") - # Fetch all channels concurrently - channels = await self.fetch_all_channels(force_refresh=True) + # Give the companion protocol a moment to settle after transport connect. + await asyncio.sleep(2) + + for attempt in range(1, max_attempts + 1): + try: + channels = await self.fetch_all_channels(force_refresh=True) + except Exception as e: + channels = [] + self.logger.warning( + "Channel fetch attempt %d/%d failed: %s", + attempt, + max_attempts, + e, + ) if channels: self.logger.info(f"Successfully fetched {len(channels)} channels from MeshCore node") @@ -68,13 +79,25 @@ class ChannelManager: self.logger.info(f" Channel {channel_idx}: {channel_name}") else: self.logger.debug(f" Channel {channel_idx}: (empty)") - else: - self.logger.warning("No channels found on MeshCore node") - self.bot.meshcore.channels = {} + return True - except Exception as e: - self.logger.error(f"Failed to fetch channels: {e}") + self._cache_valid = False self.bot.meshcore.channels = {} + if attempt < max_attempts: + self.logger.warning( + "No channels received from MeshCore node on attempt %d/%d; " + "retrying in %.1fs", + attempt, + max_attempts, + retry_delay, + ) + await asyncio.sleep(retry_delay) + + self.logger.error( + "No channels received from MeshCore node after %d attempt(s)", + max_attempts, + ) + return False async def fetch_all_channels(self, force_refresh: bool = False) -> list[dict[str, Any]]: """ @@ -136,14 +159,15 @@ class ChannelManager: break continue - self._cache_valid = True + self._cache_valid = bool(valid_channels) self.logger.info(f"Successfully fetched {len(valid_channels)} channels") # Update the bot's meshcore channels for compatibility self.bot.meshcore.channels = self._channels_cache - # Store channels in database for web viewer access - self._store_channels_in_db(valid_channels) + # Do not erase the last known channel table on a transient all-empty scan. + if valid_channels: + self._store_channels_in_db(valid_channels) return valid_channels diff --git a/modules/command_manager.py b/modules/command_manager.py index b2954bf..1cb32ec 100644 --- a/modules/command_manager.py +++ b/modules/command_manager.py @@ -5,6 +5,7 @@ Handles all bot commands, keyword matching, and response generation """ import asyncio +import contextlib import random import time from dataclasses import dataclass @@ -1347,37 +1348,50 @@ class CommandManager: scope_to_use, scope_source, ) - if not scope_is_global and not hasattr(self.bot.meshcore.commands, "set_flood_scope"): + scoped = not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope") + if not scope_is_global and not scoped: self.logger.warning( "Regional flood scope %r requested but meshcore.commands.set_flood_scope " "is unavailable; channel message will use device default (often global flood)", scope_to_use, ) - elif not scope_is_global: - _scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use) - if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR": - self.logger.warning( - "set_flood_scope(%s) failed (result=%s); " - "message will be sent with current firmware scope", - scope_to_use, _scope_result, - ) target = f"{channel} (channel {channel_num})" # Retry on no_event_received: max 2 extra attempts, 2s apart _max_retries = 2 for _attempt in range(_max_retries + 1): - try: - result = await self.bot.meshcore.commands.send_chan_msg( - channel_num, content, - timestamp=int(timestamp.timestamp()) if timestamp else None, - ) - finally: - if not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope"): - _restore_result = await self.bot.meshcore.commands.set_flood_scope("*") - if _restore_result is None or getattr(_restore_result, "type", None) == "ERROR": - self.logger.warning( - "set_flood_scope('*') restore failed (result=%s)", _restore_result - ) + # Hold the radio from set to restore so no other send goes out + # under this message's scope. + async with self.bot.radio_session() if scoped else contextlib.nullcontext(): + # The set is inside the try as well: a raising set_flood_scope + # would otherwise leave the device pinned to this region, and + # every later send would go out under it. + try: + if scoped: + _scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use) + if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR": + if _attempt == 0: + self.logger.warning( + "set_flood_scope(%s) failed (result=%s); " + "message will be sent with current firmware scope", + scope_to_use, _scope_result, + ) + else: + self.logger.warning( + "set_flood_scope(%s) failed on retry re-apply (result=%s)", + scope_to_use, _scope_result, + ) + result = await self.bot.meshcore.commands.send_chan_msg( + channel_num, content, + timestamp=int(timestamp.timestamp()) if timestamp else None, + ) + finally: + if scoped: + _restore_result = await self.bot.meshcore.commands.set_flood_scope("*") + if _restore_result is None or getattr(_restore_result, "type", None) == "ERROR": + self.logger.warning( + "set_flood_scope('*') restore failed (result=%s)", _restore_result + ) if self._is_no_event_received(result) and _attempt < _max_retries: self.logger.warning( @@ -1385,14 +1399,6 @@ class CommandManager: f"(attempt {_attempt + 1}/{_max_retries + 1}), retrying in 2s" ) await asyncio.sleep(2) - # Re-apply scope for next attempt - if not scope_is_global and hasattr(self.bot.meshcore.commands, "set_flood_scope"): - _scope_result = await self.bot.meshcore.commands.set_flood_scope(scope_to_use) - if _scope_result is None or getattr(_scope_result, "type", None) == "ERROR": - self.logger.warning( - "set_flood_scope(%s) failed on retry re-apply (result=%s)", - scope_to_use, _scope_result, - ) continue break diff --git a/modules/commands/contact_command.py b/modules/commands/contact_command.py new file mode 100644 index 0000000..1238386 --- /dev/null +++ b/modules/commands/contact_command.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +Contact command for the MeshCore Bot +Adds the bot contact info to the current channel +""" + +import re +from typing import Any, Optional + +from modules.commands.base_command import BaseCommand +from modules.models import MeshMessage + +PUBLIC_KEY_RE = re.compile(r'^[0-9a-fA-F]{64}$') + + +class ContactCommand(BaseCommand): + """Handles contact command""" + + # Plugin metadata + name = "contact" + keywords = ['contact'] + description = "Display the bot's contact information" + category = "basic" + + # Documentation + short_description = "Display the bot's contact information" + usage = "contact" + examples = [ + "contact" + ] + + def __init__(self, bot): + """Initialize the contact command. + + Args: + bot: The bot instance. + """ + super().__init__(bot) + self.enabled = self.get_config_value('Contact_Command', 'enabled', fallback=True, value_type='bool') + + def can_execute(self, message: MeshMessage, skip_channel_check: bool = False) -> bool: + """Check if this command can be executed with the given message. + + Args: + message: The message triggering the command. + skip_channel_check: If True, skip the channel check. + + Returns: + bool: True if command is enabled and checks pass, False otherwise. + """ + if not self.enabled: + return False + return super().can_execute(message, skip_channel_check=skip_channel_check) + + def get_help_text(self) -> str: + """Get help text for the contact command. + + Returns: + str: Help text string. + """ + return self.translate('commands.contact.help') + + def matches_keyword(self, message: MeshMessage) -> bool: + """Match ``contact`` (or a configured alias) on its own, with no arguments. + + Args: + message: The received message. + + Returns: + bool: True if the message is a contact command, False otherwise. + """ + def _matches(content_lower: str) -> bool: + return any(content_lower == keyword.lower() for keyword in self.keywords) + + return self._cleaned_content_matches(message, _matches) + + def _self_info_value(self, key: str) -> Optional[str]: + """Read a field from the radio's self_info, which may be a dict or an object. + + Args: + key: The self_info field name. + + Returns: + str: The field value, or None if unavailable. + """ + meshcore: Any = getattr(self.bot, 'meshcore', None) + self_info = getattr(meshcore, 'self_info', None) if meshcore else None + if not self_info: + return None + if isinstance(self_info, dict): + value = self_info.get(key) + else: + value = getattr(self_info, key, None) + return str(value).strip() if value else None + + async def execute(self, message: MeshMessage) -> bool: + """Execute the contact command. + + Args: + message: The message triggering the command. + + Returns: + bool: True if executed successfully, False otherwise. + """ + public_key = self._self_info_value('public_key') + name = self._self_info_value('name') or self._self_info_value('adv_name') + + if not public_key or not PUBLIC_KEY_RE.match(public_key): + self.logger.warning("Contact command: no usable public key in self_info") + return await self.send_response(message, self.translate('commands.contact.unavailable')) + + if not name: + self.logger.warning("Contact command: no device name in self_info") + return await self.send_response(message, self.translate('commands.contact.unavailable')) + + return await self.send_response(message, f"<{public_key.lower()}:1:{name}>") diff --git a/modules/commands/hello_command.py b/modules/commands/hello_command.py index 6d4158a..a7767cd 100644 --- a/modules/commands/hello_command.py +++ b/modules/commands/hello_command.py @@ -10,6 +10,7 @@ import re from typing import Any from ..models import MeshMessage +from ..security_utils import sanitize_name from ..utils import get_config_timezone from .base_command import BaseCommand @@ -23,6 +24,23 @@ class HelloCommand(BaseCommand): description = "Responds to greetings with robot-themed responses" category = "basic" + # How the sender is addressed when include_sender is on. The greeting already + # has a slot for the person being greeted — the random human descriptor + # ("pal", "carbon-based lifeform") -- so the mention is substituted there + # rather than prepended, which keeps the translated sentence intact. + SENDER_MENTION_FORMAT = "@[{sender}]" + + # message_handler falls back to this literal when a channel packet carries no + # "Name: body" prefix. It names nobody, so it must never be echoed as a mention. + UNKNOWN_CHANNEL_SENDER = "Channel User" + + # Web-viewer settings schema (see modules/settings_schema.py) + settings_schema = [ + {"key": "include_sender", "label": "Address the sender", "type": "bool", + "default": False, + "help": "Name the user in the reply, so a busy channel can tell whose greeting was answered."}, + ] + # Documentation short_description = "Responds to greetings with robot-themed responses" usage = "hello" @@ -38,10 +56,38 @@ class HelloCommand(BaseCommand): # Load configuration self.hello_enabled = self.get_config_value('Hello_Command', 'enabled', fallback=True, value_type='bool') + self.include_sender = self.get_config_value('Hello_Command', 'include_sender', fallback=False, value_type='bool') # Fallback arrays if translations not available self._init_fallback_arrays() + def get_sender_mention(self, message: MeshMessage) -> str: + """Build the @[name] mention for the sender of *message*. + + Args: + message: The message being answered. + + Returns: + str: The mention, or "" when the sender must not be named — the + setting is off, the reply is a DM (the recipient is the only + other party), or the name is missing/unresolvable. + """ + if not self.include_sender or message.is_dm: + return "" + + # Channel sender names are parsed straight off the wire in + # message_handler.handle_channel_message and are not sanitized there, so + # treat them as hostile. Control characters go; so do brackets, which + # would let a crafted name close the mention and forge a second one, and + # braces, which would raise from the .format(bot_name=...) the assembled + # reply still has to go through. + sender = sanitize_name(message.sender_id or "") + sender = sender.translate(str.maketrans("", "", "[]{}")).strip() + if not sender or sender == self.UNKNOWN_CHANNEL_SENDER: + return "" + + return self.SENDER_MENTION_FORMAT.format(sender=sender) + def _init_fallback_arrays(self) -> None: """Initialize fallback arrays for when translations are not available.""" # Time-neutral greeting openings @@ -297,24 +343,54 @@ class HelloCommand(BaseCommand): # Strip mentions from content for processing content = self._strip_mentions(message.content) + mention = self.get_sender_mention(message) + # Build the reply in the sender's language when auto-detection is enabled. # The full response string is assembled inside the context manager; the # await (send_response) happens after it, keeping the override window # free of suspension points. with self.respond_in_sender_language(message): - # Check if message is emoji-only (after stripping mentions) - if self.is_emoji_only_message(content): - response = self.get_emoji_response(content, bot_name) - else: - # Get random robot greeting - random_greeting = self.get_random_greeting() - response_format = self.translate('commands.hello.response_format') - response = f"{random_greeting} {response_format}".format(bot_name=bot_name) + response = self._build_response(content, bot_name, mention) + + # Nothing downstream truncates, so an over-budget reply is cut by the + # firmware mid-word. A long node name can push the reply past the + # channel body budget; drop the mention rather than lose the tail. + if mention and len(response.encode('utf-8')) > self.get_max_message_length(message): + response = self._build_response(content, bot_name, "") return await self.send_response(message, response) - def get_random_greeting(self) -> str: - """Generate a random robot greeting by combining opening and descriptor""" + def _build_response(self, content: str, bot_name: str, mention: str) -> str: + """Assemble the reply for *content*, addressing *mention* when set. + + Args: + content: The incoming message body, with mentions already stripped. + bot_name: Name to substitute into the response format. + mention: The sender's @[name] mention, or "" to leave them unnamed. + + Returns: + str: The assembled reply. + """ + # Check if message is emoji-only (after stripping mentions) + if self.is_emoji_only_message(content): + return self.get_emoji_response(content, bot_name, mention) + + # Get random robot greeting + random_greeting = self.get_random_greeting(mention) + response_format = self.translate('commands.hello.response_format') + return f"{random_greeting} {response_format}".format(bot_name=bot_name) + + def get_random_greeting(self, mention: str = "") -> str: + """Generate a random robot greeting by combining opening and descriptor. + + Args: + mention: When set, the sender's @[name] mention is used in place of + the random human descriptor, so the reply addresses the + person by name inside the sentence the translation defines. + + Returns: + str: The assembled greeting. + """ tz, _ = get_config_timezone(self.bot.config, self.logger) current_time = datetime.datetime.now(tz) @@ -339,7 +415,7 @@ class HelloCommand(BaseCommand): greeting_pool = evening_greetings + greeting_openings opening = random.choice(greeting_pool) - descriptor = random.choice(human_descriptors) + descriptor = mention or random.choice(human_descriptors) # Add some variety in punctuation and formatting punctuation_options = ["!", ".", "!", "!", "!"] # Favor exclamation marks @@ -381,8 +457,19 @@ class HelloCommand(BaseCommand): # Call parent can_execute() which includes channel checking, cooldown, etc. return super().can_execute(message) - def get_emoji_response(self, text: str, bot_name: str) -> str: - """Get appropriate response for emoji-only message""" + def get_emoji_response(self, text: str, bot_name: str, mention: str = "") -> str: + """Get appropriate response for emoji-only message. + + Args: + text: The emoji-only message body. + bot_name: Name to substitute into the response format. + mention: When set, the sender's @[name] mention. The canned emoji + lines have no descriptor slot to substitute it into, so it + is prefixed instead; they are short enough to afford it. + + Returns: + str: The assembled response. + """ import random # Get emoji responses from translations or fallback @@ -395,8 +482,9 @@ class HelloCommand(BaseCommand): # Check if this emoji has special responses if first_emoji in emoji_responses: response = random.choice(emoji_responses[first_emoji]) - return f"{response} {response_format}".format(bot_name=bot_name) + prefix = f"{mention} " if mention else "" + return f"{prefix}{response} {response_format}".format(bot_name=bot_name) else: # Use random greeting generator for general emojis - random_greeting = self.get_random_greeting() + random_greeting = self.get_random_greeting(mention) return f"{random_greeting} {response_format}".format(bot_name=bot_name) diff --git a/modules/config_schema.py b/modules/config_schema.py index f500dc2..f5489c0 100644 --- a/modules/config_schema.py +++ b/modules/config_schema.py @@ -50,8 +50,8 @@ LEGACY_ENABLED_KEY_RE = re.compile(r"^[a-z]+_enabled$") # behavior and the UI can never disagree. # Maps canonical section -> ordered ((legacy_section, legacy_key), ...). LEGACY_ENABLED_ALIASES: dict[str, tuple[tuple[str, str], ...]] = { - "Joke_Command": (("Jokes", "joke_enabled"),), - "DadJoke_Command": (("Jokes", "dadjoke_enabled"),), + "Joke_Command": (("Joke_Command", "joke_enabled"), ("Jokes", "joke_enabled")), + "DadJoke_Command": (("DadJoke_Command", "dadjoke_enabled"), ("Jokes", "dadjoke_enabled")), "Stats_Command": (("Stats_Command", "stats_enabled"), ("Stats", "stats_enabled")), "Sports_Command": (("Sports_Command", "sports_enabled"), ("Sports", "sports_enabled")), "Hacker_Command": (("Hacker_Command", "hacker_enabled"), ("Hacker", "hacker_enabled")), diff --git a/modules/core.py b/modules/core.py index ea17513..171c304 100644 --- a/modules/core.py +++ b/modules/core.py @@ -7,8 +7,9 @@ Contains the main bot class and message processing logic import asyncio import atexit import configparser +import contextlib +import contextvars import functools -import inspect import json import logging import signal @@ -119,8 +120,15 @@ class _BotAdminServer(threading.Thread): self._bot.logger.error("BotAdminServer failed to start: %s", exc) -class _SerializedCommands: - """Serializing proxy around ``meshcore.commands``. +# True while the current task holds the radio through MeshCoreBot.radio_session(), +# so the frames it sends don't try to take the (non-reentrant) lock again. +_radio_session_held: contextvars.ContextVar[bool] = contextvars.ContextVar( + "radio_session_held", default=False +) + + +def _serialize_command_frames(bot: "MeshCoreBot", commands: Any) -> bool: + """Route every host->radio frame through the bot's radio command lock. The companion firmware processes one host serial frame per main-loop iteration and has no mid-frame resync: a burst of concurrent commands can @@ -130,37 +138,33 @@ class _SerializedCommands: (sends, channel/contact ops, scheduler ops, health probes, auto message fetch) with no shared serialization. - This proxy routes every coroutine command through a single per-bot lock and - enforces a minimum inter-command interval, guaranteeing at most one - in-flight companion frame at a time. Non-coroutine attributes are passed - through untouched, so library internals that read ``_sender_func``, - ``default_timeout`` etc. are unaffected. ``meshcore_cli.next_cmd`` calls are - serialized too, since they dispatch through this same ``commands`` object. + Every meshcore command writes its frame through ``CommandHandler.send()``, + which waits for the radio's immediate reply (OK, ERROR, MSG_SENT, ...). + Wrapping ``send`` on the handler instance serializes exactly that exchange + and paces frames by a minimum interval, so there is at most one in-flight + companion frame at a time. Library methods call ``self.send``, so composite + commands (``send_msg_with_retry``, ``req_*_sync``, ``send_login_sync``) and + ``meshcore_cli.next_cmd`` are covered too, while their waits for ACKs and + remote responses happen outside the lock and don't block other senders. + + Returns False when ``commands`` is already serialized. """ + send = commands.send + if getattr(send, "_radio_serialized", False): + return False - __slots__ = ("_bot", "_commands") + @functools.wraps(send) + async def _serialized_send(*args: Any, **kwargs: Any) -> Any: + if _radio_session_held.get(): + await bot._pace_radio_command() + return await send(*args, **kwargs) + async with bot._get_radio_cmd_lock(): + await bot._pace_radio_command() + return await send(*args, **kwargs) - def __init__(self, bot: "MeshCoreBot", commands: Any) -> None: - object.__setattr__(self, "_bot", bot) - object.__setattr__(self, "_commands", commands) - - def __getattr__(self, name: str) -> Any: - commands = object.__getattribute__(self, "_commands") - attr = getattr(commands, name) - if not inspect.iscoroutinefunction(attr): - return attr - bot = object.__getattribute__(self, "_bot") - - @functools.wraps(attr) - async def _serialized(*args: Any, **kwargs: Any) -> Any: - async with bot._get_radio_cmd_lock(): - await bot._pace_radio_command() - return await attr(*args, **kwargs) - - return _serialized - - def __setattr__(self, name: str, value: Any) -> None: - setattr(object.__getattribute__(self, "_commands"), name, value) + _serialized_send._radio_serialized = True # type: ignore[attr-defined] + commands.send = _serialized_send + return True class MeshCoreBot: @@ -1752,25 +1756,46 @@ long_jokes = false await asyncio.sleep(wait) self._radio_cmd_last_ts = time.monotonic() - def _install_command_serializer(self) -> None: - """Wrap ``meshcore.commands`` so every command is serialized + paced. + @contextlib.asynccontextmanager + async def radio_session(self): + """Hold the radio for a short sequence of frames that must not interleave. - Idempotent and safe to call after each (re)connect. Wrapping the - ``commands`` attribute in place means existing call sites - (``self.meshcore.commands.*`` and ``meshcore_cli.next_cmd``) are - serialized automatically with no per-call changes. + Frames are serialized one at a time, so another task's frame can land + between two of ours. Use this when that matters, e.g. setting the flood + scope, sending, and restoring it, so no other send goes out under the + temporary scope. Keep it short: every other sender waits, so don't wait + for ACKs or remote responses inside it. Re-entering from the same task + is a no-op. Tasks created inside the session inherit it, so don't spawn + work that sends after the session ends. + """ + if _radio_session_held.get(): + yield + return + async with self._get_radio_cmd_lock(): + token = _radio_session_held.set(True) + try: + yield + finally: + _radio_session_held.reset(token) + + def _install_command_serializer(self) -> None: + """Serialize and pace every frame ``meshcore.commands`` writes. + + Idempotent and safe to call after each (re)connect. The handler is + wrapped in place, so existing call sites (``self.meshcore.commands.*`` + and ``meshcore_cli.next_cmd``) need no per-call changes. """ if not self.meshcore: return cmds = getattr(self.meshcore, "commands", None) - if cmds is None or isinstance(cmds, _SerializedCommands): + if cmds is None: return try: - self.meshcore.commands = _SerializedCommands(self, cmds) - self.logger.debug( - "Installed serialized command gateway (min interval %.0fms)", - self._radio_cmd_min_interval * 1000, - ) + if _serialize_command_frames(self, cmds): + self.logger.debug( + "Installed serialized command gateway (min interval %.0fms)", + self._radio_cmd_min_interval * 1000, + ) except (AttributeError, TypeError) as e: self.logger.warning(f"Could not install command serializer: {e}") @@ -1852,8 +1877,11 @@ long_jokes = false # Wait for contacts to load await self.wait_for_contacts() - # Fetch channels - await self.channel_manager.fetch_channels() + # A connected transport without channel data cannot route replies. + if not await self.channel_manager.fetch_channels(): + raise ConnectionError( + "MeshCore node returned no channels after retries" + ) # Setup message event handlers await self.setup_message_handlers() diff --git a/modules/message_handler.py b/modules/message_handler.py index c0d0f51..e6eeb81 100644 --- a/modules/message_handler.py +++ b/modules/message_handler.py @@ -9,6 +9,7 @@ import copy import hmac as hmac_mod import time from collections import OrderedDict +from collections.abc import Iterable from hashlib import sha256 from typing import Any, TypedDict @@ -485,6 +486,7 @@ class MessageHandler: """Return configured channel indexes and keys without logging secrets.""" meshcore = getattr(self.bot, "meshcore", None) channels = getattr(meshcore, "channels", None) + items: Iterable[tuple[int, Any]] if isinstance(channels, dict): items = channels.items() elif isinstance(channels, list): @@ -496,8 +498,11 @@ class MessageHandler: for fallback_idx, channel in items: if not isinstance(channel, dict): continue + # Annotated Any: with a mixed Any | int default, mypy matches .get()'s + # None-default overload and infers Any | None. int() rejects None anyway. + raw_idx: Any = channel.get("channel_idx", fallback_idx) try: - channel_idx = int(channel.get("channel_idx", fallback_idx)) + channel_idx = int(raw_idx) except (TypeError, ValueError): continue diff --git a/modules/neighbors_discovery.py b/modules/neighbors_discovery.py index 0ee23af..3b7b959 100644 --- a/modules/neighbors_discovery.py +++ b/modules/neighbors_discovery.py @@ -18,25 +18,23 @@ link* between two full 32-byte public keys, with a measured SNR. That is stronger evidence than anything path inference can offer, and it costs one radio command plus a passive listen window. -Stage 2 (scopes) is deliberately optional here and defaults off, for two reasons -that do not apply to the upstream capture tool: +Stage 2 (scopes) is deliberately optional here and defaults off, for a reason +that does not apply to the upstream capture tool: upstream relies on a freshly +discovered neighbor *not* being a known contact, which is what makes +``send_anon_req`` ask for a zero-hop reply path. This bot populates the +library's contact cache. For a contact with no path (``out_path_len == -1``, +the common case for a flood repeater) the library reaches zero-hop by calling +``change_contact_path()`` and then ``reset_path()`` -- i.e. it *mutates the +device's contact table* per neighbor. Those two calls are not paired by +``try``/``finally`` upstream, so a request cut short between them leaves the +contact pinned to zero-hop; collect_scopes restores it itself (see +``_restore_flood_path``). -* ``req_regions_sync`` awaits its reply inside the call, and every bot command - is serialized through ``modules.core._SerializedCommands``, so one scope - request holds the radio for its whole round trip -- stalling message sends. -* Upstream relies on a freshly discovered neighbor *not* being a known contact, - which is what makes ``send_anon_req`` ask for a zero-hop reply path. This bot - populates the library's contact cache. For a contact with no path - (``out_path_len == -1``, the common case for a flood repeater) the library - reaches zero-hop by calling ``change_contact_path()`` and then - ``reset_path()`` -- i.e. it *mutates the device's contact table* per neighbor. - Those two calls are not paired by ``try``/``finally`` upstream, so a request - cut short between them leaves the contact pinned to zero-hop; collect_scopes - restores it itself (see ``_restore_flood_path``). - -No device-command lock is passed around: unlike upstream, every coroutine on -``meshcore.commands`` is already serialized and paced by -``modules.core._SerializedCommands``. +No device-command lock is passed around: unlike upstream, every frame +``meshcore.commands`` writes is already serialized and paced by +``modules.core._serialize_command_frames``. The lock covers each frame and its +immediate reply only, so waiting for a neighbor's scope reply doesn't stall the +bot's other sends. """ from __future__ import annotations @@ -484,9 +482,9 @@ async def collect_scopes( pinned_to_zero_hop = _contact_has_no_path(meshcore, entry.pubkey) try: - # Bounded: this holds the shared radio command lock for its whole - # round trip, and an unbounded stall here would block every other - # bot command for as long as the write hangs. + # Bounded: a hung write holds the shared radio command lock, and an + # unbounded stall here would block every other bot command for as + # long as the write hangs. scopes = await asyncio.wait_for( meshcore.commands.req_regions_sync( entry.pubkey, diff --git a/pyproject.toml b/pyproject.toml index 0c1f113..719580b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,8 +33,8 @@ dependencies = [ "maidenhead>=1.4.0", "pytz>=2023.3", "aiohttp>=3.8.0", - # 2.3.8+ required for zero-hop neighbour discovery (see requirements.txt) - "meshcore>=2.3.8", + # 2.3.14+ required for send_msg_with_retry's ACK handling (see requirements.txt) + "meshcore>=2.3.14", "openmeteo-requests>=1.7.2", "requests-cache>=1.1.1", "retry-requests>=1.0.0", diff --git a/requirements.txt b/requirements.txt index ac35df8..b9de362 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,9 +13,14 @@ geopy>=2.3.0 maidenhead>=1.4.0 pytz>=2023.3 aiohttp>=3.8.0 -# 2.3.8+ is required, not merely preferred: send_node_discover_req / req_regions_sync -# (zero-hop neighbour discovery) and the bounded, serialised BLE write both landed there. -meshcore>=2.3.8 +# 2.3.14+ is required, not merely preferred. Before 2.3.13, send_msg_with_retry +# reported ACKed DMs as failures: it subscribed to the ACK only after send_msg +# returned, and each attempt accepted only its own ACK code, so an ACK that arrived +# right behind MSG_SENT or answered an earlier attempt was ignored (meshcore_py#108). +# 2.3.14 additionally fixes send_cmd's destination-type handling. 2.3.8 first brought +# send_node_discover_req / req_regions_sync (zero-hop neighbor discovery) and the +# bounded, serialized BLE write. +meshcore>=2.3.14 openmeteo-requests>=1.7.2 requests-cache>=1.1.1 retry-requests>=1.0.0 diff --git a/tests/commands/test_hello_command.py b/tests/commands/test_hello_command.py index da354bc..79f276a 100644 --- a/tests/commands/test_hello_command.py +++ b/tests/commands/test_hello_command.py @@ -81,3 +81,90 @@ class TestHelloCommand: assert call_args is not None response = call_args[0][1] assert isinstance(response, str) and len(response) > 0 + + +class TestHelloCommandIncludeSender: + """Tests for the Hello_Command include_sender setting.""" + + @staticmethod + def _command(command_mock_bot, include_sender): + command_mock_bot.config.add_section("Hello_Command") + command_mock_bot.config.set("Hello_Command", "enabled", "true") + command_mock_bot.config.set("Hello_Command", "include_sender", str(include_sender).lower()) + return HelloCommand(command_mock_bot) + + def test_mention_empty_when_setting_off(self, command_mock_bot): + cmd = self._command(command_mock_bot, False) + assert cmd.get_sender_mention(mock_message(sender_id="Bob")) == "" + + def test_mention_built_for_channel_sender(self, command_mock_bot): + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id="Bob")) == "@[Bob]" + + def test_mention_empty_for_dm(self, command_mock_bot): + """A DM has one recipient; naming them spends airtime to say nothing.""" + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id="Bob", is_dm=True)) == "" + + def test_mention_empty_for_unnamed_channel_sender(self, command_mock_bot): + """message_handler uses this literal when the packet has no name prefix.""" + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id="Channel User")) == "" + + def test_mention_empty_when_sender_id_missing(self, command_mock_bot): + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id=None)) == "" + + def test_mention_strips_brackets_that_would_forge_a_second_mention(self, command_mock_bot): + """Channel names come off the wire unsanitized (handle_channel_message).""" + cmd = self._command(command_mock_bot, True) + mention = cmd.get_sender_mention(mock_message(sender_id="Bob] hi @[Alice")) + assert mention == "@[Bob hi @Alice]" + assert mention.count("[") == 1 and mention.count("]") == 1 + + def test_mention_strips_control_characters(self, command_mock_bot): + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id="Bo\nb\x00")) == "@[Bob]" + + def test_mention_strips_braces_that_would_break_formatting(self, command_mock_bot): + """The mention is embedded before .format(bot_name=...) runs.""" + cmd = self._command(command_mock_bot, True) + assert cmd.get_sender_mention(mock_message(sender_id="Bo{b}")) == "@[Bob]" + + def test_greeting_uses_mention_in_place_of_descriptor(self, command_mock_bot): + cmd = self._command(command_mock_bot, True) + with patch("modules.commands.hello_command.random.choice", side_effect=lambda x: x[0]): + with_mention = cmd.get_random_greeting("@[Bob]") + without = cmd.get_random_greeting() + assert "@[Bob]" in with_mention + # The name takes the descriptor's slot rather than riding in front of it. + assert not with_mention.startswith("@[Bob]") + assert with_mention.split()[0] == without.split()[0] + + def test_emoji_response_prefixes_mention(self, command_mock_bot): + """Canned emoji lines have no descriptor slot, so the mention leads.""" + cmd = self._command(command_mock_bot, True) + with patch("modules.commands.hello_command.random.choice", side_effect=lambda x: x[0]): + result = cmd.get_emoji_response("🖖", "TestBot", "@[Bob]") + assert result.startswith("@[Bob] ") + + @pytest.mark.asyncio + async def test_execute_names_channel_sender(self, command_mock_bot): + cmd = self._command(command_mock_bot, True) + msg = mock_message(content="hello", sender_id="Bob") + with patch("modules.commands.hello_command.random.choice", side_effect=lambda x: x[0]): + assert await cmd.execute(msg) is True + response = command_mock_bot.command_manager.send_response.call_args[0][1] + assert "@[Bob]" in response + + @pytest.mark.asyncio + async def test_execute_drops_mention_that_would_overflow_the_budget(self, command_mock_bot): + """Nothing downstream truncates, so an over-budget reply is cut on air.""" + cmd = self._command(command_mock_bot, True) + msg = mock_message(content="hello", sender_id="B" * 60) + with patch.object(cmd, "get_max_message_length", return_value=60): + with patch("modules.commands.hello_command.random.choice", side_effect=lambda x: x[0]): + assert await cmd.execute(msg) is True + response = command_mock_bot.command_manager.send_response.call_args[0][1] + assert "@[" not in response + assert len(response.encode("utf-8")) <= 60 diff --git a/tests/integration/test_flood_scope_reply.py b/tests/integration/test_flood_scope_reply.py index 9993a90..4249586 100644 --- a/tests/integration/test_flood_scope_reply.py +++ b/tests/integration/test_flood_scope_reply.py @@ -556,3 +556,45 @@ class TestWildcardOnlyAllowlist: def test_unset_config_configures_no_allowlist(self): keys, allow_global = self._cmd_mgr("") assert bool(keys or allow_global) is False + + +@pytest.mark.asyncio +async def test_send_channel_message_restores_scope_when_set_raises(): + """A raising set_flood_scope must still restore global flood. + + Otherwise the device stays pinned to the region and every later send — + channel replies, DMs, scheduled sends — goes out under that scope. + """ + bot = MagicMock() + bot.logger = Mock() + bot.config = make_config() + bot.connected = True + bot.meshcore = MagicMock() + bot.is_radio_zombie = False + bot.is_radio_offline = False + bot.channel_manager = MagicMock() + bot.channel_manager.get_channel_number = Mock(return_value=0) + + cm = object.__new__(CommandManager) + cm.bot = bot + cm.logger = bot.logger + + async def _set_flood_scope(value): + if value != "*": + raise RuntimeError("radio went away") + return MagicMock(type="OK") + + set_flood_scope = AsyncMock(side_effect=_set_flood_scope) + send_chan_msg = AsyncMock(return_value=MagicMock(type="OK", payload={})) + bot.meshcore.commands.set_flood_scope = set_flood_scope + bot.meshcore.commands.send_chan_msg = send_chan_msg + + cm._check_rate_limits = AsyncMock(return_value=(True, None)) + cm._is_no_event_received = Mock(return_value=False) + cm._handle_send_result = Mock(return_value=True) + + assert await cm.send_channel_message("general", "hi", scope="west") is False + + scopes_set = [c.args[0] for c in set_flood_scope.await_args_list if c.args] + assert scopes_set == ["#west", "*"] + send_chan_msg.assert_not_awaited() diff --git a/tests/test_channel_manager_logic.py b/tests/test_channel_manager_logic.py index b0a08c7..bbbcbfd 100644 --- a/tests/test_channel_manager_logic.py +++ b/tests/test_channel_manager_logic.py @@ -187,6 +187,38 @@ class TestFetchAllChannelsCacheLifecycle: # Should have aborted before fetching all 8 channels assert call_count < cm.max_channels assert channels == [] + assert cm._cache_valid is False + cm._store_channels_in_db.assert_not_called() + + +class TestFetchChannelsRetry: + """A connected node with no returned channels must not look ready (#266).""" + + @pytest.mark.asyncio + async def test_retries_empty_scan_then_succeeds(self, cm): + cm.bot.connected = True + channel = {"channel_name": "general", "channel_idx": 0} + cm.fetch_all_channels = AsyncMock(side_effect=[[], [channel]]) + + with patch("asyncio.sleep", new_callable=AsyncMock): + result = await cm.fetch_channels(max_attempts=3, retry_delay=0) + + assert result is True + assert cm.fetch_all_channels.await_count == 2 + + @pytest.mark.asyncio + async def test_exhausted_empty_scans_fail_and_invalidate_cache(self, cm): + cm.bot.connected = True + cm._cache_valid = True + cm.fetch_all_channels = AsyncMock(return_value=[]) + + with patch("asyncio.sleep", new_callable=AsyncMock): + result = await cm.fetch_channels(max_attempts=3, retry_delay=0) + + assert result is False + assert cm.fetch_all_channels.await_count == 3 + assert cm._cache_valid is False + assert cm.bot.meshcore.channels == {} # --------------------------------------------------------------------------- diff --git a/tests/test_contact_command.py b/tests/test_contact_command.py new file mode 100644 index 0000000..d21a9ed --- /dev/null +++ b/tests/test_contact_command.py @@ -0,0 +1,102 @@ +"""Tests for modules.commands.contact_command.""" + +import asyncio +import configparser +from unittest.mock import AsyncMock, MagicMock, Mock + +from modules.commands.contact_command import ContactCommand +from tests.conftest import mock_message + +PUBKEY = "f5d2b56d19b24412756933e917d4632e088cdd5daeadc9002feca73bf5d2b56d" + + +def _make_bot(self_info=None): + bot = MagicMock() + bot.logger = Mock() + config = configparser.ConfigParser() + config.add_section("Bot") + config.set("Bot", "bot_name", "TestBot") + config.set("Bot", "respond_to_mentions", "also") + config.add_section("Channels") + config.set("Channels", "monitor_channels", "general") + config.set("Channels", "respond_to_dms", "true") + config.add_section("Keywords") + config.add_section("Contact_Command") + config.set("Contact_Command", "enabled", "true") + bot.config = config + bot.translator = MagicMock() + bot.translator.translate = Mock(side_effect=lambda key, **kw: key) + bot.command_manager = MagicMock() + bot.command_manager.monitor_channels = ["general"] + bot.meshcore = MagicMock() + bot.meshcore.self_info = self_info + return bot + + +def _make_command(self_info=None): + cmd = ContactCommand(_make_bot(self_info)) + cmd.send_response = AsyncMock(return_value=True) + return cmd + + +class TestMatching: + def test_matches_bare_keyword(self): + cmd = _make_command() + assert cmd.matches_keyword(mock_message("contact")) is True + + def test_does_not_match_with_arguments(self): + cmd = _make_command() + assert cmd.matches_keyword(mock_message("contact me")) is False + + def test_matches_configured_alias(self): + bot = _make_bot() + bot.config.set("Contact_Command", "aliases", "card") + cmd = ContactCommand(bot) + assert cmd.matches_keyword(mock_message("card")) is True + + def test_non_match_leaves_message_content_untouched(self): + # Regression guard for #267: the keyword scan must not rewrite + # overheard traffic when this command does not match. + cmd = _make_command() + message = mock_message("@[TestBot] wx 98101") + assert cmd.matches_keyword(message) is False + assert message.content == "@[TestBot] wx 98101" + + +class TestExecute: + def test_sends_contact_card(self): + cmd = _make_command({"public_key": PUBKEY, "name": "TestBot"}) + assert asyncio.run(cmd.execute(mock_message("contact"))) is True + cmd.send_response.assert_awaited_once() + assert cmd.send_response.await_args[0][1] == f"<{PUBKEY}:1:TestBot>" + + def test_reads_self_info_object(self): + self_info = MagicMock(spec=["public_key", "name"]) + self_info.public_key = PUBKEY.upper() + self_info.name = "TestBot" + cmd = _make_command(self_info) + assert asyncio.run(cmd.execute(mock_message("contact"))) is True + assert cmd.send_response.await_args[0][1] == f"<{PUBKEY}:1:TestBot>" + + def test_missing_self_info_reports_unavailable(self): + cmd = _make_command(None) + assert asyncio.run(cmd.execute(mock_message("contact"))) is True + assert cmd.send_response.await_args[0][1] == "commands.contact.unavailable" + + def test_malformed_public_key_reports_unavailable(self): + cmd = _make_command({"public_key": "nothex", "name": "TestBot"}) + asyncio.run(cmd.execute(mock_message("contact"))) + assert cmd.send_response.await_args[0][1] == "commands.contact.unavailable" + + def test_missing_name_reports_unavailable(self): + cmd = _make_command({"public_key": PUBKEY}) + asyncio.run(cmd.execute(mock_message("contact"))) + assert cmd.send_response.await_args[0][1] == "commands.contact.unavailable" + + +class TestEnabledFlag: + def test_disabled_blocks_execution(self): + bot = _make_bot({"public_key": PUBKEY, "name": "TestBot"}) + bot.config.set("Contact_Command", "enabled", "false") + cmd = ContactCommand(bot) + assert cmd.can_execute(mock_message("contact")) is False diff --git a/tests/test_core.py b/tests/test_core.py index fabbc9c..a60bef8 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -868,6 +868,28 @@ class TestTransportReconnect: assert bot._radio_relinks_in_progress == 0 assert bot.keep_running is False + def test_connect_fails_when_channel_fetch_never_returns_channels(self, tmp_path): + bot = self._make_bot(tmp_path, connection_type="serial") + new_meshcore = MagicMock() + new_meshcore.is_connected = True + new_meshcore.self_info = {} + new_meshcore.disconnect = AsyncMock() + + bot.wait_for_contacts = AsyncMock() + bot.channel_manager.fetch_channels = AsyncMock(return_value=False) + bot.setup_message_handlers = AsyncMock() + + with patch( + "modules.core.meshcore.MeshCore.create_serial", + AsyncMock(return_value=new_meshcore), + ): + assert asyncio.run(bot.connect()) is False + + new_meshcore.disconnect.assert_awaited_once() + bot.setup_message_handlers.assert_not_awaited() + assert bot.meshcore is None + assert bot.connected is False + def test_keep_running_through_web_viewer_reboot(self, tmp_path): bot = self._make_bot(tmp_path) bot.meshcore = MagicMock() diff --git a/tests/test_generate_website_include_css.py b/tests/test_generate_website_include_css.py new file mode 100644 index 0000000..692771a --- /dev/null +++ b/tests/test_generate_website_include_css.py @@ -0,0 +1,137 @@ +"""Tests for the --link-css and --embed-css options in generate_website.py""" + +import sys + +import pytest + +import generate_website +from generate_website import STYLES, generate_builtin_css, generate_html + + +def _render(**kwargs) -> str: + return generate_html( + bot_name="TestBot", + title="Test Site", + introduction="Test intro", + commands=[], + monitor_channels=[], + channels_data={}, + **kwargs, + ) + + +def test_generate_builtin_css_includes_css_variables(): + css = generate_builtin_css('default') + + assert ':root' in css + assert '--bg-primary' in css + assert '--accent-blue' in css + assert 'body {' in css + + +def test_generate_builtin_css_includes_style_overrides(): + css = generate_builtin_css('minimalist') + + assert STYLES['minimalist']['css_overrides'].strip() in css + + +def test_default_output_has_no_custom_css(): + html = _render(style='default') + + assert 'Custom CSS overrides' not in html + stylesheets = [line for line in html.splitlines() if 'rel="stylesheet"' in line] + assert len(stylesheets) == 1 + assert 'fonts.googleapis.com/css2' in stylesheets[0] + + +def test_backwards_compatible_call_without_css_params(): + html = generate_html("TestBot", "Test Site", "Test intro", [], [], {}, 'default') + + assert '') < html.index(link) < html.index('') + assert 'fonts.googleapis.com' in html + assert '--bg-primary' in html + + +def test_link_css_is_escaped(): + html = _render(style='default', link_css='style.css?v=1&theme=dark">') + + assert 'href="style.css?v=1&theme=dark"><script>' in html + assert '