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

Adjust async_step_reauth in samsungtv (#74165)

Browse Source
This commit is contained in:
epenet 2022-06-29 11:46:59 +02:00 committed by GitHub
parent 75efb54cc2
commit 1b85929617
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 27 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
homeassistant/components/samsungtv/config_flow.py
View File

@ -11,7 +11,7 @@ import getmac
from samsungtvws.encrypted.authenticator import SamsungTVEncryptedWSAsyncAuthenticator from samsungtvws.encrypted.authenticator import SamsungTVEncryptedWSAsyncAuthenticator
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries, data_entry_flow from homeassistant import config_entries
from homeassistant.components import dhcp, ssdp, zeroconf from homeassistant.components import dhcp, ssdp, zeroconf
from homeassistant.const import ( from homeassistant.const import (
CONF_HOST, CONF_HOST,
@ -23,6 +23,7 @@ from homeassistant.const import (
CONF_TOKEN, CONF_TOKEN,
) )
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.device_registry import format_mac
@ -123,7 +124,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
CONF_SSDP_MAIN_TV_AGENT_LOCATION: self._ssdp_main_tv_agent_location, CONF_SSDP_MAIN_TV_AGENT_LOCATION: self._ssdp_main_tv_agent_location,
} }
def _get_entry_from_bridge(self) -> data_entry_flow.FlowResult: def _get_entry_from_bridge(self) -> FlowResult:
"""Get device entry.""" """Get device entry."""
assert self._bridge assert self._bridge
data = self._base_config_entry() data = self._base_config_entry()
@ -137,7 +138,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_set_device_unique_id(self, raise_on_progress: bool = True) -> None: async def _async_set_device_unique_id(self, raise_on_progress: bool = True) -> None:
"""Set device unique_id.""" """Set device unique_id."""
if not await self._async_get_and_check_device_info(): if not await self._async_get_and_check_device_info():
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) raise AbortFlow(RESULT_NOT_SUPPORTED)
await self._async_set_unique_id_from_udn(raise_on_progress) await self._async_set_unique_id_from_udn(raise_on_progress)
self._async_update_and_abort_for_matching_unique_id() self._async_update_and_abort_for_matching_unique_id()
@ -156,7 +157,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._ssdp_rendering_control_location, self._ssdp_rendering_control_location,
self._ssdp_main_tv_agent_location, self._ssdp_main_tv_agent_location,
): ):
raise data_entry_flow.AbortFlow("already_configured") raise AbortFlow("already_configured")
# Now that we have updated the config entry, we can raise # Now that we have updated the config entry, we can raise
# if another one is progressing # if another one is progressing
if raise_on_progress: if raise_on_progress:
@ -184,7 +185,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
result, method, _info = await self._async_get_device_info_and_method() result, method, _info = await self._async_get_device_info_and_method()
if result not in SUCCESSFUL_RESULTS: if result not in SUCCESSFUL_RESULTS:
LOGGER.debug("No working config found for %s", self._host) LOGGER.debug("No working config found for %s", self._host)
raise data_entry_flow.AbortFlow(result) raise AbortFlow(result)
assert method is not None assert method is not None
self._bridge = SamsungTVBridge.get_bridge(self.hass, method, self._host) self._bridge = SamsungTVBridge.get_bridge(self.hass, method, self._host)
return return
@ -207,7 +208,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Try to get the device info.""" """Try to get the device info."""
result, _method, info = await self._async_get_device_info_and_method() result, _method, info = await self._async_get_device_info_and_method()
if result not in SUCCESSFUL_RESULTS: if result not in SUCCESSFUL_RESULTS:
raise data_entry_flow.AbortFlow(result) raise AbortFlow(result)
if not info: if not info:
return False return False
dev_info = info.get("device", {}) dev_info = info.get("device", {})
@ -216,7 +217,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
LOGGER.debug( LOGGER.debug(
"Host:%s has type: %s which is not supported", self._host, device_type "Host:%s has type: %s which is not supported", self._host, device_type
) )
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) raise AbortFlow(RESULT_NOT_SUPPORTED)
self._model = dev_info.get("modelName") self._model = dev_info.get("modelName")
name = dev_info.get("name") name = dev_info.get("name")
self._name = name.replace("[TV] ", "") if name else device_type self._name = name.replace("[TV] ", "") if name else device_type
@ -230,9 +231,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._mac = mac self._mac = mac
return True return True
async def async_step_import( async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
self, user_input: dict[str, Any]
) -> data_entry_flow.FlowResult:
"""Handle configuration by yaml file.""" """Handle configuration by yaml file."""
# We need to import even if we cannot validate # We need to import even if we cannot validate
# since the TV may be off at startup # since the TV may be off at startup
@ -257,13 +256,13 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
socket.gethostbyname, user_input[CONF_HOST] socket.gethostbyname, user_input[CONF_HOST]
) )
except socket.gaierror as err: except socket.gaierror as err:
raise data_entry_flow.AbortFlow(RESULT_UNKNOWN_HOST) from err raise AbortFlow(RESULT_UNKNOWN_HOST) from err
self._name = user_input.get(CONF_NAME, self._host) or "" self._name = user_input.get(CONF_NAME, self._host) or ""
self._title = self._name self._title = self._name
async def async_step_user( async def async_step_user(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Handle a flow initialized by the user.""" """Handle a flow initialized by the user."""
if user_input is not None: if user_input is not None:
await self._async_set_name_host_from_input(user_input) await self._async_set_name_host_from_input(user_input)
@ -281,7 +280,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_pairing( async def async_step_pairing(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Handle a pairing by accepting the message on the TV.""" """Handle a pairing by accepting the message on the TV."""
assert self._bridge is not None assert self._bridge is not None
errors: dict[str, str] = {} errors: dict[str, str] = {}
@ -290,7 +289,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if result == RESULT_SUCCESS: if result == RESULT_SUCCESS:
return self._get_entry_from_bridge() return self._get_entry_from_bridge()
if result != RESULT_AUTH_MISSING: if result != RESULT_AUTH_MISSING:
raise data_entry_flow.AbortFlow(result) raise AbortFlow(result)
errors = {"base": RESULT_AUTH_MISSING} errors = {"base": RESULT_AUTH_MISSING}
self.context["title_placeholders"] = {"device": self._title} self.context["title_placeholders"] = {"device": self._title}
@ -303,7 +302,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_encrypted_pairing( async def async_step_encrypted_pairing(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Handle a encrypted pairing.""" """Handle a encrypted pairing."""
assert self._host is not None assert self._host is not None
await self._async_start_encrypted_pairing(self._host) await self._async_start_encrypted_pairing(self._host)
@ -421,7 +420,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if (entry := self._async_update_existing_matching_entry()) and entry.unique_id: if (entry := self._async_update_existing_matching_entry()) and entry.unique_id:
# If we have the unique id and the mac we abort # If we have the unique id and the mac we abort
# as we do not need anything else # as we do not need anything else
raise data_entry_flow.AbortFlow("already_configured") raise AbortFlow("already_configured")
self._async_abort_if_host_already_in_progress() self._async_abort_if_host_already_in_progress()
@callback @callback
@ -429,18 +428,16 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.context[CONF_HOST] = self._host self.context[CONF_HOST] = self._host
for progress in self._async_in_progress(): for progress in self._async_in_progress():
if progress.get("context", {}).get(CONF_HOST) == self._host: if progress.get("context", {}).get(CONF_HOST) == self._host:
raise data_entry_flow.AbortFlow("already_in_progress") raise AbortFlow("already_in_progress")
@callback @callback
def _abort_if_manufacturer_is_not_samsung(self) -> None: def _abort_if_manufacturer_is_not_samsung(self) -> None:
if not self._manufacturer or not self._manufacturer.lower().startswith( if not self._manufacturer or not self._manufacturer.lower().startswith(
"samsung" "samsung"
): ):
raise data_entry_flow.AbortFlow(RESULT_NOT_SUPPORTED) raise AbortFlow(RESULT_NOT_SUPPORTED)
async def async_step_ssdp( async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo) -> FlowResult:
self, discovery_info: ssdp.SsdpServiceInfo
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by ssdp discovery.""" """Handle a flow initialized by ssdp discovery."""
LOGGER.debug("Samsung device found via SSDP: %s", discovery_info) LOGGER.debug("Samsung device found via SSDP: %s", discovery_info)
model_name: str = discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME) or "" model_name: str = discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME) or ""
@ -485,9 +482,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = {"device": self._title} self.context["title_placeholders"] = {"device": self._title}
return await self.async_step_confirm() return await self.async_step_confirm()
async def async_step_dhcp( async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult:
self, discovery_info: dhcp.DhcpServiceInfo
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by dhcp discovery.""" """Handle a flow initialized by dhcp discovery."""
LOGGER.debug("Samsung device found via DHCP: %s", discovery_info) LOGGER.debug("Samsung device found via DHCP: %s", discovery_info)
self._mac = discovery_info.macaddress self._mac = discovery_info.macaddress
@ -499,7 +494,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_zeroconf( async def async_step_zeroconf(
self, discovery_info: zeroconf.ZeroconfServiceInfo self, discovery_info: zeroconf.ZeroconfServiceInfo
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Handle a flow initialized by zeroconf discovery.""" """Handle a flow initialized by zeroconf discovery."""
LOGGER.debug("Samsung device found via ZEROCONF: %s", discovery_info) LOGGER.debug("Samsung device found via ZEROCONF: %s", discovery_info)
self._mac = format_mac(discovery_info.properties["deviceid"]) self._mac = format_mac(discovery_info.properties["deviceid"])
@ -511,7 +506,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_confirm( async def async_step_confirm(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Handle user-confirmation of discovered node.""" """Handle user-confirmation of discovered node."""
if user_input is not None: if user_input is not None:
await self._async_create_bridge() await self._async_create_bridge()
@ -524,24 +519,20 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
step_id="confirm", description_placeholders={"device": self._title} step_id="confirm", description_placeholders={"device": self._title}
) )
async def async_step_reauth( async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
self, data: Mapping[str, Any]
) -> data_entry_flow.FlowResult:
"""Handle configuration by re-auth.""" """Handle configuration by re-auth."""
self._reauth_entry = self.hass.config_entries.async_get_entry( self._reauth_entry = self.hass.config_entries.async_get_entry(
self.context["entry_id"] self.context["entry_id"]
) )
assert self._reauth_entry if entry_data.get(CONF_MODEL) and entry_data.get(CONF_NAME):
data = self._reauth_entry.data self._title = f"{entry_data[CONF_NAME]} ({entry_data[CONF_MODEL]})"
if data.get(CONF_MODEL) and data.get(CONF_NAME):
self._title = f"{data[CONF_NAME]} ({data[CONF_MODEL]})"
else: else:
self._title = data.get(CONF_NAME) or data[CONF_HOST] self._title = entry_data.get(CONF_NAME) or entry_data[CONF_HOST]
return await self.async_step_reauth_confirm() return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm( async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Confirm reauth.""" """Confirm reauth."""
errors = {} errors = {}
assert self._reauth_entry assert self._reauth_entry
@ -586,7 +577,7 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_reauth_confirm_encrypted( async def async_step_reauth_confirm_encrypted(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult: ) -> FlowResult:
"""Confirm reauth (encrypted method).""" """Confirm reauth (encrypted method)."""
errors = {} errors = {}
assert self._reauth_entry assert self._reauth_entry