diff --git a/README.md b/README.md index 59ee76e..3de9209 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ A Telegram bot for a personal Ecowitt weather station and AirGradient air-quality sensor. Ask it about the weather or air quality (text or charts); it also sends silent alerts to every chat it is in: rain starting/stopping, rain likely soon, gusts over 40 km/h, UV index of 10 or more, indoor/outdoor temperatures crossing (held back for the cooldown, then announced if it still stands), -and unhealthy air (with when it is safe again). Every alert carries Subscribe / Unsubscribe buttons that expand into the alert types (rain, rain predicted, gusts, UV, temperature crossing, air quality, pollen & asthma, forecast changes): Subscribe lists the types that are off, Unsubscribe the ones that are on; `/alerts` opens the same settings (and `/alerts on|off` still subscribes or unsubscribes every type). In a group only admins can change them (`lib/alerts/menu.py`). +and unhealthy air (with when it is safe again). Every alert carries Subscribe / Unsubscribe buttons that expand into the alert types (rain, rain predicted, gusts, UV, temperature crossing, particulates, pollen & asthma, forecast changes): Subscribe lists the types that are off, Unsubscribe the ones that are on; `/alerts` opens the same settings (and `/alerts on|off` still subscribes or unsubscribes every type); `/usage` is the Help & Alerts button. In a group anyone can open `/alerts`, but only admins can change anything (`lib/alerts/menu.py`). ## Run @@ -28,6 +28,8 @@ Under systemd see `ecobot.service` (it loads the environment file; the clone's l | `POLLEN=on` | optional: Melbourne grass pollen and thunderstorm asthma risk (scraped from melbournepollen.com.au) in the report, a `pollen_asthma` tool and an alert when High or Extreme. Off by default; `POLLEN_DISTRICT` picks the district (default `Central`, never printed) | | `PLACE` | the name shown after the Pollen & asthma and Forecast headings in the report (default `Melbourne`) | | `FORECAST=on` | optional: the Open-Meteo daily forecast in the report and a `weather_forecast` tool. Off by default; `FORECAST_LAT` / `FORECAST_LON` set the location, else the weather station's own is used | +| `ADMIN_ONLY` | on by default: the bot answers nobody (private chats and groups alike) except the admins and creators of the groups it is in; anyone else gets "Sorry, only group admins can use this bot." when they address it (ordinary group talk is ignored). The one exception is `/alerts` in a group, which anyone may open (its buttons and `/alerts on|off` stay admin-only). `ADMIN_ONLY=off` answers everyone, as before. The admins are asked of Telegram and remembered for 10 minutes; a group counts once the bot is added to it or someone speaks there | +| `ADMIN_CHAT_IDS` | optional, with `ADMIN_ONLY`: more ids allowed besides those admins (added to them, not instead of them), separated by commas or spaces. A user id (the same number as their private chat) lets someone in before the bot is in any group; a group id (negative) adds that group's admins and lets its anonymous admins (those who post "as the group") in | | `ALERT_COOLDOWN_MINUTES` | optional: the rain and temperature-crossing alerts: how long it must stay dry before "the rain has stopped" is sent, and the least time between alerts of the same kind (default 30, from 5 to 150). A change that comes inside the cooldown is held and announced when it ends, if the state still differs from the last one alerted; if it is back where it was, nothing is sent | | `RAIN_QUIET_HOURS` | optional: local hours with no rain alerts (started, stopped, predicted), as `0-6` (the default; `22-6` wraps midnight; `off` for none). Rain that fell in them is summed up in one message once they end | | `CHART_ALL_FEELS_LIKE=on` | optional: "weather all week" (every reading, a panel each, sent with just its title; UV is left out as it has the same shape as solar) also draws the feels-like panel; off by default, as it nearly repeats temperature. Naming either ("plot feels like", "plot solar and uv") always works | @@ -39,7 +41,8 @@ Under systemd see `ecobot.service` (it loads the environment file; the clone's l ``` ecobot.py wiring and lifecycle lib/config.py settings -lib/bot.py Telegram handlers and replies +lib/bot.py Telegram handlers and replies; every button press comes in through one entry point that checks it first +lib/admins.py who may use the bot with ADMIN_ONLY: the admins of the groups it is in, plus ADMIN_CHAT_IDS lib/llm.py, prompt.py tool-calling loop; system prompt lib/intent.py intent.read(text) -> one Reading: effort, needs data, fast path, chart asks, period hints lib/tools.py, warm.py tool registry and the per-question Turn; keep-warm helper shared by both sources diff --git a/ecobot.py b/ecobot.py index 0704a88..778b7b0 100644 --- a/ecobot.py +++ b/ecobot.py @@ -84,6 +84,12 @@ async def main(): tools = Tools([t for s in sources for t in s.tools] + (composer.tools if composer else [])) agent = Agent(AsyncOpenAI(api_key=cfg.xai_api_key, base_url=cfg.xai_base_url), cfg.xai_model, tools) state = AlertState(cfg.state_path) + if cfg.admin_only: + log.info("Admin only: the bot answers the admins of the groups it is in%s", + f" and {len(cfg.admin_chat_ids)} listed id(s)" if cfg.admin_chat_ids else "") + if not cfg.admin_chat_ids and not any(chat < 0 for chat in state.chats): + log.warning("Admin only is on but no group is known yet and ADMIN_CHAT_IDS is empty, so nobody can use the bot: " + "add it to a group, list your user id in ADMIN_CHAT_IDS, or set ADMIN_ONLY=off") irrigation = Tuya(cfg) if cfg.irrigation else None bot = Bot(cfg, agent, sources, state, irrigation) diff --git a/lib/admins.py b/lib/admins.py new file mode 100644 index 0000000..c4f0da6 --- /dev/null +++ b/lib/admins.py @@ -0,0 +1,56 @@ +"""Who may talk to the bot when ADMIN_ONLY is on: the admins of the groups it is in, and the ids listed in ADMIN_CHAT_IDS. + +A "known admin" is anyone Telegram lists as an administrator or creator of a group the bot knows (a group becomes known when the bot is +added to it or someone speaks in it) or of a group named by a negative id in ADMIN_CHAT_IDS. The lists are asked for once and kept for +ten minutes. ADMIN_CHAT_IDS is added to that, never instead of it: a positive id is a user (it bootstraps a bot that is in no group yet), +a negative id is a group, whose own admins are then looked up and whose anonymous admins are let in: a message an admin sends "as the +group" carries the group as its `sender_chat` and no real user, so it is allowed only when that group's id is listed. +""" + +import logging +import time + +log = logging.getLogger(__name__) + +TTL_SECONDS = 600 +RETRY_SECONDS = 60 # after a failed lookup, ask again after this long (not on every message) + + +class Admins: + def __init__(self, state, extra_ids=(), ttl: float = TTL_SECONDS, clock=time.monotonic): + self.state, self.extra, self.ttl, self.clock = state, frozenset(extra_ids), ttl, clock + self._cache: dict[int, tuple[float, frozenset[int]]] = {} # group -> (when fetched, its admins) + self._failing: set[int] = set() + + def groups(self) -> list[int]: + """The groups whose admins count: the ones the bot knows, and the ones ADMIN_CHAT_IDS names.""" + known = {c for c in (self.state.chats if self.state else {}) if c < 0} + return sorted(known | {i for i in self.extra if i < 0}) + + async def _admins_of(self, bot, chat_id: int) -> frozenset[int]: + cached = self._cache.get(chat_id) + if cached and self.clock() - cached[0] < self.ttl: + return cached[1] + try: + members = await bot.get_chat_administrators(chat_id) + ids = frozenset(m.user.id for m in members if not getattr(m.user, "is_bot", False)) + except Exception as e: # the bot was removed, Telegram is down: keep what was known, try again soon, say it once + if chat_id not in self._failing: + self._failing.add(chat_id) + log.warning("Couldn't list the admins of %s (%s: %s)", chat_id, type(e).__name__, e) + ids = cached[1] if cached else frozenset() + self._cache[chat_id] = (self.clock() - self.ttl + RETRY_SECONDS, ids) + return ids + self._failing.discard(chat_id) + self._cache[chat_id] = (self.clock(), ids) + return ids + + async def allowed(self, bot, sender_ids) -> bool: + """Is one of these ids (the sender's user id and, for a message sent as a chat, that chat's id) allowed?""" + ids = {i for i in sender_ids if isinstance(i, int)} + if ids & self.extra: + return True + for chat_id in self.groups(): + if ids & await self._admins_of(bot, chat_id): + return True + return False diff --git a/lib/alerts/menu.py b/lib/alerts/menu.py index 13fde7d..f7ea046 100644 --- a/lib/alerts/menu.py +++ b/lib/alerts/menu.py @@ -15,8 +15,11 @@ # kind -> the name on the button, in the order listed LABELS = {"rain": "rain", "rain_likely": "rain predicted", "gusts": "gusts", "uv": "UV", "temps": "temperature crossing", - "air": "air quality", "pollen": "pollen & asthma", "forecast": "forecast changes", + "air": "particulates", "pollen": "pollen & asthma", "forecast": "forecast changes", "irrigation": "irrigation battery", "irrigation_pause": "irrigation pause"} +# the emoji on each type's button: the one its own alerts start with (every type has one: a test holds the two lists together) +EMOJI = {"rain": "\U0001f327๏ธ", "rain_likely": "\U0001f326๏ธ", "gusts": "\U0001f4a8", "uv": "\U0001f9f4", "temps": "\U0001f321๏ธ", + "air": "\U0001f637", "pollen": "\U0001f33c", "forecast": "\U0001f504", "irrigation": "\U0001faab", "irrigation_pause": "โ˜”"} ALL = "all" TITLE = "๐Ÿ”” Alerts in this chat" PER_ROW = 2 @@ -30,6 +33,11 @@ def available_kinds(sources: set[str]) -> list[str]: return wanted +def label(kind: str) -> str: + """The name of an alert type on its button: its emoji, then its name.""" + return f"{EMOJI[kind]} {LABELS[kind]}" + + def _button(text: str, data: str) -> InlineKeyboardButton: return InlineKeyboardButton(text, callback_data=data) @@ -50,11 +58,11 @@ def keyboard(muted: set[str], available: list[str], open: str | None = None, par listed, show_all = [k for k in listed if k == parent], bool(others) else: show_all = len(listed) > 1 - mark = lambda k: f"โ— {LABELS[k]}" if open == "unsub" and k == parent else LABELS[k] # the alert these buttons are under + mark = lambda k: f"โ— {label(k)}" if open == "unsub" and k == parent else label(k) # the alert these buttons are under buttons = [_button(mark(k), f"al:{verb}:{k}:{open}{tail}") for k in listed] rows += map(list, batched(buttons, PER_ROW)) if show_all: - rows.append([_button("Subscribe to all" if open == "sub" else "Unsubscribe from all", f"al:{verb}:{ALL}:{open}{tail}")]) + rows.append([_button("๐Ÿ”” Subscribe to all" if open == "sub" else "๐Ÿ”• Unsubscribe from all", f"al:{verb}:{ALL}:{open}{tail}")]) return InlineKeyboardMarkup(rows) diff --git a/lib/bot.py b/lib/bot.py index 25ad903..a307cff 100644 --- a/lib/bot.py +++ b/lib/bot.py @@ -21,6 +21,7 @@ from telegram.ext import CallbackQueryHandler, ChatMemberHandler, CommandHandler, ContextTypes, MessageHandler, filters from . import intent, periods, prompt, report, templates +from .admins import Admins from .alerts import menu from .alerts import AlertState, with_footer from .irrigation import IrrigationError, Watchdog @@ -41,10 +42,14 @@ DRAFT_REFRESH_SECONDS = 20 # Telegram drops a draft 30 s after its last update, so it is re-sent before that DRAFT_MIN_GAP = 1.0 # at most one draft update a second while the answer streams in IRRIGATION_SETTLE_SECONDS = 3 # after any irrigation action, the controller is given this long to report its new state before it is read and the message redrawn +NOT_ADMIN = "Sorry, only group admins can use this bot." +NOT_ADMIN_TOAST = "Only group admins can use this bot" +STALE_BUTTON = "That button is out of date, please ask again" +MAX_CALLBACK_BYTES = 64 # Telegram's own limit on a button's data WATCHDOG_SECONDS = 45 # a question still running after this many seconds logs where everything is waiting HELP = ("Hi! Message me directly, or in groups @mention me or reply to me.\n" - "/reset clears this chat's memory, /alerts opens the alert settings (buttons to subscribe or unsubscribe).\n" + "/reset clears this chat's memory, /alerts opens the alert settings (buttons to subscribe or unsubscribe), /usage shows what I can do.\n" "In a private chat the buttons under the message box ask common questions; /keyboard off hides them.\n" "Your user ID: {user} | Chat ID: {chat}") @dataclass @@ -219,6 +224,8 @@ def __init__(self, cfg: Config, agent: Agent, sources: list, state: AlertState | self.cfg, self.agent, self.sources, self.state, self.irrigation = cfg, agent, sources, state, irrigation self.watchdog = Watchdog(irrigation, state) if irrigation and state else None # closes the valve after a run, however it ends self.keyboard_version = templates.version(irrigation is not None) + self.admin_only = bool(getattr(cfg, "admin_only", False)) # ADMIN_ONLY: only the admins of the groups the bot is in may use it + self.admins = Admins(state, getattr(cfg, "admin_chat_ids", ())) self.chats: dict[tuple, ChatState] = defaultdict(ChatState) self.by_name = {s.name: s for s in sources} self.images = periods.ImageCache() # charts drawn lately, so toggling the period buttons is instant @@ -229,15 +236,85 @@ def register(self, app): app.add_handler(CommandHandler("reset", self.on_reset)) app.add_handler(CommandHandler("keyboard", self.on_keyboard)) app.add_handler(CommandHandler("alerts", self.on_alerts)) - app.add_handler(CallbackQueryHandler(self.on_alert_button, pattern=r"^al:")) - if self.irrigation: - app.add_handler(CallbackQueryHandler(self.on_irrigation_button, pattern=r"^ir:")) - app.add_handler(CallbackQueryHandler(self.on_period_button, pattern=rf"^{periods.PREFIX}")) + app.add_handler(CommandHandler("usage", self.on_usage)) + app.add_handler(CallbackQueryHandler(self.on_button)) # every button press, whatever its data: see on_button app.add_handler(MessageHandler(filters.ALL & ~filters.TEXT & ~filters.COMMAND, self.on_other), group=1) app.add_handler(ChatMemberHandler(self.on_membership, ChatMemberHandler.MY_CHAT_MEMBER)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.on_message)) app.add_error_handler(self.on_error) + # ---------- who may use the bot, and which button presses are believed ---------- + @staticmethod + def _sender_ids(update: Update) -> list[int]: + """The ids that stand for the sender: their user id and, for a message an admin sends "as the group" (or as a channel), that + chat's id (a button press has no such thing: Telegram names the real user).""" + ids = [update.effective_user.id] if update.effective_user else [] + msg = update.effective_message + if update.callback_query is None and msg is not None and getattr(msg, "sender_chat", None) is not None: + ids.append(msg.sender_chat.id) + return ids + + async def _allowed(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: + """With ADMIN_ONLY on, only known group admins (and ADMIN_CHAT_IDS) get through; anyone else is told so, once, and False is + returned. A group the bot hears from is remembered first: it has to be known for its admins to count.""" + chat = update.effective_chat + if chat is not None and chat.type != ChatType.PRIVATE: + self.remember_chat(update) + if not self.admin_only or await self.admins.allowed(context.bot, self._sender_ids(update)): + return True + log.info("Refused %s: not a known group admin", describe_source(update)) + with contextlib.suppress(TelegramError): + if update.callback_query is not None: + await update.callback_query.answer(NOT_ADMIN_TOAST) + elif update.effective_message is not None: + await update.effective_message.reply_text(NOT_ADMIN) + return False + + @staticmethod + async def _is_group_admin(bot, chat_id: int, user_id: int) -> bool: + """Is this user an administrator or the creator of this group, by Telegram's own account of it?""" + member = await bot.get_chat_member(chat_id, user_id) + return member.status in ("administrator", "creator") + + @staticmethod + def _on_keyboard(message, data: str) -> bool: + """Is `data` the data of a button on this message's own inline keyboard? (An inaccessible message has no keyboard to look at.)""" + rows = getattr(getattr(message, "reply_markup", None), "inline_keyboard", None) or () + return any(getattr(button, "callback_data", None) == data for row in rows for button in row) + + async def _press(self, query) -> bool: + """Is this button press believable? A client can send any `data` for any message (Telegram does not check it against the + keyboard), so a press counts only when its data is short, plain text and is one of the buttons the message actually carries. + Otherwise it is answered with a toast, logged, and goes no further.""" + data, message = getattr(query, "data", None), getattr(query, "message", None) + believed = (isinstance(data, str) and 0 < len(data.encode()) <= MAX_CALLBACK_BYTES and data.isprintable() + and message is not None and self._on_keyboard(message, data)) + if not believed: + log.info("Ignored a button press that is not on its message: %s", _short(repr(data), 60)) + with contextlib.suppress(TelegramError): + await query.answer(STALE_BUTTON) + return believed + + def buttons(self) -> dict: + """Every family of buttons the bot draws: the start of its data -> what handles it. The only way into any of them is on_button.""" + return {"al:": self.on_alert_button, **({"ir:": self.on_irrigation_button} if self.irrigation else {}), + periods.PREFIX: self.on_period_button} + + async def on_button(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + """The one place every button press comes in: believed (_press), a known family (a prefix nobody handles is refused, so a + new kind of button is dead until it is added to buttons()), allowed (ADMIN_ONLY), then handled.""" + query = update.callback_query + if query is None or not await self._press(query): + return + handler = next((h for prefix, h in self.buttons().items() if query.data.startswith(prefix)), None) + if handler is None: + with contextlib.suppress(TelegramError): + await query.answer(STALE_BUTTON) + return + if not await self._allowed(update, context): + return + await handler(update, context) + # ---------- alert chats ---------- @staticmethod def _title(update: Update) -> str: @@ -267,17 +344,36 @@ async def on_alerts(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/alerts: the alert settings, as buttons (Subscribe | Unsubscribe). /alerts on and /alerts off still subscribe or unsubscribe every type.""" msg, chat = update.effective_message, update.effective_chat + in_group = chat.type != ChatType.PRIVATE + if not in_group and not await self._allowed(update, context): # anyone in a group may look; a private chat is gated like the rest + return if not self.state: await msg.reply_text("Alerts aren't available: no sensor is connected.") return arg = context.args[0].lower() if context.args else "" self.remember_chat(update) if arg in ("on", "off"): - self.state.set_alerts(chat.id, self._title(update), arg == "on") - log.info("/alerts %s in %s", arg, describe_source(update)) + if in_group and not (update.effective_user and await self._is_group_admin(context.bot, chat.id, update.effective_user.id)): + await msg.reply_text("Only group admins can change alerts") # the menu below is still shown + else: + self.state.set_alerts(chat.id, self._title(update), arg == "on") + log.info("/alerts %s in %s", arg, describe_source(update)) muted, kinds = self.state.muted(chat.id), self._alert_kinds() await msg.reply_text(menu.title(muted, kinds), reply_markup=menu.keyboard(muted, kinds)) + async def _help_and_alerts(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + """What the bot can do, then the alert settings: the Help & Alerts button and /usage.""" + await update.effective_message.reply_text(templates.capabilities_text("Ecowitt" in self.by_name, "AirGradient" in self.by_name, + "Pollen" in self.by_name, "Forecast" in self.by_name)) + await self.on_alerts(update, context) + + async def on_usage(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + """/usage: the same as the Help & Alerts button.""" + if not await self._allowed(update, context): + return + self.remember_chat(update) + await self._help_and_alerts(update, context) + async def on_alert_button(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """A press on the alert settings buttons (under an alert or the /alerts message): open or close a section, or subscribe or unsubscribe a type or all. In a group only admins may change them.""" @@ -290,11 +386,9 @@ async def on_alert_button(self, update: Update, context: ContextTypes.DEFAULT_TY return message = query.message chat = message.chat - if chat.type != ChatType.PRIVATE: - member = await context.bot.get_chat_member(chat.id, query.from_user.id) - if member.status not in ("administrator", "creator"): - await query.answer("Only group admins can change alerts") - return + if chat.type != ChatType.PRIVATE and not await self._is_group_admin(context.bot, chat.id, query.from_user.id): + await query.answer("Only group admins can change alerts") + return kinds = self._alert_kinds() parent = parent if parent in kinds else None toast, opened = None, section or None @@ -488,12 +582,16 @@ def _key(cls, msg: Message) -> tuple: async def on_reset(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/reset: forget this chat's conversation.""" + if not await self._allowed(update, context): + return msg = update.effective_message self.chats.pop(self._key(msg), None) log.info("/reset in %s", describe_source(update)) await msg.reply_text("Conversation memory cleared.") async def on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE): + if not await self._allowed(update, context): + return self.remember_chat(update) log.info("/start in %s", describe_source(update)) private = update.effective_chat.type == ChatType.PRIVATE @@ -530,6 +628,8 @@ def _keyboard_stale(self, msg: Message) -> bool: async def on_keyboard(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """/keyboard shows the buttons again, /keyboard off hides them (private chats only).""" + if not await self._allowed(update, context): + return msg = update.effective_message if update.effective_chat.type != ChatType.PRIVATE: await msg.reply_text("The buttons are only in private chats.") @@ -567,14 +667,15 @@ async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE): if waited > PENDING_MAX_SECONDS: log.info("Ignored a message %d minutes old from %s: %s", waited // 60, describe_source(update), _short(msg.text, 40)) return + private = msg.chat.type == ChatType.PRIVATE + if private and not await self._allowed(update, context): # before remember_chat: a stranger never becomes an alert chat + return self.remember_chat(update) - if msg.chat.type == ChatType.PRIVATE: + if private: if self.irrigation and (msg.text.strip() == templates.IRRIGATION or irrigation_menu.asked(msg.text)): # the controller's status and buttons (no model needed) await self.send_irrigation(msg) elif msg.text.strip() == templates.CAPABILITIES: # what it measures, then the alert settings (no model needed) - await msg.reply_text(templates.capabilities_text("Ecowitt" in self.by_name, "AirGradient" in self.by_name, - "Pollen" in self.by_name, "Forecast" in self.by_name)) - await self.on_alerts(update, context) + await self._help_and_alerts(update, context) else: await self.respond(update, context, templates.sentence(msg.text) or msg.text) return @@ -583,6 +684,8 @@ async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE): replied_to_bot = (msg.reply_to_message is not None and msg.reply_to_message.from_user is not None and msg.reply_to_message.from_user.id == context.bot.id) if mention.search(msg.text) or replied_to_bot: + if not await self._allowed(update, context): # only a message meant for the bot is refused, never the group's own talk + return asked = mention.sub("", msg.text) if self.irrigation and irrigation_menu.asked(asked): # "@bot irrigation": the controller, as in a private chat await self.send_irrigation(msg) diff --git a/lib/config.py b/lib/config.py index e13a077..1422751 100644 --- a/lib/config.py +++ b/lib/config.py @@ -1,12 +1,15 @@ """Settings, all from environment variables (systemd loads them from the unit's EnvironmentFile).""" +import logging import os +import re from dataclasses import dataclass from datetime import datetime, tzinfo from pathlib import Path from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parent.parent +log = logging.getLogger(__name__) def _tz() -> tzinfo: @@ -35,6 +38,24 @@ def _hour(value: str, default: int) -> int: return int(number) if number is not None and 0 <= number <= 23 else default +def _flag(value: str, default: bool) -> bool: + """A setting that is on or off, with a default when it is unset or empty.""" + return _on(value) if value else default + + +def _ids(value: str) -> frozenset[int]: + """"123, -1001 456" -> {123, -1001, 456}. An entry that is not a whole number is ignored (and logged): it must not stop the bot.""" + found, bad = set(), [] + for part in filter(None, re.split(r"[,\s]+", value)): + try: + found.add(int(part)) + except ValueError: + bad.append(part) + if bad: + log.warning("ADMIN_CHAT_IDS: ignoring %s (not whole numbers)", ", ".join(map(repr, bad))) + return frozenset(found) + + def _number(value: str) -> float | None: try: return float(value) @@ -72,6 +93,8 @@ class Config: tuya_client_secret: str = "" tuya_base_url: str = "https://openapi.tuyaeu.com" # TUYA_BASE_URL: the data centre's address (EU by default) irrigation_check_hour: int = 18 # IRRIGATION_CHECK_HOUR: the local hour of the daily check (0 to 23) + admin_only: bool = True # ADMIN_ONLY (on by default): the bot answers nobody but the admins of the groups it is in + admin_chat_ids: frozenset[int] = frozenset() # ADMIN_CHAT_IDS: more ids allowed besides those admins (a user id, or a group id for its anonymous admins) alert_cooldown_minutes: int = 30 # ALERT_COOLDOWN_MINUTES: the rain and temperature-crossing alerts: dry this long before "the rain has stopped", and this long between alerts (5 to 150) @property @@ -114,6 +137,8 @@ def from_env(cls) -> "Config": tuya_client_secret=env("TUYA_CLIENT_SECRET"), tuya_base_url=env("TUYA_BASE_URL", "https://openapi.tuyaeu.com").rstrip("/"), irrigation_check_hour=_hour(env("IRRIGATION_CHECK_HOUR"), 18), + admin_only=_flag(env("ADMIN_ONLY"), True), + admin_chat_ids=_ids(env("ADMIN_CHAT_IDS")), alert_cooldown_minutes=int(min(150, max(5, _number(env("ALERT_COOLDOWN_MINUTES")) or 30))), ) if not (cfg.ecowitt or cfg.airgradient): diff --git a/tests/fakes.py b/tests/fakes.py index 25af547..4a422e9 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -20,7 +20,7 @@ def config(tmp_path, **over) -> Config: ecowitt_api_key="a", ecowitt_app_key="b", airgradient_token="t", airgradient_location="42", airgradient_dashboard="https://example.com/live", state_path=tmp_path / "state.json", cache_path=tmp_path / "cache.sqlite", air_cache_path=tmp_path / "air.sqlite", - conditions_cache_path=tmp_path / "conditions.sqlite") + conditions_cache_path=tmp_path / "conditions.sqlite", admin_only=False) # tests that are about ADMIN_ONLY switch it on return Config(**{**base, **over}) diff --git a/tests/test_admin_only.py b/tests/test_admin_only.py new file mode 100644 index 0000000..4a59e3e --- /dev/null +++ b/tests/test_admin_only.py @@ -0,0 +1,454 @@ +"""ADMIN_ONLY (who may use the bot), /alerts in groups, /usage, and the checks every button press goes through.""" + +import random +from types import SimpleNamespace as NS + +import pytest +from telegram import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup +from telegram.ext import CallbackQueryHandler + +from lib import periods, templates +from lib.admins import Admins +from lib.alerts import AlertState, menu +from lib.bot import MAX_CALLBACK_BYTES, NOT_ADMIN, NOT_ADMIN_TOAST, STALE_BUTTON, Bot +from lib.config import Config, _ids +from lib.irrigation import menu as irrigation_menu +from tests.fakes import TZ +from tests.test_irrigation import Remote + +ADMIN, STRANGER, GROUP, OTHER_GROUP = 1, 7, -5, -6 +ANONYMOUS = 1087968824 # the user Telegram shows for an admin who posts "as the group" + + +class TG: + """The bot's Telegram client: who the admins of each group are, and the status of each member.""" + username, id = "testbot", 99 + + def __init__(self, admins=None, statuses=None): + self.admins = {GROUP: [ADMIN]} if admins is None else admins + self.statuses = statuses if statuses is not None else {(GROUP, ADMIN): "administrator"} + self.fetches = [] + + async def get_chat_administrators(self, chat_id): + self.fetches.append(chat_id) + got = self.admins[chat_id] + if isinstance(got, Exception): + raise got + return [NS(user=NS(id=u, is_bot=False)) for u in got] + + async def get_chat_member(self, chat_id, user_id): + return NS(status=self.statuses.get((chat_id, user_id), "member")) + + +class Asked(Bot): + """A Bot that records the questions it is asked instead of asking a model.""" + asked: list + + async def respond(self, update, context, text, redraw=None): + self.asked.append(text.strip()) + + +def make(tmp_path, admin_only=True, extra=(), irrigation=None, groups=(GROUP,), tg=None): + state = AlertState(tmp_path / "s.json") + for group in groups: + state.add_chat(group, "Home") + bot = Asked(NS(tz=TZ, admin_only=admin_only, admin_chat_ids=frozenset(extra)), None, [NS(name="Ecowitt")], state, irrigation) + bot.asked = [] + return bot, state, tg or TG() + + +def ctx(tg, *args): + return NS(bot=tg, args=list(args)) + + +def message(text, user=STRANGER, kind="private", chat=None, sender_chat=None, replied=False): + sent = [] + + async def reply_text(body, **kw): + sent.append((body, kw)) + chat_id = chat if chat is not None else (user if kind == "private" else GROUP) + where = NS(type=kind, id=chat_id, title=None if kind == "private" else "Home") + msg = NS(text=text, chat=where, chat_id=chat_id, reply_text=reply_text, message_thread_id=None, is_topic_message=False, date=None, + reply_to_message=NS(from_user=NS(id=99), text="hi", caption=None) if replied else None, sender_chat=sender_chat) + update = NS(effective_message=msg, effective_chat=where, effective_user=NS(id=user, full_name="U", username="u"), callback_query=None) + return update, sent + + +class Press: + """A button press on a message that carries `markup` (as a real one does) with the given data.""" + + def __init__(self, data, markup, user=STRANGER, kind="private", chat=None): + chat_id = chat if chat is not None else (user if kind == "private" else GROUP) + self.data, self.from_user, self.toast, self.edits = data, NS(id=user, full_name="U", username="u"), "unset", [] + self.message = NS(chat=NS(type=kind, id=chat_id, title=None), chat_id=chat_id, message_id=5, text="๐Ÿ”” Alerts in this chat", + reply_markup=markup) + + async def answer(self, text=None, **kw): + self.toast = text + + async def edit_message_reply_markup(self, reply_markup=None): + self.edits.append(("markup", reply_markup)) + + async def edit_message_text(self, text, reply_markup=None, **kw): + self.edits.append(("text", text, reply_markup)) + + def update(self): + where = self.message.chat if self.message is not None else None + return NS(callback_query=self, effective_user=self.from_user, effective_chat=where, effective_message=self.message) + + +def board(*data): + return InlineKeyboardMarkup([[InlineKeyboardButton(d, callback_data=d) for d in data]]) + + +# ---------- the settings ---------- +def test_admin_only_is_on_unless_switched_off_and_the_listed_ids_are_read_safely(monkeypatch, caplog): + for k, v in {"TELEGRAM_BOT_TOKEN": "t", "XAI_API_KEY": "k", "ECOWITT_API_KEY": "a", "ECOWITT_APP_KEY": "b"}.items(): + monkeypatch.setenv(k, v) + monkeypatch.delenv("ADMIN_ONLY", raising=False) + assert Config.from_env().admin_only is True and Config.from_env().admin_chat_ids == frozenset() + for value, on in (("", True), ("on", True), ("ON", True), ("1", True), ("off", False), ("Off", False), ("0", False), ("no", False), ("false", False)): + monkeypatch.setenv("ADMIN_ONLY", value) + assert Config.from_env().admin_only is on, value + assert _ids("1, -1001 22,,x, 3.5 -7") == frozenset({1, -1001, 22, -7}) and _ids("") == frozenset() + assert "ADMIN_CHAT_IDS: ignoring 'x', '3.5'" in caplog.text + monkeypatch.setenv("ADMIN_CHAT_IDS", "5 -9") + assert Config.from_env().admin_chat_ids == frozenset({5, -9}) + + +# ---------- who is a known admin ---------- +async def test_admins_of_any_known_group_are_allowed_and_the_lists_are_kept_ten_minutes(tmp_path): + clock, state, tg = [0.0], AlertState(tmp_path / "s.json"), TG({GROUP: [ADMIN, 2], OTHER_GROUP: [3]}) + state.add_chat(GROUP, "Home") + state.add_chat(OTHER_GROUP, "Away") + state.add_chat(ADMIN, "private chats are not groups") + admins = Admins(state, clock=lambda: clock[0]) + assert admins.groups() == [OTHER_GROUP, GROUP] + assert await admins.allowed(tg, [ADMIN]) and await admins.allowed(tg, [3]) and not await admins.allowed(tg, [STRANGER]) + assert not await admins.allowed(tg, []) and not await admins.allowed(tg, [None]) + asked = len(tg.fetches) + clock[0] = 599 + await admins.allowed(tg, [STRANGER]) + assert len(tg.fetches) == asked # kept + tg.admins[GROUP] = [STRANGER] + clock[0] = 601 + assert await admins.allowed(tg, [STRANGER]) and not await admins.allowed(tg, [2]) # asked again after ten minutes + + +async def test_a_failed_lookup_keeps_the_old_list_and_is_retried_soon_without_raising(tmp_path, caplog): + clock, state, tg = [0.0], AlertState(tmp_path / "s.json"), TG({GROUP: [ADMIN]}) + state.add_chat(GROUP, "Home") + admins = Admins(state, clock=lambda: clock[0]) + assert await admins.allowed(tg, [ADMIN]) + tg.admins[GROUP] = RuntimeError("bot was kicked") + clock[0] = 700 + assert await admins.allowed(tg, [ADMIN]) and not await admins.allowed(tg, [STRANGER]) # the last known list stands + fetches = len(tg.fetches) + clock[0] = 730 + await admins.allowed(tg, [ADMIN]) + assert len(tg.fetches) == fetches and caplog.text.count("Couldn't list the admins of -5") == 1 # not again for a minute, said once + clock[0] = 800 + tg.admins[GROUP] = [STRANGER] + assert await admins.allowed(tg, [STRANGER]) # and it recovers + fresh = Admins(state) + tg.admins[GROUP] = RuntimeError("down") + assert not await fresh.allowed(tg, [ADMIN]) # nothing known: nobody + + +async def test_listed_ids_are_added_to_the_known_admins_never_replacing_them(tmp_path): + state, tg = AlertState(tmp_path / "s.json"), TG({GROUP: [ADMIN], OTHER_GROUP: [3]}) + admins = Admins(state, extra_ids=[STRANGER, OTHER_GROUP]) + assert admins.groups() == [OTHER_GROUP] # no group known: only the listed group's admins are looked up + assert await admins.allowed(tg, [STRANGER]) # a listed user, with no group known at all + assert await admins.allowed(tg, [3]) and tg.fetches == [OTHER_GROUP] # a listed group's admins + assert not await admins.allowed(tg, [ADMIN]) # this group is not known yet + state.add_chat(GROUP, "Home") + assert await admins.allowed(tg, [ADMIN]) and await admins.allowed(tg, [STRANGER]) # known and listed, together + assert await admins.allowed(tg, [ANONYMOUS, OTHER_GROUP]) and not await admins.allowed(tg, [ANONYMOUS, GROUP]) # an anonymous admin: only a listed group + + +# ---------- the gate: private chats ---------- +@pytest.mark.parametrize("command, args", [("on_start", []), ("on_reset", []), ("on_keyboard", []), ("on_usage", []), ("on_alerts", [])]) +async def test_a_stranger_is_refused_every_command_in_a_private_chat_and_is_not_remembered(tmp_path, command, args): + bot, state, tg = make(tmp_path) + update, sent = message("/x", user=STRANGER) + await getattr(bot, command)(update, ctx(tg, *args)) + assert [body for body, _ in sent] == [NOT_ADMIN] and STRANGER not in state.chats + update, sent = message("/x", user=ADMIN) # an admin of a known group is let in + await getattr(bot, command)(update, ctx(tg, *args)) + assert sent and NOT_ADMIN not in [body for body, _ in sent] + if command in ("on_start", "on_usage", "on_alerts"): + assert ADMIN in state.chats + + +async def test_a_stranger_in_a_private_chat_gets_no_answer_and_an_admin_does(tmp_path): + bot, state, tg = make(tmp_path) + update, sent = message("how warm is it", user=STRANGER) + await bot.on_message(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] and bot.asked == [] and STRANGER not in state.chats + update, sent = message("how warm is it", user=ADMIN) + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == ["how warm is it"] and ADMIN in state.chats + + +async def test_with_admin_only_off_everyone_is_answered_as_before(tmp_path): + bot, state, tg = make(tmp_path, admin_only=False) + update, sent = message("how warm is it", user=STRANGER) + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == ["how warm is it"] and STRANGER in state.chats and tg.fetches == [] + + +async def test_a_listed_user_can_use_the_bot_before_it_is_in_any_group(tmp_path): + bot, state, tg = make(tmp_path, extra=[STRANGER], groups=(), tg=TG({})) + update, sent = message("hello", user=STRANGER) + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == ["hello"] and STRANGER in state.chats + update, sent = message("hello", user=2) + await bot.on_message(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] + + +# ---------- the gate: groups ---------- +async def test_in_a_group_only_a_message_meant_for_the_bot_is_refused_and_ordinary_talk_is_left_alone(tmp_path): + bot, state, tg = make(tmp_path, groups=()) + update, sent = message("anyone want lunch?", user=STRANGER, kind="supergroup") + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == [] and GROUP in state.chats # unaddressed: nothing, and the group is now known + update, sent = message("@testbot how warm is it", user=STRANGER, kind="supergroup") + await bot.on_message(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] and bot.asked == [] + update, sent = message("how warm is it", user=STRANGER, kind="supergroup", replied=True) + await bot.on_message(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] + update, sent = message("@testbot how warm is it", user=ADMIN, kind="supergroup") + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == ["how warm is it"] + + +async def test_an_anonymous_admin_gets_in_only_through_a_listed_group_id(tmp_path): + update, sent = message("@testbot hi", user=ANONYMOUS, kind="supergroup", sender_chat=NS(id=GROUP)) + bot, state, tg = make(tmp_path) + await bot.on_message(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] and bot.asked == [] # no automatic pass + bot, state, tg = make(tmp_path, extra=[GROUP]) + update, sent = message("@testbot hi", user=ANONYMOUS, kind="supergroup", sender_chat=NS(id=GROUP)) + await bot.on_message(update, ctx(tg)) + assert sent == [] and bot.asked == ["hi"] + + +# ---------- /alerts in groups, /usage ---------- +async def test_anyone_may_open_alerts_in_a_group_but_only_admins_may_change_them(tmp_path): + bot, state, tg = make(tmp_path) + update, sent = message("/alerts", user=STRANGER, kind="supergroup") + await bot.on_alerts(update, ctx(tg)) + assert [b for b, _ in sent] == [menu.title(set(), bot._alert_kinds())] and NOT_ADMIN not in [b for b, _ in sent] + assert sent[0][1]["reply_markup"] is not None + for arg in ("off", "on"): + update, sent = message(f"/alerts {arg}", user=STRANGER, kind="supergroup") + await bot.on_alerts(update, ctx(tg, arg)) + assert [b for b, _ in sent][0] == "Only group admins can change alerts" and len(sent) == 2 and state.muted(GROUP) == set() + update, sent = message("/alerts off", user=ADMIN, kind="supergroup") + await bot.on_alerts(update, ctx(tg, "off")) + assert "Only group admins" not in sent[0][0] and state.muted(GROUP) == set(menu.LABELS) # an admin may + update, sent = message("/alerts", user=STRANGER, kind="private") # a private chat is gated like the rest + await bot.on_alerts(update, ctx(tg)) + assert [b for b, _ in sent] == [NOT_ADMIN] + + +async def test_with_admin_only_off_anyone_may_open_alerts_but_telegram_s_own_admins_still_change_them(tmp_path): + bot, state, tg = make(tmp_path, admin_only=False) + update, sent = message("/alerts off", user=STRANGER, kind="supergroup") + await bot.on_alerts(update, ctx(tg, "off")) + assert sent[0][0] == "Only group admins can change alerts" and state.muted(GROUP) == set() + update, sent = message("/alerts off", user=ADMIN, kind="supergroup") + await bot.on_alerts(update, ctx(tg, "off")) + assert state.muted(GROUP) == set(menu.LABELS) + + +@pytest.mark.parametrize("kind", ["private", "supergroup"]) +async def test_usage_is_the_help_and_alerts_button(tmp_path, kind): + bot, state, tg = make(tmp_path) + update, from_button = message(templates.CAPABILITIES, user=ADMIN, kind=kind) + await bot.on_message(update, ctx(tg)) if kind == "private" else await bot._help_and_alerts(update, ctx(tg)) + update, from_command = message("/usage", user=ADMIN, kind=kind) + await bot.on_usage(update, ctx(tg)) + shown = lambda sent: [(b, [[(x.text, x.callback_data) for x in row] for row in kw["reply_markup"].inline_keyboard] if kw.get("reply_markup") else None) + for b, kw in sent] + assert len(from_command) == 2 and shown(from_command) == shown(from_button) + assert from_command[0][0] == templates.capabilities_text(True, False, False, False) + update, refused = message("/usage", user=STRANGER, kind=kind) + await bot.on_usage(update, ctx(tg)) + assert [b for b, _ in refused] == [NOT_ADMIN] # gated like the other commands + + +# ---------- every button comes in through one place ---------- +def alert_keyboard(opened="unsub", parent=None): + return menu.keyboard(set(), menu.available_kinds({"Ecowitt"}), opened, parent) + + +async def test_a_stranger_is_refused_on_every_family_of_buttons_and_an_admin_is_let_through(tmp_path): + bot, state, tg = make(tmp_path, irrigation=Remote()) + state.add_chat(ADMIN, "Rob") + cases = [("al:off:rain:unsub", alert_keyboard()), (f"{periods.PREFIX}7", periods.keyboard(1)), + ("ir:open:water", irrigation_menu.keyboard(True, True))] + for data, markup in cases: + press = Press(data, markup, user=STRANGER) + await bot.on_button(press.update(), ctx(tg)) + assert press.toast == NOT_ADMIN_TOAST and press.edits == [] and state.muted(STRANGER) == set(), data + press = Press(data, markup, user=ADMIN) + await bot.on_button(press.update(), ctx(tg)) + assert press.toast != NOT_ADMIN_TOAST, data + assert state.muted(ADMIN) == {"rain"} # the admin's alert press went through + press = Press("ir:open:water", irrigation_menu.keyboard(True, True), user=STRANGER) + off = make(tmp_path, admin_only=False, irrigation=Remote())[0] + await off.on_button(press.update(), ctx(tg)) + assert press.edits # ADMIN_ONLY=off: as before + + +async def test_alert_buttons_in_a_group_stay_admin_only_even_when_admin_only_is_off(tmp_path): + bot, state, tg = make(tmp_path, admin_only=False) + press = Press("al:off:rain:unsub", alert_keyboard(), user=STRANGER, kind="supergroup") + await bot.on_button(press.update(), ctx(tg)) + assert press.toast == "Only group admins can change alerts" and state.muted(GROUP) == set() + press = Press("al:off:rain:unsub", alert_keyboard(), user=ADMIN, kind="supergroup") + await bot.on_button(press.update(), ctx(tg)) + assert state.muted(GROUP) == {"rain"} + + +# ---------- a press is believed only if the message really has that button ---------- +def every_keyboard(bot): + """Every inline keyboard the bot can draw, in every state.""" + kinds = menu.available_kinds({"Ecowitt", "AirGradient", "Pollen", "Forecast", "Irrigation"}) + muted_sets = [set(), {"rain"}, {"rain", "uv", "air"}, set(menu.LABELS)] + for muted in muted_sets: + for opened in (None, "sub", "unsub"): + for parent in (None, *kinds): + yield menu.keyboard(muted, kinds, opened, parent) + for battery in (True, False): + for pause in (True, False): + for opened in (None, *irrigation_menu.SECTIONS, "bogus"): + for paused in (True, False): + yield irrigation_menu.keyboard(battery, pause, opened, paused) + for current in (None, *periods.PERIODS): + yield periods.keyboard(current) + + +def test_every_button_the_bot_draws_is_short_plain_and_has_exactly_one_handler(tmp_path): + bot, state, tg = make(tmp_path, irrigation=Remote()) + table, seen = bot.buttons(), set() + for markup in every_keyboard(bot): + for row in markup.inline_keyboard: + for button in row: + data = button.callback_data + seen.add(data) + assert 0 < len(data.encode()) <= MAX_CALLBACK_BYTES and data.isprintable(), data + assert sum(data.startswith(prefix) for prefix in table) == 1, data + assert bot._on_keyboard(NS(reply_markup=markup), data) + assert len(seen) > 60 and {p for p in table} == {"al:", "ir:", periods.PREFIX} + assert "ir:" not in make(tmp_path, irrigation=None)[0].buttons() + + +def test_there_is_one_callback_handler_in_the_application_and_no_other_way_in(tmp_path): + bot, state, tg = make(tmp_path, irrigation=Remote()) + added = [] + bot.register(NS(add_handler=lambda h, group=0: added.append(h), add_error_handler=lambda h: None)) + callbacks = [h for h in added if isinstance(h, CallbackQueryHandler)] + assert len(callbacks) == 1 and callbacks[0].callback == bot.on_button and callbacks[0].pattern is None # every press, whatever its data + + +FORGED = ["ir:water:30", "ir:delay:72h", "ir:delay:cancel", "ir:sw:off", "ir:refresh", "ir:batt:off", "ir:pause:off", "al:off:all:unsub", + "al:off:rain:unsub", "al:on:all:sub", f"{periods.PREFIX}30", f"{periods.PREFIX}1", "zz:1", "ir:", "al:", "", " ", + "ir:water:30\n", "ir:water:30\x00", "ir:water:" + "9" * 80, "al:" + "x" * 200, "ir:water:๏ผ“๏ผ", "ir:water:30 ", "รฏr:water:30", + "ir:water:30:extra:colons:::::", "al:off:rain:unsub:rain:more", "../../etc/passwd", "ir:water:$(reboot)", "ir:water:;rm -rf /", + "{0.__class__}", "%s%s%s%n", "ir:water:-1", "ir:water:1e9", "al:off:__class__:unsub", "pd:__import__('os')"] + + +class Spy(Remote): + """The controller, recording everything it is asked to do.""" + + +async def press_through(bot, tg, data, markup, user=ADMIN, kind="private"): + press = Press(data, markup, user=user, kind=kind) + await bot.on_button(press.update(), ctx(tg)) + return press + + +async def test_a_forged_press_does_nothing_whoever_sends_it_and_whichever_family_it_imitates(tmp_path): + spy = Spy() + bot, state, tg = make(tmp_path, irrigation=spy) + unrelated = alert_keyboard("sub") # a keyboard that has none of the forged data + before = (dict(state.chats), list(spy.commands)) + for data in FORGED: + for user in (ADMIN, STRANGER): + press = await press_through(bot, tg, data, unrelated, user=user) + assert press.edits == [] and press.toast == STALE_BUTTON, (data, user, press.toast) # believed or not comes first, for everyone + assert (state.chats, spy.commands) == before and spy.commands == [] # nothing changed, nothing sent to the controller + + +async def test_malformed_presses_are_answered_and_never_raise(tmp_path): + bot, state, tg = make(tmp_path, irrigation=Remote()) + keyboard = irrigation_menu.keyboard(True, True) + for data in (None, 5, b"ir:refresh", ["ir:refresh"], "x" * 65, "ir:rรฉ" + "รฉ" * 30, "ir:refresh\n", "\x1b[31mir:refresh", "ir:refreshโ€ฎ"): + press = Press(data, keyboard) + await bot.on_button(press.update(), ctx(tg)) + assert press.toast == STALE_BUTTON and press.edits == [], repr(data) + no_keyboard = Press("ir:refresh", None) # a message without a keyboard (or an inaccessible one) + await bot.on_button(no_keyboard.update(), ctx(tg)) + assert no_keyboard.toast == STALE_BUTTON + inaccessible = Press("ir:refresh", keyboard) + inaccessible.message = NS(chat=NS(type="private", id=ADMIN), chat_id=ADMIN, message_id=5, date=0) # Telegram's InaccessibleMessage: no keyboard to check + await bot.on_button(inaccessible.update(), ctx(tg)) + assert inaccessible.toast == STALE_BUTTON + gone = Press("ir:refresh", keyboard) + gone.message = None + await bot.on_button(gone.update(), ctx(tg)) + assert gone.toast == STALE_BUTTON + await bot.on_button(NS(callback_query=None), ctx(tg)) # no query at all + + +async def test_a_prefix_nobody_handles_is_refused_even_when_the_message_really_has_that_button(tmp_path): + bot, state, tg = make(tmp_path) + press = Press("zz:do-something", board("zz:do-something")) + await bot.on_button(press.update(), ctx(tg)) + assert press.toast == STALE_BUTTON and press.edits == [] + press = Press("ir:refresh", irrigation_menu.keyboard(True, True)) # an irrigation button when there is no controller + await bot.on_button(press.update(), ctx(tg)) + assert press.toast == STALE_BUTTON + + +async def test_a_real_press_on_a_button_the_message_carries_still_works(tmp_path): + bot, state, tg = make(tmp_path, irrigation=Remote()) + state.add_chat(ADMIN, "Rob") + for data, markup in (("ir:refresh", irrigation_menu.keyboard(True, True)), ("al:off:rain:unsub", alert_keyboard()), + ("ir:water:10", irrigation_menu.keyboard(True, True, "water"))): + press = await press_through(bot, tg, data, markup) + assert press.toast not in (STALE_BUTTON, NOT_ADMIN_TOAST) and press.edits, data + assert bot.irrigation.commands == [("water", 10)] and state.muted(ADMIN) == {"rain"} + stale = await press_through(bot, tg, "ir:water:10", irrigation_menu.keyboard(True, True)) # the Water list is not open on this message + assert stale.toast == STALE_BUTTON and bot.irrigation.commands == [("water", 10)] + + +async def test_random_data_never_gets_past_the_entry_point(tmp_path): + rng = random.Random(7) + spy = Spy() + bot, state, tg = make(tmp_path, irrigation=spy) + boards = [alert_keyboard("sub"), irrigation_menu.keyboard(True, True), periods.keyboard(1)] + known = {b.callback_data for board_ in boards for row in board_.inline_keyboard for b in row} + pieces = ["ir", "al", "pd", "on", "off", "water", "delay", "sw", "batt", "pause", "refresh", "open", "close", "all", "sub", "unsub", + "rain", "uv", "24h", "cancel", "5", "30", "99999", "-1", "", " ", "\n", "รฉ", "๐Ÿ’ง", "*", "%s", "{}", "\\", ";", "$(x)", "None"] + before = (dict(state.chats), dict(state.charts)) + for _ in range(3000): + data = rng.choice(["ir", "al", "pd", ""]) + ":" + ":".join(rng.choice(pieces) for _ in range(rng.randint(0, 4))) + if data in known: + continue + press = await press_through(bot, tg, data, rng.choice(boards), user=rng.choice([ADMIN, STRANGER])) + assert press.toast in (STALE_BUTTON, NOT_ADMIN_TOAST) and press.edits == [], repr(data) + assert spy.commands == [] and (dict(state.chats), dict(state.charts)) == before + + +def test_a_callback_query_from_python_telegram_bot_is_what_the_checks_read(): + """The fields the checks use exist on the library's real types, so the fakes above stand for the real thing.""" + assert {"data", "message", "from_user", "answer"} <= set(dir(CallbackQuery)) + markup = InlineKeyboardMarkup([[InlineKeyboardButton("x", callback_data="ir:refresh")]]) + assert Bot._on_keyboard(NS(reply_markup=markup), "ir:refresh") and not Bot._on_keyboard(NS(reply_markup=markup), "ir:water:5") + assert not Bot._on_keyboard(NS(), "ir:refresh") and not Bot._on_keyboard(None, "ir:refresh") diff --git a/tests/test_alert_menu.py b/tests/test_alert_menu.py index 9f1358d..5a08d31 100644 --- a/tests/test_alert_menu.py +++ b/tests/test_alert_menu.py @@ -22,12 +22,12 @@ def test_the_collapsed_menu_is_subscribe_and_unsubscribe(): def test_unsubscribe_opens_the_types_that_are_on_and_subscribe_the_types_that_are_off(): got = rows(keyboard({"pollen", "forecast"}, KINDS, "unsub")) assert got[0] == [("โž• Subscribe", "al:open:sub"), ("โ–พ โž– Unsubscribe", "al:close")] - assert got[1] == [("rain", "al:off:rain:unsub"), ("rain predicted", "al:off:rain_likely:unsub")] - assert [text for row in got[1:-1] for text, _ in row] == ["rain", "rain predicted", "gusts", "UV", "temperature crossing", "air quality"] - assert got[-1] == [("Unsubscribe from all", "al:off:all:unsub")] + assert got[1] == [("๐ŸŒง๏ธ rain", "al:off:rain:unsub"), ("๐ŸŒฆ๏ธ rain predicted", "al:off:rain_likely:unsub")] + assert [text for row in got[1:-1] for text, _ in row] == ["๐ŸŒง๏ธ rain", "๐ŸŒฆ๏ธ rain predicted", "๐Ÿ’จ gusts", "๐Ÿงด UV", "๐ŸŒก๏ธ temperature crossing", "๐Ÿ˜ท particulates"] + assert got[-1] == [("๐Ÿ”• Unsubscribe from all", "al:off:all:unsub")] got = rows(keyboard({"pollen", "forecast"}, KINDS, "sub")) assert got[0][0] == ("โ–พ โž• Subscribe", "al:close") - assert got[1:] == [[("pollen & asthma", "al:on:pollen:sub"), ("forecast changes", "al:on:forecast:sub")], [("Subscribe to all", "al:on:all:sub")]] + assert got[1:] == [[("๐ŸŒผ pollen & asthma", "al:on:pollen:sub"), ("๐Ÿ”„ forecast changes", "al:on:forecast:sub")], [("๐Ÿ”” Subscribe to all", "al:on:all:sub")]] assert not any(data == "al:noop" for row in got for _, data in row) # no button that does nothing assert options({"pollen"}, KINDS, "sub") == ["pollen"] and options({"pollen"}, ["rain", "pollen"], "unsub") == ["rain"] assert len(rows(keyboard(set(), KINDS, "sub"))) == 1 # nothing off: nothing listed @@ -43,7 +43,7 @@ def test_only_the_types_the_bot_has_are_listed_and_every_callback_is_short(): def test_the_title_lists_what_is_on_and_off(): assert title({"gusts", "uv"}, KINDS[:5]) == f"{TITLE}\nโœ… On: rain, rain predicted, temperature crossing\n๐Ÿ”• Off: gusts, UV" - assert title(set(), ["air"]).endswith("๐Ÿ”• Off: nothing") and title({"air"}, ["air"]).endswith("โœ… On: nothing\n๐Ÿ”• Off: air quality") + assert title(set(), ["air"]).endswith("๐Ÿ”• Off: nothing") and title({"air"}, ["air"]).endswith("โœ… On: nothing\n๐Ÿ”• Off: particulates") def test_state_mutes_one_type_or_all_and_old_files_load(tmp_path): @@ -131,12 +131,12 @@ async def test_pressing_unsubscribe_then_a_type_updates_the_state_and_redraws_th bot, state = make_bot(tmp_path) q = Query("al:open:unsub") await press(bot, q) - assert q.toast is None and rows(q.edits[0][1])[1][0] == ("rain", "al:off:rain:unsub") and q.edits[0][0] == "markup" + assert q.toast is None and rows(q.edits[0][1])[1][0] == ("๐ŸŒง๏ธ rain", "al:off:rain:unsub") and q.edits[0][0] == "markup" q = Query("al:off:rain:unsub") await press(bot, q) assert state.muted(1) == {"rain"} and q.toast == "Rain alerts off in this chat" drawn = rows(q.edits[0][1]) - assert drawn[0][1] == ("โ–พ โž– Unsubscribe", "al:close") and "rain" not in [t for row in drawn[1:] for t, _ in row] + assert drawn[0][1] == ("โ–พ โž– Unsubscribe", "al:close") and "๐ŸŒง๏ธ rain" not in [t for row in drawn[1:] for t, _ in row] q = Query("al:on:rain:sub") await press(bot, q) assert state.muted(1) == set() and q.toast == "Rain alerts on in this chat" @@ -148,7 +148,7 @@ async def test_pressing_unsubscribe_then_a_type_updates_the_state_and_redraws_th assert q.toast == "No alerts are on" and q.edits == [] # nothing to list: a toast, no dead button q = Query("al:open:sub") await press(bot, q) - assert rows(q.edits[0][1])[-1] == [("Subscribe to all", "al:on:all:sub")] + assert rows(q.edits[0][1])[-1] == [("๐Ÿ”” Subscribe to all", "al:on:all:sub")] q = Query("al:close") await press(bot, q) assert len(rows(q.edits[0][1])) == 1 @@ -203,7 +203,7 @@ async def reply_text(body, **kw): chat = NS(type="private", id=1, title=None) update = NS(effective_message=NS(reply_text=reply_text, chat_id=1), effective_chat=chat, effective_user=NS(full_name="Rob", username="rob")) await bot.on_alerts(update, NS(args=[])) - assert replies[0][0].startswith(TITLE) and "โœ… On: rain, rain predicted, gusts, UV, temperature crossing, air quality, pollen & asthma, forecast changes" in replies[0][0] + assert replies[0][0].startswith(TITLE) and "โœ… On: rain, rain predicted, gusts, UV, temperature crossing, particulates, pollen & asthma, forecast changes" in replies[0][0] assert rows(replies[0][1]["reply_markup"])[0] == [("โž• Subscribe", "al:open:sub"), ("โž– Unsubscribe", "al:open:unsub")] await bot.on_alerts(update, NS(args=["off"])) assert not state.alerts_on(1) and "โœ… On: nothing" in replies[1][0] @@ -258,23 +258,23 @@ async def spy(chat_id, text, link=None, kind=None): def test_under_an_alert_the_unsubscribe_list_is_just_its_type_and_all_and_every_button_carries_the_type(): got = rows(keyboard({"pollen", "forecast"}, KINDS, "unsub", parent="uv")) assert got == [[("โž• Subscribe", "al:open:sub:uv"), ("โ–พ โž– Unsubscribe", "al:close:uv")], - [("โ— UV", "al:off:uv:unsub:uv")], [("Unsubscribe from all", "al:off:all:unsub:uv")]] + [("โ— ๐Ÿงด UV", "al:off:uv:unsub:uv")], [("๐Ÿ”• Unsubscribe from all", "al:off:all:unsub:uv")]] only = rows(keyboard(set(KINDS) - {"uv"}, KINDS, "unsub", parent="uv")) # the alert's type is the only one on: no "all" - assert only[1:] == [[("โ— UV", "al:off:uv:unsub:uv")]] + assert only[1:] == [[("โ— ๐Ÿงด UV", "al:off:uv:unsub:uv")]] gone = rows(keyboard({"uv", "gusts"}, KINDS, "unsub", parent="uv")) # its type is already off, others on: just "all" - assert gone[1:] == [[("Unsubscribe from all", "al:off:all:unsub:uv")]] + assert gone[1:] == [[("๐Ÿ”• Unsubscribe from all", "al:off:all:unsub:uv")]] assert len(rows(keyboard(set(KINDS), KINDS, "unsub", parent="uv"))) == 1 # nothing is on at all: nothing listed sub = rows(keyboard({"pollen", "forecast"}, KINDS, "sub", parent="uv")) # Subscribe is the types that are off, as ever - assert [t for row in sub[1:-1] for t, _ in row] == ["pollen & asthma", "forecast changes"] and sub[-1][0][0] == "Subscribe to all" - assert rows(keyboard(set(), KINDS, "unsub"))[1][0][0] == "rain" # the /alerts message has no parent: every type that is on + assert [t for row in sub[1:-1] for t, _ in row] == ["๐ŸŒผ pollen & asthma", "๐Ÿ”„ forecast changes"] and sub[-1][0][0] == "๐Ÿ”” Subscribe to all" + assert rows(keyboard(set(), KINDS, "unsub"))[1][0][0] == "๐ŸŒง๏ธ rain" # the /alerts message has no parent: every type that is on def test_the_all_button_is_only_there_when_the_list_has_more_than_one_item(): assert not any("all" in data for row in rows(keyboard({"uv", "gusts", "rain", "rain_likely", "temps", "air", "pollen"}, KINDS, "unsub")) for _, data in row) # one type on (forecast changes): no "Unsubscribe from all" assert not any("all" in data for row in rows(keyboard({"uv"}, KINDS, "sub")) for _, data in row) # one type off: no "Subscribe to all" - assert rows(keyboard({"uv", "gusts"}, KINDS, "sub"))[-1][0][0] == "Subscribe to all" - assert rows(keyboard(set(), KINDS, "unsub"))[-1][0][0] == "Unsubscribe from all" + assert rows(keyboard({"uv", "gusts"}, KINDS, "sub"))[-1][0][0] == "๐Ÿ”” Subscribe to all" + assert rows(keyboard(set(), KINDS, "unsub"))[-1][0][0] == "๐Ÿ”• Unsubscribe from all" async def test_the_alerts_own_type_survives_opening_unsubscribing_and_closing_the_menu_under_it(tmp_path): @@ -283,11 +283,11 @@ async def test_the_alerts_own_type_survives_opening_unsubscribing_and_closing_th await press(bot, q) drawn = rows(q.edits[0][1]) assert drawn[0][1] == ("โ–พ โž– Unsubscribe", "al:close:uv") and drawn[1:] == [ - [("โ— UV", "al:off:uv:unsub:uv")], [("Unsubscribe from all", "al:off:all:unsub:uv")]] + [("โ— ๐Ÿงด UV", "al:off:uv:unsub:uv")], [("๐Ÿ”• Unsubscribe from all", "al:off:all:unsub:uv")]] q = Query("al:off:uv:unsub:uv", text="โ˜€๏ธ UV 10") # unsubscribing from this type leaves "all" for the rest await press(bot, q) assert "uv" in state.muted(1) and q.toast == "UV alerts off in this chat" - assert rows(q.edits[0][1])[1:] == [[("Unsubscribe from all", "al:off:all:unsub:uv")]] + assert rows(q.edits[0][1])[1:] == [[("๐Ÿ”• Unsubscribe from all", "al:off:all:unsub:uv")]] q = Query("al:off:all:unsub:uv", text="โ˜€๏ธ UV 10") # and "all" turns the rest off, closing the empty section await press(bot, q) assert rows(q.edits[0][1]) == [[("โž• Subscribe", "al:open:sub:uv"), ("โž– Unsubscribe", "al:open:unsub:uv")]] @@ -301,5 +301,19 @@ async def test_an_unknown_parent_in_a_button_is_ignored(tmp_path): q = Query("al:open:unsub:bogus", text="โ˜€๏ธ UV 10") await press(bot, q) drawn = rows(q.edits[0][1]) - assert [t for row in drawn[1:] for t, _ in row][0] == "rain" and not any(t.startswith("โ—") for row in drawn for t, _ in row) + assert [t for row in drawn[1:] for t, _ in row][0] == "๐ŸŒง๏ธ rain" and not any(t.startswith("โ—") for row in drawn for t, _ in row) assert drawn[0][0][1] == "al:open:sub" # and it is not carried on + + +def test_every_alert_type_has_its_emoji_and_every_button_in_the_lists_starts_with_it(): + from lib.alerts.menu import EMOJI, label + assert set(EMOJI) == set(LABELS) # a new type cannot be added without one + assert len(set(EMOJI.values())) == len(EMOJI) and all(not e.isascii() for e in EMOJI.values()) # all different, all emoji + assert label("rain") == "๐ŸŒง๏ธ rain" and label("irrigation") == "๐Ÿชซ irrigation battery" and label("irrigation_pause") == "โ˜” irrigation pause" + everything = list(LABELS) + for section in ("sub", "unsub"): + for parent in (None, "uv"): + drawn = rows(keyboard(set(LABELS) if section == "sub" else set(), everything, section, parent)) + texts = [t for row in drawn[1:] for t, _ in row] + assert texts and all(not t.removeprefix("โ— ")[0].isascii() for t in texts), texts # every button under it starts with an emoji + assert title(set(), ["rain"]) == f"{TITLE}\nโœ… On: rain\n๐Ÿ”• Off: nothing" # the on/off summary and the toasts stay plain words diff --git a/tests/test_irrigation.py b/tests/test_irrigation.py index d45853c..3f13c3c 100644 --- a/tests/test_irrigation.py +++ b/tests/test_irrigation.py @@ -204,7 +204,7 @@ def test_the_alert_list_gets_irrigation_battery_only_when_there_is_a_controller( from lib.alerts.menu import LABELS, keyboard assert LABELS["irrigation"] == "irrigation battery" and LABELS["irrigation_pause"] == "irrigation pause" unsub = rows(keyboard(set(), ["rain", "irrigation"], "unsub")) - assert ("irrigation battery", "al:off:irrigation:unsub") in [b for row in unsub for b in row] + assert ("๐Ÿชซ irrigation battery", "al:off:irrigation:unsub") in [b for row in unsub for b in row] # ---------- the daily check ---------- @@ -538,7 +538,8 @@ async def test_the_handler_is_registered_and_the_alert_list_includes_irrigation_ plain = Bot(NS(tz=TZ), None, [NS(name="Ecowitt")], AlertState(tmp_path / "t.json")) added_plain = [] plain.register(NS(add_handler=lambda h, group=0: added_plain.append(h), add_error_handler=lambda h: None)) - assert len(added) == len(added_plain) + 1 + assert len(added) == len(added_plain) # one entry point for every button, with or without a controller + assert "ir:" in bot.buttons() and "ir:" not in plain.buttons() and {"al:", "pd:"} <= set(plain.buttons()) assert "irrigation" in bot._alert_kinds() and "irrigation" not in plain._alert_kinds() assert bot.keyboard_version == templates.version(True) != plain.keyboard_version == templates.VERSION diff --git a/tests/test_startup.py b/tests/test_startup.py index 1ab2b99..25d03bb 100644 --- a/tests/test_startup.py +++ b/tests/test_startup.py @@ -76,7 +76,7 @@ async def stop_when_archived(): # the background archives finish, then we shut assert "Startup warm-up:" in log and " req to fetch" in log and "AirGradient archive:" in log final = next(line for line in log.splitlines() if "Ecowitt archive:" in line and " req, held " in line) assert "failed" not in final and "days (5min/30min/4h/1d)" in final - assert len(app.handlers) == 10 # start+help, reset, keyboard, alerts, alert buttons, period buttons, other updates, membership, messages, errors + assert len(app.handlers) == 10 # start+help, reset, keyboard, alerts, usage, the one button handler, other updates, membership, messages, errors async def test_a_failed_first_refresh_does_not_stop_the_alerts_or_the_archives(tmp_path, monkeypatch, caplog): @@ -213,4 +213,4 @@ async def stop_when_archived(): await ecobot.main() await stopper assert "irrigation battery" in caplog.text # in the list of alert types - assert len(app.handlers) == 11 and closed == [1] # the irrigation buttons' handler, and the client closed at shutdown + assert len(app.handlers) == 10 and closed == [1] # the same handlers (buttons share one entry point), and the client closed at shutdown