Tell Telegram which updates the bot wants (#180519)

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

homeassistant/components/telegram_bot/bot.py

@@ -123,6 +123,20 @@ from .helpers import signal
 _FILE_TYPES = ("animation", "document", "photo", "sticker", "video", "voice")
 _LOGGER = logging.getLogger(__name__)
 
+# Telegram keeps this per bot, server side, and keeps whatever it was told last
+# when the setting is omitted, so a bot narrowed by an earlier consumer of the
+# token silently drops the rest. Ask for what `handle_update` turns into events:
+# a callback query, and the updates `Update.effective_message` is drawn from.
+ALLOWED_UPDATES = [
+    Update.CALLBACK_QUERY,
+    Update.MESSAGE,
+    Update.EDITED_MESSAGE,
+    Update.CHANNEL_POST,
+    Update.EDITED_CHANNEL_POST,
+    Update.BUSINESS_MESSAGE,
+    Update.EDITED_BUSINESS_MESSAGE,
+]
+
 type TelegramBotConfigEntry = ConfigEntry[TelegramNotificationService]
 
 _RETRY_DELAY = 1  # 1 second delay between retries

homeassistant/components/telegram_bot/polling.py

@@ -9,7 +9,7 @@ from telegram.ext import ApplicationBuilder, CallbackContext, TypeHandler
 
 from homeassistant.core import HomeAssistant
 
-from .bot import BaseTelegramBot, TelegramBotConfigEntry
+from .bot import ALLOWED_UPDATES, BaseTelegramBot, TelegramBotConfigEntry
 from .helpers import get_base_url
 
 _LOGGER = logging.getLogger(__name__)
@@ -82,7 +82,8 @@ class PollBot(BaseTelegramBot):
         await self.application.initialize()
         if self.application.updater:
             await self.application.updater.start_polling(
-                error_callback=lambda error: error_callback(self.bot, error, None)
+                allowed_updates=ALLOWED_UPDATES,
+                error_callback=lambda error: error_callback(self.bot, error, None),
             )
         await self.application.start()
         _LOGGER.info(

homeassistant/components/telegram_bot/webhooks.py

@@ -17,7 +17,7 @@ from homeassistant.const import CONF_URL
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers.network import get_url
 
-from .bot import BaseTelegramBot, TelegramBotConfigEntry
+from .bot import ALLOWED_UPDATES, BaseTelegramBot, TelegramBotConfigEntry
 from .const import CONF_TRUSTED_NETWORKS
 from .helpers import get_base_url
 
@@ -107,6 +107,7 @@ class PushBot(BaseTelegramBot):
             try:
                 return await self.bot.set_webhook(
                     self.webhook_url,
+                    allowed_updates=ALLOWED_UPDATES,
                     api_kwargs={"secret_token": self.secret_token},
                     connect_timeout=5,
                 )

tests/components/telegram_bot/test_telegram_bot.py

@@ -27,6 +27,7 @@ from telegram.error import (
 )
 
 from homeassistant.components.telegram_bot import ATTR_LATITUDE, ATTR_LONGITUDE
+from homeassistant.components.telegram_bot.bot import ALLOWED_UPDATES
 from homeassistant.components.telegram_bot.const import (
     ATTR_AUTHENTICATION,
     ATTR_CALLBACK_QUERY_ID,
@@ -810,6 +811,39 @@ async def test_webhook_endpoint_generates_telegram_attachment_event(
     assert isinstance(events[0].context, Context)
 
 
+async def test_polling_platform_allowed_updates(
+    hass: HomeAssistant,
+    mock_polling_config_entry: MockConfigEntry,
+    mock_external_calls: None,
+) -> None:
+    """Test polling asks for the updates the integration handles.
+
+    Telegram keeps the setting per bot and reuses the last one it was given
+    when it is omitted, so it has to be sent on every start.
+    """
+    with patch(
+        "homeassistant.components.telegram_bot.polling.ApplicationBuilder"
+    ) as application_builder_class:
+        application = (
+            application_builder_class.return_value.bot.return_value.build.return_value
+        )
+        application.updater.start_polling = AsyncMock()
+        application.updater.stop = AsyncMock()
+        application.initialize = AsyncMock()
+        application.start = AsyncMock()
+        application.stop = AsyncMock()
+        application.shutdown = AsyncMock()
+
+        mock_polling_config_entry.add_to_hass(hass)
+        await hass.config_entries.async_setup(mock_polling_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert (
+        application.updater.start_polling.call_args.kwargs["allowed_updates"]
+        == ALLOWED_UPDATES
+    )
+
+
 async def test_polling_platform_message_text_update(
     hass: HomeAssistant,
     mock_polling_config_entry: MockConfigEntry,

tests/components/telegram_bot/test_webhooks.py

@@ -5,6 +5,7 @@ from unittest.mock import AsyncMock, patch
 
 from telegram.error import TimedOut
 
+from homeassistant.components.telegram_bot.bot import ALLOWED_UPDATES
 from homeassistant.components.telegram_bot.const import DOMAIN
 from homeassistant.components.telegram_bot.webhooks import TELEGRAM_WEBHOOK_URL
 from homeassistant.config_entries import ConfigEntryState
@@ -87,6 +88,37 @@ async def test_set_webhooks(
     mock_start.assert_called_once()
 
 
+async def test_set_webhooks_allowed_updates(
+    hass: HomeAssistant,
+    mock_webhooks_config_entry: MockConfigEntry,
+    mock_external_calls: None,
+    mock_register_webhook: None,
+    mock_generate_secret_token: str,
+) -> None:
+    """Test the webhook is registered for the updates the integration handles.
+
+    Telegram keeps the setting per bot and reuses the last one it was given
+    when it is omitted, so it has to be sent on every registration.
+    """
+    mock_webhooks_config_entry.add_to_hass(hass)
+
+    with (
+        patch(
+            "homeassistant.components.telegram_bot.webhooks.Application.start",
+            AsyncMock(),
+        ),
+        patch(
+            "homeassistant.components.telegram_bot.webhooks.Bot.set_webhook",
+            return_value=True,
+        ) as mock_set_webhook,
+    ):
+        await hass.config_entries.async_setup(mock_webhooks_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_webhooks_config_entry.state is ConfigEntryState.LOADED
+    assert mock_set_webhook.call_args.kwargs["allowed_updates"] == ALLOWED_UPDATES
+
+
 async def test_webhooks_update_invalid_json(
     hass: HomeAssistant,
     webhook_bot,