Add reauth flow to BraviaTV integration (#79405)

* Raise ConfigEntryAuthFailed

* Add reauth flow

* Add tests

* Patch pair() method to avoid IO

* Remove unused errors dict
This commit is contained in:
Maciej Bieniek 2022-10-02 20:07:57 +00:00 committed by GitHub
parent f28e3fb46c
commit 3b794038b1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 187 additions and 8 deletions

View File

@ -1,6 +1,7 @@
"""Config flow to configure the Bravia TV integration.""" """Config flow to configure the Bravia TV integration."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Mapping
from typing import Any from typing import Any
from urllib.parse import urlparse from urllib.parse import urlparse
@ -40,6 +41,7 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Initialize config flow.""" """Initialize config flow."""
self.client: BraviaTV | None = None self.client: BraviaTV | None = None
self.device_config: dict[str, Any] = {} self.device_config: dict[str, Any] = {}
self.entry: ConfigEntry | None = None
@staticmethod @staticmethod
@callback @callback
@ -177,6 +179,56 @@ class BraviaTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="confirm") return self.async_show_form(step_id="confirm")
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle configuration by re-auth."""
self.entry = self.hass.config_entries.async_get_entry(self.context["entry_id"])
self.device_config = {**entry_data}
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Dialog that informs the user that reauth is required."""
self.create_client()
assert self.client is not None
assert self.entry is not None
if user_input is not None:
pin = user_input[CONF_PIN]
use_psk = user_input[CONF_USE_PSK]
try:
if use_psk:
await self.client.connect(psk=pin)
else:
await self.client.connect(
pin=pin, clientid=CLIENTID_PREFIX, nickname=NICKNAME
)
await self.client.set_wol_mode(True)
except BraviaTVError:
return self.async_abort(reason="reauth_unsuccessful")
else:
self.hass.config_entries.async_update_entry(
self.entry, data={**self.device_config, **user_input}
)
await self.hass.config_entries.async_reload(self.entry.entry_id)
return self.async_abort(reason="reauth_successful")
try:
await self.client.pair(CLIENTID_PREFIX, NICKNAME)
except BraviaTVError:
return self.async_abort(reason="reauth_unsuccessful")
return self.async_show_form(
step_id="reauth_confirm",
data_schema=vol.Schema(
{
vol.Required(CONF_PIN, default=""): str,
vol.Required(CONF_USE_PSK, default=False): bool,
}
),
)
class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow): class BraviaTVOptionsFlowHandler(config_entries.OptionsFlow):
"""Config flow options for Bravia TV.""" """Config flow options for Bravia TV."""

View File

