Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 17:49:37 +01:00
Code Activity

Add API server endpoint to options for Telegram bot (#161580)

Browse Source
This commit is contained in:
hanwg
2026-01-28 23:16:32 +08:00
committed by GitHub
parent 8a08016fb9
commit d5e58c817d
13 changed files with 322 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/telegram_bot/__init__.py
+36
View File
@@ -87,7 +87,9 @@ from .const import (
CHAT_ACTION_UPLOAD_VIDEO,
CHAT_ACTION_UPLOAD_VIDEO_NOTE,
CHAT_ACTION_UPLOAD_VOICE,
CONF_API_ENDPOINT,
CONF_CONFIG_ENTRY_ID,
DEFAULT_API_ENDPOINT,
DOMAIN,
PLATFORM_BROADCAST,
PLATFORM_POLLING,
@@ -551,6 +553,40 @@ def _deprecate_timeout(hass: HomeAssistant, service: ServiceCall) -> None:
)
async def async_migrate_entry(
hass: HomeAssistant, config_entry: TelegramBotConfigEntry
) -> bool:
"""Migrate Telegram Bot config entry."""
version = config_entry.version
minor_version = config_entry.minor_version
_LOGGER.debug(
"Migrating configuration from version %s.%s",
version,
minor_version,
)
if config_entry.version > 1:
# This means the user has downgraded from a future version
return False
# version 1.1: to add default API endpoint
if version == 1 and minor_version == 1:
new_data = {**config_entry.data}
new_data[CONF_API_ENDPOINT] = DEFAULT_API_ENDPOINT
updated = hass.config_entries.async_update_entry(
config_entry, data=new_data, minor_version=2
)
_LOGGER.debug(
"Migrated Telegram Bot config entry to %s.%s, entry updated: %s",
config_entry.version,
config_entry.minor_version,
updated,
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: TelegramBotConfigEntry) -> bool:
"""Create the Telegram bot from config entry."""
bot: Bot = await hass.async_add_executor_job(initialize_bot, hass, entry.data)
homeassistant/components/telegram_bot/bot.py
+13 -3
View File
@@ -94,6 +94,7 @@ from .const import (
ATTR_USER_ID,
ATTR_USERNAME,
ATTR_VERIFY_SSL,
CONF_API_ENDPOINT,
CONF_CHAT_ID,
CONF_PROXY_URL,
DOMAIN,
@@ -1099,15 +1100,24 @@ class TelegramNotificationService:
def initialize_bot(hass: HomeAssistant, p_config: MappingProxyType[str, Any]) -> Bot:
"""Initialize telegram bot with proxy support."""
api_key: str = p_config[CONF_API_KEY]
proxy_url: str | None = p_config.get(CONF_PROXY_URL)
api_key: str = p_config[CONF_API_KEY]
proxy_url: str | None = p_config.get(CONF_PROXY_URL)
if proxy_url is not None:
proxy = httpx.Proxy(proxy_url)
request = HTTPXRequest(connection_pool_size=8, proxy=proxy)
else:
request = HTTPXRequest(connection_pool_size=8)
return Bot(token=api_key, request=request)
base_url: str = p_config[CONF_API_ENDPOINT]
return Bot(
token=api_key,
base_url=f"{base_url}/bot",
base_file_url=f"{base_url}/file/bot",
request=request,
)
async def load_data(
homeassistant/components/telegram_bot/config_flow.py
+68 -4
View File
@@ -12,6 +12,7 @@ import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_RECONFIGURE,
ConfigEntryState,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryFlow,
@@ -32,13 +33,15 @@ from homeassistant.helpers.selector import (
)
from . import initialize_bot
from .bot import TelegramBotConfigEntry
from .bot import TelegramBotConfigEntry, TelegramNotificationService
from .const import (
ATTR_PARSER,
BOT_NAME,
CONF_API_ENDPOINT,
CONF_CHAT_ID,
CONF_PROXY_URL,
CONF_TRUSTED_NETWORKS,
DEFAULT_API_ENDPOINT,
DEFAULT_TRUSTED_NETWORKS,
DOMAIN,
ERROR_FIELD,
@@ -62,6 +65,8 @@ DESCRIPTION_PLACEHOLDERS: dict[str, str] = {
"getidsbot_username": "@GetIDs Bot",
"getidsbot_url": "https://t.me/getidsbot",
"socks_url": "socks5://username:password@proxy_ip:proxy_port",
# used in advanced settings section
"default_api_endpoint": DEFAULT_API_ENDPOINT,
}
STEP_USER_DATA_SCHEMA: vol.Schema = vol.Schema(
@@ -85,6 +90,12 @@ STEP_USER_DATA_SCHEMA: vol.Schema = vol.Schema(
vol.Required(SECTION_ADVANCED_SETTINGS): section(
vol.Schema(
{
vol.Required(
CONF_API_ENDPOINT,
default=DEFAULT_API_ENDPOINT,
): TextSelector(
config=TextSelectorConfig(type=TextSelectorType.URL)
),
vol.Optional(CONF_PROXY_URL): TextSelector(
config=TextSelectorConfig(type=TextSelectorType.URL)
),
@@ -109,6 +120,12 @@ STEP_RECONFIGURE_USER_DATA_SCHEMA: vol.Schema = vol.Schema(
vol.Required(SECTION_ADVANCED_SETTINGS): section(
vol.Schema(
{
vol.Required(
CONF_API_ENDPOINT,
default=DEFAULT_API_ENDPOINT,
): TextSelector(
config=TextSelectorConfig(type=TextSelectorType.URL)
),
vol.Optional(CONF_PROXY_URL): TextSelector(
config=TextSelectorConfig(type=TextSelectorType.URL)
),
@@ -145,7 +162,7 @@ OPTIONS_SCHEMA: vol.Schema = vol.Schema(
options=[PARSER_MD, PARSER_MD2, PARSER_HTML, PARSER_PLAIN_TEXT],
translation_key="parse_mode",
)
)
),
}
)
@@ -174,6 +191,7 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Telegram."""
VERSION = 1
MINOR_VERSION = 2
@staticmethod
@callback
@@ -219,6 +237,9 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
# validate connection to Telegram API
errors: dict[str, str] = {}
user_input[CONF_API_ENDPOINT] = (
user_input[SECTION_ADVANCED_SETTINGS][CONF_API_ENDPOINT],
)
user_input[CONF_PROXY_URL] = user_input[SECTION_ADVANCED_SETTINGS].get(
CONF_PROXY_URL
)
@@ -243,6 +264,7 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
title=bot_name,
data={
CONF_PLATFORM: user_input[CONF_PLATFORM],
CONF_API_ENDPOINT: user_input[CONF_API_ENDPOINT],
CONF_API_KEY: user_input[CONF_API_KEY],
CONF_PROXY_URL: user_input[SECTION_ADVANCED_SETTINGS].get(
CONF_PROXY_URL
@@ -357,6 +379,9 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
data={
CONF_PLATFORM: self._step_user_data[CONF_PLATFORM],
CONF_API_KEY: self._step_user_data[CONF_API_KEY],
CONF_API_ENDPOINT: self._step_user_data[SECTION_ADVANCED_SETTINGS][
CONF_API_ENDPOINT
],
CONF_PROXY_URL: self._step_user_data[SECTION_ADVANCED_SETTINGS].get(
CONF_PROXY_URL
),
@@ -428,6 +453,9 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
{
**self._get_reconfigure_entry().data,
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: self._get_reconfigure_entry().data[
CONF_API_ENDPOINT
],
CONF_PROXY_URL: self._get_reconfigure_entry().data.get(
CONF_PROXY_URL
),
@@ -440,6 +468,10 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
CONF_PROXY_URL
)
user_input[CONF_API_ENDPOINT] = user_input[SECTION_ADVANCED_SETTINGS][
CONF_API_ENDPOINT
]
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = DESCRIPTION_PLACEHOLDERS.copy()
@@ -449,6 +481,35 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
)
self._bot_name = bot_name
existing_api_endpoint: str = self._get_reconfigure_entry().data[
CONF_API_ENDPOINT
]
if (
self._get_reconfigure_entry().state == ConfigEntryState.LOADED
and user_input[CONF_API_ENDPOINT] != DEFAULT_API_ENDPOINT
and existing_api_endpoint == DEFAULT_API_ENDPOINT
):
# logout existing bot from the official Telegram bot API
# logout is only used when changing the API endpoint from official to a custom one
# there is a 10-minute lockout period after logout so we only logout if necessary
service: TelegramNotificationService = (
self._get_reconfigure_entry().runtime_data
)
try:
is_logged_out = await service.bot.log_out()
except TelegramError as err:
errors["base"] = "telegram_error"
description_placeholders[ERROR_MESSAGE] = str(err)
else:
_LOGGER.info(
"[%s %s] Logged out: %s",
service.bot.username,
service.bot.id,
is_logged_out,
)
if not is_logged_out:
errors["base"] = "bot_logout_failed"
if errors:
return self.async_show_form(
step_id="reconfigure",
@@ -457,6 +518,7 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
{
**user_input,
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: user_input[CONF_API_ENDPOINT],
CONF_PROXY_URL: user_input.get(CONF_PROXY_URL),
},
},
@@ -496,8 +558,10 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
updated_data = {**self._get_reauth_entry().data}
updated_data[CONF_API_KEY] = user_input[CONF_API_KEY]
bot_name = await self._validate_bot(
user_input, errors, description_placeholders
updated_data, errors, description_placeholders
)
await self._shutdown_bot()
@@ -512,7 +576,7 @@ class TelgramBotConfigFlow(ConfigFlow, domain=DOMAIN):
)
return self.async_update_reload_and_abort(
self._get_reauth_entry(), title=bot_name, data_updates=user_input
self._get_reauth_entry(), title=bot_name, data_updates=updated_data
)
homeassistant/components/telegram_bot/const.py
+2
View File
@@ -13,6 +13,7 @@ SUBENTRY_TYPE_ALLOWED_CHAT_IDS = "allowed_chat_ids"
CONF_ALLOWED_CHAT_IDS = "allowed_chat_ids"
CONF_CONFIG_ENTRY_ID = "config_entry_id"
CONF_API_ENDPOINT = "api_endpoint"
CONF_PROXY_URL = "proxy_url"
CONF_TRUSTED_NETWORKS = "trusted_networks"
@@ -23,6 +24,7 @@ BOT_NAME = "telegram_bot"
ERROR_FIELD = "error_field"
ERROR_MESSAGE = "error_message"
DEFAULT_API_ENDPOINT = "https://api.telegram.org"
DEFAULT_TRUSTED_NETWORKS = [ip_network("149.154.160.0/20"), ip_network("91.108.4.0/22")]
SERVICE_SEND_CHAT_ACTION = "send_chat_action"
homeassistant/components/telegram_bot/diagnostics.py
+6 -1
View File
@@ -11,7 +11,7 @@ from homeassistant.const import CONF_API_KEY, CONF_URL
from homeassistant.core import HomeAssistant
from . import TelegramBotConfigEntry
from .const import CONF_CHAT_ID
from .const import CONF_API_ENDPOINT, CONF_CHAT_ID, DEFAULT_API_ENDPOINT
TO_REDACT = [CONF_API_KEY, CONF_CHAT_ID]
@@ -26,6 +26,11 @@ async def async_get_config_entry_diagnostics(
url = URL(config_entry.data[CONF_URL])
data[CONF_URL] = url.with_host(REDACTED).human_repr()
api_endpoint = config_entry.data.get(CONF_API_ENDPOINT)
if api_endpoint and api_endpoint != DEFAULT_API_ENDPOINT:
url = URL(config_entry.data[CONF_API_ENDPOINT])
data[CONF_API_ENDPOINT] = url.with_host(REDACTED).human_repr()
return {
"data": data,
"options": async_redact_data(config_entry.options, TO_REDACT),
homeassistant/components/telegram_bot/helpers.py
+5
View File
@@ -8,3 +8,8 @@ from .const import SIGNAL_UPDATE_EVENT
def signal(bot: Bot) -> str:
"""Define signal name."""
return f"{SIGNAL_UPDATE_EVENT}_{bot.id}"
def get_base_url(bot: Bot) -> str:
"""Return the base URL for the bot."""
return bot.base_url.replace(bot.token, "")
homeassistant/components/telegram_bot/polling.py
+7 -1
View File
@@ -9,6 +9,7 @@ from telegram.ext import ApplicationBuilder, CallbackContext, TypeHandler
from homeassistant.core import HomeAssistant
from .bot import BaseTelegramBot, TelegramBotConfigEntry
from .helpers import get_base_url
_LOGGER = logging.getLogger(__name__)
@@ -82,7 +83,12 @@ class PollBot(BaseTelegramBot):
error_callback=lambda error: error_callback(self.bot, error, None)
)
await self.application.start()
_LOGGER.info("[%s %s] Started polling", self.bot.username, self.bot.id)
_LOGGER.info(
"[%s %s] Started polling at %s",
self.bot.username,
self.bot.id,
get_base_url(self.bot),
)
async def stop_polling(self) -> None:
"""Stop the polling task."""
homeassistant/components/telegram_bot/strings.json
+7
View File
@@ -6,6 +6,7 @@
"reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]"
},
"error": {
"bot_logout_failed": "Failed to logout Telegram bot. Please try again later.",
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]",
"invalid_proxy_url": "{proxy_url_error}",
"invalid_trusted_networks": "Invalid trusted network: {error_message}",
@@ -34,9 +35,11 @@
"sections": {
"advanced_settings": {
"data": {
"api_endpoint": "[%key:component::telegram_bot::config::step::user::sections::advanced_settings::data::api_endpoint%]",
"proxy_url": "[%key:component::telegram_bot::config::step::user::sections::advanced_settings::data::proxy_url%]"
},
"data_description": {
"api_endpoint": "[%key:component::telegram_bot::config::step::user::sections::advanced_settings::data_description::api_endpoint%]",
"proxy_url": "[%key:component::telegram_bot::config::step::user::sections::advanced_settings::data_description::proxy_url%]"
},
"name": "[%key:component::telegram_bot::config::step::user::sections::advanced_settings::name%]"
@@ -57,9 +60,11 @@
"sections": {
"advanced_settings": {
"data": {
"api_endpoint": "API endpoint",
"proxy_url": "Proxy URL"
},
"data_description": {
"api_endpoint": "Telegram bot API server endpoint.\nThe bot will be **locked out for 10 minutes** if you switch back to the default.\nDefault: `{default_api_endpoint}`.",
"proxy_url": "Proxy URL if working behind one, optionally including username and password.\n({socks_url})"
},
"name": "Advanced settings"
@@ -226,9 +231,11 @@
"step": {
"init": {
"data": {
"api_endpoint": "API endpoint",
"parse_mode": "Parse mode"
},
"data_description": {
"api_endpoint": "Telegram bot API server endpoint.\nThe bot will be **locked out for 10 minutes** if you switch back to the default.\nDefault: `{default_api_endpoint}`.",
"parse_mode": "Default parse mode for messages if not explicit in message data."
},
"title": "Configure Telegram bot"
homeassistant/components/telegram_bot/webhooks.py
+12 -9
View File
@@ -19,11 +19,11 @@ from homeassistant.helpers.network import get_url
from .bot import BaseTelegramBot, TelegramBotConfigEntry
from .const import CONF_TRUSTED_NETWORKS
from .helpers import get_base_url
_LOGGER = logging.getLogger(__name__)
TELEGRAM_WEBHOOK_URL = "/api/telegram_webhooks"
REMOVE_WEBHOOK_URL = ""
SECRET_TOKEN_LENGTH = 32
@@ -39,9 +39,16 @@ async def async_setup_platform(
pushbot = PushBot(hass, bot, config, secret_token)
await pushbot.start_application()
webhook_registered = await pushbot.register_webhook()
if not webhook_registered:
raise ConfigEntryNotReady("Failed to register webhook with Telegram")
_LOGGER.info(
"[%s %s] Webhook registered with %s",
bot.username,
bot.id,
get_base_url(bot),
)
hass.http.register_view(
PushBotView(
@@ -52,6 +59,8 @@ async def async_setup_platform(
secret_token,
)
)
_LOGGER.info("[%s %s] Webhook bot ready", bot.username, bot.id)
return pushbot
@@ -87,6 +96,7 @@ class PushBot(BaseTelegramBot):
async def shutdown(self) -> None:
"""Shutdown the app."""
await self.stop_application()
_LOGGER.info("[%s %s] Webhook bot shutdown", self.bot.username, self.bot.id)
async def _try_to_set_webhook(self) -> bool:
_LOGGER.debug("Registering webhook URL: %s", self.webhook_url)
@@ -117,14 +127,7 @@ class PushBot(BaseTelegramBot):
# Some logging of Bot current status:
_LOGGER.debug("telegram webhook status: %s", current_status)
result = await self._try_to_set_webhook()
if result:
_LOGGER.debug("Set new telegram webhook %s", self.webhook_url)
else:
_LOGGER.error("Set telegram webhook failed %s", self.webhook_url)
return False
return True
return await self._try_to_set_webhook()
async def stop_application(self) -> None:
"""Handle gracefully stopping the Application object."""
tests/components/telegram_bot/conftest.py
+9
View File
@@ -20,8 +20,10 @@ from telegram.constants import ChatType
from homeassistant.components.telegram_bot.const import (
ATTR_PARSER,
CONF_ALLOWED_CHAT_IDS,
CONF_API_ENDPOINT,
CONF_CHAT_ID,
CONF_TRUSTED_NETWORKS,
DEFAULT_API_ENDPOINT,
DOMAIN,
PARSER_MD,
PLATFORM_BROADCAST,
@@ -44,6 +46,7 @@ def mock_polling_config_entry() -> MockConfigEntry:
data={
CONF_PLATFORM: PLATFORM_POLLING,
CONF_API_KEY: "mock api key",
CONF_API_ENDPOINT: DEFAULT_API_ENDPOINT,
},
options={ATTR_PARSER: PARSER_MD},
subentries_data=[
@@ -60,6 +63,7 @@ def mock_polling_config_entry() -> MockConfigEntry:
title="mock chat 2",
),
],
minor_version=2,
)
@@ -133,6 +137,7 @@ def mock_external_calls() -> Generator[None]:
patch.object(BotMock, "send_animation", return_value=message),
patch.object(BotMock, "send_location", return_value=message),
patch.object(BotMock, "send_poll", return_value=message),
patch.object(BotMock, "log_out", return_value=True),
patch("telegram.ext.Updater._bootstrap"),
):
yield
@@ -253,6 +258,7 @@ def mock_broadcast_config_entry() -> MockConfigEntry:
domain=DOMAIN,
data={
CONF_PLATFORM: PLATFORM_BROADCAST,
CONF_API_ENDPOINT: DEFAULT_API_ENDPOINT,
CONF_API_KEY: "mock api key",
},
options={ATTR_PARSER: PARSER_MD},
@@ -270,6 +276,7 @@ def mock_broadcast_config_entry() -> MockConfigEntry:
title="mock chat 2",
),
],
minor_version=2,
)
@@ -283,6 +290,7 @@ def mock_webhooks_config_entry() -> MockConfigEntry:
CONF_PLATFORM: PLATFORM_WEBHOOKS,
CONF_API_KEY: "mock api key",
CONF_URL: "https://test",
CONF_API_ENDPOINT: "http://mock/bot",
CONF_TRUSTED_NETWORKS: ["127.0.0.1"],
},
options={ATTR_PARSER: PARSER_MD},
@@ -294,6 +302,7 @@ def mock_webhooks_config_entry() -> MockConfigEntry:
title="mock chat",
)
],
minor_version=2,
)
tests/components/telegram_bot/snapshots/test_diagnostics.ambr
+1
View File
@@ -2,6 +2,7 @@
# name: test_diagnostics
dict({
'data': dict({
'api_endpoint': 'http://**redacted**/bot',
'api_key': '**REDACTED**',
'platform': 'webhooks',
'trusted_networks': list([
tests/components/telegram_bot/test_config_flow.py
+87 -2
View File
@@ -6,8 +6,10 @@ from telegram import AcceptedGiftTypes, ChatFullInfo, User
from telegram.constants import AccentColor
from telegram.error import BadRequest, InvalidToken, NetworkError
from homeassistant.components.telegram_bot.config_flow import DESCRIPTION_PLACEHOLDERS
from homeassistant.components.telegram_bot.const import (
ATTR_PARSER,
CONF_API_ENDPOINT,
CONF_CHAT_ID,
CONF_PROXY_URL,
CONF_TRUSTED_NETWORKS,
@@ -24,7 +26,7 @@ from homeassistant.const import CONF_API_KEY, CONF_PLATFORM, CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, pytest
async def test_options_flow(
@@ -130,6 +132,7 @@ async def test_reconfigure_flow_webhooks(
{
CONF_PLATFORM: PLATFORM_WEBHOOKS,
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: "http://mock_api_endpoint",
CONF_PROXY_URL: "https://test",
},
},
@@ -192,11 +195,91 @@ async def test_reconfigure_flow_webhooks(
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_broadcast_config_entry.data[CONF_URL] == "https://reconfigure"
assert (
mock_broadcast_config_entry.data[CONF_API_ENDPOINT]
== "http://mock_api_endpoint"
)
assert mock_broadcast_config_entry.data[CONF_TRUSTED_NETWORKS] == [
"149.154.160.0/20"
]
@pytest.mark.parametrize(
("side_effect", "expected_error", "expected_description_placeholders"),
[
# test case 1: logout fails with network error, then succeeds
pytest.param(
[NetworkError("mock network error"), True],
"telegram_error",
{**DESCRIPTION_PLACEHOLDERS, "error_message": "mock network error"},
),
# test case 2: logout fails with unsuccessful response, then succeeds
pytest.param(
[False, True],
"bot_logout_failed",
DESCRIPTION_PLACEHOLDERS,
),
],
)
async def test_reconfigure_flow_logout_failed(
hass: HomeAssistant,
mock_broadcast_config_entry: MockConfigEntry,
mock_external_calls: None,
side_effect: list,
expected_error: str,
expected_description_placeholders: dict[str, str],
) -> None:
"""Test reconfigure flow for with change in API endpoint and logout failed."""
mock_broadcast_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_broadcast_config_entry.entry_id)
await hass.async_block_till_done()
result = await mock_broadcast_config_entry.start_reconfigure_flow(hass)
assert result["step_id"] == "reconfigure"
assert result["type"] is FlowResultType.FORM
assert result["errors"] is None
with patch(
"homeassistant.components.telegram_bot.bot.Bot.log_out",
AsyncMock(side_effect=side_effect),
):
# first logout attempt fails
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PLATFORM: PLATFORM_BROADCAST,
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: "http://mock1",
},
},
)
await hass.async_block_till_done()
assert result["step_id"] == "reconfigure"
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": expected_error}
assert result["description_placeholders"] == expected_description_placeholders
# second logout attempt success
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PLATFORM: PLATFORM_BROADCAST,
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: "http://mock2",
},
},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reconfigure_successful"
assert mock_broadcast_config_entry.data[CONF_API_ENDPOINT] == "http://mock2"
async def test_create_entry(hass: HomeAssistant) -> None:
"""Test user flow."""
@@ -470,7 +553,9 @@ async def test_duplicate_entry(hass: HomeAssistant) -> None:
data = {
CONF_PLATFORM: PLATFORM_BROADCAST,
CONF_API_KEY: "mock api key",
SECTION_ADVANCED_SETTINGS: {},
SECTION_ADVANCED_SETTINGS: {
CONF_API_ENDPOINT: "http://mock_api_endpoint",
},
}
with patch(
tests/components/telegram_bot/test_init.py
+69
View File
@@ -0,0 +1,69 @@
"""Init tests for the Telegram Bot integration."""
from homeassistant.components.telegram_bot.const import (
ATTR_PARSER,
CONF_API_ENDPOINT,
DEFAULT_API_ENDPOINT,
DOMAIN,
PARSER_MD,
PLATFORM_BROADCAST,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_API_KEY, CONF_PLATFORM
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
async def test_migration_error(
hass: HomeAssistant,
mock_external_calls: None,
) -> None:
"""Test migrate config entry from 1.1 to 1.2."""
mock_config_entry = MockConfigEntry(
unique_id="mock api key",
domain=DOMAIN,
data={
CONF_PLATFORM: PLATFORM_BROADCAST,
CONF_API_KEY: "mock api key",
},
options={ATTR_PARSER: PARSER_MD},
version=99,
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.MIGRATION_ERROR
async def test_migrate_entry_from_1_1(
hass: HomeAssistant,
mock_external_calls: None,
) -> None:
"""Test migrate config entry from 1.1 to 1.2."""
mock_config_entry = MockConfigEntry(
unique_id="mock api key",
domain=DOMAIN,
data={
CONF_PLATFORM: PLATFORM_BROADCAST,
CONF_API_KEY: "mock api key",
},
options={ATTR_PARSER: PARSER_MD},
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert mock_config_entry.version == 1
assert mock_config_entry.minor_version == 2
assert mock_config_entry.data == {
CONF_PLATFORM: PLATFORM_BROADCAST,
CONF_API_KEY: "mock api key",
CONF_API_ENDPOINT: DEFAULT_API_ENDPOINT,
}