Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add re-auth flow for OpenUV (#79691)

pull/81824/head
Aaron Bach 2022-11-08 07:41:09 -07:00 committed by GitHub
parent 014c2d487d
commit 45be2a260e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 228 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
homeassistant/components/openuv/__init__.py
View File

@ -31,7 +31,7 @@ from .const import (
DOMAIN,
LOGGER,
)
from .coordinator import OpenUvCoordinator
from .coordinator import InvalidApiKeyMonitor, OpenUvCoordinator
PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR]
@ -53,6 +53,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
high = entry.options.get(CONF_TO_WINDOW, DEFAULT_TO_WINDOW)
return await client.uv_protection_window(low=low, high=high)
invalid_api_key_monitor = InvalidApiKeyMonitor(hass, entry)
coordinators: dict[str, OpenUvCoordinator] = {
coordinator_name: OpenUvCoordinator(
hass,
@ -60,6 +62,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
latitude=client.latitude,
longitude=client.longitude,
update_method=update_method,
invalid_api_key_monitor=invalid_api_key_monitor,
)
for coordinator_name, update_method in (
(DATA_UV, client.uv_index),

123
homeassistant/components/openuv/config_flow.py
View File

@ -1,6 +1,8 @@
"""Config flow to configure the OpenUV component."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from pyopenuv import Client
@ -27,14 +29,39 @@ from .const import (
DOMAIN,
)
STEP_REAUTH_SCHEMA = vol.Schema(
{
vol.Required(CONF_API_KEY): str,
}
)
@dataclass
class OpenUvData:
"""Define structured OpenUV data needed to create/re-auth an entry."""
api_key: str
latitude: float
longitude: float
elevation: float
@property
def unique_id(self) -> str:
"""Return the unique for this data."""
return f"{self.latitude}, {self.longitude}"
class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle an OpenUV config flow."""
VERSION = 2
def __init__(self) -> None:
"""Initialize."""
self._reauth_data: Mapping[str, Any] = {}
@property
def config_schema(self) -> vol.Schema:
def step_user_schema(self) -> vol.Schema:
"""Return the config schema."""
return vol.Schema(
{
@ -51,13 +78,41 @@ class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}
)
async def _show_form(self, errors: dict[str, Any] | None = None) -> FlowResult:
"""Show the form to the user."""
return self.async_show_form(
step_id="user",
data_schema=self.config_schema,
errors=errors if errors else {},
)
async def _async_verify(
self, data: OpenUvData, error_step_id: str, error_schema: vol.Schema
) -> FlowResult:
"""Verify the credentials and create/re-auth the entry."""
websession = aiohttp_client.async_get_clientsession(self.hass)
client = Client(data.api_key, 0, 0, session=websession)
client.disable_request_retries()
try:
await client.uv_index()
except OpenUvError:
return self.async_show_form(
step_id=error_step_id,
data_schema=error_schema,
errors={CONF_API_KEY: "invalid_api_key"},
description_placeholders={
CONF_LATITUDE: str(data.latitude),
CONF_LONGITUDE: str(data.longitude),
},
)
entry_data = {
CONF_API_KEY: data.api_key,
CONF_LATITUDE: data.latitude,
CONF_LONGITUDE: data.longitude,
CONF_ELEVATION: data.elevation,
}
if existing_entry := await self.async_set_unique_id(data.unique_id):
self.hass.config_entries.async_update_entry(existing_entry, data=entry_data)
self.hass.async_create_task(
self.hass.config_entries.async_reload(existing_entry.entry_id)
)
return self.async_abort(reason="reauth_successful")
return self.async_create_entry(title=data.unique_id, data=entry_data)
@staticmethod
@callback
@ -65,26 +120,54 @@ class OpenUvFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Define the config flow to handle options."""
return OpenUvOptionsFlowHandler(config_entry)
async def async_step_reauth(self, entry_data: Mapping[str, Any]) -> FlowResult:
"""Handle configuration by re-auth."""
self._reauth_data = entry_data
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle re-auth completion."""
if not user_input:
return self.async_show_form(
step_id="reauth_confirm",
data_schema=STEP_REAUTH_SCHEMA,
description_placeholders={
CONF_LATITUDE: self._reauth_data[CONF_LATITUDE],
CONF_LONGITUDE: self._reauth_data[CONF_LONGITUDE],
},
)
data = OpenUvData(
user_input[CONF_API_KEY],
self._reauth_data[CONF_LATITUDE],
self._reauth_data[CONF_LONGITUDE],
self._reauth_data[CONF_ELEVATION],
)
return await self._async_verify(data, "reauth_confirm", STEP_REAUTH_SCHEMA)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the start of the config flow."""
if not user_input:
return await self._show_form()
return self.async_show_form(
step_id="user", data_schema=self.step_user_schema
)
identifier = f"{user_input[CONF_LATITUDE]}, {user_input[CONF_LONGITUDE]}"
await self.async_set_unique_id(identifier)
data = OpenUvData(
user_input[CONF_API_KEY],
user_input[CONF_LATITUDE],
user_input[CONF_LONGITUDE],
user_input[CONF_ELEVATION],
)
await self.async_set_unique_id(data.unique_id)
self._abort_if_unique_id_configured()
websession = aiohttp_client.async_get_clientsession(self.hass)
client = Client(user_input[CONF_API_KEY], 0, 0, session=websession)
try:
await client.uv_index()
except OpenUvError:
return await self._show_form({CONF_API_KEY: "invalid_api_key"})
return self.async_create_entry(title=identifier, data=user_input)
return await self._async_verify(data, "user", self.step_user_schema)
class OpenUvOptionsFlowHandler(config_entries.OptionsFlow):

83
homeassistant/components/openuv/coordinator.py
View File

@ -1,23 +1,94 @@
"""Define an update coordinator for OpenUV."""
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from typing import Any, cast
from pyopenuv.errors import OpenUvError
from pyopenuv.errors import InvalidApiKeyError, OpenUvError
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER
from .const import DOMAIN, LOGGER
DEFAULT_DEBOUNCER_COOLDOWN_SECONDS = 15 * 60
class InvalidApiKeyMonitor:
"""Define a monitor for failed API calls (due to bad keys) across coordinators."""
DEFAULT_FAILED_API_CALL_THRESHOLD = 5
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize."""
self._count = 1
self._lock = asyncio.Lock()
self._reauth_flow_manager = ReauthFlowManager(hass, entry)
self.entry = entry
async def async_increment(self) -> None:
"""Increment the counter."""
LOGGER.debug("Invalid API key response detected (number %s)", self._count)
async with self._lock:
self._count += 1
if self._count > self.DEFAULT_FAILED_API_CALL_THRESHOLD:
self._reauth_flow_manager.start_reauth()
async def async_reset(self) -> None:
"""Reset the counter."""
async with self._lock:
self._count = 0
self._reauth_flow_manager.cancel_reauth()
class ReauthFlowManager:
"""Define an OpenUV reauth flow manager."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Initialize."""
self.entry = entry
self.hass = hass
@callback
def _get_active_reauth_flow(self) -> FlowResult | None:
"""Get an active reauth flow (if it exists)."""
try:
[reauth_flow] = [
flow
for flow in self.hass.config_entries.flow.async_progress_by_handler(
DOMAIN
)
if flow["context"]["source"] == "reauth"
and flow["context"]["entry_id"] == self.entry.entry_id
]
except ValueError:
return None
return reauth_flow
@callback
def cancel_reauth(self) -> None:
"""Cancel a reauth flow (if appropriate)."""
if reauth_flow := self._get_active_reauth_flow():
LOGGER.debug("API seems to have recovered; canceling reauth flow")
self.hass.config_entries.flow.async_abort(reauth_flow["flow_id"])
@callback
def start_reauth(self) -> None:
"""Start a reauth flow (if appropriate)."""
if not self._get_active_reauth_flow():
LOGGER.debug("Multiple API failures in a row; starting reauth flow")
self.entry.async_start_reauth(self.hass)
class OpenUvCoordinator(DataUpdateCoordinator):
"""Define an OpenUV data coordinator."""
config_entry: ConfigEntry
update_method: Callable[[], Awaitable[dict[str, Any]]]
def __init__(
@ -28,6 +99,7 @@ class OpenUvCoordinator(DataUpdateCoordinator):
latitude: str,
longitude: str,
update_method: Callable[[], Awaitable[dict[str, Any]]],
invalid_api_key_monitor: InvalidApiKeyMonitor,
) -> None:
"""Initialize."""
super().__init__(
@ -43,6 +115,7 @@ class OpenUvCoordinator(DataUpdateCoordinator):
),
)
self._invalid_api_key_monitor = invalid_api_key_monitor
self.latitude = latitude
self.longitude = longitude
@ -50,6 +123,10 @@ class OpenUvCoordinator(DataUpdateCoordinator):
"""Fetch data from OpenUV."""
try:
data = await self.update_method()
except InvalidApiKeyError:
await self._invalid_api_key_monitor.async_increment()
except OpenUvError as err:
raise UpdateFailed(f"Error during protection data update: {err}") from err
await self._invalid_api_key_monitor.async_reset()
return cast(dict[str, Any], data["result"])

10
homeassistant/components/openuv/strings.json
View File

@ -1,6 +1,13 @@
{
"config": {
"step": {
"reauth_confirm": {
"title": "[%key:common::config_flow::title::reauth%]",
"description": "Please re-enter the API key for {latitude}, {longitude}.",
"data": {
"api_key": "[%key:common::config_flow::data::api_key%]"
}
},
"user": {
"title": "Fill in your information",
"data": {
@ -15,7 +22,8 @@
"invalid_api_key": "[%key:common::config_flow::error::invalid_api_key%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"options": {

10
homeassistant/components/openuv/translations/en.json
View File

@ -1,12 +1,20 @@
{
"config": {
"abort": {
"already_configured": "Location is already configured"
"already_configured": "Location is already configured",
"reauth_successful": "Re-authentication was successful"
},
"error": {
"invalid_api_key": "Invalid API key"
},
"step": {
"reauth_confirm": {
"data": {
"api_key": "API Key"
},
"description": "Please re-enter the API key for {latitude}, {longitude}.",
"title": "Reauthenticate Integration"
},
"user": {
"data": {
"api_key": "API Key",

24
tests/components/openuv/test_config_flow.py
View File

@ -5,7 +5,7 @@ from pyopenuv.errors import InvalidApiKeyError
from homeassistant import data_entry_flow
from homeassistant.components.openuv import CONF_FROM_WINDOW, CONF_TO_WINDOW, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.config_entries import SOURCE_REAUTH, SOURCE_USER
from homeassistant.const import (
CONF_API_KEY,
CONF_ELEVATION,
@ -51,6 +51,28 @@ async def test_options_flow(hass, config_entry):
assert config_entry.options == {CONF_FROM_WINDOW: 3.5, CONF_TO_WINDOW: 2.0}
async def test_step_reauth(hass, config, config_entry, setup_openuv):
"""Test that the reauth step works."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_REAUTH}, data=config
)
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
with patch("homeassistant.components.openuv.async_setup_entry", return_value=True):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={CONF_API_KEY: "new_api_key"}
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert len(hass.config_entries.async_entries()) == 1
async def test_step_user(hass, config, setup_openuv):
"""Test that the user step works."""
result = await hass.config_entries.flow.async_init(