@ -9,6 +9,7 @@ from typing import Any, Final, TypeVar
from pybravia import ( from pybravia import (
BraviaTV, BraviaTV,
BraviaTVAuthError,
BraviaTVConnectionError, BraviaTVConnectionError,
BraviaTVConnectionTimeout, BraviaTVConnectionTimeout,
BraviaTVError, BraviaTVError,
@ -19,6 +20,7 @@ from typing_extensions import Concatenate, ParamSpec
from homeassistant.components.media_player import MediaType from homeassistant.components.media_player import MediaType
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.debounce import Debouncer from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -139,6 +141,8 @@ class BraviaTVCoordinator(DataUpdateCoordinator[None]):
_LOGGER.debug("Update skipped, Bravia API service is reloading") _LOGGER.debug("Update skipped, Bravia API service is reloading")
return return
raise UpdateFailed("Error communicating with device") from err raise UpdateFailed("Error communicating with device") from err
except BraviaTVAuthError as err:
raise ConfigEntryAuthFailed from err
except (BraviaTVConnectionError, BraviaTVConnectionTimeout, BraviaTVTurnedOff): except (BraviaTVConnectionError, BraviaTVConnectionTimeout, BraviaTVTurnedOff):
self.is_on = False self.is_on = False
self.connected = False self.connected = False

View File

@ -17,6 +17,13 @@
}, },
"confirm": { "confirm": {
"description": "[%key:common::config_flow::description::confirm_setup%]" "description": "[%key:common::config_flow::description::confirm_setup%]"
},
"reauth_confirm": {
"description": "Enter the PIN code shown on the Sony Bravia TV. \n\nIf the PIN code is not shown, you have to unregister Home Assistant on your TV, go to: Settings -> Network -> Remote device settings -> Deregister remote device. \n\nYou can use PSK (Pre-Shared-Key) instead of PIN. PSK is a user-defined secret key used for access control. This authentication method is recommended as more stable. To enable PSK on your TV, go to: Settings -> Network -> Home Network Setup -> IP Control. Then check «Use PSK authentication» box and enter your PSK instead of PIN.",
"data": {
"pin": "[%key:common::config_flow::data::pin%]",
"use_psk": "Use PSK authentication"
}
} }
}, },
"error": { "error": {
@ -28,7 +35,9 @@
"abort": { "abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]", "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_ip_control": "IP Control is disabled on your TV or the TV is not supported.", "no_ip_control": "IP Control is disabled on your TV or the TV is not supported.",
"not_bravia_device": "The device is not a Bravia TV." "not_bravia_device": "The device is not a Bravia TV.",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"reauth_unsuccessful": "Re-authentication was unsuccessful, please remove the integration and set it up again."
} }
}, },
"options": { "options": {

View File

@ -1,7 +1,13 @@
"""Define tests for the Bravia TV config flow.""" """Define tests for the Bravia TV config flow."""
from unittest.mock import patch from unittest.mock import patch
from pybravia import BraviaTVAuthError, BraviaTVConnectionError, BraviaTVNotSupported from pybravia import (
BraviaTVAuthError,
BraviaTVConnectionError,
BraviaTVError,
BraviaTVNotSupported,
)
import pytest
from homeassistant import data_entry_flow from homeassistant import data_entry_flow
from homeassistant.components import ssdp from homeassistant.components import ssdp
@ -10,7 +16,7 @@ from homeassistant.components.braviatv.const import (
CONF_USE_PSK, CONF_USE_PSK,
DOMAIN, DOMAIN,
) )
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
from tests.common import MockConfigEntry from tests.common import MockConfigEntry
@ -222,6 +228,7 @@ async def test_authorize_model_unsupported(hass):
async def test_authorize_no_ip_control(hass): async def test_authorize_no_ip_control(hass):
"""Test that errors are shown when IP Control is disabled on the TV.""" """Test that errors are shown when IP Control is disabled on the TV."""
with patch("pybravia.BraviaTV.pair", side_effect=BraviaTVError):
result = await hass.config_entries.flow.async_init( result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"} DOMAIN, context={"source": SOURCE_USER}, data={CONF_HOST: "bravia-host"}
) )
@ -398,3 +405,110 @@ async def test_options_flow(hass):
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert config_entry.options == {CONF_IGNORED_SOURCES: ["HDMI 1", "HDMI 2"]} assert config_entry.options == {CONF_IGNORED_SOURCES: ["HDMI 1", "HDMI 2"]}
@pytest.mark.parametrize(
"user_input",
[{CONF_PIN: "mypsk", CONF_USE_PSK: True}, {CONF_PIN: "1234", CONF_USE_PSK: False}],
)
async def test_reauth_successful(hass, user_input):
"""Test starting a reauthentication flow."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch("pybravia.BraviaTV.connect"), patch(
"pybravia.BraviaTV.get_power_status",
return_value="active",
), patch(
"pybravia.BraviaTV.get_external_status",
return_value=BRAVIA_SOURCES,
), patch(
"pybravia.BraviaTV.send_rest_req",
return_value={},
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH, "entry_id": config_entry.entry_id},
data=config_entry.data,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_reauth_unsuccessful(hass):
"""Test reauthentication flow failed."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch(
"pybravia.BraviaTV.connect",
side_effect=BraviaTVAuthError,
), patch("pybravia.BraviaTV.pair"):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH, "entry_id": config_entry.entry_id},
data=config_entry.data,
)
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_PIN: "mypsk", CONF_USE_PSK: True},
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_unsuccessful"
async def test_reauth_unsuccessful_during_pairing(hass):
"""Test reauthentication flow failed because of pairing error."""
config_entry = MockConfigEntry(
domain=DOMAIN,
unique_id="very_unique_string",
data={
CONF_HOST: "bravia-host",
CONF_PIN: "1234",
CONF_MAC: "AA:BB:CC:DD:EE:FF",
},
title="TV-Model",
)
config_entry.add_to_hass(hass)
with patch("pybravia.BraviaTV.pair", side_effect=BraviaTVError):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_REAUTH, "entry_id": config_entry.entry_id},
data=config_entry.data,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_unsuccessful"