From 380f192c93d87287782d68e1f7b7e8a6565eea14 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 24 Apr 2024 11:06:24 -0400 Subject: [PATCH] Expose the SkyConnect integration with a firmware config/options flow (#115363) Co-authored-by: Stefan Agner Co-authored-by: Martin Hjelmare Co-authored-by: Erik --- .../homeassistant_sky_connect/__init__.py | 117 +- .../homeassistant_sky_connect/config_flow.py | 630 +++++++++- .../homeassistant_sky_connect/const.py | 11 + .../homeassistant_sky_connect/hardware.py | 2 +- .../homeassistant_sky_connect/manifest.json | 2 +- .../homeassistant_sky_connect/strings.json | 128 +- .../homeassistant_sky_connect/util.py | 140 ++- .../zha/repairs/wrong_silabs_firmware.py | 9 +- homeassistant/generated/integrations.json | 5 + script/hassfest/dependencies.py | 1 + .../test_config_flow.py | 1096 ++++++++++++----- .../test_config_flow_failures.py | 920 ++++++++++++++ .../test_hardware.py | 22 +- .../homeassistant_sky_connect/test_init.py | 393 +----- .../homeassistant_sky_connect/test_util.py | 203 +++ tests/components/usb/__init__.py | 16 + tests/components/zha/test_repairs.py | 10 +- 17 files changed, 2943 insertions(+), 762 deletions(-) create mode 100644 tests/components/homeassistant_sky_connect/test_config_flow_failures.py create mode 100644 tests/components/homeassistant_sky_connect/test_util.py diff --git a/homeassistant/components/homeassistant_sky_connect/__init__.py b/homeassistant/components/homeassistant_sky_connect/__init__.py index a85a1161792..fc02f31f263 100644 --- a/homeassistant/components/homeassistant_sky_connect/__init__.py +++ b/homeassistant/components/homeassistant_sky_connect/__init__.py @@ -2,87 +2,62 @@ from __future__ import annotations -from homeassistant.components import usb -from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( - check_multi_pan_addon, - get_zigbee_socket, - multi_pan_addon_using_device, -) -from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry -from homeassistant.core import HomeAssistant, callback -from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError -from homeassistant.helpers import discovery_flow +import logging -from .const import DOMAIN -from .util import get_hardware_variant, get_usb_service_info +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from .util import guess_firmware_type -async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None: - """Finish Home Assistant SkyConnect config entry setup.""" - matcher = usb.USBCallbackMatcher( - domain=DOMAIN, - vid=entry.data["vid"].upper(), - pid=entry.data["pid"].upper(), - serial_number=entry.data["serial_number"].lower(), - manufacturer=entry.data["manufacturer"].lower(), - description=entry.data["description"].lower(), - ) - - if not usb.async_is_plugged_in(hass, matcher): - # The USB dongle is not plugged in, remove the config entry - hass.async_create_task( - hass.config_entries.async_remove(entry.entry_id), eager_start=True - ) - return - - usb_dev = entry.data["device"] - # The call to get_serial_by_id can be removed in HA Core 2024.1 - dev_path = await hass.async_add_executor_job(usb.get_serial_by_id, usb_dev) - - if not await multi_pan_addon_using_device(hass, dev_path): - usb_info = get_usb_service_info(entry) - await hass.config_entries.flow.async_init( - "zha", - context={"source": "usb"}, - data=usb_info, - ) - return - - hw_variant = get_hardware_variant(entry) - hw_discovery_data = { - "name": f"{hw_variant.short_name} Multiprotocol", - "port": { - "path": get_zigbee_socket(), - }, - "radio_type": "ezsp", - } - discovery_flow.async_create_flow( - hass, - "zha", - context={"source": SOURCE_HARDWARE}, - data=hw_discovery_data, - ) +_LOGGER = logging.getLogger(__name__) async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up a Home Assistant SkyConnect config entry.""" - - try: - await check_multi_pan_addon(hass) - except HomeAssistantError as err: - raise ConfigEntryNotReady from err - - @callback - def async_usb_scan_done() -> None: - """Handle usb discovery started.""" - hass.async_create_task(_async_usb_scan_done(hass, entry), eager_start=True) - - unsub_usb = usb.async_register_initial_scan_callback(hass, async_usb_scan_done) - entry.async_on_unload(unsub_usb) - return True async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" return True + + +async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool: + """Migrate old entry.""" + + _LOGGER.debug( + "Migrating from version %s:%s", config_entry.version, config_entry.minor_version + ) + + if config_entry.version == 1: + if config_entry.minor_version == 1: + # Add-on startup with type service get started before Core, always (e.g. the + # Multi-Protocol add-on). Probing the firmware would interfere with the add-on, + # so we can't safely probe here. Instead, we must make an educated guess! + firmware_guess = await guess_firmware_type( + hass, config_entry.data["device"] + ) + + new_data = {**config_entry.data} + new_data["firmware"] = firmware_guess.firmware_type.value + + # Copy `description` to `product` + new_data["product"] = new_data["description"] + + hass.config_entries.async_update_entry( + config_entry, + data=new_data, + version=1, + minor_version=2, + ) + + _LOGGER.debug( + "Migration to version %s.%s successful", + config_entry.version, + config_entry.minor_version, + ) + + return True + + # This means the user has downgraded from a future version + return False diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index 3a3d32c2888..6ffb2783165 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/homeassistant/components/homeassistant_sky_connect/config_flow.py @@ -2,29 +2,498 @@ from __future__ import annotations +from abc import ABC, abstractmethod +import asyncio +import logging from typing import Any +from universal_silabs_flasher.const import ApplicationType + from homeassistant.components import usb +from homeassistant.components.hassio import ( + AddonError, + AddonInfo, + AddonManager, + AddonState, + is_hassio, +) from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon -from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult +from homeassistant.components.zha.repairs.wrong_silabs_firmware import ( + probe_silabs_firmware_type, +) +from homeassistant.config_entries import ( + ConfigEntry, + ConfigEntryBaseFlow, + ConfigFlow, + ConfigFlowResult, + OptionsFlow, + OptionsFlowWithConfigEntry, +) from homeassistant.core import callback +from homeassistant.data_entry_flow import AbortFlow -from .const import DOMAIN, HardwareVariant -from .util import get_hardware_variant, get_usb_service_info +from .const import DOCS_WEB_FLASHER_URL, DOMAIN, ZHA_DOMAIN, HardwareVariant +from .util import ( + get_hardware_variant, + get_otbr_addon_manager, + get_usb_service_info, + get_zha_device_path, + get_zigbee_flasher_addon_manager, +) + +_LOGGER = logging.getLogger(__name__) + +STEP_PICK_FIRMWARE_THREAD = "pick_firmware_thread" +STEP_PICK_FIRMWARE_ZIGBEE = "pick_firmware_zigbee" -class HomeAssistantSkyConnectConfigFlow(ConfigFlow, domain=DOMAIN): +class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC): + """Base flow to install firmware.""" + + _failed_addon_name: str + _failed_addon_reason: str + + def __init__(self, *args: Any, **kwargs: Any) -> None: + """Instantiate base flow.""" + super().__init__(*args, **kwargs) + + self._usb_info: usb.UsbServiceInfo | None = None + self._hw_variant: HardwareVariant | None = None + self._probed_firmware_type: ApplicationType | None = None + + self.addon_install_task: asyncio.Task | None = None + self.addon_start_task: asyncio.Task | None = None + self.addon_uninstall_task: asyncio.Task | None = None + + def _get_translation_placeholders(self) -> dict[str, str]: + """Shared translation placeholders.""" + placeholders = { + "model": ( + self._hw_variant.full_name + if self._hw_variant is not None + else "unknown" + ), + "firmware_type": ( + self._probed_firmware_type.value + if self._probed_firmware_type is not None + else "unknown" + ), + "docs_web_flasher_url": DOCS_WEB_FLASHER_URL, + } + + self.context["title_placeholders"] = placeholders + + 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(), + ) from err + + async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo: + """Return add-on info.""" + try: + addon_info = await addon_manager.async_get_addon_info() + except AddonError as err: + _LOGGER.error(err) + raise AbortFlow( + "addon_info_failed", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": addon_manager.addon_name, + }, + ) from err + + return addon_info + + async def async_step_pick_firmware( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Pick Thread or Zigbee firmware.""" + assert self._usb_info is not None + + self._probed_firmware_type = await probe_silabs_firmware_type( + self._usb_info.device, + probe_methods=( + # We probe in order of frequency: Zigbee, Thread, then multi-PAN + ApplicationType.GECKO_BOOTLOADER, + ApplicationType.EZSP, + ApplicationType.SPINEL, + ApplicationType.CPC, + ), + ) + + if self._probed_firmware_type not in ( + ApplicationType.EZSP, + ApplicationType.SPINEL, + ApplicationType.CPC, + ): + return self.async_abort( + reason="unsupported_firmware", + description_placeholders=self._get_translation_placeholders(), + ) + + return self.async_show_menu( + step_id="pick_firmware", + menu_options=[ + STEP_PICK_FIRMWARE_THREAD, + STEP_PICK_FIRMWARE_ZIGBEE, + ], + description_placeholders=self._get_translation_placeholders(), + ) + + async def async_step_pick_firmware_zigbee( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Pick Zigbee firmware.""" + # Allow the stick to be used with ZHA without flashing + if self._probed_firmware_type == ApplicationType.EZSP: + return await self.async_step_confirm_zigbee() + + if not is_hassio(self.hass): + return self.async_abort( + reason="not_hassio", + description_placeholders=self._get_translation_placeholders(), + ) + + # 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 + return self.async_abort( + reason="addon_already_running", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": fw_flasher_manager.addon_name, + }, + ) + + async def async_step_install_zigbee_flasher_addon( + 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", + ) + + async def _install_addon( + self, + addon_manager: silabs_multiprotocol_addon.WaitingAddonManager, + step_id: str, + next_step_id: str, + ) -> ConfigFlowResult: + """Show progress dialog for installing an addon.""" + addon_info = await self._async_get_addon_info(addon_manager) + + _LOGGER.debug("Flasher addon state: %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", + ) + + if not self.addon_install_task.done(): + return self.async_show_progress( + step_id=step_id, + progress_action="install_addon", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": addon_manager.addon_name, + }, + progress_task=self.addon_install_task, + ) + + try: + await self.addon_install_task + except AddonError as err: + _LOGGER.error(err) + self._failed_addon_name = addon_manager.addon_name + self._failed_addon_reason = "addon_install_failed" + return self.async_show_progress_done(next_step_id="addon_operation_failed") + 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._usb_info is not None + new_addon_config = { + **addon_info.options, + "device": self._usb_info.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._usb_info is not None + assert self._hw_variant is not None + self._probed_firmware_type = ApplicationType.EZSP + + if user_input is not None: + await self.hass.config_entries.flow.async_init( + ZHA_DOMAIN, + context={"source": "hardware"}, + data={ + "name": self._hw_variant.full_name, + "port": { + "path": self._usb_info.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.""" + # 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" + ) + + 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._usb_info is not None + new_addon_config = { + **addon_info.options, + "device": self._usb_info.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: + self.addon_start_task = self.hass.async_create_task( + otbr_manager.async_start_addon_waiting() + ) + + if not self.addon_start_task.done(): + return self.async_show_progress( + step_id="start_otbr_addon", + progress_action="start_otbr_addon", + description_placeholders={ + **self._get_translation_placeholders(), + "addon_name": otbr_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 = otbr_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="confirm_otbr") + + async def async_step_confirm_otbr( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Confirm OTBR setup.""" + assert self._usb_info is not None + assert self._hw_variant is not None + + self._probed_firmware_type = ApplicationType.SPINEL + + 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(), + ) + + @abstractmethod + def _async_flow_finished(self) -> ConfigFlowResult: + """Finish the flow.""" + # This should be implemented by a subclass + raise NotImplementedError + + +class HomeAssistantSkyConnectConfigFlow( + BaseFirmwareInstallFlow, ConfigFlow, domain=DOMAIN +): """Handle a config flow for Home Assistant SkyConnect.""" VERSION = 1 + MINOR_VERSION = 2 @staticmethod @callback def async_get_options_flow( config_entry: ConfigEntry, - ) -> HomeAssistantSkyConnectOptionsFlow: + ) -> OptionsFlow: """Return the options flow.""" - return HomeAssistantSkyConnectOptionsFlow(config_entry) + firmware_type = ApplicationType(config_entry.data["firmware"]) + + if firmware_type is ApplicationType.CPC: + return HomeAssistantSkyConnectMultiPanOptionsFlowHandler(config_entry) + + return HomeAssistantSkyConnectOptionsFlowHandler(config_entry) async def async_step_usb( self, discovery_info: usb.UsbServiceInfo @@ -37,27 +506,62 @@ class HomeAssistantSkyConnectConfigFlow(ConfigFlow, domain=DOMAIN): manufacturer = discovery_info.manufacturer description = discovery_info.description unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}" + if await self.async_set_unique_id(unique_id): self._abort_if_unique_id_configured(updates={"device": device}) + discovery_info.device = await self.hass.async_add_executor_job( + usb.get_serial_by_id, discovery_info.device + ) + + self._usb_info = discovery_info + assert description is not None - hw_variant = HardwareVariant.from_usb_product_name(description) + self._hw_variant = HardwareVariant.from_usb_product_name(description) + + return await self.async_step_confirm() + + async def async_step_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Confirm a discovery.""" + self._set_confirm_only() + + # Without confirmation, discovery can automatically progress into parts of the + # config flow logic that interacts with hardware. + if user_input is not None: + return await self.async_step_pick_firmware() + + return self.async_show_form( + step_id="confirm", + description_placeholders=self._get_translation_placeholders(), + ) + + def _async_flow_finished(self) -> ConfigFlowResult: + """Create the config entry.""" + assert self._usb_info is not None + assert self._hw_variant is not None + assert self._probed_firmware_type is not None return self.async_create_entry( - title=hw_variant.full_name, + title=self._hw_variant.full_name, data={ - "device": device, - "vid": vid, - "pid": pid, - "serial_number": serial_number, - "manufacturer": manufacturer, - "description": description, + "vid": self._usb_info.vid, + "pid": self._usb_info.pid, + "serial_number": self._usb_info.serial_number, + "manufacturer": self._usb_info.manufacturer, + "description": self._usb_info.description, # For backwards compatibility + "product": self._usb_info.description, + "device": self._usb_info.device, + "firmware": self._probed_firmware_type.value, }, ) -class HomeAssistantSkyConnectOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler): - """Handle an option flow for Home Assistant SkyConnect.""" +class HomeAssistantSkyConnectMultiPanOptionsFlowHandler( + silabs_multiprotocol_addon.OptionsFlowHandler +): + """Multi-PAN options flow for Home Assistant SkyConnect.""" async def _async_serial_port_settings( self, @@ -92,3 +596,97 @@ class HomeAssistantSkyConnectOptionsFlow(silabs_multiprotocol_addon.OptionsFlowH def _hardware_name(self) -> str: """Return the name of the hardware.""" return self._hw_variant.full_name + + +class HomeAssistantSkyConnectOptionsFlowHandler( + BaseFirmwareInstallFlow, OptionsFlowWithConfigEntry +): + """Zigbee and Thread options flow handlers.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + """Instantiate options flow.""" + super().__init__(*args, **kwargs) + + self._usb_info = get_usb_service_info(self.config_entry) + self._probed_firmware_type = ApplicationType(self.config_entry.data["firmware"]) + self._hw_variant = HardwareVariant.from_usb_product_name( + self.config_entry.data["product"] + ) + + # Make `context` a regular dictionary + self.context = {} + + # Regenerate the translation placeholders + self._get_translation_placeholders() + + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Manage the options flow.""" + # Don't probe the running firmware, we load it from the config entry + return self.async_show_menu( + step_id="pick_firmware", + menu_options=[ + STEP_PICK_FIRMWARE_THREAD, + STEP_PICK_FIRMWARE_ZIGBEE, + ], + description_placeholders=self._get_translation_placeholders(), + ) + + async def async_step_pick_firmware_zigbee( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Pick Zigbee firmware.""" + assert self._usb_info is not None + + if is_hassio(self.hass): + otbr_manager = get_otbr_addon_manager(self.hass) + otbr_addon_info = await self._async_get_addon_info(otbr_manager) + + if ( + otbr_addon_info.state != AddonState.NOT_INSTALLED + and otbr_addon_info.options.get("device") == self._usb_info.device + ): + raise AbortFlow( + "otbr_still_using_stick", + description_placeholders=self._get_translation_placeholders(), + ) + + return await super().async_step_pick_firmware_zigbee(user_input) + + async def async_step_pick_firmware_thread( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Pick Thread firmware.""" + assert self._usb_info is not None + + zha_entries = self.hass.config_entries.async_entries( + ZHA_DOMAIN, + include_ignore=False, + include_disabled=True, + ) + + if zha_entries and get_zha_device_path(zha_entries[0]) == self._usb_info.device: + raise AbortFlow( + "zha_still_using_stick", + description_placeholders=self._get_translation_placeholders(), + ) + + return await super().async_step_pick_firmware_thread(user_input) + + def _async_flow_finished(self) -> ConfigFlowResult: + """Create the config entry.""" + assert self._usb_info is not None + assert self._hw_variant is not None + assert self._probed_firmware_type is not None + + self.hass.config_entries.async_update_entry( + entry=self.config_entry, + data={ + **self.config_entry.data, + "firmware": self._probed_firmware_type.value, + }, + options=self.config_entry.options, + ) + + return self.async_create_entry(title="", data={}) diff --git a/homeassistant/components/homeassistant_sky_connect/const.py b/homeassistant/components/homeassistant_sky_connect/const.py index 1dd1471c470..1d6c16dc528 100644 --- a/homeassistant/components/homeassistant_sky_connect/const.py +++ b/homeassistant/components/homeassistant_sky_connect/const.py @@ -5,6 +5,17 @@ import enum from typing import Self DOMAIN = "homeassistant_sky_connect" +ZHA_DOMAIN = "zha" + +DOCS_WEB_FLASHER_URL = "https://skyconnect.home-assistant.io/firmware-update/" + +OTBR_ADDON_NAME = "OpenThread Border Router" +OTBR_ADDON_MANAGER_DATA = "openthread_border_router" +OTBR_ADDON_SLUG = "core_openthread_border_router" + +ZIGBEE_FLASHER_ADDON_NAME = "Silicon Labs Flasher" +ZIGBEE_FLASHER_ADDON_MANAGER_DATA = "silabs_flasher" +ZIGBEE_FLASHER_ADDON_SLUG = "core_silabs_flasher" @dataclasses.dataclass(frozen=True) diff --git a/homeassistant/components/homeassistant_sky_connect/hardware.py b/homeassistant/components/homeassistant_sky_connect/hardware.py index a9abeb27737..2872077111a 100644 --- a/homeassistant/components/homeassistant_sky_connect/hardware.py +++ b/homeassistant/components/homeassistant_sky_connect/hardware.py @@ -25,7 +25,7 @@ def async_info(hass: HomeAssistant) -> list[HardwareInfo]: pid=entry.data["pid"], serial_number=entry.data["serial_number"], manufacturer=entry.data["manufacturer"], - description=entry.data["description"], + description=entry.data["product"], ), name=get_hardware_variant(entry).full_name, url=DOCUMENTATION_URL, diff --git a/homeassistant/components/homeassistant_sky_connect/manifest.json b/homeassistant/components/homeassistant_sky_connect/manifest.json index f56fd24de61..c90ea2c075f 100644 --- a/homeassistant/components/homeassistant_sky_connect/manifest.json +++ b/homeassistant/components/homeassistant_sky_connect/manifest.json @@ -5,7 +5,7 @@ "config_flow": true, "dependencies": ["hardware", "usb", "homeassistant_hardware"], "documentation": "https://www.home-assistant.io/integrations/homeassistant_sky_connect", - "integration_type": "hardware", + "integration_type": "device", "usb": [ { "vid": "10C4", diff --git a/homeassistant/components/homeassistant_sky_connect/strings.json b/homeassistant/components/homeassistant_sky_connect/strings.json index 825649ef0d3..792406dcb02 100644 --- a/homeassistant/components/homeassistant_sky_connect/strings.json +++ b/homeassistant/components/homeassistant_sky_connect/strings.json @@ -57,6 +57,50 @@ "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%]" + }, + "confirm": { + "title": "[%key:component::homeassistant_sky_connect::config::step::confirm::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::confirm::description%]" + }, + "pick_firmware": { + "title": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::description%]", + "menu_options": { + "pick_firmware_thread": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::menu_options::pick_firmware_thread%]", + "pick_firmware_zigbee": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::menu_options::pick_firmware_zigbee%]" + } + }, + "install_zigbee_flasher_addon": { + "title": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::description%]" + }, + "run_zigbee_flasher_addon": { + "title": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::description%]" + }, + "zigbee_flasher_failed": { + "title": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::description%]" + }, + "confirm_zigbee": { + "title": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::description%]" + }, + "install_otbr_addon": { + "title": "[%key:component::homeassistant_sky_connect::config::step::install_otbr_addon::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::install_otbr_addon::description%]" + }, + "start_otbr_addon": { + "title": "[%key:component::homeassistant_sky_connect::config::step::start_otbr_addon::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::start_otbr_addon::description%]" + }, + "otbr_failed": { + "title": "[%key:component::homeassistant_sky_connect::config::step::otbr_failed::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::otbr_failed::description%]" + }, + "confirm_otbr": { + "title": "[%key:component::homeassistant_sky_connect::config::step::confirm_otbr::title%]", + "description": "[%key:component::homeassistant_sky_connect::config::step::confirm_otbr::description%]" } }, "error": { @@ -68,12 +112,92 @@ "addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]", "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", + "zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]", "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", - "zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]" + "not_hassio_thread": "[%key:component::homeassistant_sky_connect::config::abort::not_hassio_thread%]", + "otbr_addon_already_running": "[%key:component::homeassistant_sky_connect::config::abort::otbr_addon_already_running%]", + "zha_still_using_stick": "This {model} is in use by the Zigbee Home Automation integration. Please migrate your Zigbee network to another adapter or delete the integration and try again.", + "otbr_still_using_stick": "This {model} is in use by the OpenThread Border Router add-on. If you use the Thread network, make sure you have alternative border routers. Uninstall the add-on and try again." }, "progress": { "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_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_sky_connect::config::progress::install_zigbee_flasher_addon%]", + "run_zigbee_flasher_addon": "[%key:component::homeassistant_sky_connect::config::progress::run_zigbee_flasher_addon%]", + "uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_sky_connect::config::progress::uninstall_zigbee_flasher_addon%]" + } + }, + "config": { + "flow_title": "{model}", + "step": { + "confirm": { + "title": "Set up the {model}", + "description": "The {model} can be used as either a Thread border router or a Zigbee coordinator. In the next step, you will choose which firmware will be configured." + }, + "pick_firmware": { + "title": "Pick your firmware", + "description": "The {model} can be used as a Thread border router or a Zigbee coordinator.", + "menu_options": { + "pick_firmware_thread": "Use as a Thread border router", + "pick_firmware_zigbee": "Use as a Zigbee coordinator" + } + }, + "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 once you exit." + }, + "install_otbr_addon": { + "title": "Installing OpenThread Border Router add-on", + "description": "The OpenThread Border Router (OTBR) add-on is being installed." + }, + "start_otbr_addon": { + "title": "Starting OpenThread Border Router add-on", + "description": "The OpenThread Border Router (OTBR) add-on is now starting." + }, + "otbr_failed": { + "title": "Failed to setup OpenThread Border Router", + "description": "The OpenThread Border Router add-on installation was unsuccessful. Ensure no other software is trying to communicate with the {model}, you have access to the internet and can install other add-ons, and try again. Check the Supervisor logs if the problem persists." + }, + "confirm_otbr": { + "title": "OpenThread Border Router setup complete", + "description": "Your {model} is now an OpenThread Border Router and will show up in the Thread integration once you exit." + } + }, + "abort": { + "addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]", + "addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]", + "addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]", + "addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]", + "addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]", + "zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]", + "not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]", + "not_hassio_thread": "The OpenThread Border Router addon can only be installed with Home Assistant OS. If you would like to use the {model} as an Thread border router, please flash the firmware manually using the [web flasher]({docs_web_flasher_url}) and set up OpenThread Border Router to communicate with it.", + "otbr_addon_already_running": "The OpenThread Border Router add-on is already running, it cannot be installed again." + }, + "progress": { + "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": "The Silicon Labs Flasher addon 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 addon is being removed." } } } diff --git a/homeassistant/components/homeassistant_sky_connect/util.py b/homeassistant/components/homeassistant_sky_connect/util.py index e1de1d3b442..f242416fa9a 100644 --- a/homeassistant/components/homeassistant_sky_connect/util.py +++ b/homeassistant/components/homeassistant_sky_connect/util.py @@ -2,10 +2,35 @@ from __future__ import annotations -from homeassistant.components import usb -from homeassistant.config_entries import ConfigEntry +from collections import defaultdict +from dataclasses import dataclass +import logging +from typing import cast -from .const import HardwareVariant +from universal_silabs_flasher.const import ApplicationType + +from homeassistant.components import usb +from homeassistant.components.hassio import AddonError, AddonState, is_hassio +from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( + WaitingAddonManager, + get_multiprotocol_addon_manager, +) +from homeassistant.config_entries import ConfigEntry, ConfigEntryState +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers.singleton import singleton + +from .const import ( + OTBR_ADDON_MANAGER_DATA, + OTBR_ADDON_NAME, + OTBR_ADDON_SLUG, + ZHA_DOMAIN, + ZIGBEE_FLASHER_ADDON_MANAGER_DATA, + ZIGBEE_FLASHER_ADDON_NAME, + ZIGBEE_FLASHER_ADDON_SLUG, + HardwareVariant, +) + +_LOGGER = logging.getLogger(__name__) def get_usb_service_info(config_entry: ConfigEntry) -> usb.UsbServiceInfo: @@ -16,10 +41,115 @@ def get_usb_service_info(config_entry: ConfigEntry) -> usb.UsbServiceInfo: pid=config_entry.data["pid"], serial_number=config_entry.data["serial_number"], manufacturer=config_entry.data["manufacturer"], - description=config_entry.data["description"], + description=config_entry.data["product"], ) def get_hardware_variant(config_entry: ConfigEntry) -> HardwareVariant: """Get the hardware variant from the config entry.""" - return HardwareVariant.from_usb_product_name(config_entry.data["description"]) + return HardwareVariant.from_usb_product_name(config_entry.data["product"]) + + +def get_zha_device_path(config_entry: ConfigEntry) -> str: + """Get the device path from a ZHA config entry.""" + return cast(str, config_entry.data["device"]["path"]) + + +@singleton(OTBR_ADDON_MANAGER_DATA) +@callback +def get_otbr_addon_manager(hass: HomeAssistant) -> WaitingAddonManager: + """Get the OTBR add-on manager.""" + return WaitingAddonManager( + hass, + _LOGGER, + OTBR_ADDON_NAME, + OTBR_ADDON_SLUG, + ) + + +@singleton(ZIGBEE_FLASHER_ADDON_MANAGER_DATA) +@callback +def get_zigbee_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager: + """Get the flasher add-on manager.""" + return WaitingAddonManager( + hass, + _LOGGER, + ZIGBEE_FLASHER_ADDON_NAME, + ZIGBEE_FLASHER_ADDON_SLUG, + ) + + +@dataclass(slots=True, kw_only=True) +class FirmwareGuess: + """Firmware guess.""" + + is_running: bool + firmware_type: ApplicationType + source: str + + +async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> FirmwareGuess: + """Guess the firmware type based on installed addons and other integrations.""" + device_guesses: defaultdict[str | None, list[FirmwareGuess]] = defaultdict(list) + + for zha_config_entry in hass.config_entries.async_entries(ZHA_DOMAIN): + zha_path = get_zha_device_path(zha_config_entry) + device_guesses[zha_path].append( + FirmwareGuess( + is_running=(zha_config_entry.state == ConfigEntryState.LOADED), + firmware_type=ApplicationType.EZSP, + source="zha", + ) + ) + + if is_hassio(hass): + otbr_addon_manager = get_otbr_addon_manager(hass) + + try: + otbr_addon_info = await otbr_addon_manager.async_get_addon_info() + except AddonError: + pass + else: + if otbr_addon_info.state != AddonState.NOT_INSTALLED: + otbr_path = otbr_addon_info.options.get("device") + device_guesses[otbr_path].append( + FirmwareGuess( + is_running=(otbr_addon_info.state == AddonState.RUNNING), + firmware_type=ApplicationType.SPINEL, + source="otbr", + ) + ) + + multipan_addon_manager = await get_multiprotocol_addon_manager(hass) + + try: + multipan_addon_info = await multipan_addon_manager.async_get_addon_info() + except AddonError: + pass + else: + if multipan_addon_info.state != AddonState.NOT_INSTALLED: + multipan_path = multipan_addon_info.options.get("device") + device_guesses[multipan_path].append( + FirmwareGuess( + is_running=(multipan_addon_info.state == AddonState.RUNNING), + firmware_type=ApplicationType.CPC, + source="multiprotocol", + ) + ) + + # Fall back to EZSP if we can't guess the firmware type + if device_path not in device_guesses: + return FirmwareGuess( + is_running=False, firmware_type=ApplicationType.EZSP, source="unknown" + ) + + # Prioritizes guesses that were pulled from a running addon or integration but keep + # the sort order we defined above + guesses = sorted( + device_guesses[device_path], + key=lambda guess: guess.is_running, + ) + + assert guesses + + return guesses[-1] diff --git a/homeassistant/components/zha/repairs/wrong_silabs_firmware.py b/homeassistant/components/zha/repairs/wrong_silabs_firmware.py index 5b1f85e1a29..4ee10c7bb93 100644 --- a/homeassistant/components/zha/repairs/wrong_silabs_firmware.py +++ b/homeassistant/components/zha/repairs/wrong_silabs_firmware.py @@ -74,9 +74,14 @@ def _detect_radio_hardware(hass: HomeAssistant, device: str) -> HardwareType: return HardwareType.OTHER -async def probe_silabs_firmware_type(device: str) -> ApplicationType | None: +async def probe_silabs_firmware_type( + device: str, *, probe_methods: ApplicationType | None = None +) -> ApplicationType | None: """Probe the running firmware on a Silabs device.""" - flasher = Flasher(device=device) + flasher = Flasher( + device=device, + **({"probe_methods": probe_methods} if probe_methods else {}), + ) try: await flasher.probe_app_type() diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index e6a103989d1..cf5f352f22c 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -2565,6 +2565,11 @@ "integration_type": "virtual", "supported_by": "netatmo" }, + "homeassistant_sky_connect": { + "name": "Home Assistant SkyConnect", + "integration_type": "device", + "config_flow": true + }, "homematic": { "name": "Homematic", "integrations": { diff --git a/script/hassfest/dependencies.py b/script/hassfest/dependencies.py index d4eb135a265..66796d4dd0d 100644 --- a/script/hassfest/dependencies.py +++ b/script/hassfest/dependencies.py @@ -157,6 +157,7 @@ IGNORE_VIOLATIONS = { ("zha", "homeassistant_hardware"), ("zha", "homeassistant_sky_connect"), ("zha", "homeassistant_yellow"), + ("homeassistant_sky_connect", "zha"), # This should become a helper method that integrations can submit data to ("websocket_api", "lovelace"), ("websocket_api", "shopping_list"), diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index 9647cef4721..c34e3ebe186 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -1,23 +1,31 @@ """Test the Home Assistant SkyConnect config flow.""" -from collections.abc import Generator -import copy -from unittest.mock import Mock, patch +import asyncio +from collections.abc import Awaitable, Callable +from typing import Any +from unittest.mock import AsyncMock, Mock, call, patch import pytest +from universal_silabs_flasher.const import ApplicationType -from homeassistant.components import homeassistant_sky_connect, usb +from homeassistant.components import usb +from homeassistant.components.hassio.addon_manager import AddonInfo, AddonState +from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import ( + get_multiprotocol_addon_manager, +) +from homeassistant.components.homeassistant_sky_connect.config_flow import ( + STEP_PICK_FIRMWARE_THREAD, + STEP_PICK_FIRMWARE_ZIGBEE, +) from homeassistant.components.homeassistant_sky_connect.const import DOMAIN -from homeassistant.components.zha import ( - CONF_DEVICE_PATH, - DOMAIN as ZHA_DOMAIN, - RadioType, +from homeassistant.components.homeassistant_sky_connect.util import ( + get_otbr_addon_manager, + get_zigbee_flasher_addon_manager, ) from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType -from homeassistant.setup import async_setup_component -from tests.common import MockConfigEntry, MockModule, mock_integration +from tests.common import MockConfigEntry USB_DATA_SKY = usb.UsbServiceInfo( device="/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", @@ -38,340 +46,840 @@ USB_DATA_ZBT1 = usb.UsbServiceInfo( ) -@pytest.fixture(autouse=True) -def config_flow_handler(hass: HomeAssistant) -> Generator[None, None, None]: - """Fixture for a test config flow.""" - with patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.WaitingAddonManager.async_wait_until_addon_state" - ): - yield +def delayed_side_effect() -> Callable[..., Awaitable[None]]: + """Slows down eager tasks by delaying for an event loop tick.""" + + async def side_effect(*args: Any, **kwargs: Any) -> None: + await asyncio.sleep(0) + + return side_effect @pytest.mark.parametrize( - ("usb_data", "title"), + ("usb_data", "model"), [ (USB_DATA_SKY, "Home Assistant SkyConnect"), (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), ], ) -async def test_config_flow( - usb_data: usb.UsbServiceInfo, title: str, hass: HomeAssistant +async def test_config_flow_zigbee( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant ) -> None: """Test the config flow for SkyConnect.""" - with patch( - "homeassistant.components.homeassistant_sky_connect.async_setup_entry", - return_value=True, - ) as mock_setup_entry: - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": "usb"}, data=usb_data - ) - - expected_data = { - "device": usb_data.device, - "vid": usb_data.vid, - "pid": usb_data.pid, - "serial_number": usb_data.serial_number, - "manufacturer": usb_data.manufacturer, - "description": usb_data.description, - } - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == title - assert result["data"] == expected_data - assert result["options"] == {} - assert len(mock_setup_entry.mock_calls) == 1 - - config_entry = hass.config_entries.async_entries(DOMAIN)[0] - assert config_entry.data == expected_data - assert config_entry.options == {} - assert config_entry.title == title - assert ( - config_entry.unique_id - == f"{usb_data.vid}:{usb_data.pid}_{usb_data.serial_number}_{usb_data.manufacturer}_{usb_data.description}" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data ) + # First step is confirmation, we haven't probed the firmware yet + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "confirm" + assert result["description_placeholders"]["firmware_type"] == "unknown" + assert result["description_placeholders"]["model"] == model -@pytest.mark.parametrize( - ("usb_data", "title"), - [ - (USB_DATA_SKY, "Home Assistant SkyConnect"), - (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), - ], -) -async def test_config_flow_multiple_entries( - usb_data: usb.UsbServiceInfo, title: str, hass: HomeAssistant -) -> None: - """Test multiple entries are allowed.""" - # Setup an existing config entry - config_entry = MockConfigEntry( - data={}, - domain=DOMAIN, - options={}, - title=title, - unique_id=f"{usb_data.vid}:{usb_data.pid}_{usb_data.serial_number}_{usb_data.manufacturer}_{usb_data.description}", - ) - config_entry.add_to_hass(hass) - - usb_data = copy.copy(usb_data) - usb_data.serial_number = "bla_serial_number_2" - + # Next, we probe the firmware with patch( - "homeassistant.components.homeassistant_sky_connect.async_setup_entry", - return_value=True, + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, # Ensure we re-install it ): - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": "usb"}, data=usb_data + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} ) - assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + assert result["description_placeholders"]["firmware_type"] == "spinel" - -@pytest.mark.parametrize( - ("usb_data", "title"), - [ - (USB_DATA_SKY, "Home Assistant SkyConnect"), - (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), - ], -) -async def test_config_flow_update_device( - usb_data: usb.UsbServiceInfo, title: str, hass: HomeAssistant -) -> None: - """Test updating device path.""" - # Setup an existing config entry - config_entry = MockConfigEntry( - data={}, - domain=DOMAIN, - options={}, - title=title, - unique_id=f"{usb_data.vid}:{usb_data.pid}_{usb_data.serial_number}_{usb_data.manufacturer}_{usb_data.description}", + # Set up Zigbee firmware + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() ) - config_entry.add_to_hass(hass) - - usb_data = copy.copy(usb_data) - usb_data.device = "bla_device_2" - - with patch( - "homeassistant.components.homeassistant_sky_connect.async_setup_entry", - return_value=True, - ) as mock_setup_entry: - assert await hass.config_entries.async_setup(config_entry.entry_id) - assert len(mock_setup_entry.mock_calls) == 1 with ( patch( - "homeassistant.components.homeassistant_sky_connect.async_setup_entry", - return_value=True, - ) as mock_setup_entry, + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), patch( - "homeassistant.components.homeassistant_sky_connect.async_unload_entry", - wraps=homeassistant_sky_connect.async_unload_entry, - ) as mock_unload_entry, + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), ): - result = await hass.config_entries.flow.async_init( - DOMAIN, context={"source": "usb"}, data=usb_data + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, ) - await hass.async_block_till_done() - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "already_configured" - assert len(mock_setup_entry.mock_calls) == 1 - assert len(mock_unload_entry.mock_calls) == 1 + # 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.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) -@pytest.mark.parametrize( - ("usb_data", "title"), - [ - (USB_DATA_SKY, "Home Assistant SkyConnect"), - (USB_DATA_ZBT1, "Home Assistant ZBT-1"), - ], -) -async def test_option_flow_install_multi_pan_addon( - usb_data: usb.UsbServiceInfo, - title: str, - hass: HomeAssistant, - addon_store_info, - addon_info, - install_addon, - set_addon_options, - start_addon, -) -> None: - """Test installing the multi pan addon.""" - assert await async_setup_component(hass, "usb", {}) - mock_integration(hass, MockModule("hassio")) + # 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": usb_data.device, + "baudrate": 115200, + "bootloader_baudrate": 115200, + "flow_control": True, + } + ) + ] - # Setup the config entry - config_entry = MockConfigEntry( - data={ - "device": usb_data.device, - "vid": usb_data.vid, - "pid": usb_data.pid, - "serial_number": usb_data.serial_number, - "manufacturer": usb_data.manufacturer, - "description": usb_data.description, - }, - domain=DOMAIN, - options={}, - title=title, - unique_id=f"{usb_data.vid}:{usb_data.pid}_{usb_data.serial_number}_{usb_data.manufacturer}_{usb_data.description}", - ) - config_entry.add_to_hass(hass) + await hass.async_block_till_done(wait_background_tasks=True) - with patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), - ): - result = await hass.config_entries.options.async_init(config_entry.entry_id) + # 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"] == "addon_not_installed" + assert result["step_id"] == "confirm_zigbee" - result = await hass.config_entries.options.async_configure( - result["flow_id"], - user_input={ - "enable_multi_pan": True, - }, + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "install_addon" - assert result["progress_action"] == "install_addon" - - await hass.async_block_till_done() - install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "start_addon" - set_addon_options.assert_called_once_with( - hass, - "core_silabs_multiprotocol", - { - "options": { - "autoflash_firmware": True, - "device": usb_data.device, - "baudrate": "115200", - "flow_control": True, - } - }, - ) - - await hass.async_block_till_done() - start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - - result = await hass.config_entries.options.async_configure(result["flow_id"]) assert result["type"] is FlowResultType.CREATE_ENTRY + config_entry = result["result"] + assert config_entry.data == { + "firmware": "ezsp", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + } -def mock_detect_radio_type(radio_type=RadioType.ezsp, ret=True): - """Mock `detect_radio_type` that just sets the appropriate attributes.""" - - async def detect(self): - self.radio_type = radio_type - self.device_settings = radio_type.controller.SCHEMA_DEVICE( - {CONF_DEVICE_PATH: self.device_path} - ) - - return ret - - return detect + # 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" @pytest.mark.parametrize( - ("usb_data", "title"), + ("usb_data", "model"), [ (USB_DATA_SKY, "Home Assistant SkyConnect"), (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), ], ) -@patch( - "homeassistant.components.zha.radio_manager.ZhaRadioManager.detect_radio_type", - mock_detect_radio_type(), -) -async def test_option_flow_install_multi_pan_addon_zha( - usb_data: usb.UsbServiceInfo, - title: str, - hass: HomeAssistant, - addon_store_info, - addon_info, - install_addon, - set_addon_options, - start_addon, +async def test_config_flow_zigbee_skip_step_if_installed( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant ) -> None: - """Test installing the multi pan addon when a zha config entry exists.""" - assert await async_setup_component(hass, "usb", {}) - mock_integration(hass, MockModule("hassio")) + """Test the config flow for SkyConnect, skip installing the addon if necessary.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) - # Setup the config entry - config_entry = MockConfigEntry( - data={ - "device": usb_data.device, - "vid": usb_data.vid, - "pid": usb_data.pid, - "serial_number": usb_data.serial_number, - "manufacturer": usb_data.manufacturer, - "description": usb_data.description, - }, - domain=DOMAIN, + # First step is confirmation, we haven't probed the firmware yet + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "confirm" + assert result["description_placeholders"]["firmware_type"] == "unknown" + assert result["description_placeholders"]["model"] == model + + # Next, we probe the firmware + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, # Ensure we re-install it + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + assert result["description_placeholders"]["firmware_type"] == "spinel" + + # Set up Zigbee firmware + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = 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", + ) + + # 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 mock_flasher_manager.async_set_addon_options.mock_calls == [ + call( + { + "device": usb_data.device, + "baudrate": 115200, + "bootloader_baudrate": 115200, + "flow_control": True, + } + ) + ] + + # Uninstall the addon + await hass.async_block_till_done(wait_background_tasks=True) + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + # Done + 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_zigbee" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_SKY, "Home Assistant SkyConnect"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the config flow for SkyConnect.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + # First step is confirmation, we haven't probed the firmware yet + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "confirm" + assert result["description_placeholders"]["firmware_type"] == "unknown" + assert result["description_placeholders"]["model"] == model + + # Next, we probe the firmware + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + assert result["description_placeholders"]["firmware_type"] == "ezsp" + + # Set up Thread firmware + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + + # Pick the menu option + result = await hass.config_entries.flow.async_configure( + 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" + + await hass.async_block_till_done(wait_background_tasks=True) + + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={ + "device": "", + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": True, + }, + 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"]) + + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_otbr_addon" + assert result["progress_action"] == "start_otbr_addon" + + assert mock_otbr_manager.async_set_addon_options.mock_calls == [ + call( + { + "device": usb_data.device, + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": True, + } + ) + ] + + 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" + + 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": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + } + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_SKY, "Home Assistant SkyConnect"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_addon_already_installed( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the Thread config flow for SkyConnect, addon is already installed.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + 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( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, options={}, - title=title, - unique_id=f"{usb_data.vid}:{usb_data.pid}_{usb_data.serial_number}_{usb_data.manufacturer}_{usb_data.description}", + state=AddonState.NOT_RUNNING, + update_available=False, + version=None, + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + # Pick the menu option + result = await hass.config_entries.flow.async_configure( + 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" + + assert mock_otbr_manager.async_set_addon_options.mock_calls == [ + call( + { + "device": usb_data.device, + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": True, + } + ) + ] + + 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" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_not_hassio( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test when the stick is used with a non-hassio setup.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=False, + ), + ): + 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": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "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" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_SKY, "Home Assistant SkyConnect"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_options_flow_zigbee_to_thread( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the options flow for SkyConnect, migrating Zigbee to Thread.""" + config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "ezsp", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + }, + version=1, + minor_version=2, ) config_entry.add_to_hass(hass) - zha_config_entry = MockConfigEntry( - data={"device": {"path": usb_data.device}, "radio_type": "ezsp"}, - domain=ZHA_DOMAIN, - options={}, - title="Yellow", - ) - zha_config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) - with patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), + # First step is confirmation + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + assert result["description_placeholders"]["firmware_type"] == "ezsp" + assert result["description_placeholders"]["model"] == model + + # Pick Thread + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + + result = await hass.config_entries.options.async_configure( + 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" + + await hass.async_block_till_done(wait_background_tasks=True) + + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={ + "device": "", + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": True, + }, + 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.options.async_configure(result["flow_id"]) + + assert result["type"] is FlowResultType.SHOW_PROGRESS + assert result["step_id"] == "start_otbr_addon" + assert result["progress_action"] == "start_otbr_addon" + + assert mock_otbr_manager.async_set_addon_options.mock_calls == [ + call( + { + "device": usb_data.device, + "baudrate": 460800, + "flow_control": True, + "autoflash_firmware": True, + } + ) + ] + + await hass.async_block_till_done(wait_background_tasks=True) + + # The addon is now running + result = await hass.config_entries.options.async_configure(result["flow_id"]) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "confirm_otbr" + + # We are now done + result = await hass.config_entries.options.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + # The firmware type has been updated + assert config_entry.data["firmware"] == "spinel" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_SKY, "Home Assistant SkyConnect"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_options_flow_thread_to_zigbee( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the options flow for SkyConnect, migrating Thread to Zigbee.""" + config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "spinel", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + }, + version=1, + minor_version=2, + ) + config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(config_entry.entry_id) + + # First step is confirmation + result = await hass.config_entries.options.async_init(config_entry.entry_id) + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "pick_firmware" + assert result["description_placeholders"]["firmware_type"] == "spinel" + assert result["description_placeholders"]["model"] == model + + # Set up Zigbee firmware + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + + # OTBR is not installed + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + + # 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": usb_data.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" + + # We are now done + result = await hass.config_entries.options.async_configure( + result["flow_id"], user_input={} + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + # The firmware type has been updated + assert config_entry.data["firmware"] == "ezsp" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_SKY, "Home Assistant SkyConnect"), + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_options_flow_multipan_uninstall( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test options flow for when multi-PAN firmware is installed.""" + config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "cpc", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + }, + version=1, + minor_version=2, + ) + config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(config_entry.entry_id) + + # Multi-PAN addon is running + mock_multipan_manager = Mock(spec_set=await get_multiprotocol_addon_manager(hass)) + mock_multipan_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": usb_data.device}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + with ( + patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.get_multiprotocol_addon_manager", + return_value=mock_multipan_manager, + ), + patch( + "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", + return_value=True, + ), ): result = await hass.config_entries.options.async_init(config_entry.entry_id) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "addon_not_installed" - - result = await hass.config_entries.options.async_configure( - result["flow_id"], - user_input={ - "enable_multi_pan": True, - }, - ) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "install_addon" - assert result["progress_action"] == "install_addon" - - await hass.async_block_till_done() - install_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.SHOW_PROGRESS - assert result["step_id"] == "start_addon" - set_addon_options.assert_called_once_with( - hass, - "core_silabs_multiprotocol", - { - "options": { - "autoflash_firmware": True, - "device": usb_data.device, - "baudrate": "115200", - "flow_control": True, - } - }, - ) - # Check the ZHA config entry data is updated - assert zha_config_entry.data == { - "device": { - "path": "socket://core-silabs-multiprotocol:9999", - "baudrate": 115200, - "flow_control": None, - }, - "radio_type": "ezsp", - } - - await hass.async_block_till_done() - start_addon.assert_called_once_with(hass, "core_silabs_multiprotocol") - - result = await hass.config_entries.options.async_configure(result["flow_id"]) - assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "addon_menu" + assert "uninstall_addon" in result["menu_options"] diff --git a/tests/components/homeassistant_sky_connect/test_config_flow_failures.py b/tests/components/homeassistant_sky_connect/test_config_flow_failures.py new file mode 100644 index 00000000000..128c812272f --- /dev/null +++ b/tests/components/homeassistant_sky_connect/test_config_flow_failures.py @@ -0,0 +1,920 @@ +"""Test the Home Assistant SkyConnect config flow failure cases.""" + +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from universal_silabs_flasher.const import ApplicationType + +from homeassistant.components import usb +from homeassistant.components.hassio.addon_manager import ( + AddonError, + AddonInfo, + AddonState, +) +from homeassistant.components.homeassistant_sky_connect.config_flow import ( + STEP_PICK_FIRMWARE_THREAD, + STEP_PICK_FIRMWARE_ZIGBEE, +) +from homeassistant.components.homeassistant_sky_connect.const import DOMAIN +from homeassistant.components.homeassistant_sky_connect.util import ( + get_otbr_addon_manager, + get_zigbee_flasher_addon_manager, +) +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from .test_config_flow import USB_DATA_ZBT1, delayed_side_effect + +from tests.common import MockConfigEntry + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_cannot_probe_firmware( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when firmware cannot be probed.""" + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=None, + ): + # Start the flow + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + # Probing fails + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "unsupported_firmware" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_not_hassio_wrong_firmware( + usb_data: usb.UsbServiceInfo, model: str, 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( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=False, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_addon_already_running( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon is already running.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_addon_info_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be installed.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.side_effect = AddonError() + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_addon_install_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be installed.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_addon_set_config_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be configured.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_run_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon fails to run.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_zigbee_flasher_uninstall_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon uninstall fails.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.SPINEL, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass)) + mock_flasher_manager.addon_name = "Silicon Labs Flasher" + mock_flasher_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_flasher_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager", + return_value=mock_flasher_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_not_hassio( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test when the stick is used with a non-hassio setup and Thread is selected.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=False, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "not_hassio_thread" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_addon_info_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be installed.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.side_effect = AddonError() + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) + + # Cannot get addon info + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_info_failed" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_addon_already_running( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when the Thread addon is already running.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) + + # Cannot install addon + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "otbr_addon_already_running" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_addon_install_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be installed.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) + + # Cannot install addon + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_install_failed" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_addon_set_config_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon cannot be configured.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_set_config_failed" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_flasher_run_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon fails to run.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError()) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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"]) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "addon_start_failed" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_config_flow_thread_flasher_uninstall_fails( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test failure case when flasher addon uninstall fails.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": "usb"}, data=usb_data + ) + + with patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type", + return_value=ApplicationType.EZSP, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input={} + ) + + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={}, + state=AddonState.NOT_INSTALLED, + update_available=False, + version=None, + ) + mock_otbr_manager.async_install_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_start_addon_waiting = AsyncMock( + side_effect=delayed_side_effect() + ) + mock_otbr_manager.async_uninstall_addon_waiting = AsyncMock( + side_effect=AddonError() + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + 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" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_options_flow_zigbee_to_thread_zha_configured( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the options flow migration failure, ZHA using the stick.""" + config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "ezsp", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + }, + version=1, + minor_version=2, + ) + config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(config_entry.entry_id) + + # Set up ZHA as well + zha_config_entry = MockConfigEntry( + domain="zha", + data={"device": {"path": usb_data.device}}, + ) + zha_config_entry.add_to_hass(hass) + + # Confirm options flow + result = await hass.config_entries.options.async_init(config_entry.entry_id) + + # Pick Thread + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "zha_still_using_stick" + + +@pytest.mark.parametrize( + ("usb_data", "model"), + [ + (USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"), + ], +) +async def test_options_flow_thread_to_zigbee_otbr_configured( + usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant +) -> None: + """Test the options flow migration failure, OTBR still using the stick.""" + config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "spinel", + "device": usb_data.device, + "manufacturer": usb_data.manufacturer, + "pid": usb_data.pid, + "description": usb_data.description, + "product": usb_data.description, + "serial_number": usb_data.serial_number, + "vid": usb_data.vid, + }, + version=1, + minor_version=2, + ) + config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(config_entry.entry_id) + + # Confirm options flow + result = await hass.config_entries.options.async_init(config_entry.entry_id) + + # Pick Zigbee + mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass)) + mock_otbr_manager.addon_name = "OpenThread Border Router" + mock_otbr_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": usb_data.device}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio", + return_value=True, + ), + ): + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE}, + ) + assert result["type"] == FlowResultType.ABORT + assert result["reason"] == "otbr_still_using_stick" diff --git a/tests/components/homeassistant_sky_connect/test_hardware.py b/tests/components/homeassistant_sky_connect/test_hardware.py index 6b283378045..888ed27a3c0 100644 --- a/tests/components/homeassistant_sky_connect/test_hardware.py +++ b/tests/components/homeassistant_sky_connect/test_hardware.py @@ -1,7 +1,5 @@ """Test the Home Assistant SkyConnect hardware platform.""" -from unittest.mock import patch - from homeassistant.components.homeassistant_sky_connect.const import DOMAIN from homeassistant.core import EVENT_HOMEASSISTANT_STARTED, HomeAssistant from homeassistant.setup import async_setup_component @@ -15,7 +13,8 @@ CONFIG_ENTRY_DATA = { "pid": "EA60", "serial_number": "9e2adbd75b8beb119fe564a0f320645d", "manufacturer": "Nabu Casa", - "description": "SkyConnect v1.0", + "product": "SkyConnect v1.0", + "firmware": "ezsp", } CONFIG_ENTRY_DATA_2 = { @@ -24,7 +23,8 @@ CONFIG_ENTRY_DATA_2 = { "pid": "EA60", "serial_number": "9e2adbd75b8beb119fe564a0f320645d", "manufacturer": "Nabu Casa", - "description": "Home Assistant Connect ZBT-1", + "product": "Home Assistant Connect ZBT-1", + "firmware": "ezsp", } @@ -42,22 +42,24 @@ async def test_hardware_info( options={}, title="Home Assistant SkyConnect", unique_id="unique_1", + version=1, + minor_version=2, ) config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + config_entry_2 = MockConfigEntry( data=CONFIG_ENTRY_DATA_2, domain=DOMAIN, options={}, title="Home Assistant Connect ZBT-1", unique_id="unique_2", + version=1, + minor_version=2, ) config_entry_2.add_to_hass(hass) - with patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ): - assert await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() + + assert await hass.config_entries.async_setup(config_entry_2.entry_id) client = await hass_ws_client(hass) diff --git a/tests/components/homeassistant_sky_connect/test_init.py b/tests/components/homeassistant_sky_connect/test_init.py index a6dd5100d7e..88b57f2dd64 100644 --- a/tests/components/homeassistant_sky_connect/test_init.py +++ b/tests/components/homeassistant_sky_connect/test_init.py @@ -1,377 +1,56 @@ """Test the Home Assistant SkyConnect integration.""" -from collections.abc import Generator -from typing import Any -from unittest.mock import MagicMock, Mock, patch +from unittest.mock import patch -import pytest +from universal_silabs_flasher.const import ApplicationType -from homeassistant.components import zha -from homeassistant.components.hassio.handler import HassioAPIError from homeassistant.components.homeassistant_sky_connect.const import DOMAIN -from homeassistant.config_entries import ConfigEntryState -from homeassistant.core import EVENT_HOMEASSISTANT_STARTED, HomeAssistant -from homeassistant.setup import async_setup_component +from homeassistant.components.homeassistant_sky_connect.util import FirmwareGuess +from homeassistant.core import HomeAssistant from tests.common import MockConfigEntry -CONFIG_ENTRY_DATA = { - "device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", - "vid": "10C4", - "pid": "EA60", - "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", - "manufacturer": "Nabu Casa", - "description": "SkyConnect v1.0", -} +async def test_config_entry_migration_v2(hass: HomeAssistant) -> None: + """Test migrating config entries from v1 to v2 format.""" -@pytest.fixture(autouse=True) -def disable_usb_probing() -> Generator[None, None, None]: - """Disallow touching of system USB devices during unit tests.""" - with patch("homeassistant.components.usb.comports", return_value=[]): - yield - - -@pytest.fixture -def mock_zha_config_flow_setup() -> Generator[None, None, None]: - """Mock the radio connection and probing of the ZHA config flow.""" - - def mock_probe(config: dict[str, Any]) -> None: - # The radio probing will return the correct baudrate - return {**config, "baudrate": 115200} - - mock_connect_app = MagicMock() - mock_connect_app.__aenter__.return_value.backups.backups = [] - - with ( - patch( - "bellows.zigbee.application.ControllerApplication.probe", - side_effect=mock_probe, - ), - patch( - "homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app", - return_value=mock_connect_app, - ), - ): - yield - - -@pytest.mark.parametrize( - ("onboarded", "num_entries", "num_flows"), [(False, 1, 0), (True, 0, 1)] -) -async def test_setup_entry( - mock_zha_config_flow_setup, - hass: HomeAssistant, - addon_store_info, - onboarded, - num_entries, - num_flows, -) -> None: - """Test setup of a config entry, including setup of zha.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - # Setup the config entry config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", - ) - config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ) as mock_is_plugged_in, - patch( - "homeassistant.components.onboarding.async_is_onboarded", - return_value=onboarded, - ), - ): - assert await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert len(mock_is_plugged_in.mock_calls) == 1 - - matcher = mock_is_plugged_in.mock_calls[0].args[1] - assert matcher["vid"].isupper() - assert matcher["pid"].isupper() - assert matcher["serial_number"].islower() - assert matcher["manufacturer"].islower() - assert matcher["description"].islower() - - # Finish setting up ZHA - if num_entries > 0: - zha_flows = hass.config_entries.flow.async_progress_by_handler("zha") - assert len(zha_flows) == 1 - assert zha_flows[0]["step_id"] == "choose_formation_strategy" - - await hass.config_entries.flow.async_configure( - zha_flows[0]["flow_id"], - user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS}, - ) - await hass.async_block_till_done() - - assert len(hass.config_entries.flow.async_progress_by_handler("zha")) == num_flows - assert len(hass.config_entries.async_entries("zha")) == num_entries - - -async def test_setup_zha( - mock_zha_config_flow_setup, hass: HomeAssistant, addon_store_info -) -> None: - """Test zha gets the right config.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", - ) - config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ) as mock_is_plugged_in, - patch( - "homeassistant.components.onboarding.async_is_onboarded", return_value=False - ), - ): - await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert len(mock_is_plugged_in.mock_calls) == 1 - - zha_flows = hass.config_entries.flow.async_progress_by_handler("zha") - assert len(zha_flows) == 1 - assert zha_flows[0]["step_id"] == "choose_formation_strategy" - - # Finish setting up ZHA - await hass.config_entries.flow.async_configure( - zha_flows[0]["flow_id"], - user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS}, - ) - await hass.async_block_till_done() - - config_entry = hass.config_entries.async_entries("zha")[0] - assert config_entry.data == { - "device": { - "baudrate": 115200, - "flow_control": None, - "path": CONFIG_ENTRY_DATA["device"], + unique_id="some_unique_id", + data={ + "device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + "vid": "10C4", + "pid": "EA60", + "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", + "manufacturer": "Nabu Casa", + "description": "SkyConnect v1.0", }, - "radio_type": "ezsp", - } - assert config_entry.options == {} - assert config_entry.title == CONFIG_ENTRY_DATA["description"] - - -async def test_setup_zha_multipan( - hass: HomeAssistant, addon_info, addon_running -) -> None: - """Test zha gets the right config.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - addon_info.return_value["options"]["device"] = CONFIG_ENTRY_DATA["device"] - - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", + version=1, ) + config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ) as mock_is_plugged_in, - patch( - "homeassistant.components.onboarding.async_is_onboarded", return_value=False - ), - patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), - ), - ): - assert await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert len(mock_is_plugged_in.mock_calls) == 1 - # Finish setting up ZHA - zha_flows = hass.config_entries.flow.async_progress_by_handler("zha") - assert len(zha_flows) == 1 - assert zha_flows[0]["step_id"] == "choose_formation_strategy" - - await hass.config_entries.flow.async_configure( - zha_flows[0]["flow_id"], - user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS}, - ) - await hass.async_block_till_done() - - config_entry = hass.config_entries.async_entries("zha")[0] - assert config_entry.data == { - "device": { - "baudrate": 115200, - "flow_control": None, - "path": "socket://core-silabs-multiprotocol:9999", - }, - "radio_type": "ezsp", - } - assert config_entry.options == {} - assert config_entry.title == "SkyConnect Multiprotocol" - - -async def test_setup_zha_multipan_other_device( - mock_zha_config_flow_setup, hass: HomeAssistant, addon_info, addon_running -) -> None: - """Test zha gets the right config.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - addon_info.return_value["options"]["device"] = "/dev/not_our_sky_connect" - - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant Yellow", - ) - config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ) as mock_is_plugged_in, - patch( - "homeassistant.components.onboarding.async_is_onboarded", return_value=False - ), - patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), - ), - ): - assert await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert len(mock_is_plugged_in.mock_calls) == 1 - - # Finish setting up ZHA - zha_flows = hass.config_entries.flow.async_progress_by_handler("zha") - assert len(zha_flows) == 1 - assert zha_flows[0]["step_id"] == "choose_formation_strategy" - - await hass.config_entries.flow.async_configure( - zha_flows[0]["flow_id"], - user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS}, - ) - await hass.async_block_till_done() - - config_entry = hass.config_entries.async_entries("zha")[0] - assert config_entry.data == { - "device": { - "baudrate": 115200, - "flow_control": None, - "path": CONFIG_ENTRY_DATA["device"], - }, - "radio_type": "ezsp", - } - assert config_entry.options == {} - assert config_entry.title == CONFIG_ENTRY_DATA["description"] - - -async def test_setup_entry_wait_usb(hass: HomeAssistant) -> None: - """Test setup of a config entry when the dongle is not plugged in.""" - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", - ) - config_entry.add_to_hass(hass) with patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=False, - ) as mock_is_plugged_in: + "homeassistant.components.homeassistant_sky_connect.guess_firmware_type", + return_value=FirmwareGuess( + is_running=True, + firmware_type=ApplicationType.SPINEL, + source="otbr", + ), + ): await hass.config_entries.async_setup(config_entry.entry_id) - assert config_entry.state is ConfigEntryState.LOADED - assert len(hass.config_entries.async_entries(DOMAIN)) == 1 - # USB discovery starts, config entry should be removed - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - await hass.async_block_till_done() - assert len(mock_is_plugged_in.mock_calls) == 1 - assert len(hass.config_entries.async_entries(DOMAIN)) == 0 + assert config_entry.version == 1 + assert config_entry.minor_version == 2 + assert config_entry.data == { + "description": "SkyConnect v1.0", + "device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + "vid": "10C4", + "pid": "EA60", + "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", + "manufacturer": "Nabu Casa", + "product": "SkyConnect v1.0", # `description` has been copied to `product` + "firmware": "spinel", # new key + } -async def test_setup_entry_addon_info_fails( - hass: HomeAssistant, addon_store_info -) -> None: - """Test setup of a config entry when fetching addon info fails.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - addon_store_info.side_effect = HassioAPIError("Boom") - - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", - ) - config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ), - patch( - "homeassistant.components.onboarding.async_is_onboarded", return_value=False - ), - patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), - ), - ): - assert not await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert config_entry.state is ConfigEntryState.SETUP_RETRY - - -async def test_setup_entry_addon_not_running( - hass: HomeAssistant, addon_installed, start_addon -) -> None: - """Test the addon is started if it is not running.""" - assert await async_setup_component(hass, "usb", {}) - hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED) - - # Setup the config entry - config_entry = MockConfigEntry( - data=CONFIG_ENTRY_DATA, - domain=DOMAIN, - options={}, - title="Home Assistant SkyConnect", - ) - config_entry.add_to_hass(hass) - with ( - patch( - "homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in", - return_value=True, - ), - patch( - "homeassistant.components.onboarding.async_is_onboarded", return_value=False - ), - patch( - "homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio", - side_effect=Mock(return_value=True), - ), - ): - assert not await hass.config_entries.async_setup(config_entry.entry_id) - await hass.async_block_till_done() - assert config_entry.state is ConfigEntryState.SETUP_RETRY - start_addon.assert_called_once() + await hass.config_entries.async_unload(config_entry.entry_id) diff --git a/tests/components/homeassistant_sky_connect/test_util.py b/tests/components/homeassistant_sky_connect/test_util.py new file mode 100644 index 00000000000..12ba352eb16 --- /dev/null +++ b/tests/components/homeassistant_sky_connect/test_util.py @@ -0,0 +1,203 @@ +"""Test SkyConnect utilities.""" + +from unittest.mock import AsyncMock, patch + +from universal_silabs_flasher.const import ApplicationType + +from homeassistant.components.hassio import AddonError, AddonInfo, AddonState +from homeassistant.components.homeassistant_sky_connect.const import ( + DOMAIN, + HardwareVariant, +) +from homeassistant.components.homeassistant_sky_connect.util import ( + FirmwareGuess, + get_hardware_variant, + get_usb_service_info, + get_zha_device_path, + guess_firmware_type, +) +from homeassistant.components.usb import UsbServiceInfo +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + +SKYCONNECT_CONFIG_ENTRY = MockConfigEntry( + domain=DOMAIN, + unique_id="some_unique_id", + data={ + "device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + "vid": "10C4", + "pid": "EA60", + "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", + "manufacturer": "Nabu Casa", + "product": "SkyConnect v1.0", + "firmware": "ezsp", + }, + version=2, +) + +CONNECT_ZBT1_CONFIG_ENTRY = MockConfigEntry( + domain=DOMAIN, + unique_id="some_unique_id", + data={ + "device": "/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + "vid": "10C4", + "pid": "EA60", + "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", + "manufacturer": "Nabu Casa", + "product": "Home Assistant Connect ZBT-1", + "firmware": "ezsp", + }, + version=2, +) + +ZHA_CONFIG_ENTRY = MockConfigEntry( + domain="zha", + unique_id="some_unique_id", + data={ + "device": { + "path": "/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_3c0ed67c628beb11b1cd64a0f320645d-if00-port0", + "baudrate": 115200, + "flow_control": None, + }, + "radio_type": "ezsp", + }, + version=4, +) + + +def test_get_usb_service_info() -> None: + """Test `get_usb_service_info` conversion.""" + assert get_usb_service_info(SKYCONNECT_CONFIG_ENTRY) == UsbServiceInfo( + device=SKYCONNECT_CONFIG_ENTRY.data["device"], + vid=SKYCONNECT_CONFIG_ENTRY.data["vid"], + pid=SKYCONNECT_CONFIG_ENTRY.data["pid"], + serial_number=SKYCONNECT_CONFIG_ENTRY.data["serial_number"], + manufacturer=SKYCONNECT_CONFIG_ENTRY.data["manufacturer"], + description=SKYCONNECT_CONFIG_ENTRY.data["product"], + ) + + +def test_get_hardware_variant() -> None: + """Test `get_hardware_variant` extraction.""" + assert get_hardware_variant(SKYCONNECT_CONFIG_ENTRY) == HardwareVariant.SKYCONNECT + assert ( + get_hardware_variant(CONNECT_ZBT1_CONFIG_ENTRY) == HardwareVariant.CONNECT_ZBT1 + ) + + +def test_get_zha_device_path() -> None: + """Test extracting the ZHA device path from its config entry.""" + assert ( + get_zha_device_path(ZHA_CONFIG_ENTRY) == ZHA_CONFIG_ENTRY.data["device"]["path"] + ) + + +async def test_guess_firmware_type_unknown(hass: HomeAssistant) -> None: + """Test guessing the firmware type.""" + + assert (await guess_firmware_type(hass, "/dev/missing")) == FirmwareGuess( + is_running=False, firmware_type=ApplicationType.EZSP, source="unknown" + ) + + +async def test_guess_firmware_type(hass: HomeAssistant) -> None: + """Test guessing the firmware.""" + path = ZHA_CONFIG_ENTRY.data["device"]["path"] + + ZHA_CONFIG_ENTRY.add_to_hass(hass) + + ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.NOT_LOADED) + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=False, firmware_type=ApplicationType.EZSP, source="zha" + ) + + # When ZHA is running, we indicate as such when guessing + ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.LOADED) + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.EZSP, source="zha" + ) + + mock_otbr_addon_manager = AsyncMock() + mock_multipan_addon_manager = AsyncMock() + + with ( + patch( + "homeassistant.components.homeassistant_sky_connect.util.is_hassio", + return_value=True, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.util.get_otbr_addon_manager", + return_value=mock_otbr_addon_manager, + ), + patch( + "homeassistant.components.homeassistant_sky_connect.util.get_multiprotocol_addon_manager", + return_value=mock_multipan_addon_manager, + ), + ): + mock_otbr_addon_manager.async_get_addon_info.side_effect = AddonError() + mock_multipan_addon_manager.async_get_addon_info.side_effect = AddonError() + + # Hassio errors are ignored and we still go with ZHA + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.EZSP, source="zha" + ) + + mock_otbr_addon_manager.async_get_addon_info.side_effect = None + mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": "/some/other/device"}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + # We will prefer ZHA, as it is running (and actually pointing to the device) + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.EZSP, source="zha" + ) + + mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": path}, + state=AddonState.NOT_RUNNING, + update_available=False, + version="1.0.0", + ) + + # We will still prefer ZHA, as it is the one actually running + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.EZSP, source="zha" + ) + + mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": path}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + # Finally, ZHA loses out to OTBR + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.SPINEL, source="otbr" + ) + + mock_multipan_addon_manager.async_get_addon_info.side_effect = None + mock_multipan_addon_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname=None, + options={"device": path}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + # Which will lose out to multi-PAN + assert (await guess_firmware_type(hass, path)) == FirmwareGuess( + is_running=True, firmware_type=ApplicationType.CPC, source="multiprotocol" + ) diff --git a/tests/components/usb/__init__.py b/tests/components/usb/__init__.py index f5f32336931..96d671d0958 100644 --- a/tests/components/usb/__init__.py +++ b/tests/components/usb/__init__.py @@ -26,3 +26,19 @@ electro_lama_device = USBDevice( manufacturer=None, description="USB2.0-Serial", ) +skyconnect_macos_correct = USBDevice( + device="/dev/cu.SLAB_USBtoUART", + vid="10C4", + pid="EA60", + serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d", + manufacturer="Nabu Casa", + description="SkyConnect v1.0", +) +skyconnect_macos_incorrect = USBDevice( + device="/dev/cu.usbserial-2110", + vid="10C4", + pid="EA60", + serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d", + manufacturer="Nabu Casa", + description="SkyConnect v1.0", +) diff --git a/tests/components/zha/test_repairs.py b/tests/components/zha/test_repairs.py index 5e128cc464a..5b57ec7fcc2 100644 --- a/tests/components/zha/test_repairs.py +++ b/tests/components/zha/test_repairs.py @@ -12,7 +12,7 @@ from zigpy.application import ControllerApplication import zigpy.backups from zigpy.exceptions import NetworkSettingsInconsistent -from homeassistant.components.homeassistant_sky_connect import ( +from homeassistant.components.homeassistant_sky_connect.const import ( DOMAIN as SKYCONNECT_DOMAIN, ) from homeassistant.components.repairs import DOMAIN as REPAIRS_DOMAIN @@ -59,8 +59,10 @@ def test_detect_radio_hardware(hass: HomeAssistant) -> None: "pid": "EA60", "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", "manufacturer": "Nabu Casa", - "description": "SkyConnect v1.0", + "product": "SkyConnect v1.0", + "firmware": "ezsp", }, + version=2, domain=SKYCONNECT_DOMAIN, options={}, title="Home Assistant SkyConnect", @@ -74,8 +76,10 @@ def test_detect_radio_hardware(hass: HomeAssistant) -> None: "pid": "EA60", "serial_number": "3c0ed67c628beb11b1cd64a0f320645d", "manufacturer": "Nabu Casa", - "description": "Home Assistant Connect ZBT-1", + "product": "Home Assistant Connect ZBT-1", + "firmware": "ezsp", }, + version=2, domain=SKYCONNECT_DOMAIN, options={}, title="Home Assistant Connect ZBT-1",