Add webhook tests for Telegram bot (home-assistant#153998)

Author: hanwg

tests/components/telegram_bot/conftest.py

@@ -6,7 +6,7 @@
 from unittest.mock import AsyncMock, patch
 
 import pytest
-from telegram import Bot, Chat, ChatFullInfo, Message, User
+from telegram import Bot, Chat, ChatFullInfo, Message, User, WebhookInfo
 from telegram.constants import AccentColor, ChatType
 
 from homeassistant.components.telegram_bot import (
@@ -74,11 +74,22 @@ def mock_register_webhook() -> Generator[None]:
     """Mock calls made by telegram_bot when (de)registering webhook."""
     with (
         patch(
-            "homeassistant.components.telegram_bot.webhooks.PushBot.register_webhook",
-            return_value=True,
+            "homeassistant.components.telegram_bot.webhooks.Bot.delete_webhook",
+            AsyncMock(),
+        ),
+        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,
+                )
+            ),
         ),
         patch(
-            "homeassistant.components.telegram_bot.webhooks.PushBot.deregister_webhook",
+            "homeassistant.components.telegram_bot.webhooks.Bot.set_webhook",
             return_value=True,
         ),
     ):
@@ -113,9 +124,6 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
             super().__init__(*args, **kwargs)
             self._bot_user = test_user
 
-        async def delete_webhook(self) -> bool:
-            return True
-
     with (
         patch("homeassistant.components.telegram_bot.bot.Bot", BotMock),
         patch.object(BotMock, "get_chat", return_value=test_chat),

tests/components/telegram_bot/test_webhooks.py

@@ -1,12 +1,11 @@
 """Tests for webhooks."""
 
-from datetime import datetime
 from ipaddress import IPv4Network
-from unittest.mock import AsyncMock, patch
+from unittest.mock import patch
 
-from telegram import WebhookInfo
 from telegram.error import TimedOut
 
+from homeassistant.components.telegram_bot.const import DOMAIN
 from homeassistant.components.telegram_bot.webhooks import TELEGRAM_WEBHOOK_URL
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant
@@ -19,91 +18,61 @@ async def test_set_webhooks_failed(
     hass: HomeAssistant,
     mock_webhooks_config_entry: MockConfigEntry,
     mock_external_calls: None,
-    mock_generate_secret_token,
+    mock_register_webhook: None,
 ) -> 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,
+            "homeassistant.components.telegram_bot.webhooks.secrets.choice",
+            return_value="DEADBEEF12345678DEADBEEF87654321",
+        ),
         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
+        # first fail with exception, second fail with False
+        assert mock_set_webhook.call_count == 2
 
         # SETUP_ERROR is result of ConfigEntryNotReady("Failed to register webhook with Telegram") in webhooks.py
         assert mock_webhooks_config_entry.state == ConfigEntryState.SETUP_ERROR
 
+        # test fail after retries
+
+        mock_set_webhook.reset_mock()
+        mock_set_webhook.side_effect = TimedOut("mock timeout")
+
+        await hass.config_entries.async_reload(mock_webhooks_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+        # 3 retries
+        assert mock_set_webhook.call_count == 3
+
+        assert mock_webhooks_config_entry.state == ConfigEntryState.SETUP_ERROR
+        await hass.async_block_till_done()
+
 
 async def test_set_webhooks(
     hass: HomeAssistant,
     mock_webhooks_config_entry: MockConfigEntry,
     mock_external_calls: None,
+    mock_register_webhook: None,
     mock_generate_secret_token,
 ) -> None:
     """Test set webhooks success."""
     mock_webhooks_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_webhooks_config_entry.entry_id)
 
-    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()
+    await hass.async_block_till_done()
 
-        assert mock_webhooks_config_entry.state == ConfigEntryState.LOADED
+    assert mock_webhooks_config_entry.state == ConfigEntryState.LOADED
 
 
 async def test_webhooks_update_invalid_json(
@@ -148,3 +117,24 @@ async def test_webhooks_unauthorized_network(
 
         await hass.async_block_till_done()
         mock_remote.assert_called_once()
+
+
+async def test_webhooks_deregister_failed(
+    hass: HomeAssistant,
+    webhook_platform,
+    mock_external_calls: None,
+    mock_generate_secret_token,
+) -> None:
+    """Test deregister webhooks."""
+
+    config_entry = hass.config_entries.async_entries(DOMAIN)[0]
+    assert config_entry.state == ConfigEntryState.LOADED
+
+    with patch(
+        "homeassistant.components.telegram_bot.webhooks.Bot.delete_webhook",
+    ) as mock_delete_webhook:
+        mock_delete_webhook.side_effect = TimedOut("mock timeout")
+        await hass.config_entries.async_unload(config_entry.entry_id)
+
+    mock_delete_webhook.assert_called_once()
+    assert config_entry.state == ConfigEntryState.NOT_LOADED