Add Telegram bot webhooks tests (#146436)

* add tests for webhooks

* added asserts

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)
+        _LOGGER.debug("telegram webhook status: %s", current_status)
-        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)
 
         result = await self._try_to_set_webhook()
         if result:

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=[

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()