From c8ef3f9393c1db4df77cc6f8a674b0e44b691720 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Mon, 28 Aug 2023 17:26:34 -0400 Subject: [PATCH] Automatic migration from multi-PAN back to Zigbee firmware (#93831) * Initial implementation of migration back to Zigbee firmware * Fix typo in `BACKUP_RETRIES` constant name * Name potentially long-running tasks * Add an explicit timeout to `_async_wait_until_addon_state` * Guard against the addon not being installed when uninstalling * Do not launch the progress flow unless the addon is being installed * Use a separate translation key for confirmation before disabling multi-PAN * Disable the bellows UART thread within the ZHA config flow radio manager * Enhance config flow progress keys for flasher addon installation * Allow `zha.async_unload_entry` to succeed when ZHA is not loaded * Do not endlessly spawn task when uninstalling addon synchronously * Include `uninstall_addon.data.*` in SkyConnect and Yellow translations * Make `homeassistant_hardware` unit tests pass * Fix SkyConnect unit test USB mock * Fix unit tests in related integrations * Use a separate constant for connection retrying * Unit test ZHA migration from multi-PAN * Test ZHA multi-PAN migration helper changes * Fix flaky SkyConnect unit test being affected by system USB devices * Unit test the synchronous addon uninstall helper * Test failure when flasher addon is already running * Test failure where flasher addon fails to install * Test ZHA migration failures * Rename `get_addon_manager` to `get_multiprotocol_addon_manager` * Remove stray "addon uninstall" comment * Use better variable names for the two addon managers * Remove extraneous `self.install_task = None` * Use the addon manager's `addon_name` instead of constants * Migrate synchronous addon operations into a new class * Remove wrapper functions with `finally` clause * Use a more descriptive error message when the flasher addon is stalled * Fix existing unit tests * Remove `wait_until_done` * Fully replace all addon name constants with those from managers * Fix OTBR breakage * Simplify `is_hassio` mocking * Add missing tests for `check_multi_pan_addon` * Add missing tests for `multi_pan_addon_using_device` * Use `waiting` instead of `sync` in class name and methods --- .../homeassistant_hardware/const.py | 1 + .../silabs_multiprotocol_addon.py | 423 ++++++-- .../homeassistant_hardware/strings.json | 33 +- .../homeassistant_sky_connect/strings.json | 21 +- .../homeassistant_yellow/strings.json | 21 +- homeassistant/components/otbr/util.py | 8 +- homeassistant/components/zha/__init__.py | 6 +- homeassistant/components/zha/config_flow.py | 6 +- homeassistant/components/zha/radio_manager.py | 27 +- .../homeassistant_hardware/conftest.py | 18 + .../test_silabs_multiprotocol_addon.py | 965 +++++++++++++++--- .../homeassistant_sky_connect/conftest.py | 20 +- .../test_config_flow.py | 12 + .../homeassistant_sky_connect/test_init.py | 7 + .../homeassistant_yellow/test_config_flow.py | 10 + tests/components/zha/test_radio_manager.py | 57 +- 16 files changed, 1383 insertions(+), 252 deletions(-) diff --git a/homeassistant/components/homeassistant_hardware/const.py b/homeassistant/components/homeassistant_hardware/const.py index dd3a254d097..e4aa7c80f8d 100644 --- a/homeassistant/components/homeassistant_hardware/const.py +++ b/homeassistant/components/homeassistant_hardware/const.py @@ -5,3 +5,4 @@ import logging LOGGER = logging.getLogger(__package__) SILABS_MULTIPROTOCOL_ADDON_SLUG = "core_silabs_multiprotocol" +SILABS_FLASHER_ADDON_SLUG = "core_silabs_flasher" diff --git a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py index e4d9902346c..b4723a88742 100644 --- a/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py +++ b/homeassistant/components/homeassistant_hardware/silabs_multiprotocol_addon.py @@ -3,10 +3,12 @@ from __future__ import annotations from abc import ABC, abstractmethod import asyncio +from collections.abc import Awaitable import dataclasses import logging from typing import Any, Protocol +import async_timeout import voluptuous as vol import yarl @@ -33,17 +35,19 @@ from homeassistant.helpers.selector import ( from homeassistant.helpers.singleton import singleton from homeassistant.helpers.storage import Store -from .const import LOGGER, SILABS_MULTIPROTOCOL_ADDON_SLUG +from .const import LOGGER, SILABS_FLASHER_ADDON_SLUG, SILABS_MULTIPROTOCOL_ADDON_SLUG _LOGGER = logging.getLogger(__name__) -DATA_ADDON_MANAGER = "silabs_multiprotocol_addon_manager" +DATA_MULTIPROTOCOL_ADDON_MANAGER = "silabs_multiprotocol_addon_manager" +DATA_FLASHER_ADDON_MANAGER = "silabs_flasher" -ADDON_SETUP_TIMEOUT = 5 -ADDON_SETUP_TIMEOUT_ROUNDS = 40 +ADDON_STATE_POLL_INTERVAL = 3 +ADDON_INFO_POLL_TIMEOUT = 15 * 60 CONF_ADDON_AUTOFLASH_FW = "autoflash_firmware" CONF_ADDON_DEVICE = "device" +CONF_DISABLE_MULTI_PAN = "disable_multi_pan" CONF_ENABLE_MULTI_PAN = "enable_multi_pan" DEFAULT_CHANNEL = 15 @@ -55,15 +59,64 @@ STORAGE_VERSION_MINOR = 1 SAVE_DELAY = 10 -@singleton(DATA_ADDON_MANAGER) -async def get_addon_manager(hass: HomeAssistant) -> MultiprotocolAddonManager: +@singleton(DATA_MULTIPROTOCOL_ADDON_MANAGER) +async def get_multiprotocol_addon_manager( + hass: HomeAssistant, +) -> MultiprotocolAddonManager: """Get the add-on manager.""" manager = MultiprotocolAddonManager(hass) await manager.async_setup() return manager -class MultiprotocolAddonManager(AddonManager): +class WaitingAddonManager(AddonManager): + """Addon manager which supports waiting operations for managing an addon.""" + + async def async_wait_until_addon_state(self, *states: AddonState) -> None: + """Poll an addon's info until it is in a specific state.""" + async with async_timeout.timeout(ADDON_INFO_POLL_TIMEOUT): + while True: + try: + info = await self.async_get_addon_info() + except AddonError: + info = None + + _LOGGER.debug("Waiting for addon to be in state %s: %s", states, info) + + if info is not None and info.state in states: + break + + await asyncio.sleep(ADDON_STATE_POLL_INTERVAL) + + async def async_start_addon_waiting(self) -> None: + """Start an add-on.""" + await self.async_schedule_start_addon() + await self.async_wait_until_addon_state(AddonState.RUNNING) + + async def async_install_addon_waiting(self) -> None: + """Install an add-on.""" + await self.async_schedule_install_addon() + await self.async_wait_until_addon_state( + AddonState.RUNNING, + AddonState.NOT_RUNNING, + ) + + async def async_uninstall_addon_waiting(self) -> None: + """Uninstall an add-on.""" + try: + info = await self.async_get_addon_info() + except AddonError: + info = None + + # Do not try to uninstall an addon if it is already uninstalled + if info is not None and info.state == AddonState.NOT_INSTALLED: + return + + await self.async_uninstall_addon() + await self.async_wait_until_addon_state(AddonState.NOT_INSTALLED) + + +class MultiprotocolAddonManager(WaitingAddonManager): """Silicon Labs Multiprotocol add-on manager.""" def __init__(self, hass: HomeAssistant) -> None: @@ -207,6 +260,18 @@ class MultipanProtocol(Protocol): """ +@singleton(DATA_FLASHER_ADDON_MANAGER) +@callback +def get_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager: + """Get the flasher add-on manager.""" + return WaitingAddonManager( + hass, + LOGGER, + "Silicon Labs Flasher", + SILABS_FLASHER_ADDON_SLUG, + ) + + @dataclasses.dataclass class SerialPortSettings: """Serial port settings.""" @@ -242,9 +307,9 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): ZhaMultiPANMigrationHelper, ) - # 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.stop_task: asyncio.Task | None = None self._zha_migration_mgr: ZhaMultiPANMigrationHelper | None = None self.config_entry = config_entry self.original_addon_config: dict[str, Any] | None = None @@ -275,37 +340,37 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): """Return the correct flow manager.""" return self.hass.config_entries.options - async def _async_get_addon_info(self) -> AddonInfo: + async def _resume_flow_when_done(self, awaitable: Awaitable) -> None: + try: + await awaitable + finally: + self.hass.async_create_task( + self.flow_manager.async_configure(flow_id=self.flow_id) + ) + + async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo: """Return and cache Silicon Labs Multiprotocol add-on info.""" - addon_manager: AddonManager = await get_addon_manager(self.hass) try: addon_info: AddonInfo = await addon_manager.async_get_addon_info() except AddonError as err: _LOGGER.error(err) - raise AbortFlow("addon_info_failed") from err + raise AbortFlow( + "addon_info_failed", + description_placeholders={"addon_name": addon_manager.addon_name}, + ) from err return addon_info - async def _async_set_addon_config(self, config: dict) -> None: + async def _async_set_addon_config( + self, config: dict, addon_manager: AddonManager + ) -> None: """Set Silicon Labs Multiprotocol add-on config.""" - addon_manager: AddonManager = await get_addon_manager(self.hass) try: await addon_manager.async_set_addon_options(config) except AddonError as err: _LOGGER.error(err) raise AbortFlow("addon_set_config_failed") from err - async def _async_install_addon(self) -> None: - """Install the Silicon Labs Multiprotocol add-on.""" - addon_manager: AddonManager = await get_addon_manager(self.hass) - try: - await addon_manager.async_schedule_install_addon() - finally: - # Continue the flow after show progress when the task is done. - self.hass.async_create_task( - self.flow_manager.async_configure(flow_id=self.flow_id) - ) - async def async_step_init( self, user_input: dict[str, Any] | None = None ) -> FlowResult: @@ -319,7 +384,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Handle logic when on Supervisor host.""" - addon_info = await self._async_get_addon_info() + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(multipan_manager) if addon_info.state == AddonState.NOT_INSTALLED: return await self.async_step_addon_not_installed() @@ -347,19 +413,26 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): ) -> FlowResult: """Install Silicon Labs Multiprotocol add-on.""" if not self.install_task: - self.install_task = self.hass.async_create_task(self._async_install_addon()) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + self.install_task = self.hass.async_create_task( + self._resume_flow_when_done( + multipan_manager.async_install_addon_waiting() + ), + "SiLabs Multiprotocol addon install", + ) return self.async_show_progress( - step_id="install_addon", progress_action="install_addon" + step_id="install_addon", + progress_action="install_addon", + description_placeholders={"addon_name": multipan_manager.addon_name}, ) try: await self.install_task except AddonError as err: - self.install_task = None _LOGGER.error(err) return self.async_show_progress_done(next_step_id="install_failed") - - self.install_task = None + finally: + self.install_task = None return self.async_show_progress_done(next_step_id="configure_addon") @@ -367,7 +440,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Add-on installation failed.""" - return self.async_abort(reason="addon_install_failed") + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + return self.async_abort( + reason="addon_install_failed", + description_placeholders={"addon_name": multipan_manager.addon_name}, + ) async def async_step_configure_addon( self, user_input: dict[str, Any] | None = None @@ -386,7 +463,8 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async_get_channel as async_get_zha_channel, ) - addon_info = await self._async_get_addon_info() + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(multipan_manager) addon_config = addon_info.options @@ -426,14 +504,14 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): multipan_channel = zha_channel # Initialize the shared channel - multipan_manager = await get_addon_manager(self.hass) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) multipan_manager.async_set_channel(multipan_channel) if new_addon_config != addon_config: # Copy the add-on config to keep the objects separate. self.original_addon_config = dict(addon_config) _LOGGER.debug("Reconfiguring addon with %s", new_addon_config) - await self._async_set_addon_config(new_addon_config) + await self._async_set_addon_config(new_addon_config, multipan_manager) return await self.async_step_start_addon() @@ -442,9 +520,16 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): ) -> FlowResult: """Start Silicon Labs Multiprotocol add-on.""" if not self.start_task: - self.start_task = self.hass.async_create_task(self._async_start_addon()) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + self.start_task = self.hass.async_create_task( + self._resume_flow_when_done( + multipan_manager.async_start_addon_waiting() + ) + ) return self.async_show_progress( - step_id="start_addon", progress_action="start_addon" + step_id="start_addon", + progress_action="start_addon", + description_placeholders={"addon_name": multipan_manager.addon_name}, ) try: @@ -461,18 +546,11 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Add-on start failed.""" - return self.async_abort(reason="addon_start_failed") - - async def _async_start_addon(self) -> None: - """Start Silicon Labs Multiprotocol add-on.""" - addon_manager: AddonManager = await get_addon_manager(self.hass) - try: - await addon_manager.async_schedule_start_addon() - finally: - # Continue the flow after show progress when the task is done. - self.hass.async_create_task( - self.flow_manager.async_configure(flow_id=self.flow_id) - ) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + return self.async_abort( + reason="addon_start_failed", + description_placeholders={"addon_name": multipan_manager.addon_name}, + ) async def async_step_finish_addon_setup( self, user_input: dict[str, Any] | None = None @@ -493,16 +571,25 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): return self.async_create_entry(title="", data={}) + async def async_step_addon_installed_other_device( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Show dialog explaining the addon is in use by another device.""" + if user_input is None: + return self.async_show_form(step_id="addon_installed_other_device") + return self.async_create_entry(title="", data={}) + async def async_step_addon_installed( self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Handle logic when the addon is already installed.""" - addon_info = await self._async_get_addon_info() + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(multipan_manager) serial_device = (await self._async_serial_port_settings()).device - if addon_info.options.get(CONF_ADDON_DEVICE) == serial_device: - return await self.async_step_show_addon_menu() - return await self.async_step_addon_installed_other_device() + if addon_info.options.get(CONF_ADDON_DEVICE) != serial_device: + return await self.async_step_addon_installed_other_device() + return await self.async_step_show_addon_menu() async def async_step_show_addon_menu( self, user_input: dict[str, Any] | None = None @@ -520,7 +607,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Reconfigure the addon.""" - multipan_manager = await get_addon_manager(self.hass) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) active_platforms = await multipan_manager.async_active_platforms() if set(active_platforms) != {"otbr", "zha"}: return await self.async_step_notify_unknown_multipan_user() @@ -540,7 +627,7 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): self, user_input: dict[str, Any] | None = None ) -> FlowResult: """Change the channel.""" - multipan_manager = await get_addon_manager(self.hass) + multipan_manager = await get_multiprotocol_addon_manager(self.hass) if user_input is None: channels = [str(x) for x in range(11, 27)] suggested_channel = DEFAULT_CHANNEL @@ -584,23 +671,217 @@ class OptionsFlowHandler(config_entries.OptionsFlow, ABC): async def async_step_uninstall_addon( self, user_input: dict[str, Any] | None = None ) -> FlowResult: - """Uninstall the addon (not implemented).""" - return await self.async_step_show_revert_guide() + """Uninstall the addon and revert the firmware.""" + if user_input is None: + return self.async_show_form( + step_id="uninstall_addon", + data_schema=vol.Schema( + {vol.Required(CONF_DISABLE_MULTI_PAN, default=False): bool} + ), + description_placeholders={"hardware_name": self._hardware_name()}, + ) + if not user_input[CONF_DISABLE_MULTI_PAN]: + return self.async_create_entry(title="", data={}) - async def async_step_show_revert_guide( + return await self.async_step_firmware_revert() + + async def async_step_firmware_revert( self, user_input: dict[str, Any] | None = None ) -> FlowResult: - """Link to a guide for reverting to Zigbee firmware.""" - if user_input is None: - return self.async_show_form(step_id="show_revert_guide") - return self.async_create_entry(title="", data={}) + """Install the flasher addon, if necessary.""" - async def async_step_addon_installed_other_device( + flasher_manager = get_flasher_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(flasher_manager) + + if addon_info.state == AddonState.NOT_INSTALLED: + return await self.async_step_install_flasher_addon() + + if addon_info.state == AddonState.NOT_RUNNING: + return await self.async_step_configure_flasher_addon() + + # If the addon is already installed and running, fail + return self.async_abort( + reason="addon_already_running", + description_placeholders={"addon_name": flasher_manager.addon_name}, + ) + + async def async_step_install_flasher_addon( self, user_input: dict[str, Any] | None = None ) -> FlowResult: - """Show dialog explaining the addon is in use by another device.""" - if user_input is None: - return self.async_show_form(step_id="addon_installed_other_device") + """Show progress dialog for installing flasher addon.""" + flasher_manager = get_flasher_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(flasher_manager) + + _LOGGER.debug("Flasher addon state: %s", addon_info) + + if not self.install_task: + self.install_task = self.hass.async_create_task( + self._resume_flow_when_done( + flasher_manager.async_install_addon_waiting() + ), + "SiLabs Flasher addon install", + ) + return self.async_show_progress( + step_id="install_flasher_addon", + progress_action="install_addon", + description_placeholders={"addon_name": flasher_manager.addon_name}, + ) + + try: + await self.install_task + except AddonError as err: + _LOGGER.error(err) + return self.async_show_progress_done(next_step_id="install_failed") + finally: + self.install_task = None + + return self.async_show_progress_done(next_step_id="configure_flasher_addon") + + async def async_step_configure_flasher_addon( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Perform initial backup and reconfigure ZHA.""" + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha import DOMAIN as ZHA_DOMAIN + + # pylint: disable-next=import-outside-toplevel + from homeassistant.components.zha.radio_manager import ( + ZhaMultiPANMigrationHelper, + ) + + zha_entries = self.hass.config_entries.async_entries(ZHA_DOMAIN) + new_settings = await self._async_serial_port_settings() + + _LOGGER.debug("Using new ZHA settings: %s", new_settings) + + if zha_entries: + zha_migration_mgr = ZhaMultiPANMigrationHelper(self.hass, zha_entries[0]) + migration_data = { + "new_discovery_info": { + "name": self._hardware_name(), + "port": { + "path": new_settings.device, + "baudrate": int(new_settings.baudrate), + "flow_control": ( + "hardware" if new_settings.flow_control else None + ), + }, + "radio_type": "ezsp", + }, + "old_discovery_info": { + "hw": { + "name": self._zha_name(), + "port": {"path": get_zigbee_socket()}, + "radio_type": "ezsp", + } + }, + } + _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 + + flasher_manager = get_flasher_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(flasher_manager) + new_addon_config = { + **addon_info.options, + "device": new_settings.device, + "flow_control": new_settings.flow_control, + } + + _LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config) + await self._async_set_addon_config(new_addon_config, flasher_manager) + + return await self.async_step_uninstall_multiprotocol_addon() + + async def async_step_uninstall_multiprotocol_addon( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Uninstall Silicon Labs Multiprotocol add-on.""" + + if not self.stop_task: + multipan_manager = await get_multiprotocol_addon_manager(self.hass) + self.stop_task = self.hass.async_create_task( + self._resume_flow_when_done( + multipan_manager.async_uninstall_addon_waiting() + ), + "SiLabs Multiprotocol addon uninstall", + ) + return self.async_show_progress( + step_id="uninstall_multiprotocol_addon", + progress_action="uninstall_multiprotocol_addon", + description_placeholders={"addon_name": multipan_manager.addon_name}, + ) + + try: + await self.stop_task + finally: + self.stop_task = None + + return self.async_show_progress_done(next_step_id="start_flasher_addon") + + async def async_step_start_flasher_addon( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Start Silicon Labs Flasher add-on.""" + + if not self.start_task: + flasher_manager = get_flasher_addon_manager(self.hass) + + async def start_and_wait_until_done() -> None: + await flasher_manager.async_start_addon_waiting() + # Now that the addon is running, wait for it to finish + await flasher_manager.async_wait_until_addon_state( + AddonState.NOT_RUNNING + ) + + self.start_task = self.hass.async_create_task( + self._resume_flow_when_done(start_and_wait_until_done()) + ) + return self.async_show_progress( + step_id="start_flasher_addon", + progress_action="start_flasher_addon", + description_placeholders={"addon_name": flasher_manager.addon_name}, + ) + + try: + await self.start_task + except (AddonError, AbortFlow) as err: + _LOGGER.error(err) + return self.async_show_progress_done(next_step_id="flasher_failed") + finally: + self.start_task = None + + return self.async_show_progress_done(next_step_id="flashing_complete") + + async def async_step_flasher_failed( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Flasher add-on start failed.""" + flasher_manager = get_flasher_addon_manager(self.hass) + return self.async_abort( + reason="addon_start_failed", + description_placeholders={"addon_name": flasher_manager.addon_name}, + ) + + async def async_step_flashing_complete( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: + """Finish flashing and update the config entry.""" + flasher_manager = get_flasher_addon_manager(self.hass) + await flasher_manager.async_uninstall_addon_waiting() + + # 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={}) @@ -613,18 +894,18 @@ async def check_multi_pan_addon(hass: HomeAssistant) -> None: if not is_hassio(hass): return - addon_manager: AddonManager = await get_addon_manager(hass) + multipan_manager = await get_multiprotocol_addon_manager(hass) try: - addon_info: AddonInfo = await addon_manager.async_get_addon_info() + addon_info: AddonInfo = await multipan_manager.async_get_addon_info() except AddonError as err: _LOGGER.error(err) raise HomeAssistantError from err # Request the addon to start if it's not started - # addon_manager.async_start_addon returns as soon as the start request has been sent + # `async_start_addon` returns as soon as the start request has been sent # and does not wait for the addon to be started, so we raise below if addon_info.state == AddonState.NOT_RUNNING: - await addon_manager.async_start_addon() + await multipan_manager.async_start_addon() if addon_info.state not in (AddonState.NOT_INSTALLED, AddonState.RUNNING): _LOGGER.debug("Multi pan addon installed and in state %s", addon_info.state) @@ -640,8 +921,8 @@ async def multi_pan_addon_using_device(hass: HomeAssistant, device_path: str) -> if not is_hassio(hass): return False - addon_manager: AddonManager = await get_addon_manager(hass) - addon_info: AddonInfo = await addon_manager.async_get_addon_info() + multipan_manager = await get_multiprotocol_addon_manager(hass) + addon_info: AddonInfo = await multipan_manager.async_get_addon_info() if addon_info.state != AddonState.RUNNING: return False diff --git a/homeassistant/components/homeassistant_hardware/strings.json b/homeassistant/components/homeassistant_hardware/strings.json index 24cd049668d..a66e4879f68 100644 --- a/homeassistant/components/homeassistant_hardware/strings.json +++ b/homeassistant/components/homeassistant_hardware/strings.json @@ -39,31 +39,42 @@ "reconfigure_addon": { "title": "Reconfigure IEEE 802.15.4 radio multiprotocol support" }, - "show_revert_guide": { - "title": "Multiprotocol support is enabled for this device", - "description": "If you want to change to Zigbee only firmware, please complete the following manual steps:\n\n * Remove the Silicon Labs Multiprotocol addon\n\n * Flash the Zigbee only firmware, follow the guide at https://github.com/NabuCasa/silabs-firmware/wiki/Flash-Silicon-Labs-radio-firmware-manually.\n\n * Reconfigure ZHA to migrate settings to the reflashed radio" - }, "start_addon": { "title": "The Silicon Labs Multiprotocol add-on is starting." }, "uninstall_addon": { - "title": "Remove IEEE 802.15.4 radio multiprotocol support." + "title": "Remove IEEE 802.15.4 radio multiprotocol support", + "description": "Disabling multiprotocol support will revert your {hardware_name}'s radio back to Zigbee-only firmware and will disable Thread support provided by the {hardware_name}. Your Thread devices will continue working only if you have another Thread border router nearby.\n\nIt will take a few minutes to install the Zigbee firmware and restoring a backup.", + "data": { + "disable_multi_pan": "Disable multiprotocol support" + } + }, + "install_flasher_addon": { + "title": "The Silicon Labs Flasher add-on installation has started" + }, + "configure_flasher_addon": { + "title": "The Silicon Labs Flasher add-on installation has started" + }, + "start_flasher_addon": { + "title": "Installing firmware", + "description": "Zigbee firmware is now being installed. This will take a few minutes." } }, "error": { "unknown": "[%key:common::config_flow::error::unknown%]" }, "abort": { - "addon_info_failed": "Failed to get Silicon Labs Multiprotocol add-on info.", - "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.", + "addon_info_failed": "Failed to get {addon_name} add-on info.", + "addon_install_failed": "Failed to install the {addon_name} add-on.", + "addon_already_running": "Failed to start the {addon_name} add-on because it is already running.", + "addon_set_config_failed": "Failed to set {addon_name} configuration.", + "addon_start_failed": "Failed to start the {addon_name} add-on.", "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.", - "start_addon": "Please wait while the Silicon Labs Multiprotocol add-on start completes. This may take some seconds." + "install_addon": "Please wait while the {addon_name} add-on installation finishes. This can take several minutes.", + "start_addon": "Please wait while the {addon_name} add-on start completes. This may take some seconds." } } } diff --git a/homeassistant/components/homeassistant_sky_connect/strings.json b/homeassistant/components/homeassistant_sky_connect/strings.json index 9bc1a49125b..58fc0180743 100644 --- a/homeassistant/components/homeassistant_sky_connect/strings.json +++ b/homeassistant/components/homeassistant_sky_connect/strings.json @@ -38,15 +38,25 @@ "reconfigure_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]" }, - "show_revert_guide": { - "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]", - "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]" - }, "start_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" }, "uninstall_addon": { - "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::description%]", + "data": { + "disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]" + } + }, + "install_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]" + }, + "configure_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]" + }, + "start_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]" } }, "error": { @@ -55,6 +65,7 @@ "abort": { "addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]", "addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]", + "addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]", "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", diff --git a/homeassistant/components/homeassistant_yellow/strings.json b/homeassistant/components/homeassistant_yellow/strings.json index 644a3c04553..e5250f163ce 100644 --- a/homeassistant/components/homeassistant_yellow/strings.json +++ b/homeassistant/components/homeassistant_yellow/strings.json @@ -60,15 +60,25 @@ "reconfigure_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::reconfigure_addon::title%]" }, - "show_revert_guide": { - "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::title%]", - "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::show_revert_guide::description%]" - }, "start_addon": { "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_addon::title%]" }, "uninstall_addon": { - "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]" + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::description%]", + "data": { + "disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]" + } + }, + "install_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]" + }, + "configure_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]" + }, + "start_flasher_addon": { + "title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]", + "description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]" } }, "error": { @@ -77,6 +87,7 @@ "abort": { "addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]", "addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]", + "addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]", "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", diff --git a/homeassistant/components/otbr/util.py b/homeassistant/components/otbr/util.py index 4cbf7ce6a08..067282108f1 100644 --- a/homeassistant/components/otbr/util.py +++ b/homeassistant/components/otbr/util.py @@ -14,7 +14,7 @@ from python_otbr_api.tlv_parser import MeshcopTLVType from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( MultiprotocolAddonManager, - get_addon_manager, + get_multiprotocol_addon_manager, is_multiprotocol_url, multi_pan_addon_using_device, ) @@ -146,8 +146,10 @@ async def get_allowed_channel(hass: HomeAssistant, otbr_url: str) -> int | None: # The OTBR is not sharing the radio, no restriction return None - addon_manager: MultiprotocolAddonManager = await get_addon_manager(hass) - return addon_manager.async_get_channel() + multipan_manager: MultiprotocolAddonManager = await get_multiprotocol_addon_manager( + hass + ) + return multipan_manager.async_get_channel() async def _warn_on_channel_collision( diff --git a/homeassistant/components/zha/__init__.py b/homeassistant/components/zha/__init__.py index 8a81648b580..a51d6f387e1 100644 --- a/homeassistant/components/zha/__init__.py +++ b/homeassistant/components/zha/__init__.py @@ -166,7 +166,11 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: """Unload ZHA config entry.""" - zha_gateway: ZHAGateway = hass.data[DATA_ZHA].pop(DATA_ZHA_GATEWAY) + try: + zha_gateway: ZHAGateway = hass.data[DATA_ZHA].pop(DATA_ZHA_GATEWAY) + except KeyError: + return False + await zha_gateway.shutdown() GROUP_PROBE.cleanup() diff --git a/homeassistant/components/zha/config_flow.py b/homeassistant/components/zha/config_flow.py index ba50839ee44..6ac3a155ed9 100644 --- a/homeassistant/components/zha/config_flow.py +++ b/homeassistant/components/zha/config_flow.py @@ -96,10 +96,12 @@ async def list_serial_ports(hass: HomeAssistant) -> list[ListPortInfo]: yellow_radio.manufacturer = "Nabu Casa" # Present the multi-PAN addon as a setup option, if it's available - addon_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) try: - addon_info = await addon_manager.async_get_addon_info() + addon_info = await multipan_manager.async_get_addon_info() except (AddonError, KeyError): addon_info = None diff --git a/homeassistant/components/zha/radio_manager.py b/homeassistant/components/zha/radio_manager.py index 29214083d27..4e70fc2247f 100644 --- a/homeassistant/components/zha/radio_manager.py +++ b/homeassistant/components/zha/radio_manager.py @@ -9,6 +9,7 @@ import logging import os from typing import Any +from bellows.config import CONF_USE_THREAD import voluptuous as vol from zigpy.application import ControllerApplication import zigpy.backups @@ -47,7 +48,9 @@ RECOMMENDED_RADIOS = ( ) CONNECT_DELAY_S = 1.0 +RETRY_DELAY_S = 1.0 +BACKUP_RETRIES = 5 MIGRATION_RETRIES = 100 HARDWARE_DISCOVERY_SCHEMA = vol.Schema( @@ -134,6 +137,7 @@ class ZhaRadioManager: app_config[CONF_DATABASE] = database_path app_config[CONF_DEVICE] = self.device_settings app_config[CONF_NWK_BACKUP_ENABLED] = False + app_config[CONF_USE_THREAD] = False app_config = self.radio_type.controller.SCHEMA(app_config) app = await self.radio_type.controller.new( @@ -341,7 +345,24 @@ class ZhaMultiPANMigrationHelper: old_radio_mgr.device_path = config_entry_data[CONF_DEVICE][CONF_DEVICE_PATH] old_radio_mgr.device_settings = config_entry_data[CONF_DEVICE] old_radio_mgr.radio_type = RadioType[config_entry_data[CONF_RADIO_TYPE]] - backup = await old_radio_mgr.async_load_network_settings(create_backup=True) + + for retry in range(BACKUP_RETRIES): + try: + backup = await old_radio_mgr.async_load_network_settings( + create_backup=True + ) + break + except OSError as err: + if retry >= BACKUP_RETRIES - 1: + raise + + _LOGGER.debug( + "Failed to create backup %r, retrying in %s seconds", + err, + RETRY_DELAY_S, + ) + + await asyncio.sleep(RETRY_DELAY_S) # Then configure the radio manager for the new radio to use the new settings self._radio_mgr.chosen_backup = backup @@ -381,10 +402,10 @@ class ZhaMultiPANMigrationHelper: _LOGGER.debug( "Failed to restore backup %r, retrying in %s seconds", err, - CONNECT_DELAY_S, + RETRY_DELAY_S, ) - await asyncio.sleep(CONNECT_DELAY_S) + await asyncio.sleep(RETRY_DELAY_S) _LOGGER.debug("Restored backup after %s retries", retry) diff --git a/tests/components/homeassistant_hardware/conftest.py b/tests/components/homeassistant_hardware/conftest.py index 60c766c7204..60083c2de94 100644 --- a/tests/components/homeassistant_hardware/conftest.py +++ b/tests/components/homeassistant_hardware/conftest.py @@ -147,3 +147,21 @@ def start_addon_fixture(): "homeassistant.components.hassio.addon_manager.async_start_addon" ) as start_addon: yield start_addon + + +@pytest.fixture(name="stop_addon") +def stop_addon_fixture(): + """Mock stop add-on.""" + with patch( + "homeassistant.components.hassio.addon_manager.async_stop_addon" + ) as stop_addon: + yield stop_addon + + +@pytest.fixture(name="uninstall_addon") +def uninstall_addon_fixture(): + """Mock uninstall add-on.""" + with patch( + "homeassistant.components.hassio.addon_manager.async_uninstall_addon" + ) as uninstall_addon: + yield uninstall_addon diff --git a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py index a956214c098..17cd288050c 100644 --- a/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py +++ b/tests/components/homeassistant_hardware/test_silabs_multiprotocol_addon.py @@ -3,10 +3,11 @@ from __future__ import annotations from collections.abc import Generator from typing import Any -from unittest.mock import Mock, patch +from unittest.mock import AsyncMock, Mock, patch import pytest +from homeassistant.components.hassio import AddonError, AddonInfo, AddonState, HassIO 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 @@ -14,15 +15,15 @@ from homeassistant.config_entries import ConfigEntry, ConfigFlow from homeassistant.const import EVENT_COMPONENT_LOADED from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult, FlowResultType +from homeassistant.exceptions import HomeAssistantError from homeassistant.setup import ATTR_COMPONENT from tests.common import ( MockConfigEntry, - MockModule, MockPlatform, flush_store, + mock_component, mock_config_flow, - mock_integration, mock_platform, ) @@ -101,6 +102,22 @@ def config_flow_handler( yield +@pytest.fixture +def options_flow_poll_addon_state() -> Generator[None, None, None]: + """Fixture for patching options flow addon state polling.""" + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state" + ): + yield + + +@pytest.fixture(autouse=True) +def hassio_integration(hass: HomeAssistant) -> Generator[None, None, None]: + """Fixture to mock the `hassio` integration.""" + mock_component(hass, "hassio") + hass.data["hassio"] = Mock(spec_set=HassIO) + + class MockMultiprotocolPlatform(MockPlatform): """A mock multiprotocol platform.""" @@ -149,6 +166,48 @@ def get_suggested(schema, key): raise Exception +@patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.ADDON_STATE_POLL_INTERVAL", + 0, +) +async def test_uninstall_addon_waiting( + hass: HomeAssistant, + addon_store_info, + addon_info, + install_addon, + uninstall_addon, +): + """Test the synchronous addon uninstall helper.""" + + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) + multipan_manager.async_get_addon_info = AsyncMock() + multipan_manager.async_uninstall_addon = AsyncMock( + wraps=multipan_manager.async_uninstall_addon + ) + + # First try uninstalling the addon when it is already uninstalled + multipan_manager.async_get_addon_info.side_effect = [ + Mock(state=AddonState.NOT_INSTALLED) + ] + await multipan_manager.async_uninstall_addon_waiting() + multipan_manager.async_uninstall_addon.assert_not_called() + + # Next, try uninstalling the addon but in a complex case where the API fails first + multipan_manager.async_get_addon_info.side_effect = [ + # First the API fails + AddonError(), + AddonError(), + # Then the addon is still running + Mock(state=AddonState.RUNNING), + # And finally it is uninstalled + Mock(state=AddonState.NOT_INSTALLED), + ] + await multipan_manager.async_uninstall_addon_waiting() + multipan_manager.async_uninstall_addon.assert_called_once() + + async def test_option_flow_install_multi_pan_addon( hass: HomeAssistant, addon_store_info, @@ -156,9 +215,9 @@ async def test_option_flow_install_multi_pan_addon( install_addon, set_addon_options, start_addon, + options_flow_poll_addon_state, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) # Setup the config entry config_entry = MockConfigEntry( @@ -169,13 +228,9 @@ async def test_option_flow_install_multi_pan_addon( ) 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_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"], @@ -224,9 +279,9 @@ async def test_option_flow_install_multi_pan_addon_zha( install_addon, set_addon_options, start_addon, + options_flow_poll_addon_state, ) -> 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( @@ -245,13 +300,9 @@ async def test_option_flow_install_multi_pan_addon_zha( ) 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_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"], @@ -268,7 +319,9 @@ async def test_option_flow_install_multi_pan_addon_zha( assert result["step_id"] == "configure_addon" install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) assert multipan_manager._channel is None with patch( "homeassistant.components.zha.silabs_multiprotocol.async_get_channel", @@ -318,9 +371,9 @@ async def test_option_flow_install_multi_pan_addon_zha_other_radio( install_addon, set_addon_options, start_addon, + options_flow_poll_addon_state, ) -> 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( @@ -346,13 +399,9 @@ async def test_option_flow_install_multi_pan_addon_zha_other_radio( ) 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_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"], @@ -408,31 +457,7 @@ async def test_option_flow_install_multi_pan_addon_zha_other_radio( async def test_option_flow_non_hassio( hass: HomeAssistant, ) -> 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) - - result = await hass.config_entries.options.async_init(config_entry.entry_id) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "not_hassio" - - -async def test_option_flow_addon_installed_other_device( - hass: HomeAssistant, - addon_store_info, - addon_installed, -) -> None: - """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) - + """Test installing the multi pan addon on a Core installation, without hassio.""" # Setup the config entry config_entry = MockConfigEntry( data={}, @@ -444,11 +469,33 @@ async def test_option_flow_addon_installed_other_device( with patch( "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), + return_value=False, ): result = await hass.config_entries.options.async_init(config_entry.entry_id) - assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "addon_installed_other_device" + + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "not_hassio" + + +async def test_option_flow_addon_installed_other_device( + hass: HomeAssistant, + addon_store_info, + addon_installed, +) -> None: + """Test installing the multi pan addon.""" + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "addon_installed_other_device" result = await hass.config_entries.options.async_configure(result["flow_id"], {}) assert result["type"] == FlowResultType.CREATE_ENTRY @@ -467,10 +514,12 @@ async def test_option_flow_addon_installed_same_device_reconfigure_unexpected_us suggested_channel: int, ) -> None: """Test reconfiguring the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) multipan_manager._channel = configured_channel # Setup the config entry @@ -482,13 +531,9 @@ async def test_option_flow_addon_installed_same_device_reconfigure_unexpected_us ) 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.MENU - assert result["step_id"] == "addon_menu" + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" result = await hass.config_entries.options.async_configure( result["flow_id"], @@ -528,10 +573,12 @@ async def test_option_flow_addon_installed_same_device_reconfigure_expected_user suggested_channel: int, ) -> None: """Test reconfiguring the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) multipan_manager._channel = configured_channel # Setup the config entry @@ -557,13 +604,9 @@ async def test_option_flow_addon_installed_same_device_reconfigure_expected_user hass.bus.async_fire(EVENT_COMPONENT_LOADED, {ATTR_COMPONENT: domain}) await hass.async_block_till_done() - 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.MENU - assert result["step_id"] == "addon_menu" + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" result = await hass.config_entries.options.async_configure( result["flow_id"], @@ -593,9 +636,15 @@ async def test_option_flow_addon_installed_same_device_uninstall( addon_info, addon_store_info, addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, ) -> None: - """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + """Test uninstalling the multi pan addon.""" + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" # Setup the config entry @@ -607,32 +656,97 @@ async def test_option_flow_addon_installed_same_device_uninstall( ) 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.MENU - assert result["step_id"] == "addon_menu" + zha_config_entry = MockConfigEntry( + data={ + "device": {"path": "socket://core-silabs-multiprotocol:9999"}, + "radio_type": "ezsp", + }, + domain=ZHA_DOMAIN, + options={}, + title="Test Multi-PAN", + ) + zha_config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" result = await hass.config_entries.options.async_configure( result["flow_id"], {"next_step_id": "uninstall_addon"}, ) assert result["type"] == FlowResultType.FORM - assert result["step_id"] == "show_revert_guide" + assert result["step_id"] == "uninstall_addon" - result = await hass.config_entries.options.async_configure(result["flow_id"], {}) + # Make sure the flasher addon is installed + addon_store_info.return_value = { + "installed": None, + "available": True, + "state": "not_installed", + } + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_flasher_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_flasher_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "uninstall_multiprotocol_addon" + assert result["progress_action"] == "uninstall_multiprotocol_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + uninstall_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "start_flasher_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_flasher_addon" + assert result["progress_action"] == "start_flasher_addon" + assert result["description_placeholders"] == {"addon_name": "Silicon Labs Flasher"} + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + install_addon.assert_called_once_with(hass, "core_silabs_flasher") + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "flashing_complete" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] == FlowResultType.CREATE_ENTRY + # Check the ZHA config entry data is updated + assert zha_config_entry.data == { + "device": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + } + assert zha_config_entry.title == "Test" -async def test_option_flow_do_not_install_multi_pan_addon( + +async def test_option_flow_addon_installed_same_device_do_not_uninstall_multi_pan( hass: HomeAssistant, addon_info, addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, ) -> None: - """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + """Test uninstalling the multi pan addon.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" # Setup the config entry config_entry = MockConfigEntry( @@ -643,13 +757,439 @@ async def test_option_flow_do_not_install_multi_pan_addon( ) 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_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: False} + ) + + assert result["type"] == FlowResultType.CREATE_ENTRY + + +async def test_option_flow_flasher_already_running_failure( + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test uninstalling the multi pan addon but with the flasher addon running.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + # The flasher addon is already installed and running, this is bad + addon_store_info.return_value["installed"] = True + addon_info.return_value["state"] = "started" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_already_running" + + +async def test_option_flow_addon_installed_same_device_flasher_already_installed( + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test uninstalling the multi pan addon.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "not_running", + } + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "uninstall_multiprotocol_addon" + assert result["progress_action"] == "uninstall_multiprotocol_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + uninstall_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "start_flasher_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_flasher_addon" + assert result["progress_action"] == "start_flasher_addon" + assert result["description_placeholders"] == {"addon_name": "Silicon Labs Flasher"} + + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "not_running", + } + result = await hass.config_entries.options.async_configure(result["flow_id"]) + install_addon.assert_not_called() + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "flashing_complete" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.CREATE_ENTRY + + +async def test_option_flow_flasher_install_failure( + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test uninstalling the multi pan addon, case where flasher addon fails.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # 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": "socket://core-silabs-multiprotocol:9999"}, + "radio_type": "ezsp", + }, + domain=ZHA_DOMAIN, + options={}, + title="Test Multi-PAN", + ) + zha_config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + addon_store_info.return_value = { + "installed": None, + "available": True, + "state": "not_installed", + } + install_addon.side_effect = [AddonError()] + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "install_flasher_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"] == "install_failed" + install_addon.assert_called_once_with(hass, "core_silabs_flasher") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_install_failed" + + +async def test_option_flow_flasher_addon_flash_failure( + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test where flasher addon fails to flash Zigbee firmware.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "uninstall_multiprotocol_addon" + assert result["progress_action"] == "uninstall_multiprotocol_addon" + + start_addon.side_effect = HassioAPIError("Boom") + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + uninstall_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "start_flasher_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_flasher_addon" + assert result["progress_action"] == "start_flasher_addon" + assert result["description_placeholders"] == {"addon_name": "Silicon Labs Flasher"} + + addon_store_info.return_value["installed"] = True + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "flasher_failed" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_start_failed" + assert result["description_placeholders"]["addon_name"] == "Silicon Labs Flasher" + + +@patch( + "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_initiate_migration", + side_effect=Exception("Boom!"), +) +async def test_option_flow_uninstall_migration_initiate_failure( + mock_initiate_migration, + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test uninstalling the multi pan addon, case where ZHA migration init fails.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # 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": "socket://core-silabs-multiprotocol:9999"}, + "radio_type": "ezsp", + }, + domain=ZHA_DOMAIN, + options={}, + title="Test Multi-PAN", + ) + zha_config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "zha_migration_failed" + mock_initiate_migration.assert_called_once() + + +@patch( + "homeassistant.components.zha.radio_manager.ZhaMultiPANMigrationHelper.async_finish_migration", + side_effect=Exception("Boom!"), +) +async def test_option_flow_uninstall_migration_finish_failure( + mock_finish_migration, + hass: HomeAssistant, + addon_info, + addon_store_info, + addon_installed, + install_addon, + start_addon, + stop_addon, + uninstall_addon, + set_addon_options, + options_flow_poll_addon_state, +) -> None: + """Test uninstalling the multi pan addon, case where ZHA migration init fails.""" + + addon_info.return_value["options"]["device"] = "/dev/ttyTEST123" + + # 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": "socket://core-silabs-multiprotocol:9999"}, + "radio_type": "ezsp", + }, + domain=ZHA_DOMAIN, + options={}, + title="Test Multi-PAN", + ) + zha_config_entry.add_to_hass(hass) + + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.MENU + assert result["step_id"] == "addon_menu" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + {"next_step_id": "uninstall_addon"}, + ) + assert result["type"] == FlowResultType.FORM + assert result["step_id"] == "uninstall_addon" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], {silabs_multiprotocol_addon.CONF_DISABLE_MULTI_PAN: True} + ) + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + uninstall_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "start_flasher_addon" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_flasher_addon" + assert result["progress_action"] == "start_flasher_addon" + assert result["description_placeholders"] == {"addon_name": "Silicon Labs Flasher"} + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.SHOW_PROGRESS_DONE + assert result["step_id"] == "flashing_complete" + + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "zha_migration_failed" + + +async def test_option_flow_do_not_install_multi_pan_addon( + hass: HomeAssistant, + addon_info, + addon_store_info, +) -> None: + """Test installing the multi pan addon.""" + + # Setup the config entry + config_entry = MockConfigEntry( + data={}, + domain=TEST_DOMAIN, + options={}, + title="Test HW", + ) + config_entry.add_to_hass(hass) + + 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"], @@ -669,7 +1209,7 @@ async def test_option_flow_install_multi_pan_addon_install_fails( start_addon, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + install_addon.side_effect = HassioAPIError("Boom") # Setup the config entry @@ -681,13 +1221,9 @@ async def test_option_flow_install_multi_pan_addon_install_fails( ) 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_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"], @@ -718,7 +1254,7 @@ async def test_option_flow_install_multi_pan_addon_start_fails( start_addon, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + start_addon.side_effect = HassioAPIError("Boom") # Setup the config entry @@ -730,13 +1266,9 @@ async def test_option_flow_install_multi_pan_addon_start_fails( ) 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_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"], @@ -788,7 +1320,7 @@ async def test_option_flow_install_multi_pan_addon_set_options_fails( start_addon, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + set_addon_options.side_effect = HassioAPIError("Boom") # Setup the config entry @@ -800,13 +1332,9 @@ async def test_option_flow_install_multi_pan_addon_set_options_fails( ) 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_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"], @@ -834,7 +1362,7 @@ async def test_option_flow_addon_info_fails( addon_info, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) + addon_store_info.side_effect = HassioAPIError("Boom") # Setup the config entry @@ -846,13 +1374,9 @@ async def test_option_flow_addon_info_fails( ) 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.ABORT - assert result["reason"] == "addon_info_failed" + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_info_failed" @patch( @@ -869,7 +1393,6 @@ async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_1( start_addon, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) # Setup the config entry config_entry = MockConfigEntry( @@ -888,13 +1411,9 @@ async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_1( ) 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_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"], @@ -929,9 +1448,9 @@ async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_2( install_addon, set_addon_options, start_addon, + options_flow_poll_addon_state, ) -> None: """Test installing the multi pan addon.""" - mock_integration(hass, MockModule("hassio")) # Setup the config entry config_entry = MockConfigEntry( @@ -950,13 +1469,9 @@ async def test_option_flow_install_multi_pan_addon_zha_migration_fails_step_2( ) 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_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"], @@ -1032,7 +1547,9 @@ async def test_import_channel( new_multipan_channel: int | None, ) -> None: """Test channel is initialized from first platform.""" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) multipan_manager._channel = initial_multipan_channel mock_multiprotocol_platform = MockMultiprotocolPlatform() @@ -1066,7 +1583,9 @@ async def test_change_channel( expected_calls: list[int], ) -> None: """Test channel is initialized from first platform.""" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) mock_multiprotocol_platform.using_multipan = platform_using_multipan await multipan_manager.async_change_channel(15, 10) @@ -1075,7 +1594,9 @@ async def test_change_channel( async def test_load_preferences(hass: HomeAssistant) -> None: """Make sure that we can load/save data correctly.""" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) assert multipan_manager._channel != 11 multipan_manager.async_set_channel(11) @@ -1106,7 +1627,9 @@ async def test_active_plaforms( active_platforms: list[str], ) -> None: """Test async_active_platforms.""" - multipan_manager = await silabs_multiprotocol_addon.get_addon_manager(hass) + multipan_manager = await silabs_multiprotocol_addon.get_multiprotocol_addon_manager( + hass + ) for domain, platform_using_multipan in multipan_platforms.items(): mock_multiprotocol_platform = MockMultiprotocolPlatform() @@ -1121,3 +1644,151 @@ async def test_active_plaforms( await hass.async_block_till_done() assert await multipan_manager.async_active_platforms() == active_platforms + + +async def test_check_multi_pan_addon_no_hassio(hass: HomeAssistant) -> None: + """Test `check_multi_pan_addon` without hassio.""" + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + return_value=False, + ), patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.get_multiprotocol_addon_manager", + autospec=True, + ) as mock_get_addon_manager: + await silabs_multiprotocol_addon.check_multi_pan_addon(hass) + mock_get_addon_manager.assert_not_called() + + +async def test_check_multi_pan_addon_info_error( + hass: HomeAssistant, addon_store_info +) -> None: + """Test `check_multi_pan_addon` where the addon info cannot be read.""" + + addon_store_info.side_effect = HassioAPIError("Boom") + + with pytest.raises(HomeAssistantError): + await silabs_multiprotocol_addon.check_multi_pan_addon(hass) + + +async def test_check_multi_pan_addon_bad_state(hass: HomeAssistant) -> None: + """Test `check_multi_pan_addon` where the addon is in an unexpected state.""" + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.get_multiprotocol_addon_manager", + return_value=Mock( + spec_set=silabs_multiprotocol_addon.MultiprotocolAddonManager + ), + ) as mock_get_addon_manager: + manager = mock_get_addon_manager.return_value + manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname="core_silabs_multiprotocol", + options={}, + state=AddonState.UPDATING, + update_available=False, + version="1.0.0", + ) + + with pytest.raises(HomeAssistantError): + await silabs_multiprotocol_addon.check_multi_pan_addon(hass) + + manager.async_start_addon.assert_not_called() + + +async def test_check_multi_pan_addon_auto_start( + hass: HomeAssistant, addon_info, addon_store_info, start_addon +) -> None: + """Test `check_multi_pan_addon` auto starting the addon.""" + + addon_info.return_value["state"] = "not_running" + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "not_running", + } + + # An error is raised even if we auto-start + with pytest.raises(HomeAssistantError): + await silabs_multiprotocol_addon.check_multi_pan_addon(hass) + + start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") + + +async def test_check_multi_pan_addon( + hass: HomeAssistant, addon_info, addon_store_info, start_addon +) -> None: + """Test `check_multi_pan_addon`.""" + + addon_info.return_value["state"] = "started" + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "running", + } + + await silabs_multiprotocol_addon.check_multi_pan_addon(hass) + start_addon.assert_not_called() + + +async def test_multi_pan_addon_using_device_no_hassio(hass: HomeAssistant) -> None: + """Test `multi_pan_addon_using_device` without hassio.""" + + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + return_value=False, + ): + assert ( + await silabs_multiprotocol_addon.multi_pan_addon_using_device( + hass, "/dev/ttyAMA1" + ) + is False + ) + + +async def test_multi_pan_addon_using_device_not_running( + hass: HomeAssistant, addon_info, addon_store_info +) -> None: + """Test `multi_pan_addon_using_device` when the addon isn't running.""" + + addon_info.return_value["state"] = "not_running" + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "not_running", + } + + await silabs_multiprotocol_addon.multi_pan_addon_using_device( + hass, "/dev/ttyAMA1" + ) is False + + +@pytest.mark.parametrize( + ("options_device", "expected_result"), + [("/dev/ttyAMA2", False), ("/dev/ttyAMA1", True)], +) +async def test_multi_pan_addon_using_device( + hass: HomeAssistant, + addon_info, + addon_store_info, + options_device: str, + expected_result: bool, +) -> None: + """Test `multi_pan_addon_using_device` when the addon isn't running.""" + + addon_info.return_value["state"] = "started" + addon_info.return_value["options"] = { + "autoflash_firmware": True, + "device": options_device, + "baudrate": "115200", + "flow_control": True, + } + addon_store_info.return_value = { + "installed": True, + "available": True, + "state": "running", + } + + await silabs_multiprotocol_addon.multi_pan_addon_using_device( + hass, "/dev/ttyAMA1" + ) is expected_result diff --git a/tests/components/homeassistant_sky_connect/conftest.py b/tests/components/homeassistant_sky_connect/conftest.py index 3677b4ea8f1..85017866db9 100644 --- a/tests/components/homeassistant_sky_connect/conftest.py +++ b/tests/components/homeassistant_sky_connect/conftest.py @@ -9,7 +9,7 @@ import pytest def mock_usb_serial_by_id_fixture() -> Generator[MagicMock, None, None]: """Mock usb serial by id.""" with patch( - "homeassistant.components.zwave_js.config_flow.usb.get_serial_by_id" + "homeassistant.components.zha.config_flow.usb.get_serial_by_id" ) as mock_usb_serial_by_id: mock_usb_serial_by_id.side_effect = lambda x: x yield mock_usb_serial_by_id @@ -149,3 +149,21 @@ def start_addon_fixture(): "homeassistant.components.hassio.addon_manager.async_start_addon" ) as start_addon: yield start_addon + + +@pytest.fixture(name="stop_addon") +def stop_addon_fixture(): + """Mock stop add-on.""" + with patch( + "homeassistant.components.hassio.addon_manager.async_stop_addon" + ) as stop_addon: + yield stop_addon + + +@pytest.fixture(name="uninstall_addon") +def uninstall_addon_fixture(): + """Mock uninstall add-on.""" + with patch( + "homeassistant.components.hassio.addon_manager.async_uninstall_addon" + ) as uninstall_addon: + yield uninstall_addon diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index c74adbf32ea..9e1977192e9 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -1,7 +1,10 @@ """Test the Home Assistant SkyConnect config flow.""" +from collections.abc import Generator import copy from unittest.mock import Mock, patch +import pytest + from homeassistant.components import homeassistant_sky_connect, usb from homeassistant.components.homeassistant_sky_connect.const import DOMAIN from homeassistant.components.zha.core.const import ( @@ -25,6 +28,15 @@ USB_DATA = usb.UsbServiceInfo( ) +@pytest.fixture(autouse=True) +def config_flow_handler(hass: HomeAssistant) -> Generator[None, None, None]: + """Fixture for a test config flow.""" + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state" + ): + yield + + async def test_config_flow(hass: HomeAssistant) -> None: """Test the config flow.""" with patch( diff --git a/tests/components/homeassistant_sky_connect/test_init.py b/tests/components/homeassistant_sky_connect/test_init.py index 746e119082c..cbf1cfa7d36 100644 --- a/tests/components/homeassistant_sky_connect/test_init.py +++ b/tests/components/homeassistant_sky_connect/test_init.py @@ -24,6 +24,13 @@ CONFIG_ENTRY_DATA = { } +@pytest.fixture(autouse=True) +def disable_usb_probing() -> Generator[None, None, None]: + """Disallow touching of system USB devices during unit tests.""" + with patch("homeassistant.components.usb.comports", return_value=[]): + yield + + @pytest.fixture def mock_zha_config_flow_setup() -> Generator[None, None, None]: """Mock the radio connection and probing of the ZHA config flow.""" diff --git a/tests/components/homeassistant_yellow/test_config_flow.py b/tests/components/homeassistant_yellow/test_config_flow.py index 66401bcd7bc..58d47c41987 100644 --- a/tests/components/homeassistant_yellow/test_config_flow.py +++ b/tests/components/homeassistant_yellow/test_config_flow.py @@ -1,4 +1,5 @@ """Test the Home Assistant Yellow config flow.""" +from collections.abc import Generator from unittest.mock import Mock, patch import pytest @@ -11,6 +12,15 @@ from homeassistant.data_entry_flow import FlowResultType from tests.common import MockConfigEntry, MockModule, mock_integration +@pytest.fixture(autouse=True) +def config_flow_handler(hass: HomeAssistant) -> Generator[None, None, None]: + """Fixture for a test config flow.""" + with patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state" + ): + yield + + @pytest.fixture(name="get_yellow_settings") def mock_get_yellow_settings(): """Mock getting yellow settings.""" diff --git a/tests/components/zha/test_radio_manager.py b/tests/components/zha/test_radio_manager.py index 4d90d83d483..c507db3e6ab 100644 --- a/tests/components/zha/test_radio_manager.py +++ b/tests/components/zha/test_radio_manager.py @@ -31,7 +31,9 @@ def disable_platform_only(): @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): + with patch( + "homeassistant.components.zha.radio_manager.CONNECT_DELAY_S", 0.0001 + ), patch("homeassistant.components.zha.radio_manager.RETRY_DELAY_S", 0.0001): yield @@ -83,7 +85,7 @@ def com_port(device="/dev/ttyUSB1234"): @pytest.fixture -def mock_connect_zigpy_app() -> Generator[None, None, None]: +def mock_connect_zigpy_app() -> Generator[MagicMock, None, None]: """Mock the radio connection.""" mock_connect_app = MagicMock() @@ -96,7 +98,7 @@ def mock_connect_zigpy_app() -> Generator[None, None, None]: "homeassistant.components.zha.radio_manager.ZhaRadioManager._connect_zigpy_app", return_value=mock_connect_app, ): - yield + yield mock_connect_app @patch("homeassistant.components.zha.async_setup_entry", AsyncMock(return_value=True)) @@ -370,3 +372,52 @@ async def test_migrate_non_matching_port( "radio_type": "ezsp", } assert config_entry.title == "Test" + + +async def test_migrate_initiate_failure( + hass: HomeAssistant, + mock_connect_zigpy_app, +) -> None: + """Test retries with failure.""" + # Set up 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": { + "hw": { + "name": "Test", + "port": { + "path": "/dev/ttyTEST123", + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "efr32", + } + }, + } + + mock_load_info = AsyncMock(side_effect=OSError()) + mock_connect_zigpy_app.__aenter__.return_value.load_network_info = mock_load_info + + migration_helper = radio_manager.ZhaMultiPANMigrationHelper(hass, config_entry) + + with pytest.raises(OSError): + await migration_helper.async_initiate_migration(migration_data) + + assert len(mock_load_info.mock_calls) == radio_manager.BACKUP_RETRIES