From 4d28992f2be26806301c01e3de3fa707c67b6378 Mon Sep 17 00:00:00 2001 From: hanwg Date: Wed, 11 Jun 2025 01:58:15 +0800 Subject: [PATCH] Add Telegram bot webhooks tests (#146436) * add tests for webhooks * added asserts --- .../components/telegram_bot/webhooks.py | 18 +-- tests/components/telegram_bot/conftest.py | 2 +- .../components/telegram_bot/test_webhooks.py | 149 ++++++++++++++++++ 3 files changed, 154 insertions(+), 15 deletions(-) create mode 100644 tests/components/telegram_bot/test_webhooks.py diff --git a/homeassistant/components/telegram_bot/webhooks.py b/homeassistant/components/telegram_bot/webhooks.py index b8c2cccb738..9218bcbcd67 100644 --- a/homeassistant/components/telegram_bot/webhooks.py +++ b/homeassistant/components/telegram_bot/webhooks.py @@ -1,6 +1,5 @@ """Support for Telegram bots using webhooks.""" -import datetime as dt from http import HTTPStatus from ipaddress import IPv4Network, ip_address import logging @@ -8,7 +7,7 @@ import secrets import string from telegram import Bot, Update -from telegram.error import NetworkError, TimedOut +from telegram.error import NetworkError, TelegramError from telegram.ext import ApplicationBuilder, TypeHandler from homeassistant.components.http import HomeAssistantView @@ -98,9 +97,9 @@ class PushBot(BaseTelegramBot): api_kwargs={"secret_token": self.secret_token}, connect_timeout=5, ) - except TimedOut: + except TelegramError: retry_num += 1 - _LOGGER.warning("Timeout trying to set webhook (retry #%d)", retry_num) + _LOGGER.warning("Error trying to set webhook (retry #%d)", retry_num) return False @@ -113,16 +112,7 @@ class PushBot(BaseTelegramBot): """Query telegram and register the URL for our webhook.""" current_status = await self.bot.get_webhook_info() # Some logging of Bot current status: - last_error_date = getattr(current_status, "last_error_date", None) - if (last_error_date is not None) and (isinstance(last_error_date, int)): - last_error_date = dt.datetime.fromtimestamp(last_error_date) - _LOGGER.debug( - "Telegram webhook last_error_date: %s. Status: %s", - last_error_date, - current_status, - ) - else: - _LOGGER.debug("telegram webhook status: %s", current_status) + _LOGGER.debug("telegram webhook status: %s", current_status) result = await self._try_to_set_webhook() if result: diff --git a/tests/components/telegram_bot/conftest.py b/tests/components/telegram_bot/conftest.py index 6886a8af6e5..66c3c43ea86 100644 --- a/tests/components/telegram_bot/conftest.py +++ b/tests/components/telegram_bot/conftest.py @@ -275,7 +275,7 @@ def mock_webhooks_config_entry() -> MockConfigEntry: CONF_PLATFORM: PLATFORM_WEBHOOKS, CONF_API_KEY: "mock api key", CONF_URL: "https://test", - CONF_TRUSTED_NETWORKS: "149.154.160.0/20,91.108.4.0/22", + CONF_TRUSTED_NETWORKS: ["149.154.160.0/20", "91.108.4.0/22"], }, options={ATTR_PARSER: PARSER_MD}, subentries_data=[ diff --git a/tests/components/telegram_bot/test_webhooks.py b/tests/components/telegram_bot/test_webhooks.py new file mode 100644 index 00000000000..3419d33074d --- /dev/null +++ b/tests/components/telegram_bot/test_webhooks.py @@ -0,0 +1,149 @@ +"""Tests for webhooks.""" + +from datetime import datetime +from ipaddress import IPv4Network +from unittest.mock import AsyncMock, patch + +from telegram import WebhookInfo +from telegram.error import TimedOut + +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry +from tests.typing import ClientSessionGenerator + + +async def test_set_webhooks_failed( + hass: HomeAssistant, + mock_webhooks_config_entry: MockConfigEntry, + mock_external_calls: None, + mock_generate_secret_token, +) -> None: + """Test set webhooks failed.""" + mock_webhooks_config_entry.add_to_hass(hass) + + with ( + patch( + "homeassistant.components.telegram_bot.webhooks.Bot.get_webhook_info", + AsyncMock( + return_value=WebhookInfo( + url="mock url", + last_error_date=datetime.now(), + has_custom_certificate=False, + pending_update_count=0, + ) + ), + ) as mock_webhook_info, + patch( + "homeassistant.components.telegram_bot.webhooks.Bot.set_webhook", + ) as mock_set_webhook, + patch( + "homeassistant.components.telegram_bot.webhooks.ApplicationBuilder" + ) as application_builder_class, + ): + mock_set_webhook.side_effect = [TimedOut("mock timeout"), False] + application = application_builder_class.return_value.bot.return_value.updater.return_value.build.return_value + application.initialize = AsyncMock() + application.start = AsyncMock() + + await hass.config_entries.async_setup(mock_webhooks_config_entry.entry_id) + await hass.async_block_till_done() + await hass.async_stop() + + mock_webhook_info.assert_called_once() + application.initialize.assert_called_once() + application.start.assert_called_once() + assert mock_set_webhook.call_count > 0 + + # SETUP_ERROR is result of ConfigEntryNotReady("Failed to register webhook with Telegram") in webhooks.py + assert mock_webhooks_config_entry.state == ConfigEntryState.SETUP_ERROR + + +async def test_set_webhooks( + hass: HomeAssistant, + mock_webhooks_config_entry: MockConfigEntry, + mock_external_calls: None, + mock_generate_secret_token, +) -> None: + """Test set webhooks success.""" + mock_webhooks_config_entry.add_to_hass(hass) + + with ( + patch( + "homeassistant.components.telegram_bot.webhooks.Bot.get_webhook_info", + AsyncMock( + return_value=WebhookInfo( + url="mock url", + last_error_date=datetime.now(), + has_custom_certificate=False, + pending_update_count=0, + ) + ), + ) as mock_webhook_info, + patch( + "homeassistant.components.telegram_bot.webhooks.Bot.set_webhook", + AsyncMock(return_value=True), + ) as mock_set_webhook, + patch( + "homeassistant.components.telegram_bot.webhooks.ApplicationBuilder" + ) as application_builder_class, + ): + application = application_builder_class.return_value.bot.return_value.updater.return_value.build.return_value + application.initialize = AsyncMock() + application.start = AsyncMock() + + await hass.config_entries.async_setup(mock_webhooks_config_entry.entry_id) + await hass.async_block_till_done() + await hass.async_stop() + + mock_webhook_info.assert_called_once() + application.initialize.assert_called_once() + application.start.assert_called_once() + mock_set_webhook.assert_called_once() + + assert mock_webhooks_config_entry.state == ConfigEntryState.LOADED + + +async def test_webhooks_update_invalid_json( + hass: HomeAssistant, + webhook_platform, + hass_client: ClientSessionGenerator, + mock_generate_secret_token, +) -> None: + """Test update with invalid json.""" + client = await hass_client() + + response = await client.post( + "/api/telegram_webhooks", + headers={"X-Telegram-Bot-Api-Secret-Token": mock_generate_secret_token}, + ) + assert response.status == 400 + + await hass.async_block_till_done() + + +async def test_webhooks_unauthorized_network( + hass: HomeAssistant, + webhook_platform, + mock_external_calls: None, + mock_generate_secret_token, + hass_client: ClientSessionGenerator, +) -> None: + """Test update with request outside of trusted networks.""" + + client = await hass_client() + + with patch( + "homeassistant.components.telegram_bot.webhooks.ip_address", + return_value=IPv4Network("1.2.3.4"), + ) as mock_remote: + response = await client.post( + "/api/telegram_webhooks", + json="mock json", + headers={"X-Telegram-Bot-Api-Secret-Token": mock_generate_secret_token}, + ) + assert response.status == 401 + + await hass.async_block_till_done() + mock_remote.assert_called_once()