Refactor telegram_bot polling/webhooks platforms and add tests (#66433)

Co-authored-by: Pär Berge <paer.berge@gmail.com>

tests/components/telegram_bot/__init__.py

@@ -0,0 +1 @@
+"""Tests for telegram_bot integration."""

tests/components/telegram_bot/conftest.py

@@ -0,0 +1,187 @@
+"""Tests for the telegram_bot integration."""
+from unittest.mock import patch
+
+import pytest
+from telegram.ext.dispatcher import Dispatcher
+
+from homeassistant.components.telegram_bot import (
+    CONF_ALLOWED_CHAT_IDS,
+    CONF_TRUSTED_NETWORKS,
+    DOMAIN,
+)
+from homeassistant.const import CONF_API_KEY, CONF_PLATFORM, CONF_URL
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture
+def config_webhooks():
+    """Fixture for a webhooks platform configuration."""
+    return {
+        DOMAIN: [
+            {
+                CONF_PLATFORM: "webhooks",
+                CONF_URL: "https://test",
+                CONF_TRUSTED_NETWORKS: ["127.0.0.1"],
+                CONF_API_KEY: "1234567890:ABC",
+                CONF_ALLOWED_CHAT_IDS: [
+                    # "me"
+                    12345678,
+                    # Some chat
+                    -123456789,
+                ],
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def config_polling():
+    """Fixture for a polling platform configuration."""
+    return {
+        DOMAIN: [
+            {
+                CONF_PLATFORM: "polling",
+                CONF_API_KEY: "1234567890:ABC",
+                CONF_ALLOWED_CHAT_IDS: [
+                    # "me"
+                    12345678,
+                    # Some chat
+                    -123456789,
+                ],
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def mock_register_webhook():
+    """Mock calls made by telegram_bot when (de)registering webhook."""
+    with patch(
+        "homeassistant.components.telegram_bot.webhooks.PushBot.register_webhook",
+        return_value=True,
+    ), patch(
+        "homeassistant.components.telegram_bot.webhooks.PushBot.deregister_webhook",
+        return_value=True,
+    ):
+        yield
+
+
+@pytest.fixture
+def update_message_command():
+    """Fixture for mocking an incoming update of type message/command."""
+    return {
+        "update_id": 1,
+        "message": {
+            "message_id": 1,
+            "from": {
+                "id": 12345678,
+                "is_bot": False,
+                "first_name": "Firstname",
+                "username": "some_username",
+                "language_code": "en",
+            },
+            "chat": {
+                "id": -123456789,
+                "title": "SomeChat",
+                "type": "group",
+                "all_members_are_administrators": True,
+            },
+            "date": 1644518189,
+            "text": "/command",
+            "entities": [
+                {
+                    "type": "bot_command",
+                    "offset": 0,
+                    "length": 7,
+                }
+            ],
+        },
+    }
+
+
+@pytest.fixture
+def update_message_text():
+    """Fixture for mocking an incoming update of type message/text."""
+    return {
+        "update_id": 1,
+        "message": {
+            "message_id": 1,
+            "date": 1441645532,
+            "from": {
+                "id": 12345678,
+                "is_bot": False,
+                "last_name": "Test Lastname",
+                "first_name": "Test Firstname",
+                "username": "Testusername",
+            },
+            "chat": {
+                "last_name": "Test Lastname",
+                "id": 1111111,
+                "type": "private",
+                "first_name": "Test Firstname",
+                "username": "Testusername",
+            },
+            "text": "HELLO",
+        },
+    }
+
+
+@pytest.fixture
+def unauthorized_update_message_text(update_message_text):
+    """Fixture for mocking an incoming update of type message/text that is not in our `allowed_chat_ids`."""
+    update_message_text["message"]["from"]["id"] = 1234
+    update_message_text["message"]["chat"]["id"] = 1234
+    return update_message_text
+
+
+@pytest.fixture
+def update_callback_query():
+    """Fixture for mocking an incoming update of type callback_query."""
+    return {
+        "update_id": 1,
+        "callback_query": {
+            "id": "4382bfdwdsb323b2d9",
+            "from": {
+                "id": 12345678,
+                "type": "private",
+                "is_bot": False,
+                "last_name": "Test Lastname",
+                "first_name": "Test Firstname",
+                "username": "Testusername",
+            },
+            "chat_instance": "aaa111",
+            "data": "Data from button callback",
+            "inline_message_id": "1234csdbsk4839",
+        },
+    }
+
+
+@pytest.fixture
+async def webhook_platform(hass, config_webhooks, mock_register_webhook):
+    """Fixture for setting up the webhooks platform using appropriate config and mocks."""
+    await async_setup_component(
+        hass,
+        DOMAIN,
+        config_webhooks,
+    )
+    await hass.async_block_till_done()
+
+
+@pytest.fixture
+async def polling_platform(hass, config_polling):
+    """Fixture for setting up the polling platform using appropriate config and mocks."""
+    await async_setup_component(
+        hass,
+        DOMAIN,
+        config_polling,
+    )
+    await hass.async_block_till_done()
+
+
+@pytest.fixture(autouse=True)
+def clear_dispatcher():
+    """Clear the singleton that telegram.ext.dispatcher.Dispatcher sets on itself."""
+    yield
+    Dispatcher._set_singleton(None)
+    # This is how python-telegram-bot resets the dispatcher in their test suite
+    Dispatcher._Dispatcher__singleton_semaphore.release()

tests/components/telegram_bot/test_telegram_bot.py

@@ -0,0 +1,112 @@
+"""Tests for the telegram_bot component."""
+from telegram import Update
+from telegram.ext.dispatcher import Dispatcher
+
+from homeassistant.components.telegram_bot import DOMAIN, SERVICE_SEND_MESSAGE
+from homeassistant.components.telegram_bot.webhooks import TELEGRAM_WEBHOOK_URL
+
+from tests.common import async_capture_events
+
+
+async def test_webhook_platform_init(hass, webhook_platform):
+    """Test initialization of the webhooks platform."""
+    assert hass.services.has_service(DOMAIN, SERVICE_SEND_MESSAGE) is True
+
+
+async def test_polling_platform_init(hass, polling_platform):
+    """Test initialization of the polling platform."""
+    assert hass.services.has_service(DOMAIN, SERVICE_SEND_MESSAGE) is True
+
+
+async def test_webhook_endpoint_generates_telegram_text_event(
+    hass, webhook_platform, hass_client, update_message_text
+):
+    """POST to the configured webhook endpoint and assert fired `telegram_text` event."""
+    client = await hass_client()
+    events = async_capture_events(hass, "telegram_text")
+
+    response = await client.post(TELEGRAM_WEBHOOK_URL, json=update_message_text)
+    assert response.status == 200
+    assert (await response.read()).decode("utf-8") == ""
+
+    # Make sure event has fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 1
+    assert events[0].data["text"] == update_message_text["message"]["text"]
+
+
+async def test_webhook_endpoint_generates_telegram_command_event(
+    hass, webhook_platform, hass_client, update_message_command
+):
+    """POST to the configured webhook endpoint and assert fired `telegram_command` event."""
+    client = await hass_client()
+    events = async_capture_events(hass, "telegram_command")
+
+    response = await client.post(TELEGRAM_WEBHOOK_URL, json=update_message_command)
+    assert response.status == 200
+    assert (await response.read()).decode("utf-8") == ""
+
+    # Make sure event has fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 1
+    assert events[0].data["command"] == update_message_command["message"]["text"]
+
+
+async def test_webhook_endpoint_generates_telegram_callback_event(
+    hass, webhook_platform, hass_client, update_callback_query
+):
+    """POST to the configured webhook endpoint and assert fired `telegram_callback` event."""
+    client = await hass_client()
+    events = async_capture_events(hass, "telegram_callback")
+
+    response = await client.post(TELEGRAM_WEBHOOK_URL, json=update_callback_query)
+    assert response.status == 200
+    assert (await response.read()).decode("utf-8") == ""
+
+    # Make sure event has fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 1
+    assert events[0].data["data"] == update_callback_query["callback_query"]["data"]
+
+
+async def test_polling_platform_message_text_update(
+    hass, polling_platform, update_message_text
+):
+    """Provide the `PollBot`s `Dispatcher` with an `Update` and assert fired `telegram_text` event."""
+    events = async_capture_events(hass, "telegram_text")
+
+    def telegram_dispatcher_callback():
+        dispatcher = Dispatcher.get_instance()
+        update = Update.de_json(update_message_text, dispatcher.bot)
+        dispatcher.process_update(update)
+
+    # python-telegram-bots `Updater` uses threading, so we need to schedule its callback in a sync context.
+    await hass.async_add_executor_job(telegram_dispatcher_callback)
+
+    # Make sure event has fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 1
+    assert events[0].data["text"] == update_message_text["message"]["text"]
+
+
+async def test_webhook_endpoint_unauthorized_update_doesnt_generate_telegram_text_event(
+    hass, webhook_platform, hass_client, unauthorized_update_message_text
+):
+    """Update with unauthorized user/chat should not trigger event."""
+    client = await hass_client()
+    events = async_capture_events(hass, "telegram_text")
+
+    response = await client.post(
+        TELEGRAM_WEBHOOK_URL, json=unauthorized_update_message_text
+    )
+    assert response.status == 200
+    assert (await response.read()).decode("utf-8") == ""
+
+    # Make sure any events would have fired
+    await hass.async_block_till_done()
+
+    assert len(events) == 0