From bb64b39d0e6d41f531af9c63b69d1ce243a2751b Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Wed, 16 Nov 2022 17:13:23 +0100 Subject: [PATCH] Minor refactor of zha config flow (#82200) * Minor refactor of zha config flow * Move ZhaRadioManager to a separate module --- homeassistant/components/zha/config_flow.py | 300 ++++++------------ homeassistant/components/zha/radio_manager.py | 158 +++++++++ tests/components/zha/test_config_flow.py | 40 ++- 3 files changed, 283 insertions(+), 215 deletions(-) create mode 100644 homeassistant/components/zha/radio_manager.py diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index be422778e2d..e8745684275 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -1,44 +1,34 @@ """Config flow for ZHA.""" from __future__ import annotations -import asyncio 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.core import callback +from homeassistant.core import HomeAssistant, 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, EZSP_OVERWRITE_EUI64, RadioType, ) +from .radio_manager import ZhaRadioManager CONF_MANUAL_PATH = "Enter Manually" SUPPORTED_PORT_SETTINGS = ( @@ -47,16 +37,6 @@ 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, -) - FORMATION_STRATEGY = "formation_strategy" FORMATION_FORM_NEW_NETWORK = "form_new_network" FORMATION_REUSE_SETTINGS = "reuse_settings" @@ -74,8 +54,6 @@ UPLOADED_BACKUP_FILE = "uploaded_backup_file" DEFAULT_ZHA_ZEROCONF_PORT = 6638 ESPHOME_API_PORT = 6053 -CONNECT_DELAY_S = 1.0 - HARDWARE_DISCOVERY_SCHEMA = vol.Schema( { vol.Required("name"): str, @@ -84,8 +62,6 @@ HARDWARE_DISCOVERY_SCHEMA = vol.Schema( } ) -_LOGGER = logging.getLogger(__name__) - def _format_backup_choice( backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True @@ -134,110 +110,44 @@ def _prevent_overwrite_ezsp_ieee( class BaseZhaFlow(FlowHandler): """Mixin for common ZHA flow steps and forms.""" + _hass: HomeAssistant + def __init__(self) -> None: """Initialize flow instance.""" super().__init__() - self._device_path: str | None = None - self._device_settings: dict[str, Any] | None = None - self._radio_type: RadioType | None = None + self._hass = None # type: ignore[assignment] + self._radio_mgr = ZhaRadioManager() 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 + @property + def hass(self): + """Return hass.""" + return self._hass - config = self.hass.data.get(DATA_ZHA, {}).get(DATA_ZHA_CONFIG, {}) - app_config = config.get(CONF_ZIGPY, {}).copy() + @hass.setter + def hass(self, hass): + """Set hass.""" + self._hass = hass + self._radio_mgr.hass = hass - 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() - await asyncio.sleep(CONNECT_DELAY_S) - - 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) - - def _parse_radio_type(self, radio_type: str) -> RadioType: - """Parse a radio type name, accounting for past aliases.""" - if radio_type == "efr32": - return RadioType.ezsp - - return RadioType[radio_type] - - 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.""" + async def _async_create_radio_entry(self) -> FlowResult: + """Create a config entry 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 + assert self._radio_mgr.radio_type is not None + assert self._radio_mgr.device_path is not None + assert self._radio_mgr.device_settings is not None - device_settings = self._device_settings.copy() + device_settings = self._radio_mgr.device_settings.copy() device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job( - usb.get_serial_by_id, self._device_path + usb.get_serial_by_id, self._radio_mgr.device_path ) return self.async_create_entry( title=self._title, data={ CONF_DEVICE: device_settings, - CONF_RADIO_TYPE: self._radio_type.name, + CONF_RADIO_TYPE: self._radio_mgr.radio_type.name, }, ) @@ -264,9 +174,9 @@ class BaseZhaFlow(FlowHandler): return await self.async_step_manual_pick_radio_type() port = ports[list_of_ports.index(user_selection)] - self._device_path = port.device + self._radio_mgr.device_path = port.device - if not await self._detect_radio_type(): + if not await self._radio_mgr.detect_radio_type(): # Did not autodetect anything, proceed to manual selection return await self.async_step_manual_pick_radio_type() @@ -282,9 +192,9 @@ class BaseZhaFlow(FlowHandler): # Pre-select the currently configured port default_port = vol.UNDEFINED - if self._device_path is not None: + if self._radio_mgr.device_path is not None: for description, port in zip(list_of_ports, ports): - if port.device == self._device_path: + if port.device == self._radio_mgr.device_path: default_port = description break else: @@ -304,14 +214,16 @@ class BaseZhaFlow(FlowHandler): ) -> FlowResult: """Manually select the radio type.""" if user_input is not None: - self._radio_type = RadioType.get_by_description(user_input[CONF_RADIO_TYPE]) + self._radio_mgr.radio_type = RadioType.get_by_description( + user_input[CONF_RADIO_TYPE] + ) 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 + if self._radio_mgr.radio_type is not None: + default = self._radio_mgr.radio_type.description schema = { vol.Required(CONF_RADIO_TYPE, default=default): vol.In(RadioType.list()) @@ -326,35 +238,43 @@ class BaseZhaFlow(FlowHandler): 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 + assert self._radio_mgr.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() + self._radio_mgr.device_path = user_input[CONF_DEVICE_PATH] + self._radio_mgr.device_settings = user_input.copy() - if await self._radio_type.controller.probe(user_input): + if await self._radio_mgr.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 + CONF_DEVICE_PATH, default=self._radio_mgr.device_path or vol.UNDEFINED ): str } source = self.context.get("source") - for param, value in self._radio_type.controller.SCHEMA_DEVICE.schema.items(): + for ( + param, + value, + ) in self._radio_mgr.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]) + elif ( + self._radio_mgr.device_settings is not None + and param in self._radio_mgr.device_settings + ): + param = vol.Required( + str(param), default=self._radio_mgr.device_settings[param] + ) schema[param] = value @@ -364,43 +284,26 @@ class BaseZhaFlow(FlowHandler): 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() + await self._radio_mgr.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 + if self._radio_mgr.backups and ( + self._radio_mgr.current_settings is None or any( - not backup.is_compatible_with(self._current_settings) - for backup in self._backups + not backup.is_compatible_with(self._radio_mgr.current_settings) + for backup in self._radio_mgr.backups ) ): strategies.append(CHOOSE_AUTOMATIC_BACKUP) - if self._current_settings is not None: + if self._radio_mgr.current_settings is not None: strategies.append(FORMATION_REUSE_SETTINGS) strategies.append(FORMATION_UPLOAD_MANUAL_BACKUP) @@ -415,16 +318,14 @@ class BaseZhaFlow(FlowHandler): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Reuse the existing network settings on the stick.""" - return await self._async_create_radio_entity() + return await self._async_create_radio_entry() 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() + await self._radio_mgr.async_form_network() + return await self._async_create_radio_entry() def _parse_uploaded_backup( self, uploaded_file_id: str @@ -443,7 +344,7 @@ class BaseZhaFlow(FlowHandler): if user_input is not None: try: - self._chosen_backup = await self.hass.async_add_executor_job( + self._radio_mgr.chosen_backup = await self.hass.async_add_executor_job( self._parse_uploaded_backup, user_input[UPLOADED_BACKUP_FILE] ) except ValueError: @@ -470,23 +371,24 @@ class BaseZhaFlow(FlowHandler): 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 + _format_backup_choice(backup, pan_ids=True) + for backup in self._radio_mgr.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 + backup.backup_time.date() for backup in self._radio_mgr.backups ) choices = [ _format_backup_choice( backup, pan_ids=(num_backups_on_date[backup.backup_time.date()] > 1) ) - for backup in self._backups + for backup in self._radio_mgr.backups ] if user_input is not None: index = choices.index(user_input[CHOOSE_AUTOMATIC_BACKUP]) - self._chosen_backup = self._backups[index] + self._radio_mgr.chosen_backup = self._radio_mgr.backups[index] return await self.async_step_maybe_confirm_ezsp_restore() @@ -505,46 +407,47 @@ class BaseZhaFlow(FlowHandler): 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 + assert self._radio_mgr.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() + if self._radio_mgr.radio_type != RadioType.ezsp: + await self._radio_mgr.restore_backup(self._radio_mgr.chosen_backup) + return await self._async_create_radio_entry() # We have no way to partially load network settings if no network is formed - if self._current_settings is None: + if self._radio_mgr.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() + temp_backup = _prevent_overwrite_ezsp_ieee(self._radio_mgr.chosen_backup) + await self._radio_mgr.restore_backup(temp_backup, create_new=False) + await self._radio_mgr.async_load_network_settings() - assert self._current_settings is not None + assert self._radio_mgr.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"][ + self._radio_mgr.current_settings.node_info.ieee + == self._radio_mgr.chosen_backup.node_info.ieee + or not self._radio_mgr.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) + await self._radio_mgr.restore_backup(self._radio_mgr.chosen_backup) - return await self._async_create_radio_entity() + return await self._async_create_radio_entry() if user_input is not None: - backup = self._chosen_backup + backup = self._radio_mgr.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) + await self._radio_mgr.restore_backup(backup) - return await self._async_create_radio_entity() + return await self._async_create_radio_entry() return self.async_show_form( step_id="maybe_confirm_ezsp_restore", @@ -608,13 +511,16 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN # config flow logic that interacts with hardware! if user_input is not None or not onboarding.async_is_onboarded(self.hass): # Probe the radio type if we don't have one yet - if self._radio_type is None and not await self._detect_radio_type(): + if ( + self._radio_mgr.radio_type is None + and not await self._radio_mgr.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") - if self._device_settings is None: + if self._radio_mgr.device_settings is None: return await self.async_step_manual_port_config() return await self.async_step_choose_formation_strategy() @@ -647,7 +553,7 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN if entry.source != config_entries.SOURCE_IGNORE: return self.async_abort(reason="not_zha_device") - self._device_path = dev_path + self._radio_mgr.device_path = dev_path self._title = description or usb.human_readable_device_name( dev_path, serial_number, @@ -673,13 +579,13 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN port = DEFAULT_ZHA_ZEROCONF_PORT if "radio_type" in discovery_info.properties: - self._radio_type = self._parse_radio_type( + self._radio_mgr.radio_type = self._radio_mgr.parse_radio_type( discovery_info.properties["radio_type"] ) elif "efr32" in local_name: - self._radio_type = RadioType.ezsp + self._radio_mgr.radio_type = RadioType.ezsp else: - self._radio_type = RadioType.znp + self._radio_mgr.radio_type = RadioType.znp node_name = local_name[: -len(".local")] device_path = f"socket://{discovery_info.host}:{port}" @@ -691,7 +597,7 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN self.context["title_placeholders"] = {CONF_NAME: node_name} self._title = device_path - self._device_path = device_path + self._radio_mgr.device_path = device_path return await self.async_step_confirm() @@ -705,7 +611,7 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN return self.async_abort(reason="invalid_hardware_data") name = discovery_data["name"] - radio_type = self._parse_radio_type(discovery_data["radio_type"]) + radio_type = self._radio_mgr.parse_radio_type(discovery_data["radio_type"]) try: device_settings = radio_type.controller.SCHEMA_DEVICE( @@ -720,9 +626,9 @@ class ZhaConfigFlowHandler(BaseZhaFlow, config_entries.ConfigFlow, domain=DOMAIN ) self._title = name - self._radio_type = radio_type - self._device_path = device_settings[CONF_DEVICE_PATH] - self._device_settings = device_settings + self._radio_mgr.radio_type = radio_type + self._radio_mgr.device_path = device_settings[CONF_DEVICE_PATH] + self._radio_mgr.device_settings = device_settings self.context["title_placeholders"] = {CONF_NAME: name} return await self.async_step_confirm() @@ -736,9 +642,9 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow): 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._radio_mgr.device_path = config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH] + self._radio_mgr.device_settings = config_entry.data[CONF_DEVICE] + self._radio_mgr.radio_type = RadioType[config_entry.data[CONF_RADIO_TYPE]] self._title = config_entry.title async def async_step_init( @@ -781,9 +687,7 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow): """Confirm the user wants to reset their current radio.""" if user_input is not None: - # Reset the current adapter - async with self._connect_zigpy_app() as app: - await app.reset_network_info() + await self._radio_mgr.async_reset_adapter() return await self.async_step_instruct_unplug() @@ -800,11 +704,11 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow): return self.async_show_form(step_id="instruct_unplug") - async def _async_create_radio_entity(self): + async def _async_create_radio_entry(self): """Re-implementation of the base flow's final step to update the config.""" - device_settings = self._device_settings.copy() + device_settings = self._radio_mgr.device_settings.copy() device_settings[CONF_DEVICE_PATH] = await self.hass.async_add_executor_job( - usb.get_serial_by_id, self._device_path + usb.get_serial_by_id, self._radio_mgr.device_path ) # Avoid creating both `.options` and `.data` by directly writing `data` here @@ -812,7 +716,7 @@ class ZhaOptionsFlowHandler(BaseZhaFlow, config_entries.OptionsFlow): entry=self.config_entry, data={ CONF_DEVICE: device_settings, - CONF_RADIO_TYPE: self._radio_type.name, + CONF_RADIO_TYPE: self._radio_mgr.radio_type.name, }, options=self.config_entry.options, ) diff --git a/homeassistant/components/zha/radio_manager.py b/homeassistant/components/zha/radio_manager.py new file mode 100644 index 00000000000..05629ec8249 --- /dev/null +++ b/homeassistant/components/zha/radio_manager.py @@ -0,0 +1,158 @@ +"""Config flow for ZHA.""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +import os +from typing import Any + +from zigpy.application import ControllerApplication +import zigpy.backups +from zigpy.config import CONF_DEVICE, CONF_DEVICE_PATH +from zigpy.exceptions import NetworkNotFormed + +from homeassistant.core import HomeAssistant + +from .core.const import ( + CONF_DATABASE, + CONF_ZIGPY, + DATA_ZHA, + DATA_ZHA_CONFIG, + DEFAULT_DATABASE_NAME, + RadioType, +) + +# 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, +) + +CONNECT_DELAY_S = 1.0 + +_LOGGER = logging.getLogger(__name__) + + +class ZhaRadioManager: + """Helper class with radio related functionality.""" + + hass: HomeAssistant + + def __init__(self) -> None: + """Initialize ZhaRadioManager instance.""" + self.device_path: str | None = None + self.device_settings: dict[str, Any] | None = None + self.radio_type: RadioType | 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() + await asyncio.sleep(CONNECT_DELAY_S) + + 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) + + def parse_radio_type(self, radio_type: str) -> RadioType: + """Parse a radio type name, accounting for past aliases.""" + if radio_type == "efr32": + return RadioType.ezsp + + return RadioType[radio_type] + + 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_load_network_settings(self, create_backup: bool = False) -> 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, + ) + + if create_backup: + await app.backups.create_backup() + + # The list of backups will always exist + self.backups = app.backups.backups.copy() + + async def async_form_network(self) -> None: + """Form a brand new network.""" + async with self._connect_zigpy_app() as app: + await app.form_network() + + async def async_reset_adapter(self) -> None: + """Reset the current adapter.""" + async with self._connect_zigpy_app() as app: + await app.reset_network_info() diff --git a/tests/components/zha/test_config_flow.py b/tests/components/zha/test_config_flow.py index dd6498fde43..d2001dfda1b 100644 --- a/tests/components/zha/test_config_flow.py +++ b/tests/components/zha/test_config_flow.py @@ -49,7 +49,7 @@ def disable_platform_only(): @pytest.fixture(autouse=True) def reduce_reconnect_timeout(): """Reduces reconnect timeout to speed up tests.""" - with patch("homeassistant.components.zha.config_flow.CONNECT_DELAY_S", 0.01): + with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.01): yield @@ -76,12 +76,12 @@ def backup(): def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True): - """Mock `_detect_radio_type` that just sets the appropriate attributes.""" + """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} + self.radio_type = radio_type + self.device_settings = radio_type.controller.SCHEMA_DEVICE( + {CONF_DEVICE_PATH: self.device_path} ) return ret @@ -669,7 +669,7 @@ async def test_discovery_already_setup(hass): @patch( - "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type", mock_detect_radio_type(radio_type=RadioType.deconz), ) @patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @@ -707,7 +707,7 @@ async def test_user_flow(hass): @patch( - "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type", mock_detect_radio_type(ret=False), ) @patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @@ -799,12 +799,14 @@ async def test_detect_radio_type_success( """Test detect radios successfully.""" handler = config_flow.ZhaConfigFlowHandler() - handler._device_path = "/dev/null" + handler._radio_mgr.device_path = "/dev/null" - await handler._detect_radio_type() + await handler._radio_mgr.detect_radio_type() - assert handler._radio_type == RadioType.znp - assert handler._device_settings[zigpy.config.CONF_DEVICE_PATH] == "/dev/null" + assert handler._radio_mgr.radio_type == RadioType.znp + assert ( + handler._radio_mgr.device_settings[zigpy.config.CONF_DEVICE_PATH] == "/dev/null" + ) assert bellows_probe.await_count == 1 assert znp_probe.await_count == 1 @@ -825,12 +827,14 @@ async def test_detect_radio_type_success_with_settings( """Test detect radios successfully but probing returns new settings.""" handler = config_flow.ZhaConfigFlowHandler() - handler._device_path = "/dev/null" - await handler._detect_radio_type() + handler._radio_mgr.device_path = "/dev/null" + await handler._radio_mgr.detect_radio_type() - 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 handler._radio_mgr.radio_type == RadioType.ezsp + assert handler._radio_mgr.device_settings["new_setting"] == 123 + assert ( + handler._radio_mgr.device_settings[zigpy.config.CONF_DEVICE_PATH] == "/dev/null" + ) assert bellows_probe.await_count == 1 assert znp_probe.await_count == 0 @@ -1047,7 +1051,7 @@ def pick_radio(hass): port_select = f"{port}, s/n: {port.serial_number} - {port.manufacturer}" with patch( - "homeassistant.components.zha.config_flow.ZhaConfigFlowHandler._detect_radio_type", + "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type", mock_detect_radio_type(radio_type=radio_type), ): result = await hass.config_entries.flow.async_init( @@ -1631,6 +1635,7 @@ async def test_options_flow_defaults_socket(hass): assert result5["step_id"] == "choose_formation_strategy" +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @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.""" @@ -1683,6 +1688,7 @@ async def test_options_flow_restarts_running_zha_if_cancelled(async_setup_entry, async_setup_entry.assert_called_once_with(hass, entry) +@patch("serial.tools.list_ports.comports", MagicMock(return_value=[com_port()])) @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) async def test_options_flow_migration_reset_old_adapter(hass, mock_app): """Test options flow for migrating from an old radio."""