From f78b39bdbfbe151e8bab72610b6fe03afc8c0747 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Tue, 30 Aug 2022 12:40:16 -0400 Subject: [PATCH] ZHA backup/restore config flow (#77044) --- homeassistant/components/zha/config_flow.py | 706 +++++++++-- homeassistant/components/zha/core/const.py | 12 +- homeassistant/components/zha/manifest.json | 1 + homeassistant/components/zha/strings.json | 122 +- .../components/zha/translations/en.json | 124 +- tests/components/zha/test_config_flow.py | 1085 ++++++++++++++--- 6 files changed, 1736 insertions(+), 314 deletions(-) diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index 9c7ec46a386..5684e784a6a 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -1,21 +1,39 @@ """Config flow for ZHA.""" from __future__ import annotations +import collections +import contextlib +import copy +import json +import logging +import os from typing import Any import serial.tools.list_ports import voluptuous as vol +from zigpy.application import ControllerApplication +import zigpy.backups from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH +from zigpy.exceptions import NetworkNotFormed from homeassistant import config_entries from homeassistant.components import onboarding, usb, zeroconf +from homeassistant.components.file_upload import process_uploaded_file from homeassistant.const import CONF_NAME -from homeassistant.data_entry_flow import FlowResult +from homeassistant.core import callback +from homeassistant.data_entry_flow import FlowHandler, FlowResult +from homeassistant.helpers.selector import FileSelector, FileSelectorConfig +from homeassistant.util import dt from .core.const import ( CONF_BAUDRATE, + CONF_DATABASE, CONF_FLOWCONTROL, CONF_RADIO_TYPE, + CONF_ZIGPY, + DATA_ZHA, + DATA_ZHA_CONFIG, + DEFAULT_DATABASE_NAME, DOMAIN, RadioType, ) @@ -27,24 +45,184 @@ SUPPORTED_PORT_SETTINGS = ( ) DECONZ_DOMAIN = "deconz" +# Only the common radio types will be autoprobed, ordered by new device popularity. +# XBee takes too long to probe since it scans through all possible bauds and likely has +# very few users to begin with. +AUTOPROBE_RADIOS = ( + RadioType.ezsp, + RadioType.znp, + RadioType.deconz, + RadioType.zigate, +) -class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): - """Handle a config flow.""" +FORMATION_STRATEGY = "formation_strategy" +FORMATION_FORM_NEW_NETWORK = "form_new_network" +FORMATION_REUSE_SETTINGS = "reuse_settings" +FORMATION_CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup" +FORMATION_UPLOAD_MANUAL_BACKUP = "upload_manual_backup" - VERSION = 3 +CHOOSE_AUTOMATIC_BACKUP = "choose_automatic_backup" +OVERWRITE_COORDINATOR_IEEE = "overwrite_coordinator_ieee" - def __init__(self): +UPLOADED_BACKUP_FILE = "uploaded_backup_file" + +_LOGGER = logging.getLogger(__name__) + + +def _format_backup_choice( + backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True +) -> str: + """Format network backup info into a short piece of text.""" + if not pan_ids: + return dt.as_local(backup.backup_time).strftime("%c") + + identifier = ( + # PAN ID + f"{str(backup.network_info.pan_id)[2:]}" + # EPID + f":{str(backup.network_info.extended_pan_id).replace(':', '')}" + ).lower() + + return f"{dt.as_local(backup.backup_time).strftime('%c')} ({identifier})" + + +def _allow_overwrite_ezsp_ieee( + backup: zigpy.backups.NetworkBackup, +) -> zigpy.backups.NetworkBackup: + """Return a new backup with the flag to allow overwriting the EZSP EUI64.""" + new_stack_specific = copy.deepcopy(backup.network_info.stack_specific) + new_stack_specific.setdefault("ezsp", {})[ + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it" + ] = True + + return backup.replace( + network_info=backup.network_info.replace(stack_specific=new_stack_specific) + ) + + +def _prevent_overwrite_ezsp_ieee( + backup: zigpy.backups.NetworkBackup, +) -> zigpy.backups.NetworkBackup: + """Return a new backup without the flag to allow overwriting the EZSP EUI64.""" + if "ezsp" not in backup.network_info.stack_specific: + return backup + + new_stack_specific = copy.deepcopy(backup.network_info.stack_specific) + new_stack_specific.setdefault("ezsp", {}).pop( + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it", None + ) + + return backup.replace( + network_info=backup.network_info.replace(stack_specific=new_stack_specific) + ) + + +class BaseZhaFlow(FlowHandler): + """Mixin for common ZHA flow steps and forms.""" + + def __init__(self) -> None: """Initialize flow instance.""" - self._device_path = None - self._device_settings = None - self._radio_type = None - self._title = None + super().__init__() - async def async_step_user(self, user_input=None): - """Handle a zha config flow start.""" - if self._async_current_entries(): - return self.async_abort(reason="single_instance_allowed") + self._device_path: str | None = None + self._device_settings: dict[str, Any] | None = None + self._radio_type: RadioType | None = None + self._title: str | None = None + self._current_settings: zigpy.backups.NetworkBackup | None = None + self._backups: list[zigpy.backups.NetworkBackup] = [] + self._chosen_backup: zigpy.backups.NetworkBackup | None = None + @contextlib.asynccontextmanager + async def _connect_zigpy_app(self) -> ControllerApplication: + """Connect to the radio with the current config and then clean up.""" + assert self._radio_type is not None + + config = self.hass.data.get(DATA_ZHA, {}).get(DATA_ZHA_CONFIG, {}) + app_config = config.get(CONF_ZIGPY, {}).copy() + + database_path = config.get( + CONF_DATABASE, + self.hass.config.path(DEFAULT_DATABASE_NAME), + ) + + # Don't create `zigbee.db` if it doesn't already exist + if not await self.hass.async_add_executor_job(os.path.exists, database_path): + database_path = None + + app_config[CONF_DATABASE] = database_path + app_config[CONF_DEVICE] = self._device_settings + app_config = self._radio_type.controller.SCHEMA(app_config) + + app = await self._radio_type.controller.new( + app_config, auto_form=False, start_radio=False + ) + + try: + await app.connect() + yield app + finally: + await app.disconnect() + + async def _restore_backup( + self, backup: zigpy.backups.NetworkBackup, **kwargs: Any + ) -> None: + """Restore the provided network backup, passing through kwargs.""" + if self._current_settings is not None and self._current_settings.supersedes( + self._chosen_backup + ): + return + + async with self._connect_zigpy_app() as app: + await app.backups.restore_backup(backup, **kwargs) + + async def _detect_radio_type(self) -> bool: + """Probe all radio types on the current port.""" + for radio in AUTOPROBE_RADIOS: + _LOGGER.debug("Attempting to probe radio type %s", radio) + + dev_config = radio.controller.SCHEMA_DEVICE( + {CONF_DEVICE_PATH: self._device_path} + ) + probe_result = await radio.controller.probe(dev_config) + + if not probe_result: + continue + + # Radio library probing can succeed and return new device settings + if isinstance(probe_result, dict): + dev_config = probe_result + + self._radio_type = radio + self._device_settings = dev_config + + return True + + return False + + async def _async_create_radio_entity(self) -> FlowResult: + """Create a config entity with the current flow state.""" + assert self._title is not None + assert self._radio_type is not None + assert self._device_path is not None + assert self._device_settings is not None + + device_settings = self._device_settings.copy() + device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job( + usb.get_serial_by_id, self._device_path + ) + + return self.async_create_entry( + title=self._title, + data={ + CONF_DEVICE: device_settings, + CONF_RADIO_TYPE: self._radio_type.name, + }, + ) + + async def async_step_choose_serial_port( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Choose a serial port.""" ports = await self.hass.async_add_executor_job(serial.tools.list_ports.comports) list_of_ports = [ f"{p}, s/n: {p.serial_number or 'n/a'}" @@ -53,48 +231,329 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): ] if not list_of_ports: - return await self.async_step_pick_radio() + return await self.async_step_manual_pick_radio_type() list_of_ports.append(CONF_MANUAL_PATH) if user_input is not None: user_selection = user_input[CONF_DEVICE_PATH] + if user_selection == CONF_MANUAL_PATH: - return await self.async_step_pick_radio() + return await self.async_step_manual_pick_radio_type() port = ports[list_of_ports.index(user_selection)] - dev_path = await self.hass.async_add_executor_job( - usb.get_serial_by_id, port.device + self._device_path = port.device + + if not await self._detect_radio_type(): + # Did not autodetect anything, proceed to manual selection + return await self.async_step_manual_pick_radio_type() + + self._title = ( + f"{port.description}, s/n: {port.serial_number or 'n/a'}" + f" - {port.manufacturer}" + if port.manufacturer + else "" ) - auto_detected_data = await detect_radios(dev_path) - if auto_detected_data is not None: - title = f"{port.description}, s/n: {port.serial_number or 'n/a'}" - title += f" - {port.manufacturer}" if port.manufacturer else "" - return self.async_create_entry( - title=title, - data=auto_detected_data, + + return await self.async_step_choose_formation_strategy() + + # Pre-select the currently configured port + default_port = vol.UNDEFINED + + if self._device_path is not None: + for description, port in zip(list_of_ports, ports): + if port.device == self._device_path: + default_port = description + break + else: + default_port = CONF_MANUAL_PATH + + schema = vol.Schema( + { + vol.Required(CONF_DEVICE_PATH, default=default_port): vol.In( + list_of_ports ) + } + ) + return self.async_show_form(step_id="choose_serial_port", data_schema=schema) - # did not detect anything - self._device_path = dev_path - return await self.async_step_pick_radio() - - schema = vol.Schema({vol.Required(CONF_DEVICE_PATH): vol.In(list_of_ports)}) - return self.async_show_form(step_id="user", data_schema=schema) - - async def async_step_pick_radio(self, user_input=None): - """Select radio type.""" - + async def async_step_manual_pick_radio_type( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Manually select the radio type.""" if user_input is not None: self._radio_type = RadioType.get_by_description(user_input[CONF_RADIO_TYPE]) - return await self.async_step_port_config() + return await self.async_step_manual_port_config() + + # Pre-select the current radio type + default = vol.UNDEFINED + + if self._radio_type is not None: + default = self._radio_type.description + + schema = { + vol.Required(CONF_RADIO_TYPE, default=default): vol.In(RadioType.list()) + } - schema = {vol.Required(CONF_RADIO_TYPE): vol.In(sorted(RadioType.list()))} return self.async_show_form( - step_id="pick_radio", + step_id="manual_pick_radio_type", data_schema=vol.Schema(schema), ) + async def async_step_manual_port_config( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Enter port settings specific for this type of radio.""" + assert self._radio_type is not None + errors = {} + + if user_input is not None: + self._title = user_input[CONF_DEVICE_PATH] + self._device_path = user_input[CONF_DEVICE_PATH] + self._device_settings = user_input.copy() + + if await self._radio_type.controller.probe(user_input): + return await self.async_step_choose_formation_strategy() + + errors["base"] = "cannot_connect" + + schema = { + vol.Required( + CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED + ): str + } + + source = self.context.get("source") + for param, value in self._radio_type.controller.SCHEMA_DEVICE.schema.items(): + if param not in SUPPORTED_PORT_SETTINGS: + continue + + if source == config_entries.SOURCE_ZEROCONF and param == CONF_BAUDRATE: + value = 115200 + param = vol.Required(CONF_BAUDRATE, default=value) + elif self._device_settings is not None and param in self._device_settings: + param = vol.Required(str(param), default=self._device_settings[param]) + + schema[param] = value + + return self.async_show_form( + step_id="manual_port_config", + data_schema=vol.Schema(schema), + errors=errors, + ) + + async def _async_load_network_settings(self) -> None: + """Connect to the radio and load its current network settings.""" + async with self._connect_zigpy_app() as app: + # Check if the stick has any settings and load them + try: + await app.load_network_info() + except NetworkNotFormed: + pass + else: + self._current_settings = zigpy.backups.NetworkBackup( + network_info=app.state.network_info, + node_info=app.state.node_info, + ) + + # The list of backups will always exist + self._backups = app.backups.backups.copy() + + async def async_step_choose_formation_strategy( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Choose how to deal with the current radio's settings.""" + await self._async_load_network_settings() + + strategies = [] + + # Check if we have any automatic backups *and* if the backups differ from + # the current radio settings, if they exist (since restoring would be redundant) + if self._backups and ( + self._current_settings is None + or any( + not backup.is_compatible_with(self._current_settings) + for backup in self._backups + ) + ): + strategies.append(CHOOSE_AUTOMATIC_BACKUP) + + if self._current_settings is not None: + strategies.append(FORMATION_REUSE_SETTINGS) + + strategies.append(FORMATION_UPLOAD_MANUAL_BACKUP) + strategies.append(FORMATION_FORM_NEW_NETWORK) + + return self.async_show_menu( + step_id="choose_formation_strategy", + menu_options=strategies, + ) + + async def async_step_reuse_settings( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Reuse the existing network settings on the stick.""" + return await self._async_create_radio_entity() + + async def async_step_form_new_network( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Form a brand new network.""" + async with self._connect_zigpy_app() as app: + await app.form_network() + + return await self._async_create_radio_entity() + + def _parse_uploaded_backup( + self, uploaded_file_id: str + ) -> zigpy.backups.NetworkBackup: + """Read and parse an uploaded backup JSON file.""" + with process_uploaded_file(self.hass, uploaded_file_id) as file_path: + contents = file_path.read_text() + + return zigpy.backups.NetworkBackup.from_dict(json.loads(contents)) + + async def async_step_upload_manual_backup( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Upload and restore a coordinator backup JSON file.""" + errors = {} + + if user_input is not None: + try: + self._chosen_backup = await self.hass.async_add_executor_job( + self._parse_uploaded_backup, user_input[UPLOADED_BACKUP_FILE] + ) + except ValueError: + errors["base"] = "invalid_backup_json" + else: + return await self.async_step_maybe_confirm_ezsp_restore() + + return self.async_show_form( + step_id="upload_manual_backup", + data_schema=vol.Schema( + { + vol.Required(UPLOADED_BACKUP_FILE): FileSelector( + FileSelectorConfig(accept=".json,application/json") + ) + } + ), + errors=errors, + ) + + async def async_step_choose_automatic_backup( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Choose an automatic backup.""" + if self.show_advanced_options: + # Always show the PAN IDs when in advanced mode + choices = [ + _format_backup_choice(backup, pan_ids=True) for backup in self._backups + ] + else: + # Only show the PAN IDs for multiple backups taken on the same day + num_backups_on_date = collections.Counter( + backup.backup_time.date() for backup in self._backups + ) + choices = [ + _format_backup_choice( + backup, pan_ids=(num_backups_on_date[backup.backup_time.date()] > 1) + ) + for backup in self._backups + ] + + if user_input is not None: + index = choices.index(user_input[CHOOSE_AUTOMATIC_BACKUP]) + self._chosen_backup = self._backups[index] + + return await self.async_step_maybe_confirm_ezsp_restore() + + return self.async_show_form( + step_id="choose_automatic_backup", + data_schema=vol.Schema( + { + vol.Required(CHOOSE_AUTOMATIC_BACKUP, default=choices[0]): vol.In( + choices + ), + } + ), + ) + + async def async_step_maybe_confirm_ezsp_restore( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Confirm restore for EZSP radios that require permanent IEEE writes.""" + assert self._chosen_backup is not None + + if self._radio_type != RadioType.ezsp: + await self._restore_backup(self._chosen_backup) + return await self._async_create_radio_entity() + + # We have no way to partially load network settings if no network is formed + if self._current_settings is None: + # Since we are going to be restoring the backup anyways, write it to the + # radio without overwriting the IEEE but don't take a backup with these + # temporary settings + temp_backup = _prevent_overwrite_ezsp_ieee(self._chosen_backup) + await self._restore_backup(temp_backup, create_new=False) + await self._async_load_network_settings() + + assert self._current_settings is not None + + if ( + self._current_settings.node_info.ieee == self._chosen_backup.node_info.ieee + or not self._current_settings.network_info.metadata["ezsp"][ + "can_write_custom_eui64" + ] + ): + # No point in prompting the user if the backup doesn't have a new IEEE + # address or if there is no way to overwrite the IEEE address a second time + await self._restore_backup(self._chosen_backup) + + return await self._async_create_radio_entity() + + if user_input is not None: + backup = self._chosen_backup + + if user_input[OVERWRITE_COORDINATOR_IEEE]: + backup = _allow_overwrite_ezsp_ieee(backup) + + # If the user declined to overwrite the IEEE *and* we wrote the backup to + # their empty radio above, restoring it again would be redundant. + await self._restore_backup(backup) + + return await self._async_create_radio_entity() + + return self.async_show_form( + step_id="maybe_confirm_ezsp_restore", + data_schema=vol.Schema( + {vol.Required(OVERWRITE_COORDINATOR_IEEE, default=True): bool} + ), + ) + + +class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow.""" + + VERSION = 3 + + @staticmethod + @callback + def async_get_options_flow( + config_entry: config_entries.ConfigEntry, + ) -> config_entries.OptionsFlow: + """Create the options flow.""" + return ZhaOptionsFlowHandler(config_entry) + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Handle a zha config flow start.""" + if self._async_current_entries(): + return self.async_abort(reason="single_instance_allowed") + + return await self.async_step_choose_serial_port(user_input) + async def async_step_usb(self, discovery_info: usb.UsbServiceInfo) -> FlowResult: """Handle usb discovery.""" vid = discovery_info.vid @@ -118,9 +577,8 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") - # If they already have a discovery for deconz - # we ignore the usb discovery as they probably - # want to use it there instead + # If they already have a discovery for deconz we ignore the usb discovery as + # they probably want to use it there instead if self.hass.config_entries.flow.async_progress_by_handler(DECONZ_DOMAIN): return self.async_abort(reason="not_zha_device") for entry in self.hass.config_entries.async_entries(DECONZ_DOMAIN): @@ -140,19 +598,18 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): self.context["title_placeholders"] = {CONF_NAME: self._title} return await self.async_step_confirm() - async def async_step_confirm(self, user_input=None): - """Confirm a USB discovery.""" + async def async_step_confirm( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Confirm a discovery.""" if user_input is not None or not onboarding.async_is_onboarded(self.hass): - auto_detected_data = await detect_radios(self._device_path) - if auto_detected_data is None: + if not await self._detect_radio_type(): # This path probably will not happen now that we have # more precise USB matching unless there is a problem # with the device return self.async_abort(reason="usb_probe_failed") - return self.async_create_entry( - title=self._title, - data=auto_detected_data, - ) + + return await self.async_step_choose_formation_strategy() return self.async_show_form( step_id="confirm", @@ -188,61 +645,22 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") - self.context["title_placeholders"] = { - CONF_NAME: node_name, - } - + self.context["title_placeholders"] = {CONF_NAME: node_name} + self._title = device_path self._device_path = device_path + if "efr32" in radio_type: - self._radio_type = RadioType.ezsp.name + self._radio_type = RadioType.ezsp elif "zigate" in radio_type: - self._radio_type = RadioType.zigate.name + self._radio_type = RadioType.zigate else: - self._radio_type = RadioType.znp.name + self._radio_type = RadioType.znp - return await self.async_step_port_config() + return await self.async_step_manual_port_config() - async def async_step_port_config(self, user_input=None): - """Enter port settings specific for this type of radio.""" - errors = {} - app_cls = RadioType[self._radio_type].controller - - if user_input is not None: - self._device_path = user_input.get(CONF_DEVICE_PATH) - if await app_cls.probe(user_input): - serial_by_id = await self.hass.async_add_executor_job( - usb.get_serial_by_id, user_input[CONF_DEVICE_PATH] - ) - user_input[CONF_DEVICE_PATH] = serial_by_id - return self.async_create_entry( - title=user_input[CONF_DEVICE_PATH], - data={CONF_DEVICE: user_input, CONF_RADIO_TYPE: self._radio_type}, - ) - errors["base"] = "cannot_connect" - - schema = { - vol.Required( - CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED - ): str - } - radio_schema = app_cls.SCHEMA_DEVICE.schema - if isinstance(radio_schema, vol.Schema): - radio_schema = radio_schema.schema - - source = self.context.get("source") - for param, value in radio_schema.items(): - if param in SUPPORTED_PORT_SETTINGS: - schema[param] = value - if source == config_entries.SOURCE_ZEROCONF and param == CONF_BAUDRATE: - schema[param] = 115200 - - return self.async_show_form( - step_id="port_config", - data_schema=vol.Schema(schema), - errors=errors, - ) - - async def async_step_hardware(self, data=None): + async def async_step_hardware( + self, data: dict[str, Any] | None = None + ) -> FlowResult: """Handle hardware flow.""" if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") @@ -250,40 +668,39 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): return self.async_abort(reason="invalid_hardware_data") if data.get("radio_type") != "efr32": return self.async_abort(reason="invalid_hardware_data") - self._radio_type = RadioType.ezsp.name - app_cls = RadioType[self._radio_type].controller + self._radio_type = RadioType.ezsp schema = { vol.Required( CONF_DEVICE_PATH, default=self._device_path or vol.UNDEFINED ): str } - radio_schema = app_cls.SCHEMA_DEVICE.schema + + radio_schema = self._radio_type.controller.SCHEMA_DEVICE.schema assert not isinstance(radio_schema, vol.Schema) for param, value in radio_schema.items(): if param in SUPPORTED_PORT_SETTINGS: schema[param] = value + try: - self._device_settings = vol.Schema(schema)(data.get("port")) + device_settings = vol.Schema(schema)(data.get("port")) except vol.Invalid: return self.async_abort(reason="invalid_hardware_data") self._title = data.get("name", data["port"]["path"]) + self._device_path = device_settings.pop(CONF_DEVICE_PATH) + self._device_settings = device_settings self._set_confirm_only() return await self.async_step_confirm_hardware() - async def async_step_confirm_hardware(self, user_input=None): + async def async_step_confirm_hardware( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Confirm a hardware discovery.""" if user_input is not None or not onboarding.async_is_onboarded(self.hass): - return self.async_create_entry( - title=self._title, - data={ - CONF_DEVICE: self._device_settings, - CONF_RADIO_TYPE: self._radio_type, - }, - ) + return await self._async_create_radio_entity() return self.async_show_form( step_id="confirm_hardware", @@ -291,14 +708,65 @@ class ZhaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): ) -async def detect_radios(dev_path: str) -> dict[str, Any] | None: - """Probe all radio types on the device port.""" - for radio in RadioType: - dev_config = radio.controller.SCHEMA_DEVICE({CONF_DEVICE_PATH: dev_path}) - probe_result = await radio.controller.probe(dev_config) - if probe_result: - if isinstance(probe_result, dict): - return {CONF_RADIO_TYPE: radio.name, CONF_DEVICE: probe_result} - return {CONF_RADIO_TYPE: radio.name, CONF_DEVICE: dev_config} +class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow): + """Handle an options flow.""" - return None + def __init__(self, config_entry: config_entries.ConfigEntry) -> None: + """Initialize options flow.""" + super().__init__() + self.config_entry = config_entry + + self._device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH] + self._device_settings = config_entry.data[CONF_DEVICE] + self._radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]] + self._title = config_entry.title + + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Launch the options flow.""" + if user_input is not None: + try: + await self.hass.config_entries.async_unload(self.config_entry.entry_id) + except config_entries.OperationNotAllowed: + # ZHA is not running + pass + + return await self.async_step_choose_serial_port() + + return self.async_show_form(step_id="init") + + async def _async_create_radio_entity(self): + """Re-implementation of the base flow's final step to update the config.""" + device_settings = self._device_settings.copy() + device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job( + usb.get_serial_by_id, self._device_path + ) + + # Avoid creating both `.options` and `.data` by directly writing `data` here + self.hass.config_entries.async_update_entry( + entry=self.config_entry, + data={ + CONF_DEVICE: device_settings, + CONF_RADIO_TYPE: self._radio_type.name, + }, + options=self.config_entry.options, + ) + + # Reload ZHA after we finish + await self.hass.config_entries.async_setup(self.config_entry.entry_id) + + # Intentionally do not set `data` to avoid creating `options`, we set it above + return self.async_create_entry(title=self._title, data={}) + + def async_remove(self): + """Maybe reload ZHA if the flow is aborted.""" + if self.config_entry.state not in ( + config_entries.ConfigEntryState.SETUP_ERROR, + config_entries.ConfigEntryState.NOT_LOADED, + ): + return + + self.hass.async_create_task( + self.hass.config_entries.async_setup(self.config_entry.entry_id) + ) diff --git a/homeassistant/components/zha/core/const.py b/homeassistant/components/zha/core/const.py index 4a48c254b40..fa8b7148c77 100644 --- a/homeassistant/components/zha/core/const.py +++ b/homeassistant/components/zha/core/const.py @@ -236,14 +236,14 @@ _ControllerClsType = type[zigpy.application.ControllerApplication] class RadioType(enum.Enum): """Possible options for radio type.""" - znp = ( - "ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2", - zigpy_znp.zigbee.application.ControllerApplication, - ) ezsp = ( "EZSP = Silicon Labs EmberZNet protocol: Elelabs, HUSBZB-1, Telegesis", bellows.zigbee.application.ControllerApplication, ) + znp = ( + "ZNP = Texas Instruments Z-Stack ZNP protocol: CC253x, CC26x2, CC13x2", + zigpy_znp.zigbee.application.ControllerApplication, + ) deconz = ( "deCONZ = dresden elektronik deCONZ protocol: ConBee I/II, RaspBee I/II", zigpy_deconz.zigbee.application.ControllerApplication, @@ -263,11 +263,11 @@ class RadioType(enum.Enum): return [e.description for e in RadioType] @classmethod - def get_by_description(cls, description: str) -> str: + def get_by_description(cls, description: str) -> RadioType: """Get radio by description.""" for radio in cls: if radio.description == description: - return radio.name + return radio raise ValueError def __init__(self, description: str, controller_cls: _ControllerClsType) -> None: diff --git a/homeassistant/components/zha/manifest.json b/homeassistant/components/zha/manifest.json index 607fb351838..2e35427a70c 100644 --- a/homeassistant/components/zha/manifest.json +++ b/homeassistant/components/zha/manifest.json @@ -93,6 +93,7 @@ "name": "*zigate*" } ], + "dependencies": ["file_upload"], "after_dependencies": ["onboarding", "usb", "zeroconf"], "iot_class": "local_polling", "loggers": [ diff --git a/homeassistant/components/zha/strings.json b/homeassistant/components/zha/strings.json index 37be80e9b56..1de5e164fee 100644 --- a/homeassistant/components/zha/strings.json +++ b/homeassistant/components/zha/strings.json @@ -2,10 +2,10 @@ "config": { "flow_title": "{name}", "step": { - "user": { - "title": "ZHA", + "choose_serial_port": { + "title": "Select a Serial Port", "data": { "path": "Serial Device Path" }, - "description": "Select serial port for Zigbee radio" + "description": "Select the serial port for your Zigbee radio" }, "confirm": { "description": "Do you want to setup {name}?" @@ -13,23 +13,55 @@ "confirm_hardware": { "description": "Do you want to setup {name}?" }, - "pick_radio": { + "manual_pick_radio_type": { "data": { "radio_type": "Radio Type" }, "title": "Radio Type", - "description": "Pick a type of your Zigbee radio" + "description": "Pick your Zigbee radio type" }, - "port_config": { - "title": "Settings", - "description": "Enter port specific settings", + "manual_port_config": { + "title": "Serial Port Settings", + "description": "Enter the serial port settings", "data": { "path": "Serial device path", "baudrate": "port speed", "flow_control": "data flow control" } + }, + "choose_formation_strategy": { + "title": "Network Formation", + "description": "Choose the network settings for your radio.", + "menu_options": { + "form_new_network": "Erase network settings and form a new network", + "reuse_settings": "Keep radio network settings", + "choose_automatic_backup": "Restore an automatic backup", + "upload_manual_backup": "Upload a manual backup" + } + }, + "choose_automatic_backup": { + "title": "Restore Automatic Backup", + "description": "Restore your network settings from an automatic backup", + "data": { + "choose_automatic_backup": "Choose an automatic backup" + } + }, + "upload_manual_backup": { + "title": "Upload a Manual Backup", + "description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.", + "data": { + "uploaded_backup_file": "Upload a file" + } + }, + "maybe_confirm_ezsp_restore": { + "title": "Overwrite Radio IEEE Address", + "description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.", + "data": { + "overwrite_coordinator_ieee": "Permanently replace the radio IEEE address" + } } }, "error": { - "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_backup_json": "Invalid backup JSON" }, "abort": { "single_instance_allowed": "[%key:common::config_flow::abort::single_instance_allowed%]", @@ -37,6 +69,78 @@ "usb_probe_failed": "Failed to probe the usb device" } }, + "options": { + "flow_title": "[%key:component::zha::config::flow_title%]", + "step": { + "init": { + "title": "Reconfigure ZHA", + "description": "ZHA will be stopped. Do you wish to continue?" + }, + "choose_serial_port": { + "title": "[%key:component::zha::config::step::choose_serial_port::title%]", + "data": { + "path": "[%key:component::zha::config::step::choose_serial_port::data::path%]" + }, + "description": "[%key:component::zha::config::step::choose_serial_port::description%]" + }, + "manual_pick_radio_type": { + "data": { + "radio_type": "[%key:component::zha::config::step::manual_pick_radio_type::data::radio_type%]" + }, + "title": "[%key:component::zha::config::step::manual_pick_radio_type::title%]", + "description": "[%key:component::zha::config::step::manual_pick_radio_type::description%]" + }, + "manual_port_config": { + "title": "[%key:component::zha::config::step::manual_port_config::title%]", + "description": "[%key:component::zha::config::step::manual_port_config::description%]", + "data": { + "path": "[%key:component::zha::config::step::manual_port_config::data::path%]", + "baudrate": "[%key:component::zha::config::step::manual_port_config::data::baudrate%]", + "flow_control": "[%key:component::zha::config::step::manual_port_config::data::flow_control%]" + } + }, + "choose_formation_strategy": { + "title": "[%key:component::zha::config::step::choose_formation_strategy::title%]", + "description": "[%key:component::zha::config::step::choose_formation_strategy::description%]", + "menu_options": { + "form_new_network": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::form_new_network%]", + "reuse_settings": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::reuse_settings%]", + "choose_automatic_backup": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::choose_automatic_backup%]", + "upload_manual_backup": "[%key:component::zha::config::step::choose_formation_strategy::menu_options::upload_manual_backup%]" + } + }, + "choose_automatic_backup": { + "title": "[%key:component::zha::config::step::choose_automatic_backup::title%]", + "description": "[%key:component::zha::config::step::choose_automatic_backup::description%]", + "data": { + "choose_automatic_backup": "[%key:component::zha::config::step::choose_automatic_backup::data::choose_automatic_backup%]" + } + }, + "upload_manual_backup": { + "title": "[%key:component::zha::config::step::upload_manual_backup::title%]", + "description": "[%key:component::zha::config::step::upload_manual_backup::description%]", + "data": { + "uploaded_backup_file": "[%key:component::zha::config::step::upload_manual_backup::data::uploaded_backup_file%]" + } + }, + "maybe_confirm_ezsp_restore": { + "title": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::title%]", + "description": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::description%]", + "data": { + "overwrite_coordinator_ieee": "[%key:component::zha::config::step::maybe_confirm_ezsp_restore::data::overwrite_coordinator_ieee%]" + } + } + }, + "error": { + "cannot_connect": "[%key:component::zha::config::error::cannot_connect%]", + "invalid_backup_json": "[%key:component::zha::config::error::invalid_backup_json%]" + }, + "abort": { + "single_instance_allowed": "[%key:component::zha::config::abort::single_instance_allowed%]", + "not_zha_device": "[%key:component::zha::config::abort::not_zha_device%]", + "usb_probe_failed": "[%key:component::zha::config::abort::usb_probe_failed%]" + } + }, "config_panel": { "zha_options": { "title": "Global Options", diff --git a/homeassistant/components/zha/translations/en.json b/homeassistant/components/zha/translations/en.json index 46e897167b0..27a4db9ef02 100644 --- a/homeassistant/components/zha/translations/en.json +++ b/homeassistant/components/zha/translations/en.json @@ -6,38 +6,70 @@ "usb_probe_failed": "Failed to probe the usb device" }, "error": { - "cannot_connect": "Failed to connect" + "cannot_connect": "Failed to connect", + "invalid_backup_json": "Invalid backup JSON" }, "flow_title": "{name}", "step": { + "choose_automatic_backup": { + "data": { + "choose_automatic_backup": "Choose an automatic backup" + }, + "description": "Restore your network settings from an automatic backup", + "title": "Restore Automatic Backup" + }, + "choose_formation_strategy": { + "description": "Choose the network settings for your radio.", + "menu_options": { + "choose_automatic_backup": "Restore an automatic backup", + "form_new_network": "Erase network settings and form a new network", + "reuse_settings": "Keep radio network settings", + "upload_manual_backup": "Upload a manual backup" + }, + "title": "Network Formation" + }, + "choose_serial_port": { + "data": { + "path": "Serial Device Path" + }, + "description": "Select the serial port for your Zigbee radio", + "title": "Select a Serial Port" + }, "confirm": { "description": "Do you want to setup {name}?" }, "confirm_hardware": { "description": "Do you want to setup {name}?" }, - "pick_radio": { + "manual_pick_radio_type": { "data": { "radio_type": "Radio Type" }, - "description": "Pick a type of your Zigbee radio", + "description": "Pick your Zigbee radio type", "title": "Radio Type" }, - "port_config": { + "manual_port_config": { "data": { "baudrate": "port speed", "flow_control": "data flow control", "path": "Serial device path" }, - "description": "Enter port specific settings", - "title": "Settings" + "description": "Enter the serial port settings", + "title": "Serial Port Settings" }, - "user": { + "maybe_confirm_ezsp_restore": { "data": { - "path": "Serial Device Path" + "overwrite_coordinator_ieee": "Permanently replace the radio IEEE address" }, - "description": "Select serial port for Zigbee radio", - "title": "ZHA" + "description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.", + "title": "Overwrite Radio IEEE Address" + }, + "upload_manual_backup": { + "data": { + "uploaded_backup_file": "Upload a file" + }, + "description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.", + "title": "Upload a Manual Backup" } } }, @@ -114,5 +146,77 @@ "remote_button_short_release": "\"{subtype}\" button released", "remote_button_triple_press": "\"{subtype}\" button triple clicked" } + }, + "options": { + "abort": { + "not_zha_device": "This device is not a zha device", + "single_instance_allowed": "Already configured. Only a single configuration possible.", + "usb_probe_failed": "Failed to probe the usb device" + }, + "error": { + "cannot_connect": "Failed to connect", + "invalid_backup_json": "Invalid backup JSON" + }, + "flow_title": "{name}", + "step": { + "choose_automatic_backup": { + "data": { + "choose_automatic_backup": "Choose an automatic backup" + }, + "description": "Restore your network settings from an automatic backup", + "title": "Restore Automatic Backup" + }, + "choose_formation_strategy": { + "description": "Choose the network settings for your radio.", + "menu_options": { + "choose_automatic_backup": "Restore an automatic backup", + "form_new_network": "Erase network settings and form a new network", + "reuse_settings": "Keep radio network settings", + "upload_manual_backup": "Upload a manual backup" + }, + "title": "Network Formation" + }, + "choose_serial_port": { + "data": { + "path": "Serial Device Path" + }, + "description": "Select the serial port for your Zigbee radio", + "title": "Select a Serial Port" + }, + "init": { + "description": "ZHA will be stopped. Do you wish to continue?", + "title": "Reconfigure ZHA" + }, + "manual_pick_radio_type": { + "data": { + "radio_type": "Radio Type" + }, + "description": "Pick your Zigbee radio type", + "title": "Radio Type" + }, + "manual_port_config": { + "data": { + "baudrate": "port speed", + "flow_control": "data flow control", + "path": "Serial device path" + }, + "description": "Enter the serial port settings", + "title": "Serial Port Settings" + }, + "maybe_confirm_ezsp_restore": { + "data": { + "overwrite_coordinator_ieee": "Permanently replace the radio IEEE address" + }, + "description": "Your backup has a different IEEE address than your radio. For your network to function properly, the IEEE address of your radio should also be changed.\n\nThis is a permanent operation.", + "title": "Overwrite Radio IEEE Address" + }, + "upload_manual_backup": { + "data": { + "uploaded_backup_file": "Upload a file" + }, + "description": "Restore your network settings from an uploaded backup JSON file. You can download one from a different ZHA installation from **Network Settings**, or use a Zigbee2MQTT `coordinator_backup.json` file.", + "title": "Upload a Manual Backup" + } + } } } \ No newline at end of file diff --git a/tests/components/zha/test_config_flow.py b/tests/components/zha/test_config_flow.py index a769303a4c4..12f5434abd4 100644 --- a/tests/components/zha/test_config_flow.py +++ b/tests/components/zha/test_config_flow.py @@ -1,11 +1,16 @@ """Tests for ZHA config flow.""" -from unittest.mock import AsyncMock, MagicMock, patch, sentinel +import copy +import json +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch +import uuid import pytest import serial.tools.list_ports import zigpy.config from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH +from zigpy.exceptions import NetworkNotFormed +import zigpy.types from homeassistant import config_entries from homeassistant.components import ssdp, usb, zeroconf @@ -29,6 +34,8 @@ from homeassistant.data_entry_flow import FlowResultType from tests.common import MockConfigEntry +PROBE_FUNCTION_PATH = "zigbee.application.ControllerApplication.probe" + @pytest.fixture(autouse=True) def disable_platform_only(): @@ -37,20 +44,55 @@ def disable_platform_only(): yield -def com_port(): +@pytest.fixture(autouse=True) +def mock_app(): + """Mock zigpy app interface.""" + mock_app = AsyncMock() + mock_app.backups.backups = [] + + with patch( + "zigpy.application.ControllerApplication.new", AsyncMock(return_value=mock_app) + ): + yield mock_app + + +@pytest.fixture +def backup(): + """Zigpy network backup with non-default settings.""" + backup = zigpy.backups.NetworkBackup() + backup.node_info.ieee = zigpy.types.EUI64.convert("AA:BB:CC:DD:11:22:33:44") + + return backup + + +def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True): + """Mock `_detect_radio_type` that just sets the appropriate attributes.""" + + async def detect(self): + self._radio_type = radio_type + self._device_settings = radio_type.controller.SCHEMA_DEVICE( + {CONF_DEVICE_PATH: self._device_path} + ) + + return ret + + return detect + + +def com_port(device="/dev/ttyUSB1234"): """Mock of a serial port.""" port = serial.tools.list_ports_common.ListPortInfo("/dev/ttyUSB1234") port.serial_number = "1234" port.manufacturer = "Virtual serial port" - port.device = "/dev/ttyUSB1234" + port.device = device port.description = "Some serial port" return port @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_zeroconf_discovery_znp(hass): """Test zeroconf flow -- radio detected.""" service_info = zeroconf.ZeroconfServiceInfo( host="192.168.1.200", @@ -62,15 +104,24 @@ async def test_discovery(detect_mock, hass): type="mock_type", ) flow = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) - result = await hass.config_entries.flow.async_configure( + result1 = await hass.config_entries.flow.async_configure( flow["flow_id"], user_input={} ) - assert result["type"] == FlowResultType.CREATE_ENTRY - assert result["title"] == "socket://192.168.1.200:6638" - assert result["data"] == { + assert result1["type"] == FlowResultType.MENU + assert result1["step_id"] == "choose_formation_strategy" + + result2 = await hass.config_entries.flow.async_configure( + result1["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "socket://192.168.1.200:6638" + assert result2["data"] == { CONF_DEVICE: { CONF_BAUDRATE: 115200, CONF_FLOWCONTROL: None, @@ -81,8 +132,8 @@ async def test_discovery(detect_mock, hass): @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("zigpy_zigate.zigbee.application.ControllerApplication.probe") -async def test_zigate_via_zeroconf(probe_mock, hass): +@patch(f"zigpy_zigate.{PROBE_FUNCTION_PATH}") +async def test_zigate_via_zeroconf(setup_entry_mock, hass): """Test zeroconf flow -- zigate radio detected.""" service_info = zeroconf.ZeroconfServiceInfo( host="192.168.1.200", @@ -94,15 +145,24 @@ async def test_zigate_via_zeroconf(probe_mock, hass): type="mock_type", ) flow = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) - result = await hass.config_entries.flow.async_configure( + result1 = await hass.config_entries.flow.async_configure( flow["flow_id"], user_input={} ) - assert result["type"] == FlowResultType.CREATE_ENTRY - assert result["title"] == "socket://192.168.1.200:1234" - assert result["data"] == { + assert result1["type"] == FlowResultType.MENU + assert result1["step_id"] == "choose_formation_strategy" + + result2 = await hass.config_entries.flow.async_configure( + result1["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "socket://192.168.1.200:1234" + assert result2["data"] == { CONF_DEVICE: { CONF_DEVICE_PATH: "socket://192.168.1.200:1234", }, @@ -111,8 +171,8 @@ async def test_zigate_via_zeroconf(probe_mock, hass): @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("bellows.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_efr32_via_zeroconf(probe_mock, hass): +@patch(f"bellows.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_efr32_via_zeroconf(hass): """Test zeroconf flow -- efr32 radio detected.""" service_info = zeroconf.ZeroconfServiceInfo( host="192.168.1.200", @@ -124,15 +184,24 @@ async def test_efr32_via_zeroconf(probe_mock, hass): type="mock_type", ) flow = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) - result = await hass.config_entries.flow.async_configure( - flow["flow_id"], user_input={"baudrate": 115200} + result1 = await hass.config_entries.flow.async_configure( + flow["flow_id"], user_input={} ) - assert result["type"] == FlowResultType.CREATE_ENTRY - assert result["title"] == "socket://192.168.1.200:6638" - assert result["data"] == { + assert result1["type"] == FlowResultType.MENU + assert result1["step_id"] == "choose_formation_strategy" + + result2 = await hass.config_entries.flow.async_configure( + result1["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"] == "socket://192.168.1.200:6638" + assert result2["data"] == { CONF_DEVICE: { CONF_DEVICE_PATH: "socket://192.168.1.200:6638", CONF_BAUDRATE: 115200, @@ -143,8 +212,8 @@ async def test_efr32_via_zeroconf(probe_mock, hass): @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_zeroconf_ip_change(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_zeroconf_ip_change(hass): """Test zeroconf flow -- radio detected.""" entry = MockConfigEntry( domain=DOMAIN, @@ -169,7 +238,7 @@ async def test_discovery_via_zeroconf_ip_change(detect_mock, hass): type="mock_type", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) assert result["type"] == FlowResultType.ABORT @@ -182,8 +251,8 @@ async def test_discovery_via_zeroconf_ip_change(detect_mock, hass): @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_zeroconf_ip_change_ignored(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_zeroconf_ip_change_ignored(hass): """Test zeroconf flow that was ignored gets updated.""" entry = MockConfigEntry( domain=DOMAIN, @@ -202,7 +271,7 @@ async def test_discovery_via_zeroconf_ip_change_ignored(detect_mock, hass): type="mock_type", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) assert result["type"] == FlowResultType.ABORT @@ -212,8 +281,8 @@ async def test_discovery_via_zeroconf_ip_change_ignored(detect_mock, hass): } -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb(hass): """Test usb flow -- radio detected.""" discovery_info = usb.UsbServiceInfo( device="/dev/ttyZIGBEE", @@ -224,21 +293,30 @@ async def test_discovery_via_usb(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() assert result["type"] == FlowResultType.FORM assert result["step_id"] == "confirm" - with patch("homeassistant.components.zha.async_setup_entry"): - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.MENU + assert result2["step_id"] == "choose_formation_strategy" + + with patch("homeassistant.components.zha.async_setup_entry", return_value=True): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, ) await hass.async_block_till_done() - assert result2["type"] == FlowResultType.CREATE_ENTRY - assert result2["title"] == "zigbee radio" - assert result2["data"] == { + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["title"] == "zigbee radio" + assert result3["data"] == { "device": { "baudrate": 115200, "flow_control": None, @@ -248,8 +326,8 @@ async def test_discovery_via_usb(detect_mock, hass): } -@patch("zigpy_zigate.zigbee.application.ControllerApplication.probe") -async def test_zigate_discovery_via_usb(detect_mock, hass): +@patch(f"zigpy_zigate.{PROBE_FUNCTION_PATH}", return_value=True) +async def test_zigate_discovery_via_usb(probe_mock, hass): """Test zigate usb flow -- radio detected.""" discovery_info = usb.UsbServiceInfo( device="/dev/ttyZIGBEE", @@ -260,21 +338,30 @@ async def test_zigate_discovery_via_usb(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() assert result["type"] == FlowResultType.FORM assert result["step_id"] == "confirm" - with patch("homeassistant.components.zha.async_setup_entry"): - result2 = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.MENU + assert result2["step_id"] == "choose_formation_strategy" + + with patch("homeassistant.components.zha.async_setup_entry", return_value=True): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, ) await hass.async_block_till_done() - assert result2["type"] == FlowResultType.CREATE_ENTRY - assert result2["title"] == "zigate radio" - assert result2["data"] == { + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["title"] == "zigate radio" + assert result3["data"] == { "device": { "path": "/dev/ttyZIGBEE", }, @@ -282,8 +369,8 @@ async def test_zigate_discovery_via_usb(detect_mock, hass): } -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=False) -async def test_discovery_via_usb_no_radio(detect_mock, hass): +@patch(f"bellows.{PROBE_FUNCTION_PATH}", return_value=False) +async def test_discovery_via_usb_no_radio(probe_mock, hass): """Test usb flow -- no radio detected.""" discovery_info = usb.UsbServiceInfo( device="/dev/null", @@ -294,13 +381,13 @@ async def test_discovery_via_usb_no_radio(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() assert result["type"] == FlowResultType.FORM assert result["step_id"] == "confirm" - with patch("homeassistant.components.zha.async_setup_entry"): + with patch("homeassistant.components.zha.async_setup_entry", return_value=True): result2 = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) @@ -310,8 +397,8 @@ async def test_discovery_via_usb_no_radio(detect_mock, hass): assert result2["reason"] == "usb_probe_failed" -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb_already_setup(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb_already_setup(hass): """Test usb flow -- already setup.""" MockConfigEntry( @@ -327,7 +414,7 @@ async def test_discovery_via_usb_already_setup(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -361,7 +448,7 @@ async def test_discovery_via_usb_path_changes(hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -374,8 +461,8 @@ async def test_discovery_via_usb_path_changes(hass): } -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb_deconz_already_discovered(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb_deconz_already_discovered(hass): """Test usb flow -- deconz discovered.""" result = await hass.config_entries.flow.async_init( "deconz", @@ -400,7 +487,7 @@ async def test_discovery_via_usb_deconz_already_discovered(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -408,8 +495,8 @@ async def test_discovery_via_usb_deconz_already_discovered(detect_mock, hass): assert result["reason"] == "not_zha_device" -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb_deconz_already_setup(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb_deconz_already_setup(hass): """Test usb flow -- deconz setup.""" MockConfigEntry(domain="deconz", data={}).add_to_hass(hass) await hass.async_block_till_done() @@ -422,7 +509,7 @@ async def test_discovery_via_usb_deconz_already_setup(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -430,8 +517,8 @@ async def test_discovery_via_usb_deconz_already_setup(detect_mock, hass): assert result["reason"] == "not_zha_device" -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb_deconz_ignored(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb_deconz_ignored(hass): """Test usb flow -- deconz ignored.""" MockConfigEntry( domain="deconz", source=config_entries.SOURCE_IGNORE, data={} @@ -446,7 +533,7 @@ async def test_discovery_via_usb_deconz_ignored(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -454,8 +541,8 @@ async def test_discovery_via_usb_deconz_ignored(detect_mock, hass): assert result["step_id"] == "confirm" -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_via_usb_zha_ignored_updates(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_via_usb_zha_ignored_updates(hass): """Test usb flow that was ignored gets updated.""" entry = MockConfigEntry( domain=DOMAIN, @@ -474,7 +561,7 @@ async def test_discovery_via_usb_zha_ignored_updates(detect_mock, hass): manufacturer="test", ) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_USB}, data=discovery_info + DOMAIN, context={"source": SOURCE_USB}, data=discovery_info ) await hass.async_block_till_done() @@ -486,8 +573,8 @@ async def test_discovery_via_usb_zha_ignored_updates(detect_mock, hass): @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=True) -async def test_discovery_already_setup(detect_mock, hass): +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)) +async def test_discovery_already_setup(hass): """Test zeroconf flow -- radio detected.""" service_info = zeroconf.ZeroconfServiceInfo( host="192.168.1.200", @@ -504,7 +591,7 @@ async def test_discovery_already_setup(detect_mock, hass): ).add_to_hass(hass) result = await hass.config_entries.flow.async_init( - "zha", context={"source": SOURCE_ZEROCONF}, data=service_info + DOMAIN, context={"source": SOURCE_ZEROCONF}, data=service_info ) await hass.async_block_till_done() @@ -512,12 +599,12 @@ async def test_discovery_already_setup(detect_mock, hass): assert result["reason"] == "single_instance_allowed" -@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @patch( - "homeassistant.components.zha.config_flow.detect_radios", - return_value={CONF_RADIO_TYPE: "test_radio"}, + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + mock_detect_radio_type(radio_type=RadioType.deconz), ) -async def test_user_flow(detect_mock, hass): +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) +async def test_user_flow(hass): """Test user flow -- radio detected.""" port = com_port() @@ -526,21 +613,36 @@ async def test_user_flow(detect_mock, hass): result = await hass.config_entries.flow.async_init( DOMAIN, context={CONF_SOURCE: SOURCE_USER}, - data={zigpy.config.CONF_DEVICE_PATH: port_select}, + data={ + zigpy.config.CONF_DEVICE_PATH: port_select, + }, ) - assert result["type"] == FlowResultType.CREATE_ENTRY - assert result["title"].startswith(port.description) - assert result["data"] == {CONF_RADIO_TYPE: "test_radio"} - assert detect_mock.await_count == 1 - assert detect_mock.await_args[0][0] == port.device + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "choose_formation_strategy" + + with patch("homeassistant.components.zha.async_setup_entry", return_value=True): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + assert result2["title"].startswith(port.description) + assert result2["data"] == { + "device": { + "path": port.device, + }, + CONF_RADIO_TYPE: "deconz", + } -@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @patch( - "homeassistant.components.zha.config_flow.detect_radios", - return_value=None, + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + mock_detect_radio_type(ret=False), ) -async def test_user_flow_not_detected(detect_mock, hass): +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) +async def test_user_flow_not_detected(hass): """Test user flow, radio not detected.""" port = com_port() @@ -553,9 +655,7 @@ async def test_user_flow_not_detected(detect_mock, hass): ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "pick_radio" - assert detect_mock.await_count == 1 - assert detect_mock.await_args[0][0] == port.device + assert result["step_id"] == "manual_pick_radio_type" @patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @@ -567,7 +667,7 @@ async def test_user_flow_show_form(hass): ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "user" + assert result["step_id"] == "choose_serial_port" @patch("serial.tools.list_ports.comports", MagicMock(return_value=[])) @@ -579,7 +679,7 @@ async def test_user_flow_show_manual(hass): ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "pick_radio" + assert result["step_id"] == "manual_pick_radio_type" async def test_user_flow_manual(hass): @@ -591,7 +691,7 @@ async def test_user_flow_manual(hass): data={zigpy.config.CONF_DEVICE_PATH: config_flow.CONF_MANUAL_PATH}, ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "pick_radio" + assert result["step_id"] == "manual_pick_radio_type" @pytest.mark.parametrize("radio_type", RadioType.list()) @@ -599,10 +699,12 @@ async def test_pick_radio_flow(hass, radio_type): """Test radio picker.""" result = await hass.config_entries.flow.async_init( - DOMAIN, context={CONF_SOURCE: "pick_radio"}, data={CONF_RADIO_TYPE: radio_type} + DOMAIN, + context={CONF_SOURCE: "manual_pick_radio_type"}, + data={CONF_RADIO_TYPE: radio_type}, ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "port_config" + assert result["step_id"] == "manual_port_config" async def test_user_flow_existing_config_entry(hass): @@ -618,82 +720,62 @@ async def test_user_flow_existing_config_entry(hass): assert result["type"] == "abort" -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=False) +@patch(f"bellows.{PROBE_FUNCTION_PATH}", return_value=False) +@patch(f"zigpy_deconz.{PROBE_FUNCTION_PATH}", return_value=False) +@patch(f"zigpy_zigate.{PROBE_FUNCTION_PATH}", return_value=False) +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", return_value=True) +async def test_detect_radio_type_success( + znp_probe, zigate_probe, deconz_probe, bellows_probe, hass +): + """Test detect radios successfully.""" + + handler = config_flow.ZhaConfigFlowHandler() + handler._device_path = "/dev/null" + + await handler._detect_radio_type() + + assert handler._radio_type == RadioType.znp + assert handler._device_settings[zigpy.config.CONF_DEVICE_PATH] == "/dev/null" + + assert bellows_probe.await_count == 1 + assert znp_probe.await_count == 1 + assert deconz_probe.await_count == 0 + assert zigate_probe.await_count == 0 + + @patch( - "zigpy_deconz.zigbee.application.ControllerApplication.probe", return_value=False + f"bellows.{PROBE_FUNCTION_PATH}", + return_value={"new_setting": 123, zigpy.config.CONF_DEVICE_PATH: "/dev/null"}, ) -@patch( - "zigpy_zigate.zigbee.application.ControllerApplication.probe", return_value=False -) -@patch("zigpy_xbee.zigbee.application.ControllerApplication.probe", return_value=False) -async def test_probe_radios(xbee_probe, zigate_probe, deconz_probe, znp_probe, hass): - """Test detect radios.""" - app_ctrl_cls = MagicMock() - app_ctrl_cls.SCHEMA_DEVICE = zigpy.config.SCHEMA_DEVICE - app_ctrl_cls.probe = AsyncMock(side_effect=(True, False)) +@patch(f"zigpy_deconz.{PROBE_FUNCTION_PATH}", return_value=False) +@patch(f"zigpy_zigate.{PROBE_FUNCTION_PATH}", return_value=False) +@patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", return_value=False) +async def test_detect_radio_type_success_with_settings( + znp_probe, zigate_probe, deconz_probe, bellows_probe, hass +): + """Test detect radios successfully but probing returns new settings.""" - p1 = patch( - "bellows.zigbee.application.ControllerApplication.probe", - side_effect=(True, False), - ) - with p1 as probe_mock: - res = await config_flow.detect_radios("/dev/null") - assert probe_mock.await_count == 1 - assert znp_probe.await_count == 1 # ZNP appears earlier in the radio list - assert res[CONF_RADIO_TYPE] == "ezsp" - assert zigpy.config.CONF_DEVICE in res - assert ( - res[zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH] == "/dev/null" - ) + handler = config_flow.ZhaConfigFlowHandler() + handler._device_path = "/dev/null" + await handler._detect_radio_type() - res = await config_flow.detect_radios("/dev/null") - assert res is None - assert xbee_probe.await_count == 1 - assert zigate_probe.await_count == 1 - assert deconz_probe.await_count == 1 - assert znp_probe.await_count == 2 + assert handler._radio_type == RadioType.ezsp + assert handler._device_settings["new_setting"] == 123 + assert handler._device_settings[zigpy.config.CONF_DEVICE_PATH] == "/dev/null" + + assert bellows_probe.await_count == 1 + assert znp_probe.await_count == 0 + assert deconz_probe.await_count == 0 + assert zigate_probe.await_count == 0 -@patch("zigpy_znp.zigbee.application.ControllerApplication.probe", return_value=False) -@patch( - "zigpy_deconz.zigbee.application.ControllerApplication.probe", return_value=False -) -@patch( - "zigpy_zigate.zigbee.application.ControllerApplication.probe", return_value=False -) -@patch("zigpy_xbee.zigbee.application.ControllerApplication.probe", return_value=False) -async def test_probe_new_ezsp(xbee_probe, zigate_probe, deconz_probe, znp_probe, hass): - """Test detect radios.""" - app_ctrl_cls = MagicMock() - app_ctrl_cls.SCHEMA_DEVICE = zigpy.config.SCHEMA_DEVICE - app_ctrl_cls.probe = AsyncMock(side_efferct=(True, False)) - - p1 = patch( - "bellows.zigbee.application.ControllerApplication.probe", - return_value={ - zigpy.config.CONF_DEVICE_PATH: sentinel.usb_port, - "baudrate": 33840, - }, - ) - with p1 as probe_mock: - res = await config_flow.detect_radios("/dev/null") - assert probe_mock.await_count == 1 - assert res[CONF_RADIO_TYPE] == "ezsp" - assert zigpy.config.CONF_DEVICE in res - assert ( - res[zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH] - is sentinel.usb_port - ) - assert res[zigpy.config.CONF_DEVICE]["baudrate"] == 33840 - - -@patch("bellows.zigbee.application.ControllerApplication.probe", return_value=False) +@patch(f"bellows.{PROBE_FUNCTION_PATH}", return_value=False) async def test_user_port_config_fail(probe_mock, hass): """Test port config flow.""" result = await hass.config_entries.flow.async_init( DOMAIN, - context={CONF_SOURCE: "pick_radio"}, + context={CONF_SOURCE: "manual_pick_radio_type"}, data={CONF_RADIO_TYPE: RadioType.ezsp.description}, ) @@ -702,19 +784,19 @@ async def test_user_port_config_fail(probe_mock, hass): user_input={zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB33"}, ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "port_config" + assert result["step_id"] == "manual_port_config" assert result["errors"]["base"] == "cannot_connect" assert probe_mock.await_count == 1 @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) -@patch("bellows.zigbee.application.ControllerApplication.probe", return_value=True) +@patch(f"bellows.{PROBE_FUNCTION_PATH}", return_value=True) async def test_user_port_config(probe_mock, hass): """Test port config.""" result = await hass.config_entries.flow.async_init( DOMAIN, - context={CONF_SOURCE: "pick_radio"}, + context={CONF_SOURCE: "manual_pick_radio_type"}, data={CONF_RADIO_TYPE: RadioType.ezsp.description}, ) @@ -723,13 +805,20 @@ async def test_user_port_config(probe_mock, hass): user_input={zigpy.config.CONF_DEVICE_PATH: "/dev/ttyUSB33"}, ) - assert result["type"] == "create_entry" - assert result["title"].startswith("/dev/ttyUSB33") + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "choose_formation_strategy" + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + assert ( - result["data"][zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH] + result2["data"][zigpy.config.CONF_DEVICE][zigpy.config.CONF_DEVICE_PATH] == "/dev/ttyUSB33" ) - assert result["data"][CONF_RADIO_TYPE] == "ezsp" + assert result2["data"][CONF_RADIO_TYPE] == "ezsp" assert probe_mock.await_count == 1 @@ -784,7 +873,7 @@ async def test_hardware_not_onboarded(hass): "homeassistant.components.onboarding.async_is_onboarded", return_value=False ): result = await hass.config_entries.flow.async_init( - "zha", context={"source": "hardware"}, data=data + DOMAIN, context={"source": "hardware"}, data=data ) assert result["type"] == FlowResultType.CREATE_ENTRY @@ -814,7 +903,7 @@ async def test_hardware_onboarded(hass): "homeassistant.components.onboarding.async_is_onboarded", return_value=True ): result = await hass.config_entries.flow.async_init( - "zha", context={"source": "hardware"}, data=data + DOMAIN, context={"source": "hardware"}, data=data ) assert result["type"] == FlowResultType.FORM @@ -852,7 +941,7 @@ async def test_hardware_already_setup(hass): }, } result = await hass.config_entries.flow.async_init( - "zha", context={"source": "hardware"}, data=data + DOMAIN, context={"source": "hardware"}, data=data ) assert result["type"] == FlowResultType.ABORT @@ -866,8 +955,664 @@ async def test_hardware_invalid_data(hass, data): """Test onboarding flow -- invalid data.""" result = await hass.config_entries.flow.async_init( - "zha", context={"source": "hardware"}, data=data + DOMAIN, context={"source": "hardware"}, data=data ) assert result["type"] == FlowResultType.ABORT assert result["reason"] == "invalid_hardware_data" + + +def test_allow_overwrite_ezsp_ieee(): + """Test modifying the backup to allow bellows to override the IEEE address.""" + backup = zigpy.backups.NetworkBackup() + new_backup = config_flow._allow_overwrite_ezsp_ieee(backup) + + assert backup != new_backup + assert ( + new_backup.network_info.stack_specific["ezsp"][ + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it" + ] + is True + ) + + +def test_prevent_overwrite_ezsp_ieee(): + """Test modifying the backup to prevent bellows from overriding the IEEE address.""" + backup = zigpy.backups.NetworkBackup() + backup.network_info.stack_specific["ezsp"] = { + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it": True + } + new_backup = config_flow._prevent_overwrite_ezsp_ieee(backup) + + assert backup != new_backup + assert not new_backup.network_info.stack_specific.get("ezsp", {}).get( + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it" + ) + + +@pytest.fixture +def pick_radio(hass): + """Fixture for the first step of the config flow (where a radio is picked).""" + + async def wrapper(radio_type): + port = com_port() + port_select = f"{port}, s/n: {port.serial_number} - {port.manufacturer}" + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + mock_detect_radio_type(radio_type=radio_type), + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={CONF_SOURCE: SOURCE_USER}, + data={ + zigpy.config.CONF_DEVICE_PATH: port_select, + }, + ) + + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "choose_formation_strategy" + + return result, port + + p1 = patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) + p2 = patch("homeassistant.components.zha.async_setup_entry") + + with p1, p2: + yield wrapper + + +async def test_strategy_no_network_settings(pick_radio, mock_app, hass): + """Test formation strategy when no network settings are present.""" + mock_app.load_network_info = MagicMock(side_effect=NetworkNotFormed()) + + result, port = await pick_radio(RadioType.ezsp) + assert ( + config_flow.FORMATION_REUSE_SETTINGS + not in result["data_schema"].schema["next_step_id"].container + ) + + +async def test_formation_strategy_form_new_network(pick_radio, mock_app, hass): + """Test forming a new network.""" + result, port = await pick_radio(RadioType.ezsp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_FORM_NEW_NETWORK}, + ) + await hass.async_block_till_done() + + # A new network will be formed + mock_app.form_network.assert_called_once() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + + +async def test_formation_strategy_reuse_settings(pick_radio, mock_app, hass): + """Test reusing existing network settings.""" + result, port = await pick_radio(RadioType.ezsp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + # Nothing will be written when settings are reused + mock_app.write_network_info.assert_not_called() + + assert result2["type"] == FlowResultType.CREATE_ENTRY + + +@patch("homeassistant.components.zha.config_flow.process_uploaded_file") +def test_parse_uploaded_backup(process_mock): + """Test parsing uploaded backup files.""" + backup = zigpy.backups.NetworkBackup() + + text = json.dumps(backup.as_dict()) + process_mock.return_value.__enter__.return_value.read_text.return_value = text + + handler = config_flow.ZhaConfigFlowHandler() + parsed_backup = handler._parse_uploaded_backup(str(uuid.uuid4())) + + assert backup == parsed_backup + + +@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +async def test_formation_strategy_restore_manual_backup_non_ezsp( + allow_overwrite_ieee_mock, pick_radio, mock_app, hass +): + """Test restoring a manual backup on non-EZSP coordinators.""" + result, port = await pick_radio(RadioType.znp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_UPLOAD_MANUAL_BACKUP}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "upload_manual_backup" + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._parse_uploaded_backup", + return_value=zigpy.backups.NetworkBackup(), + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={config_flow.UPLOADED_BACKUP_FILE: str(uuid.uuid4())}, + ) + + mock_app.backups.restore_backup.assert_called_once() + allow_overwrite_ieee_mock.assert_not_called() + + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["data"][CONF_RADIO_TYPE] == "znp" + + +@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +async def test_formation_strategy_restore_manual_backup_overwrite_ieee_ezsp( + allow_overwrite_ieee_mock, pick_radio, mock_app, backup, hass +): + """Test restoring a manual backup on EZSP coordinators (overwrite IEEE).""" + result, port = await pick_radio(RadioType.ezsp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_UPLOAD_MANUAL_BACKUP}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "upload_manual_backup" + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._parse_uploaded_backup", + return_value=backup, + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={config_flow.UPLOADED_BACKUP_FILE: str(uuid.uuid4())}, + ) + + assert result3["type"] == FlowResultType.FORM + assert result3["step_id"] == "maybe_confirm_ezsp_restore" + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], + user_input={config_flow.OVERWRITE_COORDINATOR_IEEE: True}, + ) + + allow_overwrite_ieee_mock.assert_called_once() + mock_app.backups.restore_backup.assert_called_once() + + assert result4["type"] == FlowResultType.CREATE_ENTRY + assert result4["data"][CONF_RADIO_TYPE] == "ezsp" + + +@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +async def test_formation_strategy_restore_manual_backup_ezsp( + allow_overwrite_ieee_mock, pick_radio, mock_app, hass +): + """Test restoring a manual backup on EZSP coordinators (don't overwrite IEEE).""" + result, port = await pick_radio(RadioType.ezsp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_UPLOAD_MANUAL_BACKUP}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "upload_manual_backup" + + backup = zigpy.backups.NetworkBackup() + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._parse_uploaded_backup", + return_value=backup, + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={config_flow.UPLOADED_BACKUP_FILE: str(uuid.uuid4())}, + ) + + assert result3["type"] == FlowResultType.FORM + assert result3["step_id"] == "maybe_confirm_ezsp_restore" + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], + user_input={config_flow.OVERWRITE_COORDINATOR_IEEE: False}, + ) + + allow_overwrite_ieee_mock.assert_not_called() + mock_app.backups.restore_backup.assert_called_once_with(backup) + + assert result4["type"] == FlowResultType.CREATE_ENTRY + assert result4["data"][CONF_RADIO_TYPE] == "ezsp" + + +async def test_formation_strategy_restore_manual_backup_invalid_upload( + pick_radio, mock_app, hass +): + """Test restoring a manual backup but an invalid file is uploaded.""" + result, port = await pick_radio(RadioType.ezsp) + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_UPLOAD_MANUAL_BACKUP}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "upload_manual_backup" + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._parse_uploaded_backup", + side_effect=ValueError("Invalid backup JSON"), + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={config_flow.UPLOADED_BACKUP_FILE: str(uuid.uuid4())}, + ) + + mock_app.backups.restore_backup.assert_not_called() + + assert result3["type"] == FlowResultType.FORM + assert result3["step_id"] == "upload_manual_backup" + assert result3["errors"]["base"] == "invalid_backup_json" + + +def test_format_backup_choice(): + """Test formatting zigpy NetworkBackup objects.""" + backup = zigpy.backups.NetworkBackup() + backup.network_info.pan_id = zigpy.types.PanId(0x1234) + backup.network_info.extended_pan_id = zigpy.types.EUI64.convert( + "aa:bb:cc:dd:ee:ff:00:11" + ) + + with_ids = config_flow._format_backup_choice(backup, pan_ids=True) + without_ids = config_flow._format_backup_choice(backup, pan_ids=False) + + assert with_ids.startswith(without_ids) + assert "1234:aabbccddeeff0011" in with_ids + assert "1234:aabbccddeeff0011" not in without_ids + + +@patch( + "homeassistant.components.zha.config_flow._format_backup_choice", + lambda s, **kwargs: "choice:" + repr(s), +) +@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) +async def test_formation_strategy_restore_automatic_backup_ezsp( + pick_radio, mock_app, hass +): + """Test restoring an automatic backup (EZSP radio).""" + mock_app.backups.backups = [ + MagicMock(), + MagicMock(), + MagicMock(), + ] + backup = mock_app.backups.backups[1] # pick the second one + backup.is_compatible_with = MagicMock(return_value=False) + + result, port = await pick_radio(RadioType.ezsp) + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": (config_flow.FORMATION_CHOOSE_AUTOMATIC_BACKUP)}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "choose_automatic_backup" + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={ + config_flow.CHOOSE_AUTOMATIC_BACKUP: "choice:" + repr(backup), + }, + ) + + assert result3["type"] == FlowResultType.FORM + assert result3["step_id"] == "maybe_confirm_ezsp_restore" + + result4 = await hass.config_entries.flow.async_configure( + result3["flow_id"], + user_input={config_flow.OVERWRITE_COORDINATOR_IEEE: True}, + ) + + mock_app.backups.restore_backup.assert_called_once() + + assert result4["type"] == FlowResultType.CREATE_ENTRY + assert result4["data"][CONF_RADIO_TYPE] == "ezsp" + + +@patch( + "homeassistant.components.zha.config_flow._format_backup_choice", + lambda s, **kwargs: "choice:" + repr(s), +) +@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) +@pytest.mark.parametrize("is_advanced", [True, False]) +async def test_formation_strategy_restore_automatic_backup_non_ezsp( + is_advanced, pick_radio, mock_app, hass +): + """Test restoring an automatic backup (non-EZSP radio).""" + mock_app.backups.backups = [ + MagicMock(), + MagicMock(), + MagicMock(), + ] + backup = mock_app.backups.backups[1] # pick the second one + backup.is_compatible_with = MagicMock(return_value=False) + + result, port = await pick_radio(RadioType.znp) + + with patch( + "homeassistant.config_entries.ConfigFlow.show_advanced_options", + new_callable=PropertyMock(return_value=is_advanced), + ): + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + "next_step_id": (config_flow.FORMATION_CHOOSE_AUTOMATIC_BACKUP) + }, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "choose_automatic_backup" + + # We must prompt for overwriting the IEEE address + assert config_flow.OVERWRITE_COORDINATOR_IEEE not in result2["data_schema"].schema + + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={ + config_flow.CHOOSE_AUTOMATIC_BACKUP: "choice:" + repr(backup), + }, + ) + + mock_app.backups.restore_backup.assert_called_once_with(backup) + + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["data"][CONF_RADIO_TYPE] == "znp" + + +@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +async def test_ezsp_restore_without_settings_change_ieee( + allow_overwrite_ieee_mock, pick_radio, mock_app, backup, hass +): + """Test a manual backup on EZSP coordinators without settings (no IEEE write).""" + # Fail to load settings + with patch.object( + mock_app, "load_network_info", MagicMock(side_effect=NetworkNotFormed()) + ): + result, port = await pick_radio(RadioType.ezsp) + + # Set the network state, it'll be picked up later after the load "succeeds" + mock_app.state.node_info = backup.node_info + mock_app.state.network_info = copy.deepcopy(backup.network_info) + mock_app.state.network_info.network_key.tx_counter += 10000 + + # Include the overwrite option, just in case someone uploads a backup with it + backup.network_info.metadata["ezsp"] = { + "i_understand_i_can_update_eui64_only_once_and_i_still_want_to_do_it": True + } + + result2 = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_UPLOAD_MANUAL_BACKUP}, + ) + await hass.async_block_till_done() + + assert result2["type"] == FlowResultType.FORM + assert result2["step_id"] == "upload_manual_backup" + + with patch( + "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._parse_uploaded_backup", + return_value=backup, + ): + result3 = await hass.config_entries.flow.async_configure( + result2["flow_id"], + user_input={config_flow.UPLOADED_BACKUP_FILE: str(uuid.uuid4())}, + ) + + # We wrote settings when connecting + allow_overwrite_ieee_mock.assert_not_called() + mock_app.backups.restore_backup.assert_called_once_with(backup, create_new=False) + + assert result3["type"] == FlowResultType.CREATE_ENTRY + assert result3["data"][CONF_RADIO_TYPE] == "ezsp" + + +@pytest.mark.parametrize( + "async_unload_effect", [True, config_entries.OperationNotAllowed()] +) +@patch( + "serial.tools.list_ports.comports", + MagicMock( + return_value=[ + com_port("/dev/SomePort"), + com_port("/dev/ttyUSB0"), + com_port("/dev/SomeOtherPort"), + ] + ), +) +@patch("homeassistant.components.zha.async_setup_entry", return_value=True) +async def test_options_flow_defaults(async_setup_entry, async_unload_effect, hass): + """Test options flow defaults match radio defaults.""" + + entry = MockConfigEntry( + version=config_flow.ZhaConfigFlowHandler.VERSION, + domain=DOMAIN, + data={ + CONF_DEVICE: { + CONF_DEVICE_PATH: "/dev/ttyUSB0", + CONF_BAUDRATE: 12345, + CONF_FLOWCONTROL: None, + }, + CONF_RADIO_TYPE: "znp", + }, + ) + entry.add_to_hass(hass) + + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + flow = await hass.config_entries.options.async_init(entry.entry_id) + + async_setup_entry.reset_mock() + + # ZHA gets unloaded + with patch( + "homeassistant.config_entries.ConfigEntries.async_unload", + side_effect=[async_unload_effect], + ) as mock_async_unload: + result1 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + mock_async_unload.assert_called_once_with(entry.entry_id) + + # Unload it ourselves + entry.state = config_entries.ConfigEntryState.NOT_LOADED + + # Current path is the default + assert result1["step_id"] == "choose_serial_port" + assert "/dev/ttyUSB0" in result1["data_schema"]({})[CONF_DEVICE_PATH] + + # Autoprobing fails, we have to manually choose the radio type + result2 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + # Current radio type is the default + assert result2["step_id"] == "manual_pick_radio_type" + assert result2["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description + + # Continue on to port settings + result3 = await hass.config_entries.options.async_configure( + flow["flow_id"], + user_input={ + CONF_RADIO_TYPE: RadioType.znp.description, + }, + ) + + # The defaults match our current settings + assert result3["step_id"] == "manual_port_config" + assert result3["data_schema"]({}) == entry.data[CONF_DEVICE] + + with patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)): + # Change the serial port path + result4 = await hass.config_entries.options.async_configure( + flow["flow_id"], + user_input={ + # Change everything + CONF_DEVICE_PATH: "/dev/new_serial_port", + CONF_BAUDRATE: 54321, + CONF_FLOWCONTROL: "software", + }, + ) + + # The radio has been detected, we can move on to creating the config entry + assert result4["step_id"] == "choose_formation_strategy" + + async_setup_entry.assert_not_called() + + result5 = await hass.config_entries.options.async_configure( + result1["flow_id"], + user_input={"next_step_id": config_flow.FORMATION_REUSE_SETTINGS}, + ) + await hass.async_block_till_done() + + assert result5["type"] == FlowResultType.CREATE_ENTRY + assert result5["data"] == {} + + # The updated entry contains correct settings + assert entry.data == { + CONF_DEVICE: { + CONF_DEVICE_PATH: "/dev/new_serial_port", + CONF_BAUDRATE: 54321, + CONF_FLOWCONTROL: "software", + }, + CONF_RADIO_TYPE: "znp", + } + + # ZHA was started again + assert async_setup_entry.call_count == 1 + + +@patch( + "serial.tools.list_ports.comports", + MagicMock( + return_value=[ + com_port("/dev/SomePort"), + com_port("/dev/SomeOtherPort"), + ] + ), +) +@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) +async def test_options_flow_defaults_socket(hass): + """Test options flow defaults work even for serial ports that can't be listed.""" + + entry = MockConfigEntry( + version=config_flow.ZhaConfigFlowHandler.VERSION, + domain=DOMAIN, + data={ + CONF_DEVICE: { + CONF_DEVICE_PATH: "socket://localhost:5678", + CONF_BAUDRATE: 12345, + CONF_FLOWCONTROL: None, + }, + CONF_RADIO_TYPE: "znp", + }, + ) + entry.add_to_hass(hass) + + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + flow = await hass.config_entries.options.async_init(entry.entry_id) + + # ZHA gets unloaded + with patch( + "homeassistant.config_entries.ConfigEntries.async_unload", return_value=True + ): + result1 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + # Radio path must be manually entered + assert result1["step_id"] == "choose_serial_port" + assert result1["data_schema"]({})[CONF_DEVICE_PATH] == config_flow.CONF_MANUAL_PATH + + result2 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + # Current radio type is the default + assert result2["step_id"] == "manual_pick_radio_type" + assert result2["data_schema"]({})[CONF_RADIO_TYPE] == RadioType.znp.description + + # Continue on to port settings + result3 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + # The defaults match our current settings + assert result3["step_id"] == "manual_port_config" + assert result3["data_schema"]({}) == entry.data[CONF_DEVICE] + + with patch(f"zigpy_znp.{PROBE_FUNCTION_PATH}", AsyncMock(return_value=True)): + result4 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + assert result4["step_id"] == "choose_formation_strategy" + + +@patch("homeassistant.components.zha.async_setup_entry", return_value=True) +async def test_options_flow_restarts_running_zha_if_cancelled(async_setup_entry, hass): + """Test options flow restarts a previously-running ZHA if it's cancelled.""" + + entry = MockConfigEntry( + version=config_flow.ZhaConfigFlowHandler.VERSION, + domain=DOMAIN, + data={ + CONF_DEVICE: { + CONF_DEVICE_PATH: "socket://localhost:5678", + CONF_BAUDRATE: 12345, + CONF_FLOWCONTROL: None, + }, + CONF_RADIO_TYPE: "znp", + }, + ) + entry.add_to_hass(hass) + + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + flow = await hass.config_entries.options.async_init(entry.entry_id) + + # ZHA gets unloaded + with patch( + "homeassistant.config_entries.ConfigEntries.async_unload", return_value=True + ): + result1 = await hass.config_entries.options.async_configure( + flow["flow_id"], user_input={} + ) + + entry.state = config_entries.ConfigEntryState.NOT_LOADED + + # Radio path must be manually entered + assert result1["step_id"] == "choose_serial_port" + + async_setup_entry.reset_mock() + + # Abort the flow + hass.config_entries.options.async_abort(result1["flow_id"]) + await hass.async_block_till_done() + + # ZHA was set up once more + async_setup_entry.assert_called_once_with(hass, entry)