core/tests/components/telegram_bot/test_webhooks.py

"""Tests for webhooks."""

from ipaddress import IPv4Network
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
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_register_webhook: None,
) -> None:
    """Test set webhooks failed."""
    mock_webhooks_config_entry.add_to_hass(hass)

    with (
        patch(
            "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.Application.start",
            AsyncMock(),
        ) as mock_start,
    ):
        mock_set_webhook.side_effect = [TimedOut("mock timeout"), False]

        await hass.config_entries.async_setup(mock_webhooks_config_entry.entry_id)
        await hass.async_block_till_done()
        await hass.async_stop()

        # first fail with exception, second fail with False
        assert mock_set_webhook.call_count == 2

        # SETUP_ERROR is result of RuntimeError(
        # "Failed to register webhook with Telegram") in webhooks.py
        assert mock_webhooks_config_entry.state is 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 is ConfigEntryState.SETUP_ERROR
        await hass.async_block_till_done()

        assert mock_start.call_count == 0


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)

    with patch(
        "homeassistant.components.telegram_bot.webhooks.Application.start", AsyncMock()
    ) as mock_start:
        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
    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,
    hass_client: ClientSessionGenerator,
    mock_generate_secret_token,
) -> None:
    """Test update with invalid json."""
    client = await hass_client()

    response = await client.post(
        f"{TELEGRAM_WEBHOOK_URL}_123456",
        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_bot,
    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(
            f"{TELEGRAM_WEBHOOK_URL}_123456",
            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()


async def test_webhooks_deregister_failed(
    hass: HomeAssistant,
    webhook_bot,
) -> None:
    """Test deregister webhooks."""

    config_entry = hass.config_entries.async_entries(DOMAIN)[0]
    assert config_entry.state is 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 is ConfigEntryState.NOT_LOADED