jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 19:27:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Clean up Freebox config flow (#97347)

Browse Source 
Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Joost Lekkerkerker 2023-08-23 11:52:28 +02:00 committed by GitHub
parent a2b0149677
commit 5b3c60527a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
homeassistant/components/freebox/config_flow.py
View File

@ -1,5 +1,6 @@
"""Config flow to configure the Freebox integration.""" """Config flow to configure the Freebox integration."""
import logging import logging
from typing import Any
from freebox_api.exceptions import AuthorizationError, HttpRequestError from freebox_api.exceptions import AuthorizationError, HttpRequestError
import voluptuous as vol import voluptuous as vol
@ -20,45 +21,35 @@ class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self) -> None: _data: dict[str, Any] = {}
"""Initialize Freebox config flow."""
self._host: str
self._port = None
def _show_setup_form(self, user_input=None, errors=None): async def async_step_user(
"""Show the setup form to the user.""" self, user_input: dict[str, Any] | None = None
) -> FlowResult:
if user_input is None:
user_input = {}
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
vol.Required(CONF_PORT, default=user_input.get(CONF_PORT, "")): int,
}
),
errors=errors or {},
)
async def async_step_user(self, user_input=None) -> FlowResult:
"""Handle a flow initiated by the user.""" """Handle a flow initiated by the user."""
errors: dict[str, str] = {}
if user_input is None: if user_input is None:
return self._show_setup_form(user_input, errors) return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT): int,
}
),
errors={},
)
self._host = user_input[CONF_HOST] self._data = user_input
self._port = user_input[CONF_PORT]
# Check if already configured # Check if already configured
await self.async_set_unique_id(self._host) await self.async_set_unique_id(self._data[CONF_HOST])
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
return await self.async_step_link() return await self.async_step_link()
async def async_step_link(self, user_input=None) -> FlowResult: async def async_step_link(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Attempt to link with the Freebox router. """Attempt to link with the Freebox router.
Given a configured host, will ask the user to press the button Given a configured host, will ask the user to press the button
@ -69,10 +60,10 @@ class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors = {} errors = {}
fbx = await get_api(self.hass, self._host) fbx = await get_api(self.hass, self._data[CONF_HOST])
try: try:
# Open connection and check authentication # Open connection and check authentication
await fbx.open(self._host, self._port) await fbx.open(self._data[CONF_HOST], self._data[CONF_PORT])
# Check permissions # Check permissions
await fbx.system.get_config() await fbx.system.get_config()
@ -82,8 +73,8 @@ class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
await fbx.close() await fbx.close()
return self.async_create_entry( return self.async_create_entry(
title=self._host, title=self._data[CONF_HOST],
data={CONF_HOST: self._host, CONF_PORT: self._port}, data=self._data,
) )
except AuthorizationError as error: except AuthorizationError as error:
@ -91,18 +82,23 @@ class FreeboxFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors["base"] = "register_failed" errors["base"] = "register_failed"
except HttpRequestError: except HttpRequestError:
_LOGGER.error("Error connecting to the Freebox router at %s", self._host) _LOGGER.error(
"Error connecting to the Freebox router at %s", self._data[CONF_HOST]
)
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except except Exception: # pylint: disable=broad-except
_LOGGER.exception( _LOGGER.exception(
"Unknown error connecting with Freebox router at %s", self._host "Unknown error connecting with Freebox router at %s",
self._data[CONF_HOST],
) )
errors["base"] = "unknown" errors["base"] = "unknown"
return self.async_show_form(step_id="link", errors=errors) return self.async_show_form(step_id="link", errors=errors)
async def async_step_import(self, user_input=None) -> FlowResult: async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Import a config entry.""" """Import a config entry."""
return await self.async_step_user(user_input) return await self.async_step_user(user_input)