From 20c8a0ec52d92a01edbc864a701a6a753fc52cf8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 8 Oct 2026 23:35:06 +0000 Subject: [PATCH 1/4] Admin-only by default (ADMIN_ONLY, ADMIN_CHAT_IDS), /usage, /alerts open in groups, one checked entry point for every button - Only the admins of the groups the bot is in (plus ADMIN_CHAT_IDS) can use it; anyone else gets a short refusal when they address it. - /alerts may be opened by anyone in a group; its buttons and '/alerts on|off' stay admin-only. - /usage is the Help & Alerts button. - Every button press comes in through one handler that believes it only if its data is short, plain text and is one of the buttons its message really carries, then dispatches by prefix (an unknown prefix is refused). Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WBwHofjb1g9onaYRdLZMbf --- README.md | 7 +- ecobot.py | 6 + lib/admins.py | 56 +++++ lib/bot.py | 135 ++++++++++-- lib/config.py | 25 +++ tests/fakes.py | 2 +- tests/test_admin_only.py | 454 +++++++++++++++++++++++++++++++++++++++ tests/test_irrigation.py | 3 +- tests/test_startup.py | 4 +- 9 files changed, 670 insertions(+), 22 deletions(-) create mode 100644 lib/admins.py create mode 100644 tests/test_admin_only.py diff --git a/README.md b/README.md index 59ee76e..649f428 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, 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); `/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; Telegram's @userinfobot tells you yours) 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/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_irrigation.py b/tests/test_irrigation.py index d45853c..c574693 100644 --- a/tests/test_irrigation.py +++ b/tests/test_irrigation.py @@ -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 From 54125316d7fe99b3bec9157408e16870968e53ec Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 8 Oct 2026 23:47:43 +0000 Subject: [PATCH 2/4] README: drop the pointer to an outside bot for finding a user id Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WBwHofjb1g9onaYRdLZMbf --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index 649f428..b7e5a97 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ Under systemd see `ecobot.service` (it loads the environment file; the clone's l | `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; Telegram's @userinfobot tells you yours) 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 | +| `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 | From 879a65250470fe30ce2babc15378e6ff6d70f822 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 00:02:21 +0000 Subject: [PATCH 3/4] Alert buttons: an emoji on each alert type (the one its own alerts start with) and on Subscribe/Unsubscribe to all Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WBwHofjb1g9onaYRdLZMbf --- lib/alerts/menu.py | 12 ++++++++-- tests/test_alert_menu.py | 48 ++++++++++++++++++++++++++-------------- tests/test_irrigation.py | 2 +- 3 files changed, 42 insertions(+), 20 deletions(-) diff --git a/lib/alerts/menu.py b/lib/alerts/menu.py index 13fde7d..4be7439 100644 --- a/lib/alerts/menu.py +++ b/lib/alerts/menu.py @@ -17,6 +17,9 @@ LABELS = {"rain": "rain", "rain_likely": "rain predicted", "gusts": "gusts", "uv": "UV", "temps": "temperature crossing", "air": "air quality", "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/tests/test_alert_menu.py b/tests/test_alert_menu.py index 9f1358d..21cf7a3 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", "๐Ÿ˜ท air quality"] + 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 @@ -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 @@ -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 c574693..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 ---------- From 4d15e4dbaef3358943331a4a0b48db1564dba571 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 9 Oct 2026 00:05:43 +0000 Subject: [PATCH 4/4] Alerts: the air alert type is called 'particulates' Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WBwHofjb1g9onaYRdLZMbf --- README.md | 2 +- lib/alerts/menu.py | 2 +- tests/test_alert_menu.py | 6 +++--- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index b7e5a97..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); `/usage` is the Help & Alerts button. In a group anyone can open `/alerts`, but only admins can change anything (`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 diff --git a/lib/alerts/menu.py b/lib/alerts/menu.py index 4be7439..f7ea046 100644 --- a/lib/alerts/menu.py +++ b/lib/alerts/menu.py @@ -15,7 +15,7 @@ # 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๏ธ", diff --git a/tests/test_alert_menu.py b/tests/test_alert_menu.py index 21cf7a3..5a08d31 100644 --- a/tests/test_alert_menu.py +++ b/tests/test_alert_menu.py @@ -23,7 +23,7 @@ def test_unsubscribe_opens_the_types_that_are_on_and_subscribe_the_types_that_ar 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 [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") @@ -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): @@ -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]