diff --git a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py index 1b4840e5a98..7519e0ae394 100644 --- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py +++ b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py @@ -7,6 +7,8 @@ import asyncio import logging from typing import Any +from ha_silabs_firmware_client import FirmwareUpdateClient + from homeassistant.components.hassio import ( AddonError, AddonInfo, @@ -22,17 +24,17 @@ from homeassistant.config_entries import ( ) from homeassistant.core import callback from homeassistant.data_entry_flow import AbortFlow +from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.hassio import is_hassio -from . import silabs_multiprotocol_addon from .const import OTBR_DOMAIN, ZHA_DOMAIN from .util import ( ApplicationType, FirmwareInfo, OwningAddon, OwningIntegration, + async_flash_silabs_firmware, get_otbr_addon_manager, - get_zigbee_flasher_addon_manager, guess_firmware_info, guess_hardware_owners, probe_silabs_firmware_info, @@ -61,6 +63,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): self.addon_install_task: asyncio.Task | None = None self.addon_start_task: asyncio.Task | None = None self.addon_uninstall_task: asyncio.Task | None = None + self.firmware_install_task: asyncio.Task | None = None def _get_translation_placeholders(self) -> dict[str, str]: """Shared translation placeholders.""" @@ -77,22 +80,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): return placeholders - async def _async_set_addon_config( - self, config: dict, addon_manager: AddonManager - ) -> None: - """Set add-on config.""" - try: - await addon_manager.async_set_addon_options(config) - except AddonError as err: - _LOGGER.error(err) - raise AbortFlow( - "addon_set_config_failed", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": addon_manager.addon_name, - }, - ) from err - async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo: """Return add-on info.""" try: @@ -150,6 +137,54 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): ) ) + async def _install_firmware_step( + self, + fw_update_url: str, + fw_type: str, + firmware_name: str, + expected_installed_firmware_type: ApplicationType, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: + assert self._device is not None + + if not self.firmware_install_task: + session = async_get_clientsession(self.hass) + client = FirmwareUpdateClient(fw_update_url, session) + manifest = await client.async_update_data() + + fw_meta = next( + fw for fw in manifest.firmwares if fw.filename.startswith(fw_type) + ) + + fw_data = await client.async_fetch_firmware(fw_meta) + self.firmware_install_task = self.hass.async_create_task( + async_flash_silabs_firmware( + hass=self.hass, + device=self._device, + fw_data=fw_data, + expected_installed_firmware_type=expected_installed_firmware_type, + bootloader_reset_type=None, + progress_callback=lambda offset, total: self.async_update_progress( + offset / total + ), + ), + f"Flash {firmware_name} firmware", + ) + + if not self.firmware_install_task.done(): + return self.async_show_progress( + step_id=step_id, + progress_action="install_firmware", + description_placeholders={ + **self._get_translation_placeholders(), + "firmware_name": firmware_name, + }, + progress_task=self.firmware_install_task, + ) + + return self.async_show_progress_done(next_step_id=next_step_id) + async def async_step_pick_firmware_zigbee( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: @@ -160,68 +195,133 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): description_placeholders=self._get_translation_placeholders(), ) - # Allow the stick to be used with ZHA without flashing - if ( - self._probed_firmware_info is not None - and self._probed_firmware_info.firmware_type == ApplicationType.EZSP - ): - return await self.async_step_confirm_zigbee() + return await self.async_step_install_zigbee_firmware() - if not is_hassio(self.hass): - return self.async_abort( - reason="not_hassio", - description_placeholders=self._get_translation_placeholders(), - ) + async def async_step_install_zigbee_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Zigbee firmware.""" + raise NotImplementedError - # Only flash new firmware if we need to - fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass) - addon_info = await self._async_get_addon_info(fw_flasher_manager) - - if addon_info.state == AddonState.NOT_INSTALLED: - return await self.async_step_install_zigbee_flasher_addon() - - if addon_info.state == AddonState.NOT_RUNNING: - return await self.async_step_run_zigbee_flasher_addon() - - # If the addon is already installed and running, fail + async def async_step_addon_operation_failed( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Abort when add-on installation or start failed.""" return self.async_abort( - reason="addon_already_running", + reason=self._failed_addon_reason, description_placeholders={ **self._get_translation_placeholders(), - "addon_name": fw_flasher_manager.addon_name, + "addon_name": self._failed_addon_name, }, ) - async def async_step_install_zigbee_flasher_addon( + async def async_step_confirm_zigbee( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Show progress dialog for installing the Zigbee flasher addon.""" - return await self._install_addon( - get_zigbee_flasher_addon_manager(self.hass), - "install_zigbee_flasher_addon", - "run_zigbee_flasher_addon", + """Confirm Zigbee setup.""" + assert self._device is not None + assert self._hardware_name is not None + + if user_input is None: + return self.async_show_form( + step_id="confirm_zigbee", + description_placeholders=self._get_translation_placeholders(), + ) + + if not await self._probe_firmware_info(probe_methods=(ApplicationType.EZSP,)): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + + await self.hass.config_entries.flow.async_init( + ZHA_DOMAIN, + context={"source": "hardware"}, + data={ + "name": self._hardware_name, + "port": { + "path": self._device, + "baudrate": 115200, + "flow_control": "hardware", + }, + "radio_type": "ezsp", + }, ) - async def _install_addon( - self, - addon_manager: silabs_multiprotocol_addon.WaitingAddonManager, - step_id: str, - next_step_id: str, + return self._async_flow_finished() + + async def _ensure_thread_addon_setup(self) -> ConfigFlowResult | None: + """Ensure the OTBR addon is set up and not running.""" + + # We install the OTBR addon no matter what, since it is required to use Thread + if not is_hassio(self.hass): + return self.async_abort( + reason="not_hassio_thread", + description_placeholders=self._get_translation_placeholders(), + ) + + otbr_manager = get_otbr_addon_manager(self.hass) + addon_info = await self._async_get_addon_info(otbr_manager) + + if addon_info.state == AddonState.NOT_INSTALLED: + return await self.async_step_install_otbr_addon() + + if addon_info.state == AddonState.RUNNING: + # We only fail setup if we have an instance of OTBR running *and* it's + # pointing to different hardware + if addon_info.options["device"] != self._device: + return self.async_abort( + reason="otbr_addon_already_running", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": otbr_manager.addon_name, + }, + ) + + # Otherwise, stop the addon before continuing to flash firmware + await otbr_manager.async_stop_addon() + + return None + + async def async_step_pick_firmware_thread( + self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: - """Show progress dialog for installing an addon.""" + """Pick Thread firmware.""" + if not await self._probe_firmware_info(): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + + if result := await self._ensure_thread_addon_setup(): + return result + + return await self.async_step_install_thread_firmware() + + async def async_step_install_thread_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Thread firmware.""" + raise NotImplementedError + + async def async_step_install_otbr_addon( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Show progress dialog for installing the OTBR addon.""" + addon_manager = get_otbr_addon_manager(self.hass) addon_info = await self._async_get_addon_info(addon_manager) - _LOGGER.debug("Flasher addon state: %s", addon_info) + _LOGGER.debug("OTBR addon info: %s", addon_info) if not self.addon_install_task: self.addon_install_task = self.hass.async_create_task( addon_manager.async_install_addon_waiting(), - "Addon install", + "OTBR addon install", ) if not self.addon_install_task.done(): return self.async_show_progress( - step_id=step_id, + step_id="install_otbr_addon", progress_action="install_addon", description_placeholders={ **self._get_translation_placeholders(), @@ -240,208 +340,50 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): finally: self.addon_install_task = None - return self.async_show_progress_done(next_step_id=next_step_id) - - async def async_step_addon_operation_failed( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Abort when add-on installation or start failed.""" - return self.async_abort( - reason=self._failed_addon_reason, - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": self._failed_addon_name, - }, - ) - - async def async_step_run_zigbee_flasher_addon( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Configure the flasher addon to point to the SkyConnect and run it.""" - fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass) - addon_info = await self._async_get_addon_info(fw_flasher_manager) - - assert self._device is not None - new_addon_config = { - **addon_info.options, - "device": self._device, - "baudrate": 115200, - "bootloader_baudrate": 115200, - "flow_control": True, - } - - _LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config) - await self._async_set_addon_config(new_addon_config, fw_flasher_manager) - - if not self.addon_start_task: - - async def start_and_wait_until_done() -> None: - await fw_flasher_manager.async_start_addon_waiting() - # Now that the addon is running, wait for it to finish - await fw_flasher_manager.async_wait_until_addon_state( - AddonState.NOT_RUNNING - ) - - self.addon_start_task = self.hass.async_create_task( - start_and_wait_until_done() - ) - - if not self.addon_start_task.done(): - return self.async_show_progress( - step_id="run_zigbee_flasher_addon", - progress_action="run_zigbee_flasher_addon", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": fw_flasher_manager.addon_name, - }, - progress_task=self.addon_start_task, - ) - - try: - await self.addon_start_task - except (AddonError, AbortFlow) as err: - _LOGGER.error(err) - self._failed_addon_name = fw_flasher_manager.addon_name - self._failed_addon_reason = "addon_start_failed" - return self.async_show_progress_done(next_step_id="addon_operation_failed") - finally: - self.addon_start_task = None - - return self.async_show_progress_done( - next_step_id="uninstall_zigbee_flasher_addon" - ) - - async def async_step_uninstall_zigbee_flasher_addon( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Uninstall the flasher addon.""" - fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass) - - if not self.addon_uninstall_task: - _LOGGER.debug("Uninstalling flasher addon") - self.addon_uninstall_task = self.hass.async_create_task( - fw_flasher_manager.async_uninstall_addon_waiting() - ) - - if not self.addon_uninstall_task.done(): - return self.async_show_progress( - step_id="uninstall_zigbee_flasher_addon", - progress_action="uninstall_zigbee_flasher_addon", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": fw_flasher_manager.addon_name, - }, - progress_task=self.addon_uninstall_task, - ) - - try: - await self.addon_uninstall_task - except (AddonError, AbortFlow) as err: - _LOGGER.error(err) - # The uninstall failing isn't critical so we can just continue - finally: - self.addon_uninstall_task = None - - return self.async_show_progress_done(next_step_id="confirm_zigbee") - - async def async_step_confirm_zigbee( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Confirm Zigbee setup.""" - assert self._device is not None - assert self._hardware_name is not None - - if not await self._probe_firmware_info(probe_methods=(ApplicationType.EZSP,)): - return self.async_abort( - reason="unsupported_firmware", - description_placeholders=self._get_translation_placeholders(), - ) - - if user_input is not None: - await self.hass.config_entries.flow.async_init( - ZHA_DOMAIN, - context={"source": "hardware"}, - data={ - "name": self._hardware_name, - "port": { - "path": self._device, - "baudrate": 115200, - "flow_control": "hardware", - }, - "radio_type": "ezsp", - }, - ) - - return self._async_flow_finished() - - return self.async_show_form( - step_id="confirm_zigbee", - description_placeholders=self._get_translation_placeholders(), - ) - - async def async_step_pick_firmware_thread( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Pick Thread firmware.""" - if not await self._probe_firmware_info(): - return self.async_abort( - reason="unsupported_firmware", - description_placeholders=self._get_translation_placeholders(), - ) - - # We install the OTBR addon no matter what, since it is required to use Thread - if not is_hassio(self.hass): - return self.async_abort( - reason="not_hassio_thread", - description_placeholders=self._get_translation_placeholders(), - ) - - otbr_manager = get_otbr_addon_manager(self.hass) - addon_info = await self._async_get_addon_info(otbr_manager) - - if addon_info.state == AddonState.NOT_INSTALLED: - return await self.async_step_install_otbr_addon() - - if addon_info.state == AddonState.NOT_RUNNING: - return await self.async_step_start_otbr_addon() - - # If the addon is already installed and running, fail - return self.async_abort( - reason="otbr_addon_already_running", - description_placeholders={ - **self._get_translation_placeholders(), - "addon_name": otbr_manager.addon_name, - }, - ) - - async def async_step_install_otbr_addon( - self, user_input: dict[str, Any] | None = None - ) -> ConfigFlowResult: - """Show progress dialog for installing the OTBR addon.""" - return await self._install_addon( - get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon" - ) + return self.async_show_progress_done(next_step_id="install_thread_firmware") async def async_step_start_otbr_addon( self, user_input: dict[str, Any] | None = None ) -> ConfigFlowResult: """Configure OTBR to point to the SkyConnect and run the addon.""" otbr_manager = get_otbr_addon_manager(self.hass) - addon_info = await self._async_get_addon_info(otbr_manager) - - assert self._device is not None - new_addon_config = { - **addon_info.options, - "device": self._device, - "baudrate": 460800, - "flow_control": True, - "autoflash_firmware": True, - } - - _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config) - await self._async_set_addon_config(new_addon_config, otbr_manager) if not self.addon_start_task: + # Before we start the addon, confirm that the correct firmware is running + # and populate `self._probed_firmware_info` with the correct information + if not await self._probe_firmware_info( + probe_methods=(ApplicationType.SPINEL,) + ): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + + addon_info = await self._async_get_addon_info(otbr_manager) + + assert self._device is not None + new_addon_config = { + **addon_info.options, + "device": self._device, + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": False, + } + + _LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config) + + try: + await otbr_manager.async_set_addon_options(new_addon_config) + except AddonError as err: + _LOGGER.error(err) + raise AbortFlow( + "addon_set_config_failed", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": otbr_manager.addon_name, + }, + ) from err + self.addon_start_task = self.hass.async_create_task( otbr_manager.async_start_addon_waiting() ) @@ -475,20 +417,14 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): """Confirm OTBR setup.""" assert self._device is not None - if not await self._probe_firmware_info(probe_methods=(ApplicationType.SPINEL,)): - return self.async_abort( - reason="unsupported_firmware", + if user_input is None: + return self.async_show_form( + step_id="confirm_otbr", description_placeholders=self._get_translation_placeholders(), ) - if user_input is not None: - # OTBR discovery is done automatically via hassio - return self._async_flow_finished() - - return self.async_show_form( - step_id="confirm_otbr", - description_placeholders=self._get_translation_placeholders(), - ) + # OTBR discovery is done automatically via hassio + return self._async_flow_finished() @abstractmethod def _async_flow_finished(self) -> ConfigFlowResult: diff --git a/homeassistant/components/homeassistant_hardware/strings.json b/homeassistant/components/homeassistant_hardware/strings.json index e184f9b3a85..99172c963b8 100644 --- a/homeassistant/components/homeassistant_hardware/strings.json +++ b/homeassistant/components/homeassistant_hardware/strings.json @@ -10,22 +10,6 @@ "pick_firmware_thread": "Thread" } }, - "install_zigbee_flasher_addon": { - "title": "Installing flasher", - "description": "Installing the Silicon Labs Flasher add-on." - }, - "run_zigbee_flasher_addon": { - "title": "Installing Zigbee firmware", - "description": "Installing Zigbee firmware. This will take about a minute." - }, - "uninstall_zigbee_flasher_addon": { - "title": "Removing flasher", - "description": "Removing the Silicon Labs Flasher add-on." - }, - "zigbee_flasher_failed": { - "title": "Zigbee installation failed", - "description": "The Zigbee firmware installation process was unsuccessful. Ensure no other software is trying to communicate with the {model} and try again." - }, "confirm_zigbee": { "title": "Zigbee setup complete", "description": "Your {model} is now a Zigbee coordinator and will be shown as discovered by the Zigbee Home Automation integration." @@ -55,9 +39,7 @@ "unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device. If you are running Home Assistant OS in a virtual machine or in Docker, please make sure that permissions are set correctly for the device." }, "progress": { - "install_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is installed, this may take a few minutes.", - "run_zigbee_flasher_addon": "Please wait while Zigbee firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes.", - "uninstall_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is being removed." + "install_firmware": "Please wait while {firmware_name} firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes." } } }, @@ -110,16 +92,6 @@ "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": { diff --git a/homeassistant/components/homeassistant_hardware/update.py b/homeassistant/components/homeassistant_hardware/update.py index 1b0f15ca021..831d9f3f4da 100644 --- a/homeassistant/components/homeassistant_hardware/update.py +++ b/homeassistant/components/homeassistant_hardware/update.py @@ -2,15 +2,12 @@ from __future__ import annotations -from collections.abc import AsyncIterator, Callable -from contextlib import AsyncExitStack, asynccontextmanager +from collections.abc import Callable from dataclasses import dataclass import logging from typing import Any, cast from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata -from universal_silabs_flasher.firmware import parse_firmware_image -from universal_silabs_flasher.flasher import Flasher from yarl import URL from homeassistant.components.update import ( @@ -20,18 +17,12 @@ from homeassistant.components.update import ( ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import CALLBACK_TYPE, callback -from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.restore_state import ExtraStoredData from homeassistant.helpers.update_coordinator import CoordinatorEntity from .coordinator import FirmwareUpdateCoordinator from .helpers import async_register_firmware_info_callback -from .util import ( - ApplicationType, - FirmwareInfo, - guess_firmware_info, - probe_silabs_firmware_info, -) +from .util import ApplicationType, FirmwareInfo, async_flash_silabs_firmware _LOGGER = logging.getLogger(__name__) @@ -249,19 +240,11 @@ class BaseFirmwareUpdateEntity( self._attr_update_percentage = round((offset * 100) / total_size) self.async_write_ha_state() - @asynccontextmanager - async def _temporarily_stop_hardware_owners( - self, device: str - ) -> AsyncIterator[None]: - """Temporarily stop addons and integrations communicating with the device.""" - firmware_info = await guess_firmware_info(self.hass, device) - _LOGGER.debug("Identified firmware info: %s", firmware_info) - - async with AsyncExitStack() as stack: - for owner in firmware_info.owners: - await stack.enter_async_context(owner.temporarily_stop(self.hass)) - - yield + # Switch to an indeterminate progress bar after installation is complete, since + # we probe the firmware after flashing + if offset == total_size: + self._attr_update_percentage = None + self.async_write_ha_state() async def async_install( self, version: str | None, backup: bool, **kwargs: Any @@ -278,49 +261,18 @@ class BaseFirmwareUpdateEntity( fw_data = await self.coordinator.client.async_fetch_firmware( self._latest_firmware ) - fw_image = await self.hass.async_add_executor_job(parse_firmware_image, fw_data) - device = self._current_device + try: + firmware_info = await async_flash_silabs_firmware( + hass=self.hass, + device=self._current_device, + fw_data=fw_data, + expected_installed_firmware_type=self.entity_description.expected_firmware_type, + bootloader_reset_type=self.bootloader_reset_type, + progress_callback=self._update_progress, + ) + finally: + self._attr_in_progress = False + self.async_write_ha_state() - flasher = Flasher( - device=device, - probe_methods=( - ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(), - ApplicationType.EZSP.as_flasher_application_type(), - ApplicationType.SPINEL.as_flasher_application_type(), - ApplicationType.CPC.as_flasher_application_type(), - ), - bootloader_reset=self.bootloader_reset_type, - ) - - async with self._temporarily_stop_hardware_owners(device): - try: - try: - # Enter the bootloader with indeterminate progress - await flasher.enter_bootloader() - - # Flash the firmware, with progress - await flasher.flash_firmware( - fw_image, progress_callback=self._update_progress - ) - except Exception as err: - raise HomeAssistantError("Failed to flash firmware") from err - - # Probe the running application type with indeterminate progress - self._attr_update_percentage = None - self.async_write_ha_state() - - firmware_info = await probe_silabs_firmware_info( - device, - probe_methods=(self.entity_description.expected_firmware_type,), - ) - - if firmware_info is None: - raise HomeAssistantError( - "Failed to probe the firmware after flashing" - ) - - self._firmware_info_callback(firmware_info) - finally: - self._attr_in_progress = False - self.async_write_ha_state() + self._firmware_info_callback(firmware_info) diff --git a/homeassistant/components/homeassistant_hardware/util.py b/homeassistant/components/homeassistant_hardware/util.py index 64f363e4f23..d84f4f75ff7 100644 --- a/homeassistant/components/homeassistant_hardware/util.py +++ b/homeassistant/components/homeassistant_hardware/util.py @@ -4,18 +4,20 @@ from __future__ import annotations import asyncio from collections import defaultdict -from collections.abc import AsyncIterator, Iterable -from contextlib import asynccontextmanager +from collections.abc import AsyncIterator, Callable, Iterable +from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass from enum import StrEnum import logging from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType +from universal_silabs_flasher.firmware import parse_firmware_image from universal_silabs_flasher.flasher import Flasher from homeassistant.components.hassio import AddonError, AddonManager, AddonState from homeassistant.config_entries import ConfigEntryState from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.hassio import is_hassio from homeassistant.helpers.singleton import singleton @@ -333,3 +335,52 @@ async def probe_silabs_firmware_type( return None return fw_info.firmware_type + + +async def async_flash_silabs_firmware( + hass: HomeAssistant, + device: str, + fw_data: bytes, + expected_installed_firmware_type: ApplicationType, + bootloader_reset_type: str | None = None, + progress_callback: Callable[[int, int], None] | None = None, +) -> FirmwareInfo: + """Flash firmware to the SiLabs device.""" + firmware_info = await guess_firmware_info(hass, device) + _LOGGER.debug("Identified firmware info: %s", firmware_info) + + fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data) + + flasher = Flasher( + device=device, + probe_methods=( + ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(), + ApplicationType.EZSP.as_flasher_application_type(), + ApplicationType.SPINEL.as_flasher_application_type(), + ApplicationType.CPC.as_flasher_application_type(), + ), + bootloader_reset=bootloader_reset_type, + ) + + async with AsyncExitStack() as stack: + for owner in firmware_info.owners: + await stack.enter_async_context(owner.temporarily_stop(hass)) + + try: + # Enter the bootloader with indeterminate progress + await flasher.enter_bootloader() + + # Flash the firmware, with progress + await flasher.flash_firmware(fw_image, progress_callback=progress_callback) + except Exception as err: + raise HomeAssistantError("Failed to flash firmware") from err + + probed_firmware_info = await probe_silabs_firmware_info( + device, + probe_methods=(expected_installed_firmware_type,), + ) + + if probed_firmware_info is None: + raise HomeAssistantError("Failed to probe the firmware after flashing") + + return probed_firmware_info diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index eb5ea214b3e..997edb54b18 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/homeassistant/components/homeassistant_sky_connect/config_flow.py @@ -32,6 +32,7 @@ from .const import ( FIRMWARE, FIRMWARE_VERSION, MANUFACTURER, + NABU_CASA_FIRMWARE_RELEASES_URL, PID, PRODUCT, SERIAL_NUMBER, @@ -45,19 +46,29 @@ _LOGGER = logging.getLogger(__name__) if TYPE_CHECKING: - class TranslationPlaceholderProtocol(Protocol): - """Protocol describing `BaseFirmwareInstallFlow`'s translation placeholders.""" + class FirmwareInstallFlowProtocol(Protocol): + """Protocol describing `BaseFirmwareInstallFlow` for a mixin.""" def _get_translation_placeholders(self) -> dict[str, str]: return {} + async def _install_firmware_step( + self, + fw_update_url: str, + fw_type: str, + firmware_name: str, + expected_installed_firmware_type: ApplicationType, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: ... + else: # Multiple inheritance with `Protocol` seems to break - TranslationPlaceholderProtocol = object + FirmwareInstallFlowProtocol = object -class SkyConnectTranslationMixin(ConfigEntryBaseFlow, TranslationPlaceholderProtocol): - """Translation placeholder mixin for Home Assistant SkyConnect.""" +class SkyConnectFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol): + """Mixin for Home Assistant SkyConnect firmware methods.""" context: ConfigFlowContext @@ -72,9 +83,35 @@ class SkyConnectTranslationMixin(ConfigEntryBaseFlow, TranslationPlaceholderProt return placeholders + async def async_step_install_zigbee_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Zigbee firmware.""" + return await self._install_firmware_step( + fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL, + fw_type="skyconnect_zigbee_ncp", + firmware_name="Zigbee", + expected_installed_firmware_type=ApplicationType.EZSP, + step_id="install_zigbee_firmware", + next_step_id="confirm_zigbee", + ) + + async def async_step_install_thread_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Thread firmware.""" + return await self._install_firmware_step( + fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL, + fw_type="skyconnect_openthread_rcp", + firmware_name="OpenThread", + expected_installed_firmware_type=ApplicationType.SPINEL, + step_id="install_thread_firmware", + next_step_id="start_otbr_addon", + ) + class HomeAssistantSkyConnectConfigFlow( - SkyConnectTranslationMixin, + SkyConnectFirmwareMixin, firmware_config_flow.BaseFirmwareConfigFlow, domain=DOMAIN, ): @@ -207,7 +244,7 @@ class HomeAssistantSkyConnectMultiPanOptionsFlowHandler( class HomeAssistantSkyConnectOptionsFlowHandler( - SkyConnectTranslationMixin, firmware_config_flow.BaseFirmwareOptionsFlow + SkyConnectFirmwareMixin, firmware_config_flow.BaseFirmwareOptionsFlow ): """Zigbee and Thread options flow handlers.""" diff --git a/homeassistant/components/homeassistant_sky_connect/strings.json b/homeassistant/components/homeassistant_sky_connect/strings.json index a990f025e8d..08c8a56c30d 100644 --- a/homeassistant/components/homeassistant_sky_connect/strings.json +++ b/homeassistant/components/homeassistant_sky_connect/strings.json @@ -48,16 +48,6 @@ "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%]" - }, "pick_firmware": { "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]", "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::description%]", @@ -66,18 +56,6 @@ "pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]" } }, - "install_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]" - }, - "run_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]" - }, - "zigbee_flasher_failed": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]" - }, "confirm_zigbee": { "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]", "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]" @@ -120,9 +98,7 @@ "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]", "start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", "start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", - "install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]", - "run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]", - "uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]" + "install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]" } }, "config": { @@ -136,22 +112,6 @@ "pick_firmware_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_thread%]" } }, - "install_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]" - }, - "run_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]" - }, - "uninstall_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::description%]" - }, - "zigbee_flasher_failed": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]" - }, "confirm_zigbee": { "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]", "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]" @@ -191,9 +151,7 @@ "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]", "start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", "start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", - "install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]", - "run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]", - "uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]" + "install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]" } }, "exceptions": { diff --git a/homeassistant/components/homeassistant_yellow/config_flow.py b/homeassistant/components/homeassistant_yellow/config_flow.py index 1fac6bcac96..db844d0b0e9 100644 --- a/homeassistant/components/homeassistant_yellow/config_flow.py +++ b/homeassistant/components/homeassistant_yellow/config_flow.py @@ -5,7 +5,7 @@ from __future__ import annotations from abc import ABC, abstractmethod import asyncio import logging -from typing import Any, final +from typing import TYPE_CHECKING, Any, Protocol, final import aiohttp import voluptuous as vol @@ -31,6 +31,7 @@ from homeassistant.components.homeassistant_hardware.util import ( from homeassistant.config_entries import ( SOURCE_HARDWARE, ConfigEntry, + ConfigEntryBaseFlow, ConfigFlowResult, OptionsFlow, ) @@ -41,6 +42,7 @@ from .const import ( DOMAIN, FIRMWARE, FIRMWARE_VERSION, + NABU_CASA_FIRMWARE_RELEASES_URL, RADIO_DEVICE, ZHA_DOMAIN, ZHA_HW_DISCOVERY_DATA, @@ -57,8 +59,59 @@ STEP_HW_SETTINGS_SCHEMA = vol.Schema( } ) +if TYPE_CHECKING: -class HomeAssistantYellowConfigFlow(BaseFirmwareConfigFlow, domain=DOMAIN): + class FirmwareInstallFlowProtocol(Protocol): + """Protocol describing `BaseFirmwareInstallFlow` for a mixin.""" + + async def _install_firmware_step( + self, + fw_update_url: str, + fw_type: str, + firmware_name: str, + expected_installed_firmware_type: ApplicationType, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: ... + +else: + # Multiple inheritance with `Protocol` seems to break + FirmwareInstallFlowProtocol = object + + +class YellowFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol): + """Mixin for Home Assistant Yellow firmware methods.""" + + async def async_step_install_zigbee_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Zigbee firmware.""" + return await self._install_firmware_step( + fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL, + fw_type="yellow_zigbee_ncp", + firmware_name="Zigbee", + expected_installed_firmware_type=ApplicationType.EZSP, + step_id="install_zigbee_firmware", + next_step_id="confirm_zigbee", + ) + + async def async_step_install_thread_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Thread firmware.""" + return await self._install_firmware_step( + fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL, + fw_type="yellow_openthread_rcp", + firmware_name="OpenThread", + expected_installed_firmware_type=ApplicationType.SPINEL, + step_id="install_thread_firmware", + next_step_id="start_otbr_addon", + ) + + +class HomeAssistantYellowConfigFlow( + YellowFirmwareMixin, BaseFirmwareConfigFlow, domain=DOMAIN +): """Handle a config flow for Home Assistant Yellow.""" VERSION = 1 @@ -275,7 +328,9 @@ class HomeAssistantYellowMultiPanOptionsFlowHandler( class HomeAssistantYellowOptionsFlowHandler( - BaseHomeAssistantYellowOptionsFlow, BaseFirmwareOptionsFlow + YellowFirmwareMixin, + BaseHomeAssistantYellowOptionsFlow, + BaseFirmwareOptionsFlow, ): """Handle a firmware options flow for Home Assistant Yellow.""" diff --git a/homeassistant/components/homeassistant_yellow/strings.json b/homeassistant/components/homeassistant_yellow/strings.json index 41c1438b234..980052f9ffb 100644 --- a/homeassistant/components/homeassistant_yellow/strings.json +++ b/homeassistant/components/homeassistant_yellow/strings.json @@ -71,16 +71,6 @@ "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%]" - }, "pick_firmware": { "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]", "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::description%]", @@ -89,18 +79,6 @@ "pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]" } }, - "install_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]" - }, - "run_zigbee_flasher_addon": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]" - }, - "zigbee_flasher_failed": { - "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]", - "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]" - }, "confirm_zigbee": { "title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]", "description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]" @@ -145,9 +123,7 @@ "install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]", "start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", "start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]", - "install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]", - "run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]", - "uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]" + "install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]" } }, "entity": { diff --git a/tests/components/homeassistant_hardware/test_config_flow.py b/tests/components/homeassistant_hardware/test_config_flow.py index 2d5067bea3e..530308fdf41 100644 --- a/tests/components/homeassistant_hardware/test_config_flow.py +++ b/tests/components/homeassistant_hardware/test_config_flow.py @@ -4,9 +4,15 @@ import asyncio from collections.abc import Awaitable, Callable, Generator, Iterator import contextlib from typing import Any -from unittest.mock import AsyncMock, Mock, call, patch +from unittest.mock import AsyncMock, MagicMock, Mock, call, patch +from ha_silabs_firmware_client import ( + FirmwareManifest, + FirmwareMetadata, + FirmwareUpdateClient, +) import pytest +from yarl import URL from homeassistant.components.hassio import AddonInfo, AddonState from homeassistant.components.homeassistant_hardware.firmware_config_flow import ( @@ -19,12 +25,13 @@ from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, FirmwareInfo, get_otbr_addon_manager, - get_zigbee_flasher_addon_manager, ) from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResultType +from homeassistant.exceptions import HomeAssistantError from homeassistant.setup import async_setup_component +from homeassistant.util.dt import utcnow from tests.common import ( MockConfigEntry, @@ -37,6 +44,7 @@ from tests.common import ( TEST_DOMAIN = "test_firmware_domain" TEST_DEVICE = "/dev/SomeDevice123" TEST_HARDWARE_NAME = "Some Hardware Name" +TEST_RELEASES_URL = URL("http://invalid/releases") class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN): @@ -62,6 +70,32 @@ class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN): return await self.async_step_confirm() + async def async_step_install_zigbee_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Zigbee firmware.""" + return await self._install_firmware_step( + fw_update_url=TEST_RELEASES_URL, + fw_type="fake_zigbee_ncp", + firmware_name="Zigbee", + expected_installed_firmware_type=ApplicationType.EZSP, + step_id="install_zigbee_firmware", + next_step_id="confirm_zigbee", + ) + + async def async_step_install_thread_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Thread firmware.""" + return await self._install_firmware_step( + fw_update_url=TEST_RELEASES_URL, + fw_type="fake_openthread_rcp", + firmware_name="Thread", + expected_installed_firmware_type=ApplicationType.SPINEL, + step_id="install_thread_firmware", + next_step_id="start_otbr_addon", + ) + def _async_flow_finished(self) -> ConfigFlowResult: """Create the config entry.""" assert self._device is not None @@ -99,6 +133,18 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow): # Regenerate the translation placeholders self._get_translation_placeholders() + async def async_step_install_zigbee_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Zigbee firmware.""" + return await self.async_step_confirm_zigbee() + + async def async_step_install_thread_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Install Thread firmware.""" + return await self.async_step_start_otbr_addon() + def _async_flow_finished(self) -> ConfigFlowResult: """Create the config entry.""" assert self._probed_firmware_info is not None @@ -146,12 +192,22 @@ def delayed_side_effect() -> Callable[..., Awaitable[None]]: return side_effect +def create_mock_owner() -> Mock: + """Mock for OwningAddon / OwningIntegration.""" + owner = Mock() + owner.is_running = AsyncMock(return_value=True) + owner.temporarily_stop = MagicMock() + owner.temporarily_stop.return_value.__aenter__.return_value = AsyncMock() + + return owner + + @contextlib.contextmanager -def mock_addon_info( +def mock_firmware_info( hass: HomeAssistant, *, is_hassio: bool = True, - app_type: ApplicationType | None = ApplicationType.EZSP, + probe_app_type: ApplicationType | None = ApplicationType.EZSP, otbr_addon_info: AddonInfo = AddonInfo( available=True, hostname=None, @@ -160,29 +216,9 @@ def mock_addon_info( update_available=False, version=None, ), - flasher_addon_info: AddonInfo = AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.NOT_INSTALLED, - update_available=False, - version=None, - ), + flash_app_type: ApplicationType = ApplicationType.EZSP, ) -> Iterator[tuple[Mock, Mock]]: """Mock the main addon states for the config flow.""" - mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) - mock_flasher_manager.addon_name = "Silicon Labs Flasher" - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_get_addon_info.return_value = flasher_addon_info - mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) mock_otbr_manager.addon_name = "OpenThread Border Router" mock_otbr_manager.async_install_addon_waiting = AsyncMock( @@ -196,17 +232,73 @@ def mock_addon_info( ) mock_otbr_manager.async_get_addon_info.return_value = otbr_addon_info - if app_type is None: - firmware_info_result = None + mock_update_client = AsyncMock(spec_set=FirmwareUpdateClient) + mock_update_client.async_update_data.return_value = FirmwareManifest( + url=TEST_RELEASES_URL, + html_url=TEST_RELEASES_URL / "html", + created_at=utcnow(), + firmwares=[ + FirmwareMetadata( + filename="fake_openthread_rcp_7.4.4.0_variant.gbl", + checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef", + size=123, + release_notes="Some release notes", + metadata={}, + url=TEST_RELEASES_URL / "fake_openthread_rcp_7.4.4.0_variant.gbl", + ), + FirmwareMetadata( + filename="fake_zigbee_ncp_7.4.4.0_variant.gbl", + checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef", + size=123, + release_notes="Some release notes", + metadata={}, + url=TEST_RELEASES_URL / "fake_zigbee_ncp_7.4.4.0_variant.gbl", + ), + ], + ) + + if probe_app_type is None: + probed_firmware_info = None else: - firmware_info_result = FirmwareInfo( + probed_firmware_info = FirmwareInfo( device="/dev/ttyUSB0", # Not used - firmware_type=app_type, + firmware_type=probe_app_type, firmware_version=None, owners=[], source="probe", ) + if flash_app_type is None: + flashed_firmware_info = None + else: + flashed_firmware_info = FirmwareInfo( + device=TEST_DEVICE, + firmware_type=flash_app_type, + firmware_version="7.4.4.0", + owners=[create_mock_owner()], + source="probe", + ) + + async def mock_flash_firmware( + hass: HomeAssistant, + device: str, + fw_data: bytes, + expected_installed_firmware_type: ApplicationType, + bootloader_reset_type: str | None = None, + progress_callback: Callable[[int, int], None] | None = None, + ) -> FirmwareInfo: + await asyncio.sleep(0) + progress_callback(0, 100) + await asyncio.sleep(0) + progress_callback(50, 100) + await asyncio.sleep(0) + progress_callback(100, 100) + + if flashed_firmware_info is None: + raise HomeAssistantError("Failed to probe the firmware after flashing") + + return flashed_firmware_info + with ( patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager", @@ -216,10 +308,6 @@ def mock_addon_info( "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager", return_value=mock_otbr_manager, ), - patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_zigbee_flasher_addon_manager", - return_value=mock_flasher_manager, - ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio", return_value=is_hassio, @@ -229,81 +317,85 @@ def mock_addon_info( return_value=is_hassio, ), patch( + # We probe once before installation and once after "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info", - return_value=firmware_info_result, + side_effect=(probed_firmware_info, flashed_firmware_info), + ), + patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.FirmwareUpdateClient", + return_value=mock_update_client, + ), + patch( + "homeassistant.components.homeassistant_hardware.util.parse_firmware_image" + ), + patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.async_flash_silabs_firmware", + side_effect=mock_flash_firmware, ), ): - yield mock_otbr_manager, mock_flasher_manager + yield mock_otbr_manager + + +async def consume_progress_flow( + hass: HomeAssistant, + flow_id: str, + valid_step_ids: tuple[str], +) -> ConfigFlowResult: + """Consume a progress flow until it is done.""" + while True: + result = await hass.config_entries.flow.async_configure(flow_id) + flow_id = result["flow_id"] + + if result["type"] != FlowResultType.SHOW_PROGRESS: + break + + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] in valid_step_ids + + await asyncio.sleep(0.1) + + return result async def test_config_flow_zigbee(hass: HomeAssistant) -> None: """Test the config flow.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - assert result["type"] is FlowResultType.MENU - assert result["step_id"] == "pick_firmware" + assert init_result["type"] is FlowResultType.MENU + assert init_result["step_id"] == "pick_firmware" - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): - # Pick the menu option: we are now installing the addon - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + probe_app_type=ApplicationType.SPINEL, + flash_app_type=ApplicationType.EZSP, + ): + # Pick the menu option: we are flashing the firmware + pick_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["progress_action"] == "install_addon" - assert result["step_id"] == "install_zigbee_flasher_addon" - assert result["description_placeholders"]["firmware_type"] == "spinel" - await hass.async_block_till_done(wait_background_tasks=True) + assert pick_result["type"] is FlowResultType.SHOW_PROGRESS + assert pick_result["progress_action"] == "install_firmware" + assert pick_result["step_id"] == "install_zigbee_firmware" - # Progress the flow, we are now configuring the addon and running it - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "run_zigbee_flasher_addon" - assert result["progress_action"] == "run_zigbee_flasher_addon" - assert mock_flasher_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, - "baudrate": 115200, - "bootloader_baudrate": 115200, - "flow_control": True, - } - ) - ] - - await hass.async_block_till_done(wait_background_tasks=True) - - # Progress the flow, we are now uninstalling the addon - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "uninstall_zigbee_flasher_addon" - assert result["progress_action"] == "uninstall_zigbee_flasher_addon" - - await hass.async_block_till_done(wait_background_tasks=True) - - # We are finally done with the addon - assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()] - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_zigbee" - - with mock_addon_info( - hass, - app_type=ApplicationType.EZSP, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + confirm_result = await consume_progress_flow( + hass, + flow_id=pick_result["flow_id"], + valid_step_ids=("install_zigbee_firmware",), ) - assert result["type"] is FlowResultType.CREATE_ENTRY - config_entry = result["result"] + assert confirm_result["type"] is FlowResultType.FORM + assert confirm_result["step_id"] == "confirm_zigbee" + + create_result = await hass.config_entries.flow.async_configure( + confirm_result["flow_id"], user_input={} + ) + assert create_result["type"] is FlowResultType.CREATE_ENTRY + + config_entry = create_result["result"] assert config_entry.data == { "firmware": "ezsp", "device": TEST_DEVICE, @@ -328,52 +420,20 @@ async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> assert result["type"] is FlowResultType.MENU assert result["step_id"] == "pick_firmware" - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - flasher_addon_info=AddonInfo( - available=True, - hostname=None, - options={ - "device": "", - "baudrate": 115200, - "bootloader_baudrate": 115200, - "flow_control": True, - }, - state=AddonState.NOT_RUNNING, - update_available=False, - version="1.2.3", - ), - ) as (mock_otbr_manager, mock_flasher_manager): + with mock_firmware_info(hass, probe_app_type=ApplicationType.SPINEL): # Pick the menu option: we skip installation, instead we directly run it result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "run_zigbee_flasher_addon" - assert result["progress_action"] == "run_zigbee_flasher_addon" - assert result["description_placeholders"]["firmware_type"] == "spinel" - assert mock_flasher_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, - "baudrate": 115200, - "bootloader_baudrate": 115200, - "flow_control": True, - } - ) - ] - - # Uninstall the addon - await hass.async_block_till_done(wait_background_tasks=True) + # Confirm result = await hass.config_entries.flow.async_configure(result["flow_id"]) # Done - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, + probe_app_type=ApplicationType.EZSP, ): await hass.async_block_till_done(wait_background_tasks=True) result = await hass.config_entries.flow.async_configure(result["flow_id"]) @@ -409,28 +469,29 @@ async def test_config_flow_auto_confirm_if_running(hass: HomeAssistant) -> None: async def test_config_flow_thread(hass: HomeAssistant) -> None: """Test the config flow.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - assert result["type"] is FlowResultType.MENU - assert result["step_id"] == "pick_firmware" + assert init_result["type"] is FlowResultType.MENU + assert init_result["step_id"] == "pick_firmware" - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + flash_app_type=ApplicationType.SPINEL, + ) as mock_otbr_manager: # Pick the menu option - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + pick_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["progress_action"] == "install_addon" - assert result["step_id"] == "install_otbr_addon" - assert result["description_placeholders"]["firmware_type"] == "ezsp" - assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME + assert pick_result["type"] is FlowResultType.SHOW_PROGRESS + assert pick_result["progress_action"] == "install_addon" + assert pick_result["step_id"] == "install_otbr_addon" + assert pick_result["description_placeholders"]["firmware_type"] == "ezsp" + assert pick_result["description_placeholders"]["model"] == TEST_HARDWARE_NAME await hass.async_block_till_done(wait_background_tasks=True) @@ -441,19 +502,37 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None: "device": "", "baudrate": 460800, "flow_control": True, - "autoflash_firmware": True, + "autoflash_firmware": False, }, state=AddonState.NOT_RUNNING, update_available=False, version="1.2.3", ) - # Progress the flow, it is now configuring the addon and running it - result = await hass.config_entries.flow.async_configure(result["flow_id"]) + # Progress the flow, it is now installing firmware + confirm_otbr_result = await consume_progress_flow( + hass, + flow_id=pick_result["flow_id"], + valid_step_ids=( + "pick_firmware_thread", + "install_otbr_addon", + "install_thread_firmware", + "start_otbr_addon", + ), + ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "start_otbr_addon" - assert result["progress_action"] == "start_otbr_addon" + # Installation will conclude with the config entry being created + create_result = await hass.config_entries.flow.async_configure( + confirm_otbr_result["flow_id"], user_input={} + ) + assert create_result["type"] is FlowResultType.CREATE_ENTRY + + config_entry = create_result["result"] + assert config_entry.data == { + "firmware": "spinel", + "device": TEST_DEVICE, + "hardware": TEST_HARDWARE_NAME, + } assert mock_otbr_manager.async_set_addon_options.mock_calls == [ call( @@ -461,44 +540,22 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None: "device": TEST_DEVICE, "baudrate": 460800, "flow_control": True, - "autoflash_firmware": True, + "autoflash_firmware": False, } ) ] - await hass.async_block_till_done(wait_background_tasks=True) - - # The addon is now running - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_otbr" - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.CREATE_ENTRY - - config_entry = result["result"] - assert config_entry.data == { - "firmware": "spinel", - "device": TEST_DEVICE, - "hardware": TEST_HARDWARE_NAME, - } - async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -> None: """Test the Thread config flow, addon is already installed.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, + probe_app_type=ApplicationType.EZSP, + flash_app_type=ApplicationType.SPINEL, otbr_addon_info=AddonInfo( available=True, hostname=None, @@ -507,81 +564,50 @@ async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) - update_available=False, version=None, ), - ) as (mock_otbr_manager, mock_flasher_manager): + ) as mock_otbr_manager: # Pick the menu option - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + pick_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "start_otbr_addon" - assert result["progress_action"] == "start_otbr_addon" + # Progress + confirm_otbr_result = await consume_progress_flow( + hass, + flow_id=pick_result["flow_id"], + valid_step_ids=( + "pick_firmware_thread", + "install_thread_firmware", + "start_otbr_addon", + ), + ) + + # We're now waiting to confirm OTBR + assert confirm_otbr_result["type"] is FlowResultType.FORM + assert confirm_otbr_result["step_id"] == "confirm_otbr" + + # The addon has been installed assert mock_otbr_manager.async_set_addon_options.mock_calls == [ call( { "device": TEST_DEVICE, "baudrate": 460800, "flow_control": True, - "autoflash_firmware": True, + "autoflash_firmware": False, # And firmware flashing is disabled } ) ] - await hass.async_block_till_done(wait_background_tasks=True) - - # The addon is now running - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_otbr" - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + # Finally, create the config entry + create_result = await hass.config_entries.flow.async_configure( + confirm_otbr_result["flow_id"], user_input={} ) - assert result["type"] is FlowResultType.CREATE_ENTRY - - -async def test_config_flow_zigbee_not_hassio(hass: HomeAssistant) -> None: - """Test when the stick is used with a non-hassio setup.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - is_hassio=False, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_zigbee" - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.CREATE_ENTRY - - config_entry = result["result"] - assert config_entry.data == { - "firmware": "ezsp", - "device": TEST_DEVICE, - "hardware": TEST_HARDWARE_NAME, - } - - # Ensure a ZHA discovery flow has been created - flows = hass.config_entries.flow.async_progress() - assert len(flows) == 1 - zha_flow = flows[0] - assert zha_flow["handler"] == "zha" - assert zha_flow["context"]["source"] == "hardware" - assert zha_flow["step_id"] == "confirm" + assert create_result["type"] is FlowResultType.CREATE_ENTRY + assert create_result["result"].data == { + "firmware": "spinel", + "device": TEST_DEVICE, + "hardware": TEST_HARDWARE_NAME, + } @pytest.mark.usefixtures("addon_store_info") @@ -601,10 +627,11 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: assert await hass.config_entries.async_setup(config_entry.entry_id) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + flash_app_type=ApplicationType.SPINEL, + ) as mock_otbr_manager: # First step is confirmation result = await hass.config_entries.options.async_init(config_entry.entry_id) assert result["type"] is FlowResultType.MENU @@ -630,7 +657,7 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: "device": "", "baudrate": 460800, "flow_control": True, - "autoflash_firmware": True, + "autoflash_firmware": False, }, state=AddonState.NOT_RUNNING, update_available=False, @@ -650,7 +677,7 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: "device": TEST_DEVICE, "baudrate": 460800, "flow_control": True, - "autoflash_firmware": True, + "autoflash_firmware": False, } ) ] @@ -662,10 +689,6 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None: assert result["type"] is FlowResultType.FORM assert result["step_id"] == "confirm_otbr" - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ): # We are now done result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={} @@ -700,57 +723,23 @@ async def test_options_flow_thread_to_zigbee(hass: HomeAssistant) -> None: assert result["description_placeholders"]["firmware_type"] == "spinel" assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.SPINEL, + ): # Pick the menu option: we are now installing the addon result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["progress_action"] == "install_addon" - assert result["step_id"] == "install_zigbee_flasher_addon" - - await hass.async_block_till_done(wait_background_tasks=True) - - # Progress the flow, we are now configuring the addon and running it - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "run_zigbee_flasher_addon" - assert result["progress_action"] == "run_zigbee_flasher_addon" - assert mock_flasher_manager.async_set_addon_options.mock_calls == [ - call( - { - "device": TEST_DEVICE, - "baudrate": 115200, - "bootloader_baudrate": 115200, - "flow_control": True, - } - ) - ] - - await hass.async_block_till_done(wait_background_tasks=True) - - # Progress the flow, we are now uninstalling the addon - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "uninstall_zigbee_flasher_addon" - assert result["progress_action"] == "uninstall_zigbee_flasher_addon" - - await hass.async_block_till_done(wait_background_tasks=True) - - # We are finally done with the addon - assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()] result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] is FlowResultType.FORM assert result["step_id"] == "confirm_zigbee" - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, + probe_app_type=ApplicationType.EZSP, ): # We are now done result = await hass.config_entries.options.async_configure( diff --git a/tests/components/homeassistant_hardware/test_config_flow_failures.py b/tests/components/homeassistant_hardware/test_config_flow_failures.py index 38c2696a62a..65a5f58b17d 100644 --- a/tests/components/homeassistant_hardware/test_config_flow_failures.py +++ b/tests/components/homeassistant_hardware/test_config_flow_failures.py @@ -21,8 +21,8 @@ from .test_config_flow import ( TEST_DEVICE, TEST_DOMAIN, TEST_HARDWARE_NAME, - delayed_side_effect, - mock_addon_info, + consume_progress_flow, + mock_firmware_info, mock_test_firmware_platform, # noqa: F401 ) @@ -51,10 +51,10 @@ async def test_config_flow_cannot_probe_firmware( ) -> None: """Test failure case when firmware cannot be probed.""" - with mock_addon_info( + with mock_firmware_info( hass, - app_type=None, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=None, + ): # Start the flow result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} @@ -69,283 +69,6 @@ async def test_config_flow_cannot_probe_firmware( assert result["reason"] == "unsupported_firmware" -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_not_hassio_wrong_firmware( - hass: HomeAssistant, -) -> None: - """Test when the stick is used with a non-hassio setup but the firmware is bad.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - is_hassio=False, - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "not_hassio" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_flasher_addon_already_running( - hass: HomeAssistant, -) -> None: - """Test failure case when flasher addon is already running.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - flasher_addon_info=AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ), - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - - # Cannot get addon info - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_already_running" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_flasher_addon_info_fails(hass: HomeAssistant) -> None: - """Test failure case when flasher addon cannot be installed.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - flasher_addon_info=AddonInfo( - available=True, - hostname=None, - options={}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ), - ) as (mock_otbr_manager, mock_flasher_manager): - mock_flasher_manager.async_get_addon_info.side_effect = AddonError() - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - - # Cannot get addon info - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_info_failed" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_flasher_addon_install_fails( - hass: HomeAssistant, -) -> None: - """Test failure case when flasher addon cannot be installed.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - - # Cannot install addon - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_install_failed" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_flasher_addon_set_config_fails( - hass: HomeAssistant, -) -> None: - """Test failure case when flasher addon cannot be configured.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): - mock_flasher_manager.async_install_addon_waiting = AsyncMock( - side_effect=delayed_side_effect() - ) - mock_flasher_manager.async_set_addon_options = AsyncMock( - side_effect=AddonError() - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_set_config_failed" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_flasher_run_fails(hass: HomeAssistant) -> None: - """Test failure case when flasher addon fails to run.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): - mock_flasher_manager.async_start_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_start_failed" - - -async def test_config_flow_zigbee_flasher_uninstall_fails(hass: HomeAssistant) -> None: - """Test failure case when flasher addon uninstall fails.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.SPINEL, - ) as (mock_otbr_manager, mock_flasher_manager): - mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=AddonError() - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - # Uninstall failure isn't critical - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_zigbee" - - -@pytest.mark.parametrize( - "ignore_translations_for_mock_domains", - ["test_firmware_domain"], -) -async def test_config_flow_zigbee_confirmation_fails(hass: HomeAssistant) -> None: - """Test the config flow failing due to Zigbee firmware not being detected.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - assert result["type"] is FlowResultType.MENU - assert result["step_id"] == "pick_firmware" - - with mock_addon_info( - hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): - # Pick the menu option: we are now installing the addon - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, - ) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_zigbee" - - with mock_addon_info( - hass, - app_type=None, # Probing fails - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "unsupported_firmware" - - @pytest.mark.parametrize( "ignore_translations_for_mock_domains", ["test_firmware_domain"], @@ -356,11 +79,11 @@ async def test_config_flow_thread_not_hassio(hass: HomeAssistant) -> None: TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, is_hassio=False, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + ): result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} ) @@ -383,10 +106,10 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None: TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + ) as mock_otbr_manager: mock_otbr_manager.async_get_addon_info.side_effect = AddonError() result = await hass.config_entries.flow.async_configure( result["flow_id"], user_input={} @@ -405,24 +128,26 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None: "ignore_translations_for_mock_domains", ["test_firmware_domain"], ) -async def test_config_flow_thread_addon_already_running(hass: HomeAssistant) -> None: +async def test_config_flow_thread_addon_already_configured(hass: HomeAssistant) -> None: """Test failure case when the Thread addon is already running.""" result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, + probe_app_type=ApplicationType.EZSP, otbr_addon_info=AddonInfo( available=True, hostname=None, - options={}, + options={ + "device": TEST_DEVICE + "2", # A different device + }, state=AddonState.RUNNING, update_available=False, version="1.0.0", ), - ) as (mock_otbr_manager, mock_flasher_manager): + ) as mock_otbr_manager: mock_otbr_manager.async_install_addon_waiting = AsyncMock( side_effect=AddonError() ) @@ -450,10 +175,10 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + ) as mock_otbr_manager: mock_otbr_manager.async_install_addon_waiting = AsyncMock( side_effect=AddonError() ) @@ -477,29 +202,51 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No ) async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> None: """Test failure case when flasher addon cannot be configured.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + ) as mock_otbr_manager: + + async def install_addon() -> None: + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": TEST_DEVICE}, + state=AddonState.NOT_RUNNING, + update_available=False, + version="1.0.0", + ) + + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=install_addon + ) mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + confirm_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={} ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + + pick_thread_result = await hass.config_entries.flow.async_configure( + confirm_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_set_config_failed" + pick_thread_progress_result = await consume_progress_flow( + hass, + flow_id=pick_thread_result["flow_id"], + valid_step_ids=( + "pick_firmware_thread", + "install_thread_firmware", + "start_otbr_addon", + ), + ) + + assert pick_thread_progress_result["type"] == FlowResultType.ABORT + assert pick_thread_progress_result["reason"] == "addon_set_config_failed" @pytest.mark.parametrize( @@ -508,63 +255,45 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> ) async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None: """Test failure case when flasher addon fails to run.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): + probe_app_type=ApplicationType.EZSP, + otbr_addon_info=AddonInfo( + available=True, + hostname=None, + options={"device": TEST_DEVICE}, + state=AddonState.NOT_RUNNING, + update_available=False, + version="1.0.0", + ), + ) as mock_otbr_manager: mock_otbr_manager.async_start_addon_waiting = AsyncMock( side_effect=AddonError() ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + confirm_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={} ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + pick_thread_result = await hass.config_entries.flow.async_configure( + confirm_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] == FlowResultType.ABORT - assert result["reason"] == "addon_start_failed" - - -async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -> None: - """Test failure case when flasher addon uninstall fails.""" - result = await hass.config_entries.flow.async_init( - TEST_DOMAIN, context={"source": "hardware"} - ) - - with mock_addon_info( - hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): - mock_otbr_manager.async_uninstall_addon_waiting = AsyncMock( - side_effect=AddonError() + pick_thread_progress_result = await consume_progress_flow( + hass, + flow_id=pick_thread_result["flow_id"], + valid_step_ids=( + "pick_firmware_thread", + "install_thread_firmware", + "start_otbr_addon", + ), ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} - ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, - ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - # Uninstall failure isn't critical - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_otbr" + assert pick_thread_progress_result["type"] == FlowResultType.ABORT + assert pick_thread_progress_result["reason"] == "addon_start_failed" @pytest.mark.parametrize( @@ -573,40 +302,43 @@ async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) - ) async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> None: """Test the config flow failing due to OpenThread firmware not being detected.""" - result = await hass.config_entries.flow.async_init( + init_result = await hass.config_entries.flow.async_init( TEST_DOMAIN, context={"source": "hardware"} ) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.EZSP, - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + probe_app_type=ApplicationType.EZSP, + flash_app_type=None, + otbr_addon_info=AddonInfo( + available=True, + hostname=None, + options={"device": TEST_DEVICE}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ), + ): + confirm_result = await hass.config_entries.flow.async_configure( + init_result["flow_id"], user_input={} ) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], + pick_thread_result = await hass.config_entries.flow.async_configure( + confirm_result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, ) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - await hass.async_block_till_done(wait_background_tasks=True) - - result = await hass.config_entries.flow.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "confirm_otbr" - - with mock_addon_info( - hass, - app_type=None, # Probing fails - ) as (mock_otbr_manager, mock_flasher_manager): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], user_input={} + pick_thread_progress_result = await consume_progress_flow( + hass, + flow_id=pick_thread_result["flow_id"], + valid_step_ids=( + "pick_firmware_thread", + "install_thread_firmware", + "start_otbr_addon", + ), ) - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "unsupported_firmware" + + assert pick_thread_progress_result["type"] is FlowResultType.ABORT + assert pick_thread_progress_result["reason"] == "unsupported_firmware" @pytest.mark.parametrize( @@ -683,9 +415,9 @@ async def test_options_flow_thread_to_zigbee_otbr_configured( # Confirm options flow result = await hass.config_entries.options.async_init(config_entry.entry_id) - with mock_addon_info( + with mock_firmware_info( hass, - app_type=ApplicationType.SPINEL, + probe_app_type=ApplicationType.SPINEL, otbr_addon_info=AddonInfo( available=True, hostname=None, @@ -694,7 +426,7 @@ async def test_options_flow_thread_to_zigbee_otbr_configured( update_available=False, version="1.0.0", ), - ) as (mock_otbr_manager, mock_flasher_manager): + ): result = await hass.config_entries.options.async_configure( result["flow_id"], user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, diff --git a/tests/components/homeassistant_hardware/test_update.py b/tests/components/homeassistant_hardware/test_update.py index 81c6f2e0459..aacc064e4f2 100644 --- a/tests/components/homeassistant_hardware/test_update.py +++ b/tests/components/homeassistant_hardware/test_update.py @@ -3,10 +3,10 @@ from __future__ import annotations import asyncio -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, Callable import dataclasses import logging -from unittest.mock import AsyncMock, Mock, patch +from unittest.mock import Mock, patch import aiohttp from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata @@ -355,10 +355,14 @@ async def test_update_entity_installation( "https://example.org/release_notes" ) - mock_firmware = Mock() - mock_flasher = AsyncMock() - - async def mock_flash_firmware(fw_image, progress_callback): + async def mock_flash_firmware( + hass: HomeAssistant, + device: str, + fw_data: bytes, + expected_installed_firmware_type: ApplicationType, + bootloader_reset_type: str | None = None, + progress_callback: Callable[[int, int], None] | None = None, + ) -> FirmwareInfo: await asyncio.sleep(0) progress_callback(0, 100) await asyncio.sleep(0) @@ -366,31 +370,20 @@ async def test_update_entity_installation( await asyncio.sleep(0) progress_callback(100, 100) - mock_flasher.flash_firmware = mock_flash_firmware + return FirmwareInfo( + device=TEST_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="7.4.4.0 build 0", + owners=[], + source="probe", + ) # When we install it, the other integration is reloaded with ( patch( - "homeassistant.components.homeassistant_hardware.update.parse_firmware_image", - return_value=mock_firmware, + "homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware", + side_effect=mock_flash_firmware, ), - patch( - "homeassistant.components.homeassistant_hardware.update.Flasher", - return_value=mock_flasher, - ), - patch( - "homeassistant.components.homeassistant_hardware.update.probe_silabs_firmware_info", - return_value=FirmwareInfo( - device=TEST_DEVICE, - firmware_type=ApplicationType.EZSP, - firmware_version="7.4.4.0 build 0", - owners=[], - source="probe", - ), - ), - patch.object( - owning_config_entry, "async_unload", wraps=owning_config_entry.async_unload - ) as owning_config_entry_unload, ): state_changes: list[Event[EventStateChangedData]] = async_capture_events( hass, EVENT_STATE_CHANGED @@ -423,9 +416,6 @@ async def test_update_entity_installation( assert state_changes[6].data["new_state"].attributes["update_percentage"] is None assert state_changes[6].data["new_state"].attributes["in_progress"] is False - # The owning integration was unloaded and is again running - assert len(owning_config_entry_unload.mock_calls) == 1 - # After the firmware update, the entity has the new version and the correct state state_after_install = hass.states.get(TEST_UPDATE_ENTITY_ID) assert state_after_install is not None @@ -456,19 +446,10 @@ async def test_update_entity_installation_failure( assert state_before_install.attributes["installed_version"] == "7.3.1.0" assert state_before_install.attributes["latest_version"] == "7.4.4.0" - mock_flasher = AsyncMock() - mock_flasher.flash_firmware.side_effect = RuntimeError( - "Something broke during flashing!" - ) - with ( patch( - "homeassistant.components.homeassistant_hardware.update.parse_firmware_image", - return_value=Mock(), - ), - patch( - "homeassistant.components.homeassistant_hardware.update.Flasher", - return_value=mock_flasher, + "homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware", + side_effect=HomeAssistantError("Failed to flash firmware"), ), pytest.raises(HomeAssistantError, match="Failed to flash firmware"), ): @@ -511,16 +492,10 @@ async def test_update_entity_installation_probe_failure( with ( patch( - "homeassistant.components.homeassistant_hardware.update.parse_firmware_image", - return_value=Mock(), - ), - patch( - "homeassistant.components.homeassistant_hardware.update.Flasher", - return_value=AsyncMock(), - ), - patch( - "homeassistant.components.homeassistant_hardware.update.probe_silabs_firmware_info", - return_value=None, + "homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware", + side_effect=HomeAssistantError( + "Failed to probe the firmware after flashing" + ), ), pytest.raises( HomeAssistantError, match="Failed to probe the firmware after flashing" diff --git a/tests/components/homeassistant_hardware/test_util.py b/tests/components/homeassistant_hardware/test_util.py index 1b7bfe4a8ac..048bf998d13 100644 --- a/tests/components/homeassistant_hardware/test_util.py +++ b/tests/components/homeassistant_hardware/test_util.py @@ -1,10 +1,13 @@ """Test hardware utilities.""" -from unittest.mock import AsyncMock, MagicMock, patch +import asyncio +from collections.abc import Callable +from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch import pytest from universal_silabs_flasher.common import Version as FlasherVersion from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType +from universal_silabs_flasher.firmware import GBLImage from homeassistant.components.hassio import ( AddonError, @@ -20,6 +23,7 @@ from homeassistant.components.homeassistant_hardware.util import ( FirmwareInfo, OwningAddon, OwningIntegration, + async_flash_silabs_firmware, get_otbr_addon_firmware_info, guess_firmware_info, probe_silabs_firmware_info, @@ -27,8 +31,11 @@ from homeassistant.components.homeassistant_hardware.util import ( ) from homeassistant.config_entries import ConfigEntry, ConfigEntryState from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError from homeassistant.setup import async_setup_component +from .test_config_flow import create_mock_owner + from tests.common import MockConfigEntry ZHA_CONFIG_ENTRY = MockConfigEntry( @@ -526,3 +533,201 @@ async def test_probe_silabs_firmware_type( ): result = await probe_silabs_firmware_type("/dev/ttyUSB0") assert result == expected + + +async def test_async_flash_silabs_firmware(hass: HomeAssistant) -> None: + """Test async_flash_silabs_firmware.""" + owner1 = create_mock_owner() + owner2 = create_mock_owner() + + progress_callback = Mock() + + async def mock_flash_firmware( + fw_image: GBLImage, progress_callback: Callable[[int, int], None] + ) -> None: + """Mock flash firmware function.""" + await asyncio.sleep(0) + progress_callback(0, 100) + await asyncio.sleep(0) + progress_callback(50, 100) + await asyncio.sleep(0) + progress_callback(100, 100) + await asyncio.sleep(0) + + mock_flasher = Mock() + mock_flasher.enter_bootloader = AsyncMock() + mock_flasher.flash_firmware = AsyncMock(side_effect=mock_flash_firmware) + + expected_firmware_info = FirmwareInfo( + device="/dev/ttyUSB0", + firmware_type=ApplicationType.SPINEL, + firmware_version=None, + source="probe", + owners=[], + ) + + with ( + patch( + "homeassistant.components.homeassistant_hardware.util.guess_firmware_info", + return_value=FirmwareInfo( + device="/dev/ttyUSB0", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="unknown", + owners=[owner1, owner2], + ), + ), + patch( + "homeassistant.components.homeassistant_hardware.util.Flasher", + return_value=mock_flasher, + ), + patch( + "homeassistant.components.homeassistant_hardware.util.parse_firmware_image" + ), + patch( + "homeassistant.components.homeassistant_hardware.util.probe_silabs_firmware_info", + return_value=expected_firmware_info, + ), + ): + after_flash_info = await async_flash_silabs_firmware( + hass=hass, + device="/dev/ttyUSB0", + fw_data=b"firmware contents", + expected_installed_firmware_type=ApplicationType.SPINEL, + bootloader_reset_type=None, + progress_callback=progress_callback, + ) + + assert progress_callback.mock_calls == [call(0, 100), call(50, 100), call(100, 100)] + assert after_flash_info == expected_firmware_info + + # Both owning integrations/addons are stopped and restarted + assert owner1.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, None, None, None), + ] + + assert owner2.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, None, None, None), + ] + + +async def test_async_flash_silabs_firmware_flash_failure(hass: HomeAssistant) -> None: + """Test async_flash_silabs_firmware flash failure.""" + owner1 = create_mock_owner() + owner2 = create_mock_owner() + + mock_flasher = Mock() + mock_flasher.enter_bootloader = AsyncMock() + mock_flasher.flash_firmware = AsyncMock(side_effect=RuntimeError("Failure!")) + + with ( + patch( + "homeassistant.components.homeassistant_hardware.util.guess_firmware_info", + return_value=FirmwareInfo( + device="/dev/ttyUSB0", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="unknown", + owners=[owner1, owner2], + ), + ), + patch( + "homeassistant.components.homeassistant_hardware.util.Flasher", + return_value=mock_flasher, + ), + patch( + "homeassistant.components.homeassistant_hardware.util.parse_firmware_image" + ), + pytest.raises(HomeAssistantError, match="Failed to flash firmware") as exc, + ): + await async_flash_silabs_firmware( + hass=hass, + device="/dev/ttyUSB0", + fw_data=b"firmware contents", + expected_installed_firmware_type=ApplicationType.SPINEL, + bootloader_reset_type=None, + ) + + # Both owning integrations/addons are stopped and restarted + assert owner1.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, HomeAssistantError, exc.value, ANY), + ] + assert owner2.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, HomeAssistantError, exc.value, ANY), + ] + + +async def test_async_flash_silabs_firmware_probe_failure(hass: HomeAssistant) -> None: + """Test async_flash_silabs_firmware probe failure.""" + owner1 = create_mock_owner() + owner2 = create_mock_owner() + + mock_flasher = Mock() + mock_flasher.enter_bootloader = AsyncMock() + mock_flasher.flash_firmware = AsyncMock() + + with ( + patch( + "homeassistant.components.homeassistant_hardware.util.guess_firmware_info", + return_value=FirmwareInfo( + device="/dev/ttyUSB0", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="unknown", + owners=[owner1, owner2], + ), + ), + patch( + "homeassistant.components.homeassistant_hardware.util.Flasher", + return_value=mock_flasher, + ), + patch( + "homeassistant.components.homeassistant_hardware.util.parse_firmware_image" + ), + patch( + "homeassistant.components.homeassistant_hardware.util.probe_silabs_firmware_info", + return_value=None, + ), + pytest.raises( + HomeAssistantError, match="Failed to probe the firmware after flashing" + ), + ): + await async_flash_silabs_firmware( + hass=hass, + device="/dev/ttyUSB0", + fw_data=b"firmware contents", + expected_installed_firmware_type=ApplicationType.SPINEL, + bootloader_reset_type=None, + ) + + # Both owning integrations/addons are stopped and restarted + assert owner1.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, None, None, None), + ] + assert owner2.temporarily_stop.mock_calls == [ + call(hass), + # pylint: disable-next=unnecessary-dunder-call + call().__aenter__(ANY), + # pylint: disable-next=unnecessary-dunder-call + call().__aexit__(ANY, None, None, None), + ] diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index 44a5e0029c3..9dcac0732c9 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -6,6 +6,7 @@ import pytest from homeassistant.components.hassio import AddonInfo, AddonState from homeassistant.components.homeassistant_hardware.firmware_config_flow import ( + STEP_PICK_FIRMWARE_THREAD, STEP_PICK_FIRMWARE_ZIGBEE, ) from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( @@ -18,6 +19,7 @@ from homeassistant.components.homeassistant_hardware.util import ( FirmwareInfo, ) from homeassistant.components.homeassistant_sky_connect.const import DOMAIN +from homeassistant.config_entries import ConfigFlowResult from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from homeassistant.helpers.service_info.usb import UsbServiceInfo @@ -28,14 +30,31 @@ from tests.common import MockConfigEntry @pytest.mark.parametrize( - ("usb_data", "model"), + ("step", "usb_data", "model", "fw_type", "fw_version"), [ - (USB_DATA_SKY, "Home Assistant SkyConnect"), - (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ( + STEP_PICK_FIRMWARE_ZIGBEE, + USB_DATA_SKY, + "Home Assistant SkyConnect", + ApplicationType.EZSP, + "7.4.4.0 build 0", + ), + ( + STEP_PICK_FIRMWARE_THREAD, + USB_DATA_ZBT1, + "Home Assistant Connect ZBT-1", + ApplicationType.SPINEL, + "2.4.4.0", + ), ], ) async def test_config_flow( - usb_data: UsbServiceInfo, model: str, hass: HomeAssistant + step: str, + usb_data: UsbServiceInfo, + model: str, + fw_type: ApplicationType, + fw_version: str, + hass: HomeAssistant, ) -> None: """Test the config flow for SkyConnect.""" result = await hass.config_entries.flow.async_init( @@ -46,21 +65,36 @@ async def test_config_flow( assert result["step_id"] == "pick_firmware" assert result["description_placeholders"]["model"] == model - async def mock_async_step_pick_firmware_zigbee(self, data): - return await self.async_step_confirm_zigbee(user_input={}) + async def mock_install_firmware_step( + self, + fw_update_url: str, + fw_type: str, + firmware_name: str, + expected_installed_firmware_type: ApplicationType, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: + if next_step_id == "start_otbr_addon": + next_step_id = "confirm_otbr" + + return await getattr(self, f"async_step_{next_step_id}")(user_input={}) with ( patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow.async_step_pick_firmware_zigbee", + "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", + return_value=None, + ), + patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step", autospec=True, - side_effect=mock_async_step_pick_firmware_zigbee, + side_effect=mock_install_firmware_step, ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info", return_value=FirmwareInfo( device=usb_data.device, - firmware_type=ApplicationType.EZSP, - firmware_version="7.4.4.0 build 0", + firmware_type=fw_type, + firmware_version=fw_version, owners=[], source="probe", ), @@ -68,15 +102,15 @@ async def test_config_flow( ): result = await hass.config_entries.flow.async_configure( result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, + user_input={"next_step_id": step}, ) assert result["type"] is FlowResultType.CREATE_ENTRY config_entry = result["result"] assert config_entry.data == { - "firmware": "ezsp", - "firmware_version": "7.4.4.0 build 0", + "firmware": fw_type.value, + "firmware_version": fw_version, "device": usb_data.device, "manufacturer": usb_data.manufacturer, "pid": usb_data.pid, @@ -86,13 +120,17 @@ async def test_config_flow( "vid": usb_data.vid, } - # Ensure a ZHA discovery flow has been created flows = hass.config_entries.flow.async_progress() - assert len(flows) == 1 - zha_flow = flows[0] - assert zha_flow["handler"] == "zha" - assert zha_flow["context"]["source"] == "hardware" - assert zha_flow["step_id"] == "confirm" + + if step == STEP_PICK_FIRMWARE_ZIGBEE: + # Ensure a ZHA discovery flow has been created + assert len(flows) == 1 + zha_flow = flows[0] + assert zha_flow["handler"] == "zha" + assert zha_flow["context"]["source"] == "hardware" + assert zha_flow["step_id"] == "confirm" + else: + assert len(flows) == 0 @pytest.mark.parametrize( diff --git a/tests/components/homeassistant_yellow/test_config_flow.py b/tests/components/homeassistant_yellow/test_config_flow.py index 1d5a64eafb9..cd4a1941050 100644 --- a/tests/components/homeassistant_yellow/test_config_flow.py +++ b/tests/components/homeassistant_yellow/test_config_flow.py @@ -11,6 +11,7 @@ from homeassistant.components.hassio import ( AddonState, ) from homeassistant.components.homeassistant_hardware.firmware_config_flow import ( + STEP_PICK_FIRMWARE_THREAD, STEP_PICK_FIRMWARE_ZIGBEE, ) from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( @@ -23,6 +24,7 @@ from homeassistant.components.homeassistant_hardware.util import ( FirmwareInfo, ) from homeassistant.components.homeassistant_yellow.const import DOMAIN, RADIO_DEVICE +from homeassistant.config_entries import ConfigFlowResult from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from homeassistant.setup import async_setup_component @@ -305,7 +307,16 @@ async def test_option_flow_led_settings_fail_2( assert result["reason"] == "write_hw_settings_error" -async def test_firmware_options_flow(hass: HomeAssistant) -> None: +@pytest.mark.parametrize( + ("step", "fw_type", "fw_version"), + [ + (STEP_PICK_FIRMWARE_ZIGBEE, ApplicationType.EZSP, "7.4.4.0 build 0"), + (STEP_PICK_FIRMWARE_THREAD, ApplicationType.SPINEL, "2.4.4.0"), + ], +) +async def test_firmware_options_flow( + step: str, fw_type: ApplicationType, fw_version: str, hass: HomeAssistant +) -> None: """Test the firmware options flow for Yellow.""" mock_integration(hass, MockModule("hassio")) await async_setup_component(hass, HASSIO_DOMAIN, {}) @@ -339,18 +350,36 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None: async def mock_async_step_pick_firmware_zigbee(self, data): return await self.async_step_confirm_zigbee(user_input={}) + async def mock_install_firmware_step( + self, + fw_update_url: str, + fw_type: str, + firmware_name: str, + expected_installed_firmware_type: ApplicationType, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: + if next_step_id == "start_otbr_addon": + next_step_id = "confirm_otbr" + + return await getattr(self, f"async_step_{next_step_id}")(user_input={}) + with ( patch( - "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow.async_step_pick_firmware_zigbee", + "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup", + return_value=None, + ), + patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._install_firmware_step", autospec=True, - side_effect=mock_async_step_pick_firmware_zigbee, + side_effect=mock_install_firmware_step, ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info", return_value=FirmwareInfo( device=RADIO_DEVICE, - firmware_type=ApplicationType.EZSP, - firmware_version="7.4.4.0 build 0", + firmware_type=fw_type, + firmware_version=fw_version, owners=[], source="probe", ), @@ -358,15 +387,15 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None: ): result = await hass.config_entries.options.async_configure( result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, + user_input={"next_step_id": step}, ) assert result["type"] is FlowResultType.CREATE_ENTRY assert result["result"] is True assert config_entry.data == { - "firmware": "ezsp", - "firmware_version": "7.4.4.0 build 0", + "firmware": fw_type.value, + "firmware_version": fw_version, }