From be7e76f302f670da61f24b96332a061a5032f479 Mon Sep 17 00:00:00 2001 From: Erik Montnemery Date: Tue, 22 Nov 2022 11:17:23 +0100 Subject: [PATCH] Migrate ZHA when enabling multi-PAN support on HA Yellow (#82213) * Migrate ZHA when enabling multi-PAN support on HA Yellow * Refactor BaseZhaFlow.async_step_maybe_confirm_ezsp_restore * Change data passed to ZHA to initiate migration * Catch errors during ZHA migration * Fix ZhaMigrationHelper.async_prepare_yellow_migration return value * Improve test coverage * Improve test coverage * Fix spelling * Rename some none HA yellow specifics * Rename again * Increase number of migration retries + refactor * Suppress OperationNotAllowed when reloading * Adjust tests --- .../homeassistant_hardware/manifest.json | 3 +- .../silabs_multiprotocol_addon.py | 48 +++ .../homeassistant_yellow/__init__.py | 22 +- .../homeassistant_yellow/config_flow.py | 14 +- .../components/homeassistant_yellow/const.py | 10 + .../homeassistant_yellow/strings.json | 5 +- .../homeassistant_yellow/translations/en.json | 5 +- homeassistant/components/zha/config_flow.py | 81 +--- homeassistant/components/zha/radio_manager.py | 230 ++++++++++- .../homeassistant_hardware/conftest.py | 30 +- .../test_silabs_multiprotocol_addon.py | 371 +++++++++++++++++- .../homeassistant_yellow/conftest.py | 5 +- .../homeassistant_yellow/test_config_flow.py | 86 ++++ .../homeassistant_yellow/test_init.py | 2 +- tests/components/zha/test_config_flow.py | 14 +- tests/components/zha/test_radio_manager.py | 311 +++++++++++++++ 16 files changed, 1120 insertions(+), 117 deletions(-) create mode 100644 tests/components/zha/test_radio_manager.py diff --git a/homeassistant/components/homeassistant_hardware/manifest.json b/homeassistant/components/homeassistant_hardware/manifest.json index 32472bcdded..722306720a5 100644 --- a/homeassistant/components/homeassistant_hardware/manifest.json +++ b/homeassistant/components/homeassistant_hardware/manifest.json @@ -3,5 +3,6 @@ "name": "Home Assistant Hardware", "documentation": "https://www.home-assistant.io/integrations/homeassistant_hardware", "codeowners": ["@home-assistant/core"], - "integration_type": "system" + "integration_type": "system", + "after_dependencies": ["zha"] } diff --git a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py index b4dcca0e329..0724eee9ed5 100644 --- a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py +++ b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py @@ -17,6 +17,8 @@ from homeassistant.components.hassio import ( AddonState, is_hassio, ) +from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN +from homeassistant.components.zha.radio_manager import ZhaMultiPANMigrationHelper from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import ( AbortFlow, @@ -77,6 +79,7 @@ class BaseMultiPanFlow(FlowHandler): # If we install the add-on we should uninstall it on entry remove. self.install_task: asyncio.Task | None = None self.start_task: asyncio.Task | None = None + self._zha_migration_mgr: ZhaMultiPANMigrationHelper | None = None @property @abstractmethod @@ -87,6 +90,18 @@ class BaseMultiPanFlow(FlowHandler): async def _async_serial_port_settings(self) -> SerialPortSettings: """Return the radio serial port settings.""" + @abstractmethod + async def _async_zha_physical_discovery(self) -> dict[str, Any]: + """Return ZHA discovery data when multiprotocol FW is not used. + + Passed to ZHA do determine if the ZHA config entry is connected to the radio + being migrated. + """ + + @abstractmethod + def _zha_name(self) -> str: + """Return the ZHA name.""" + async def async_step_install_addon( self, user_input: dict[str, Any] | None = None ) -> FlowResult: @@ -260,6 +275,31 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow): **dataclasses.asdict(serial_port_settings), } + # Initiate ZHA migration + zha_entries = self.hass.config_entries.async_entries(ZHA_DOMAIN) + + if zha_entries: + zha_migration_mgr = ZhaMultiPANMigrationHelper(self.hass, zha_entries[0]) + migration_data = { + "new_discovery_info": { + "name": self._zha_name(), + "port": { + "path": get_zigbee_socket(self.hass, addon_info), + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + "old_discovery_info": await self._async_zha_physical_discovery(), + } + _LOGGER.debug("Starting ZHA migration with: %s", migration_data) + try: + if await zha_migration_mgr.async_initiate_migration(migration_data): + self._zha_migration_mgr = zha_migration_mgr + except Exception as err: + _LOGGER.exception("Unexpected exception during ZHA migration") + raise AbortFlow("zha_migration_failed") from err + if new_addon_config != addon_config: # Copy the add-on config to keep the objects separate. self.original_addon_config = dict(addon_config) @@ -277,6 +317,14 @@ class OptionsFlowHandler(BaseMultiPanFlow, config_entries.OptionsFlow): self.hass.config_entries.async_reload(self.config_entry.entry_id) ) + # Finish ZHA migration if needed + if self._zha_migration_mgr: + try: + await self._zha_migration_mgr.async_finish_migration() + except Exception as err: + _LOGGER.exception("Unexpected exception during ZHA migration") + raise AbortFlow("zha_migration_failed") from err + return self.async_create_entry(title="", data={}) async def async_step_addon_installed( diff --git a/homeassistant/components/homeassistant_yellow/__init__.py b/homeassistant/components/homeassistant_yellow/__init__.py index 83a3d9dad32..a7e351535aa 100644 --- a/homeassistant/components/homeassistant_yellow/__init__.py +++ b/homeassistant/components/homeassistant_yellow/__init__.py @@ -18,6 +18,8 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady +from .const import ZHA_HW_DISCOVERY_DATA + _LOGGER = logging.getLogger(__name__) @@ -52,22 +54,22 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: raise ConfigEntryNotReady if addon_info.state == AddonState.NOT_INSTALLED: - path = "/dev/ttyAMA1" + hw_discovery_data = ZHA_HW_DISCOVERY_DATA else: - path = get_zigbee_socket(hass, addon_info) - - await hass.config_entries.flow.async_init( - "zha", - context={"source": "hardware"}, - data={ - "name": "Yellow", + hw_discovery_data = { + "name": "Yellow Multi-PAN", "port": { - "path": path, + "path": get_zigbee_socket(hass, addon_info), "baudrate": 115200, "flow_control": "hardware", }, "radio_type": "efr32", - }, + } + + await hass.config_entries.flow.async_init( + "zha", + context={"source": "hardware"}, + data=hw_discovery_data, ) return True diff --git a/homeassistant/components/homeassistant_yellow/config_flow.py b/homeassistant/components/homeassistant_yellow/config_flow.py index 48a797a2aab..e5e6bb3b2c6 100644 --- a/homeassistant/components/homeassistant_yellow/config_flow.py +++ b/homeassistant/components/homeassistant_yellow/config_flow.py @@ -8,7 +8,7 @@ from homeassistant.config_entries import ConfigEntry, ConfigFlow from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult -from .const import DOMAIN +from .const import DOMAIN, ZHA_HW_DISCOVERY_DATA class HomeAssistantYellowConfigFlow(ConfigFlow, domain=DOMAIN): @@ -44,3 +44,15 @@ class HomeAssistantYellowOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandl baudrate="115200", flow_control=True, ) + + async def _async_zha_physical_discovery(self) -> dict[str, Any]: + """Return ZHA discovery data when multiprotocol FW is not used. + + Passed to ZHA do determine if the ZHA config entry is connected to the radio + being migrated. + """ + return ZHA_HW_DISCOVERY_DATA + + def _zha_name(self) -> str: + """Return the ZHA name.""" + return "Yellow Multi-PAN" diff --git a/homeassistant/components/homeassistant_yellow/const.py b/homeassistant/components/homeassistant_yellow/const.py index 41eae70b3f2..5d693d819b6 100644 --- a/homeassistant/components/homeassistant_yellow/const.py +++ b/homeassistant/components/homeassistant_yellow/const.py @@ -1,3 +1,13 @@ """Constants for the Home Assistant Yellow integration.""" DOMAIN = "homeassistant_yellow" + +ZHA_HW_DISCOVERY_DATA = { + "name": "Yellow", + "port": { + "path": "/dev/ttyAMA1", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", +} diff --git a/homeassistant/components/homeassistant_yellow/strings.json b/homeassistant/components/homeassistant_yellow/strings.json index df1eb780a96..1810597b242 100644 --- a/homeassistant/components/homeassistant_yellow/strings.json +++ b/homeassistant/components/homeassistant_yellow/strings.json @@ -3,7 +3,7 @@ "step": { "addon_not_installed": { "title": "Enable multiprotocol support on the IEEE 802.15.4 radio", - "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. Note: This is an experimental feature.", + "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. If the radio is already used by the ZHA Zigbee integration, ZHA will be reconfigured to use the multiprotocol firmware.\n\nNote: This is an experimental feature.", "data": { "enable_multi_pan": "Enable multiprotocol support" } @@ -30,7 +30,8 @@ "addon_install_failed": "Failed to install the Silicon Labs Multiprotocol add-on.", "addon_set_config_failed": "Failed to set Silicon Labs Multiprotocol configuration.", "addon_start_failed": "Failed to start the Silicon Labs Multiprotocol add-on.", - "not_hassio": "The hardware options can only be configured on HassOS installations." + "not_hassio": "The hardware options can only be configured on HassOS installations.", + "zha_migration_failed": "The ZHA migration did not succeed." }, "progress": { "install_addon": "Please wait while the Silicon Labs Multiprotocol add-on installation finishes. This can take several minutes.", diff --git a/homeassistant/components/homeassistant_yellow/translations/en.json b/homeassistant/components/homeassistant_yellow/translations/en.json index 252303ea428..0b074452652 100644 --- a/homeassistant/components/homeassistant_yellow/translations/en.json +++ b/homeassistant/components/homeassistant_yellow/translations/en.json @@ -5,7 +5,8 @@ "addon_install_failed": "Failed to install the Silicon Labs Multiprotocol add-on.", "addon_set_config_failed": "Failed to set Silicon Labs Multiprotocol configuration.", "addon_start_failed": "Failed to start the Silicon Labs Multiprotocol add-on.", - "not_hassio": "The hardware options can only be configured on HassOS installations." + "not_hassio": "The hardware options can only be configured on HassOS installations.", + "zha_migration_failed": "The ZHA migration did not succeed." }, "error": { "unknown": "Unexpected error" @@ -22,7 +23,7 @@ "data": { "enable_multi_pan": "Enable multiprotocol support" }, - "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. Note: This is an experimental feature.", + "description": "When multiprotocol support is enabled, the Home Assistant Yellow's IEEE 802.15.4 radio can be used for both Zigbee and Thread (used by Matter) at the same time. If the radio is already used by the ZHA Zigbee integration, ZHA will be reconfigured to use the multiprotocol firmware.\n\nNote: This is an experimental feature.", "title": "Enable multiprotocol support on the IEEE 802.15.4 radio" }, "install_addon": { diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index e8745684275..df5fa047c99 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -2,7 +2,6 @@ from __future__ import annotations import collections -import copy import json from typing import Any @@ -25,10 +24,9 @@ from .core.const import ( CONF_FLOWCONTROL, CONF_RADIO_TYPE, DOMAIN, - EZSP_OVERWRITE_EUI64, RadioType, ) -from .radio_manager import ZhaRadioManager +from .radio_manager import HARDWARE_DISCOVERY_SCHEMA, ZhaRadioManager CONF_MANUAL_PATH = "Enter Manually" SUPPORTED_PORT_SETTINGS = ( @@ -54,14 +52,6 @@ UPLOADED_BACKUP_FILE = "uploaded_backup_file" DEFAULT_ZHA_ZEROCONF_PORT = 6638 ESPHOME_API_PORT = 6053 -HARDWARE_DISCOVERY_SCHEMA = vol.Schema( - { - vol.Required("name"): str, - vol.Required("port"): dict, - vol.Required("radio_type"): str, - } -) - def _format_backup_choice( backup: zigpy.backups.NetworkBackup, *, pan_ids: bool = True @@ -80,33 +70,6 @@ def _format_backup_choice( 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", {})[EZSP_OVERWRITE_EUI64] = 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(EZSP_OVERWRITE_EUI64, 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.""" @@ -407,46 +370,14 @@ class BaseZhaFlow(FlowHandler): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Confirm restore for EZSP radios that require permanent IEEE writes.""" - assert self._radio_mgr.chosen_backup is not None - - 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._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._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._radio_mgr.current_settings is not None - - if ( - 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._radio_mgr.restore_backup(self._radio_mgr.chosen_backup) - + call_step_2 = await self._radio_mgr.async_restore_backup_step_1() + if not call_step_2: return await self._async_create_radio_entry() if user_input is not None: - 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._radio_mgr.restore_backup(backup) - + await self._radio_mgr.async_restore_backup_step_2( + user_input[OVERWRITE_COORDINATOR_IEEE] + ) return await self._async_create_radio_entry() return self.async_show_form( diff --git a/homeassistant/components/zha/radio_manager.py b/homeassistant/components/zha/radio_manager.py index 05629ec8249..512f26a139b 100644 --- a/homeassistant/components/zha/radio_manager.py +++ b/homeassistant/components/zha/radio_manager.py @@ -3,23 +3,28 @@ from __future__ import annotations import asyncio import contextlib +import copy import logging import os from typing import Any +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.core import HomeAssistant from .core.const import ( CONF_DATABASE, + CONF_RADIO_TYPE, CONF_ZIGPY, DATA_ZHA, DATA_ZHA_CONFIG, DEFAULT_DATABASE_NAME, + EZSP_OVERWRITE_EUI64, RadioType, ) @@ -35,9 +40,53 @@ AUTOPROBE_RADIOS = ( CONNECT_DELAY_S = 1.0 +MIGRATION_RETRIES = 100 + +HARDWARE_DISCOVERY_SCHEMA = vol.Schema( + { + vol.Required("name"): str, + vol.Required("port"): dict, + vol.Required("radio_type"): str, + } +) + +HARDWARE_MIGRATION_SCHEMA = vol.Schema( + { + vol.Required("new_discovery_info"): HARDWARE_DISCOVERY_SCHEMA, + vol.Required("old_discovery_info"): HARDWARE_DISCOVERY_SCHEMA, + } +) + _LOGGER = logging.getLogger(__name__) +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", {})[EZSP_OVERWRITE_EUI64] = 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(EZSP_OVERWRITE_EUI64, None) + + return backup.replace( + network_info=backup.network_info.replace(stack_specific=new_stack_specific) + ) + + class ZhaRadioManager: """Helper class with radio related functionality.""" @@ -96,7 +145,8 @@ class ZhaRadioManager: async with self._connect_zigpy_app() as app: await app.backups.restore_backup(backup, **kwargs) - def parse_radio_type(self, radio_type: str) -> RadioType: + @staticmethod + def parse_radio_type(radio_type: str) -> RadioType: """Parse a radio type name, accounting for past aliases.""" if radio_type == "efr32": return RadioType.ezsp @@ -127,8 +177,12 @@ class ZhaRadioManager: return False - async def async_load_network_settings(self, create_backup: bool = False) -> None: + async def async_load_network_settings( + self, *, create_backup: bool = False + ) -> zigpy.backups.NetworkBackup | None: """Connect to the radio and load its current network settings.""" + backup = None + async with self._connect_zigpy_app() as app: # Check if the stick has any settings and load them try: @@ -142,11 +196,13 @@ class ZhaRadioManager: ) if create_backup: - await app.backups.create_backup() + backup = await app.backups.create_backup() # The list of backups will always exist self.backups = app.backups.backups.copy() + return backup + async def async_form_network(self) -> None: """Form a brand new network.""" async with self._connect_zigpy_app() as app: @@ -156,3 +212,171 @@ class ZhaRadioManager: """Reset the current adapter.""" async with self._connect_zigpy_app() as app: await app.reset_network_info() + + async def async_restore_backup_step_1(self) -> bool: + """Prepare restoring backup. + + Returns True if async_restore_backup_step_2 should be called. + """ + assert self.chosen_backup is not None + + if self.radio_type != RadioType.ezsp: + await self.restore_backup(self.chosen_backup) + return False + + # 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 False + + return True + + async def async_restore_backup_step_2(self, overwrite_ieee: bool) -> None: + """Restore backup and optionally overwrite IEEE.""" + assert self.chosen_backup is not None + + backup = self.chosen_backup + + if overwrite_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) + + +class ZhaMultiPANMigrationHelper: + """Helper class for automatic migration when upgrading the firmware of a radio. + + This class is currently only intended to be used when changing the firmware on the + radio used in the Home Assistant Sky Connect USB stick and the Home Asssistant Yellow + from Zigbee only firmware to firmware supporting both Zigbee and Thread. + """ + + def __init__( + self, hass: HomeAssistant, config_entry: config_entries.ConfigEntry + ) -> None: + """Initialize MigrationHelper instance.""" + self._config_entry = config_entry + self._hass = hass + self._radio_mgr = ZhaRadioManager() + self._radio_mgr.hass = hass + + async def async_initiate_migration(self, data: dict[str, Any]) -> bool: + """Initiate ZHA migration. + + The passed data should contain: + - Discovery data identifying the device being firmware updated + - Discovery data for connecting to the device after the firmware update is + completed. + + Returns True if async_finish_migration should be called after the firmware + update is completed. + """ + migration_data = HARDWARE_MIGRATION_SCHEMA(data) + + name = migration_data["new_discovery_info"]["name"] + new_radio_type = ZhaRadioManager.parse_radio_type( + migration_data["new_discovery_info"]["radio_type"] + ) + old_radio_type = ZhaRadioManager.parse_radio_type( + migration_data["old_discovery_info"]["radio_type"] + ) + + new_device_settings = new_radio_type.controller.SCHEMA_DEVICE( + migration_data["new_discovery_info"]["port"] + ) + old_device_settings = old_radio_type.controller.SCHEMA_DEVICE( + migration_data["old_discovery_info"]["port"] + ) + + if ( + self._config_entry.data[CONF_DEVICE][CONF_DEVICE_PATH] + != old_device_settings[CONF_DEVICE_PATH] + ): + # ZHA is using another radio, do nothing + return False + + try: + await self._hass.config_entries.async_unload(self._config_entry.entry_id) + except config_entries.OperationNotAllowed: + # ZHA is not running + pass + + # Temporarily connect to the old radio to read its settings + old_radio_mgr = ZhaRadioManager() + old_radio_mgr.hass = self._hass + old_radio_mgr.radio_type = old_radio_type + old_radio_mgr.device_path = old_device_settings[CONF_DEVICE_PATH] + old_radio_mgr.device_settings = old_device_settings + backup = await old_radio_mgr.async_load_network_settings(create_backup=True) + + # Then configure the radio manager for the new radio to use the new settings + self._radio_mgr.chosen_backup = backup + self._radio_mgr.radio_type = new_radio_type + self._radio_mgr.device_path = new_device_settings[CONF_DEVICE_PATH] + self._radio_mgr.device_settings = new_device_settings + device_settings = self._radio_mgr.device_settings.copy() # type: ignore[union-attr] + + # Update the config entry settings + self._hass.config_entries.async_update_entry( + entry=self._config_entry, + data={ + CONF_DEVICE: device_settings, + CONF_RADIO_TYPE: self._radio_mgr.radio_type.name, + }, + options=self._config_entry.options, + title=name, + ) + return True + + async def async_finish_migration(self) -> None: + """Finish ZHA migration. + + Throws an exception if the migration did not succeed. + """ + # Restore the backup, permanently overwriting the device IEEE address + for retry in range(MIGRATION_RETRIES): + try: + if await self._radio_mgr.async_restore_backup_step_1(): + await self._radio_mgr.async_restore_backup_step_2(True) + + break + except OSError as err: + if retry >= MIGRATION_RETRIES - 1: + raise + + _LOGGER.debug( + "Failed to restore backup %r, retrying in %s seconds", + err, + CONNECT_DELAY_S, + ) + + await asyncio.sleep(CONNECT_DELAY_S) + + _LOGGER.debug("Restored backup after %s retries", retry) + + # Launch ZHA again + try: + await self._hass.config_entries.async_setup(self._config_entry.entry_id) + except config_entries.OperationNotAllowed: + # ZHA is not unloaded + pass diff --git a/tests/components/homeassistant_hardware/conftest.py b/tests/components/homeassistant_hardware/conftest.py index d46241b46e2..fd0ce2e761b 100644 --- a/tests/components/homeassistant_hardware/conftest.py +++ b/tests/components/homeassistant_hardware/conftest.py @@ -1,9 +1,37 @@ """Test fixtures for the Home Assistant Hardware integration.""" -from unittest.mock import patch +from collections.abc import Generator +from typing import Any +from unittest.mock import MagicMock, patch import pytest +@pytest.fixture(autouse=True) +def mock_zha_config_flow_setup() -> Generator[None, None, None]: + """Mock the radio connection and probing of the ZHA config flow.""" + + def mock_probe(config: dict[str, Any]) -> None: + # The radio probing will return the correct baudrate + return {**config, "baudrate": 115200} + + mock_connect_app = MagicMock() + mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()] + mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = ( + MagicMock() + ) + + with patch( + "bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe + ), patch( + "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app", + return_value=mock_connect_app, + ), patch( + "homeassistant.components.zha.async_setup_entry", + return_value=True, + ): + yield + + @pytest.fixture(name="addon_running") def mock_addon_running(addon_store_info, addon_info): """Mock add-on already running.""" diff --git a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py index 4b966033857..f61fd5e8d76 100644 --- a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py +++ b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py @@ -1,4 +1,4 @@ -"""Test the Home Assistant Yellow config flow.""" +"""Test the Home Assistant Hardware silabs multiprotocol addon manager.""" from __future__ import annotations from collections.abc import Generator @@ -10,6 +10,7 @@ import pytest from homeassistant import config_entries from homeassistant.components.hassio.handler import HassioAPIError from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon +from homeassistant.components.zha.core.const import DOMAIN as ZHA_DOMAIN from homeassistant.config_entries import ConfigEntry, ConfigFlow from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult, FlowResultType @@ -20,7 +21,7 @@ TEST_DOMAIN = "test" class TestConfigFlow(ConfigFlow, domain=TEST_DOMAIN): - """Handle a config flow for Home Assistant Yellow.""" + """Handle a config flow for the silabs multiprotocol add-on.""" VERSION = 1 @@ -37,11 +38,11 @@ class TestConfigFlow(ConfigFlow, domain=TEST_DOMAIN): if self._async_current_entries(): return self.async_abort(reason="single_instance_allowed") - return self.async_create_entry(title="Home Assistant Yellow", data={}) + return self.async_create_entry(title="Test HW", data={}) class TestOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler): - """Handle an option flow for Home Assistant Yellow.""" + """Handle an option flow for the silabs multiprotocol add-on.""" async def _async_serial_port_settings( self, @@ -53,6 +54,26 @@ class TestOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler): flow_control=True, ) + async def _async_zha_physical_discovery(self) -> dict[str, Any]: + """Return ZHA discovery data when multiprotocol FW is not used. + + Passed to ZHA do determine if the ZHA config entry is connected to the radio + being migrated. + """ + return { + "name": "Test", + "port": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + } + + def _zha_name(self) -> str: + """Return the ZHA name.""" + return "Test Multi-PAN" + @pytest.fixture(autouse=True) def config_flow_handler( @@ -80,7 +101,7 @@ async def test_option_flow_install_multi_pan_addon( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -132,6 +153,186 @@ async def test_option_flow_install_multi_pan_addon( assert result["type"] == FlowResultType.CREATE_ENTRY +async def test_option_flow_install_multi_pan_addon_zha( + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + set_addon_options, + start_addon, +) -> None: + """Test installing the multi pan addon when a zha config entry exists.""" + mock_integration(hass, MockModule("hassio")) + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + zha_config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=ZHA_DOMAIN, + options={}, + title="Test", + ) + zha_config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_not_installed" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + "enable_multi_pan": True, + }, + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_addon" + assert result["progress_action"] == "install_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "configure_addon" + install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_addon" + set_addon_options.assert_called_once_with( + hass, + "core_silabs_multiprotocol", + { + "options": { + "autoflash_firmware": True, + "device": "/dev/ttyTEST123", + "baudrate": "115200", + "flow_control": True, + } + }, + ) + # Check the ZHA config entry data is updated + assert zha_config_entry.data == { + "device": { + "path": "socket://core-silabs-multiprotocol:9999", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + assert zha_config_entry.title == "Test Multi-PAN" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "finish_addon_setup" + start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.CREATE_ENTRY + + +async def test_option_flow_install_multi_pan_addon_zha_other_radio( + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + set_addon_options, + start_addon, +) -> None: + """Test installing the multi pan addon when a zha config entry exists.""" + mock_integration(hass, MockModule("hassio")) + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + zha_config_entry = MockConfigEntry( + data={ + "device": { + "path": "/dev/other_radio", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + }, + domain=ZHA_DOMAIN, + options={}, + title="Test HW", + ) + zha_config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_not_installed" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + "enable_multi_pan": True, + }, + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_addon" + assert result["progress_action"] == "install_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "configure_addon" + install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + addon_info.return_value["hostname"] = "core-silabs-multiprotocol" + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_addon" + set_addon_options.assert_called_once_with( + hass, + "core_silabs_multiprotocol", + { + "options": { + "autoflash_firmware": True, + "device": "/dev/ttyTEST123", + "baudrate": "115200", + "flow_control": True, + } + }, + ) + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "finish_addon_setup" + start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.CREATE_ENTRY + + # Check the ZHA entry data is not changed + assert zha_config_entry.data == { + "device": { + "path": "/dev/other_radio", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + + async def test_option_flow_non_hassio( hass: HomeAssistant, ) -> None: @@ -143,7 +344,7 @@ async def test_option_flow_non_hassio( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -165,7 +366,7 @@ async def test_option_flow_addon_installed_other_device( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -196,7 +397,7 @@ async def test_option_flow_addon_installed_same_device( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -225,7 +426,7 @@ async def test_option_flow_do_not_install_multi_pan_addon( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -263,7 +464,7 @@ async def test_option_flow_install_multi_pan_addon_install_fails( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -312,7 +513,7 @@ async def test_option_flow_install_multi_pan_addon_start_fails( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -382,7 +583,7 @@ async def test_option_flow_install_multi_pan_addon_set_options_fails( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -428,7 +629,7 @@ async def test_option_flow_addon_info_fails( data={}, domain=TEST_DOMAIN, options={}, - title="Home Assistant Yellow", + title="Test HW", ) config_entry.add_to_hass(hass) @@ -439,3 +640,147 @@ async def test_option_flow_addon_info_fails( result = await hass.config_entries.options.async_init(config_entry.entry_id) assert result["type"] == FlowResultType.ABORT assert result["reason"] == "addon_info_failed" + + +@patch( + "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_initiate_migration", + side_effect=Exception("Boom!"), +) +async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_1( + mock_initiate_migration, + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + set_addon_options, + start_addon, +) -> None: + """Test installing the multi pan addon.""" + mock_integration(hass, MockModule("hassio")) + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + zha_config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=ZHA_DOMAIN, + options={}, + title="Test", + ) + zha_config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_not_installed" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + "enable_multi_pan": True, + }, + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_addon" + assert result["progress_action"] == "install_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "configure_addon" + install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "zha_migration_failed" + set_addon_options.assert_not_called() + + +@patch( + "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_finish_migration", + side_effect=Exception("Boom!"), +) +async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_2( + mock_finish_migration, + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + set_addon_options, + start_addon, +) -> None: + """Test installing the multi pan addon.""" + mock_integration(hass, MockModule("hassio")) + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + zha_config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=ZHA_DOMAIN, + options={}, + title="Test", + ) + zha_config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_not_installed" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + "enable_multi_pan": True, + }, + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_addon" + assert result["progress_action"] == "install_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "configure_addon" + install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_addon" + set_addon_options.assert_called_once_with( + hass, + "core_silabs_multiprotocol", + { + "options": { + "autoflash_firmware": True, + "device": "/dev/ttyTEST123", + "baudrate": "115200", + "flow_control": True, + } + }, + ) + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "finish_addon_setup" + start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "zha_migration_failed" diff --git a/tests/components/homeassistant_yellow/conftest.py b/tests/components/homeassistant_yellow/conftest.py index 23ef93ee8c6..62595c11fe1 100644 --- a/tests/components/homeassistant_yellow/conftest.py +++ b/tests/components/homeassistant_yellow/conftest.py @@ -15,7 +15,10 @@ def mock_zha_config_flow_setup() -> Generator[None, None, None]: return {**config, "baudrate": 115200} mock_connect_app = MagicMock() - mock_connect_app.__aenter__.return_value.backups.backups = [] + mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()] + mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = ( + MagicMock() + ) with patch( "bellows.zigbee.application.ControllerApplication.probe", side_effect=mock_probe diff --git a/tests/components/homeassistant_yellow/test_config_flow.py b/tests/components/homeassistant_yellow/test_config_flow.py index 91f15f792d5..a956f812d8a 100644 --- a/tests/components/homeassistant_yellow/test_config_flow.py +++ b/tests/components/homeassistant_yellow/test_config_flow.py @@ -2,6 +2,7 @@ from unittest.mock import Mock, patch from homeassistant.components.homeassistant_yellow.const import DOMAIN +from homeassistant.components.zha.core.const import DOMAIN as ZHA_DOMAIN from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType @@ -124,3 +125,88 @@ async def test_option_flow_install_multi_pan_addon( result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] == FlowResultType.CREATE_ENTRY + + +async def test_option_flow_install_multi_pan_addon_zha( + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + set_addon_options, + start_addon, +) -> None: + """Test installing the multi pan addon when a zha config entry exists.""" + mock_integration(hass, MockModule("hassio")) + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=DOMAIN, + options={}, + title="Home Assistant Yellow", + ) + config_entry.add_to_hass(hass) + + zha_config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyAMA1"}, "radio_type": "ezsp"}, + domain=ZHA_DOMAIN, + options={}, + title="Yellow", + ) + zha_config_entry.add_to_hass(hass) + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + side_effect=Mock(return_value=True), + ): + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_not_installed" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={ + "enable_multi_pan": True, + }, + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_addon" + assert result["progress_action"] == "install_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "configure_addon" + install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_addon" + set_addon_options.assert_called_once_with( + hass, + "core_silabs_multiprotocol", + { + "options": { + "autoflash_firmware": True, + "device": "/dev/ttyAMA1", + "baudrate": "115200", + "flow_control": True, + } + }, + ) + # Check the ZHA config entry data is updated + assert zha_config_entry.data == { + "device": { + "path": "socket://core-silabs-multiprotocol:9999", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "finish_addon_setup" + start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.CREATE_ENTRY diff --git a/tests/components/homeassistant_yellow/test_init.py b/tests/components/homeassistant_yellow/test_init.py index 06fe9465d2d..6df8977b35d 100644 --- a/tests/components/homeassistant_yellow/test_init.py +++ b/tests/components/homeassistant_yellow/test_init.py @@ -149,7 +149,7 @@ async def test_setup_zha_multipan( "radio_type": "ezsp", } assert config_entry.options == {} - assert config_entry.title == "Yellow" + assert config_entry.title == "Yellow Multi-PAN" async def test_setup_entry_wrong_board(hass: HomeAssistant) -> None: diff --git a/tests/components/zha/test_config_flow.py b/tests/components/zha/test_config_flow.py index d2001dfda1b..d457e0b6b8c 100644 --- a/tests/components/zha/test_config_flow.py +++ b/tests/components/zha/test_config_flow.py @@ -16,7 +16,7 @@ import zigpy.types from homeassistant import config_entries from homeassistant.components import ssdp, usb, zeroconf from homeassistant.components.ssdp import ATTR_UPNP_MANUFACTURER_URL, ATTR_UPNP_SERIAL -from homeassistant.components.zha import config_flow +from homeassistant.components.zha import config_flow, radio_manager from homeassistant.components.zha.core.const import ( CONF_BAUDRATE, CONF_FLOWCONTROL, @@ -1024,7 +1024,7 @@ async def test_hardware_invalid_data(hass, 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) + new_backup = radio_manager._allow_overwrite_ezsp_ieee(backup) assert backup != new_backup assert new_backup.network_info.stack_specific["ezsp"][EZSP_OVERWRITE_EUI64] is True @@ -1034,7 +1034,7 @@ 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"] = {EZSP_OVERWRITE_EUI64: True} - new_backup = config_flow._prevent_overwrite_ezsp_ieee(backup) + new_backup = radio_manager._prevent_overwrite_ezsp_ieee(backup) assert backup != new_backup assert not new_backup.network_info.stack_specific.get("ezsp", {}).get( @@ -1131,7 +1131,7 @@ def test_parse_uploaded_backup(process_mock): assert backup == parsed_backup -@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee") async def test_formation_strategy_restore_manual_backup_non_ezsp( allow_overwrite_ieee_mock, pick_radio, mock_app, hass ): @@ -1163,7 +1163,7 @@ async def test_formation_strategy_restore_manual_backup_non_ezsp( assert result3["data"][CONF_RADIO_TYPE] == "znp" -@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +@patch("homeassistant.components.zha.radio_manager._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 ): @@ -1203,7 +1203,7 @@ async def test_formation_strategy_restore_manual_backup_overwrite_ieee_ezsp( assert result4["data"][CONF_RADIO_TYPE] == "ezsp" -@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee") async def test_formation_strategy_restore_manual_backup_ezsp( allow_overwrite_ieee_mock, pick_radio, mock_app, hass ): @@ -1391,7 +1391,7 @@ async def test_formation_strategy_restore_automatic_backup_non_ezsp( assert result3["data"][CONF_RADIO_TYPE] == "znp" -@patch("homeassistant.components.zha.config_flow._allow_overwrite_ezsp_ieee") +@patch("homeassistant.components.zha.radio_manager._allow_overwrite_ezsp_ieee") async def test_ezsp_restore_without_settings_change_ieee( allow_overwrite_ieee_mock, pick_radio, mock_app, backup, hass ): diff --git a/tests/components/zha/test_radio_manager.py b/tests/components/zha/test_radio_manager.py new file mode 100644 index 00000000000..505118524b6 --- /dev/null +++ b/tests/components/zha/test_radio_manager.py @@ -0,0 +1,311 @@ +"""Tests for ZHA config flow.""" + +from collections.abc import Generator +from unittest.mock import AsyncMock, MagicMock, create_autospec, patch + +import pytest +import serial.tools.list_ports +from zigpy.backups import BackupManager +import zigpy.config +from zigpy.config import CONF_DEVICE_PATH +import zigpy.types + +from homeassistant import config_entries +from homeassistant.components.zha import radio_manager +from homeassistant.components.zha.core.const import DOMAIN, RadioType +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + +PROBE_FUNCTION_PATH = "zigbee.application.ControllerApplication.probe" + + +@pytest.fixture(autouse=True) +def disable_platform_only(): + """Disable platforms to speed up tests.""" + with patch("homeassistant.components.zha.PLATFORMS", []): + yield + + +@pytest.fixture(autouse=True) +def reduce_reconnect_timeout(): + """Reduces reconnect timeout to speed up tests.""" + with patch("homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001): + yield + + +@pytest.fixture(autouse=True) +def mock_app(): + """Mock zigpy app interface.""" + mock_app = AsyncMock() + mock_app.backups = create_autospec(BackupManager, instance=True) + 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 = device + port.description = "Some serial port" + + return port + + +@pytest.fixture() +def mock_connect_zigpy_app() -> Generator[None, None, None]: + """Mock the radio connection.""" + + mock_connect_app = MagicMock() + mock_connect_app.__aenter__.return_value.backups.backups = [MagicMock()] + mock_connect_app.__aenter__.return_value.backups.create_backup.return_value = ( + MagicMock() + ) + + with patch( + "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app", + return_value=mock_connect_app, + ): + yield + + +@patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) +async def test_migrate_matching_port( + hass: HomeAssistant, + mock_connect_zigpy_app, +) -> None: + """Test automatic migration.""" + # Setup the config entry + config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=DOMAIN, + options={}, + title="Test", + version=3, + ) + config_entry.add_to_hass(hass) + + migration_data = { + "new_discovery_info": { + "name": "Test Updated", + "port": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + "old_discovery_info": { + "name": "Test", + "port": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + } + + migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry) + assert await migration_helper.async_initiate_migration(migration_data) + + # Check the ZHA config entry data is updated + assert config_entry.data == { + "device": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + assert config_entry.title == "Test Updated" + + await migration_helper.async_finish_migration() + + +async def test_migrate_matching_port_config_entry_not_loaded( + hass: HomeAssistant, + mock_connect_zigpy_app, +) -> None: + """Test automatic migration.""" + # Setup the config entry + config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=DOMAIN, + options={}, + title="Test", + ) + config_entry.add_to_hass(hass) + config_entry.state = config_entries.ConfigEntryState.SETUP_IN_PROGRESS + + migration_data = { + "new_discovery_info": { + "name": "Test Updated", + "port": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + "old_discovery_info": { + "name": "Test", + "port": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + } + + migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry) + assert await migration_helper.async_initiate_migration(migration_data) + + # Check the ZHA config entry data is updated + assert config_entry.data == { + "device": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + assert config_entry.title == "Test Updated" + + await migration_helper.async_finish_migration() + + +@patch( + "homeassistant.components.zha.radio_manager.ZhaRadioManager.async_restore_backup_step_1", + side_effect=OSError, +) +async def test_migrate_matching_port_retry( + mock_restore_backup_step_1, + hass: HomeAssistant, + mock_connect_zigpy_app, +) -> None: + """Test automatic migration.""" + # Setup the config entry + config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=DOMAIN, + options={}, + title="Test", + ) + config_entry.add_to_hass(hass) + config_entry.state = config_entries.ConfigEntryState.SETUP_IN_PROGRESS + + migration_data = { + "new_discovery_info": { + "name": "Test Updated", + "port": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + "old_discovery_info": { + "name": "Test", + "port": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + } + + migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry) + assert await migration_helper.async_initiate_migration(migration_data) + + # Check the ZHA config entry data is updated + assert config_entry.data == { + "device": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + assert config_entry.title == "Test Updated" + + with pytest.raises(OSError): + await migration_helper.async_finish_migration() + assert mock_restore_backup_step_1.call_count == 100 + + +async def test_migrate_non_matching_port( + hass: HomeAssistant, + mock_connect_zigpy_app, +) -> None: + """Test automatic migration.""" + # Setup the config entry + config_entry = MockConfigEntry( + data={"device": {"path": "/dev/ttyTEST123"}, "radio_type": "ezsp"}, + domain=DOMAIN, + options={}, + title="Test", + ) + config_entry.add_to_hass(hass) + + migration_data = { + "new_discovery_info": { + "name": "Test Updated", + "port": { + "path": "socket://some/virtual_port", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + "old_discovery_info": { + "name": "Test", + "port": { + "path": "/dev/ttyTEST456", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + }, + } + + migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry) + assert not await migration_helper.async_initiate_migration(migration_data) + + # Check the ZHA config entry data is not updated + assert config_entry.data == { + "device": {"path": "/dev/ttyTEST123"}, + "radio_type": "ezsp", + } + assert config_entry.title == "Test"