Use SchemaOptionsFlowHandler in blink (#82903)

* Use SchemaOptionsFlowHandler in blink

* Use config entry update listener

* Adjust tests

* Simplify options init

* Don't reload

* Use selector
This commit is contained in:
epenet 2022-11-30 14:59:26 +01:00 committed by GitHub
parent 89a7d5fecc
commit 5be961f07f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 51 additions and 51 deletions

View file

@ -30,7 +30,7 @@ SERVICE_SAVE_VIDEO_SCHEMA = vol.Schema(
SERVICE_SEND_PIN_SCHEMA = vol.Schema({vol.Optional(CONF_PIN): cv.string})
def _blink_startup_wrapper(hass, entry):
def _blink_startup_wrapper(hass: HomeAssistant, entry: ConfigEntry) -> Blink:
"""Startup wrapper for blink."""
blink = Blink()
auth_data = deepcopy(dict(entry.data))
@ -87,6 +87,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
raise ConfigEntryNotReady
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(update_listener))
def blink_refresh(event_time=None):
"""Call blink to refresh info."""
@ -116,7 +117,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@callback
def _async_import_options_from_data_if_missing(hass, entry):
def _async_import_options_from_data_if_missing(
hass: HomeAssistant, entry: ConfigEntry
) -> None:
options = dict(entry.options)
if CONF_SCAN_INTERVAL not in entry.options:
options[CONF_SCAN_INTERVAL] = entry.data.get(
@ -144,6 +147,12 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Handle options update."""
blink: Blink = hass.data[DOMAIN][entry.entry_id]
blink.refresh_rate = entry.options[CONF_SCAN_INTERVAL]
async def async_handle_save_video_service(hass, entry, call):
"""Handle save video service calls."""
camera_name = call.data[CONF_NAME]

View file

@ -18,11 +18,35 @@ from homeassistant.const import (
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import selector
from homeassistant.helpers.schema_config_entry_flow import (
SchemaFlowFormStep,
SchemaOptionsFlowHandler,
)
from .const import DEFAULT_SCAN_INTERVAL, DEVICE_ID, DOMAIN
_LOGGER = logging.getLogger(__name__)
SIMPLE_OPTIONS_SCHEMA = vol.Schema(
{
vol.Optional(
CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
): selector.NumberSelector(
selector.NumberSelectorConfig(
mode=selector.NumberSelectorMode.BOX,
unit_of_measurement="seconds",
),
),
}
)
OPTIONS_FLOW = {
"init": SchemaFlowFormStep(next_step="simple_options"),
"simple_options": SchemaFlowFormStep(SIMPLE_OPTIONS_SCHEMA),
}
def validate_input(hass: core.HomeAssistant, auth):
"""Validate the user input allows us to connect."""
@ -56,9 +80,9 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@callback
def async_get_options_flow(
config_entry: config_entries.ConfigEntry,
) -> BlinkOptionsFlowHandler:
) -> SchemaOptionsFlowHandler:
"""Get options flow for this handler."""
return BlinkOptionsFlowHandler(config_entry)
return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
@ -133,45 +157,6 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=DOMAIN, data=self.auth.login_attributes)
class BlinkOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle Blink options."""
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize Blink options flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
self.blink = None
async def async_step_init(self, user_input=None):
"""Manage the Blink options."""
self.blink = self.hass.data[DOMAIN][self.config_entry.entry_id]
self.options[CONF_SCAN_INTERVAL] = self.blink.refresh_rate
return await self.async_step_simple_options()
async def async_step_simple_options(self, user_input=None):
"""For simple options."""
if user_input is not None:
self.options.update(user_input)
self.blink.refresh_rate = user_input[CONF_SCAN_INTERVAL]
return self.async_create_entry(title="", data=self.options)
options = self.config_entry.options
scan_interval = options.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
return self.async_show_form(
step_id="simple_options",
data_schema=vol.Schema(
{
vol.Optional(
CONF_SCAN_INTERVAL,
default=scan_interval,
): int
}
),
)
class Require2FA(exceptions.HomeAssistantError):
"""Error to indicate we require 2FA."""

View file

@ -6,6 +6,7 @@ from blinkpy.blinkpy import BlinkSetupError
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.blink import DOMAIN
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
@ -254,7 +255,7 @@ async def test_reauth_shows_user_step(hass):
assert result["step_id"] == "user"
async def test_options_flow(hass):
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test config flow options."""
config_entry = MockConfigEntry(
domain=DOMAIN,
@ -283,11 +284,16 @@ async def test_options_flow(hass):
assert result["type"] == data_entry_flow.FlowResultType.FORM
assert result["step_id"] == "simple_options"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"scan_interval": 5},
)
with patch("homeassistant.components.blink.Auth", return_value=mock_auth), patch(
"homeassistant.components.blink.Blink", return_value=mock_blink
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"scan_interval": 5},
)
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"] == {"scan_interval": 5}
assert mock_blink.refresh_rate == 5
assert result["type"] == data_entry_flow.FlowResultType.CREATE_ENTRY
assert result["data"] == {"scan_interval": 5}
await hass.async_block_till_done()
assert mock_blink.refresh_rate == 5