Abstract SkyConnect firmware config flow to the hardware platform (#122140)

* Move the SkyConnect config flow to hardware;

* Clean up

* Get SkyConnect unit tests passing

* Split apart `test_util.py`

* Migrate `test_config_flow`

* Remove unnecessary constants

* Re-apply `contextmanager` typing from #122250

* Move the SkyConnect translation strings into hardware
pull/122800/head
puddly 2024-07-29 12:39:25 -04:00 committed by GitHub
parent 570725293c
commit 1f488b00f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 1803 additions and 1628 deletions

View File

@ -4,5 +4,15 @@ import logging
LOGGER = logging.getLogger(__package__)
ZHA_DOMAIN = "zha"
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"
SILABS_MULTIPROTOCOL_ADDON_SLUG = "core_silabs_multiprotocol"
SILABS_FLASHER_ADDON_SLUG = "core_silabs_flasher"

View File

@ -0,0 +1,557 @@
"""Config flow for the Home Assistant SkyConnect integration."""
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.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
is_hassio,
)
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 . import silabs_multiprotocol_addon
from .const import ZHA_DOMAIN
from .util import (
get_otbr_addon_manager,
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 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._probed_firmware_type: ApplicationType | None = None
self._device: str | None = None # To be set in a subclass
self._hardware_name: str = "unknown" # To be set in a subclass
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 = {
"firmware_type": (
self._probed_firmware_type.value
if self._probed_firmware_type is not None
else "unknown"
),
"model": self._hardware_name,
}
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(),
"addon_name": addon_manager.addon_name,
},
) 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."""
return self.async_show_menu(
step_id="pick_firmware",
menu_options=[
STEP_PICK_FIRMWARE_ZIGBEE,
STEP_PICK_FIRMWARE_THREAD,
],
description_placeholders=self._get_translation_placeholders(),
)
async def _probe_firmware_type(self) -> bool:
"""Probe the firmware currently on the device."""
assert self._device is not None
self._probed_firmware_type = await probe_silabs_firmware_type(
self._device,
probe_methods=(
# We probe in order of frequency: Zigbee, Thread, then multi-PAN
ApplicationType.GECKO_BOOTLOADER,
ApplicationType.EZSP,
ApplicationType.SPINEL,
ApplicationType.CPC,
),
)
return self._probed_firmware_type in (
ApplicationType.EZSP,
ApplicationType.SPINEL,
ApplicationType.CPC,
)
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Zigbee firmware."""
if not await self._probe_firmware_type():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
# 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._device is not None
new_addon_config = {
**addon_info.options,
"device": self._device,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
_LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, fw_flasher_manager)
if not self.addon_start_task:
async def start_and_wait_until_done() -> None:
await fw_flasher_manager.async_start_addon_waiting()
# Now that the addon is running, wait for it to finish
await fw_flasher_manager.async_wait_until_addon_state(
AddonState.NOT_RUNNING
)
self.addon_start_task = self.hass.async_create_task(
start_and_wait_until_done()
)
if not self.addon_start_task.done():
return self.async_show_progress(
step_id="run_zigbee_flasher_addon",
progress_action="run_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_start_task,
)
try:
await self.addon_start_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
self._failed_addon_name = fw_flasher_manager.addon_name
self._failed_addon_reason = "addon_start_failed"
return self.async_show_progress_done(next_step_id="addon_operation_failed")
finally:
self.addon_start_task = None
return self.async_show_progress_done(
next_step_id="uninstall_zigbee_flasher_addon"
)
async def async_step_uninstall_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Uninstall the flasher addon."""
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
if not self.addon_uninstall_task:
_LOGGER.debug("Uninstalling flasher addon")
self.addon_uninstall_task = self.hass.async_create_task(
fw_flasher_manager.async_uninstall_addon_waiting()
)
if not self.addon_uninstall_task.done():
return self.async_show_progress(
step_id="uninstall_zigbee_flasher_addon",
progress_action="uninstall_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_uninstall_task,
)
try:
await self.addon_uninstall_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
# The uninstall failing isn't critical so we can just continue
finally:
self.addon_uninstall_task = None
return self.async_show_progress_done(next_step_id="confirm_zigbee")
async def async_step_confirm_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm Zigbee setup."""
assert self._device is not None
assert self._hardware_name is not None
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._hardware_name,
"port": {
"path": self._device,
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
},
)
return self._async_flow_finished()
return self.async_show_form(
step_id="confirm_zigbee",
description_placeholders=self._get_translation_placeholders(),
)
async def async_step_pick_firmware_thread(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Thread firmware."""
if not await self._probe_firmware_type():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
# We install the OTBR addon no matter what, since it is required to use Thread
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio_thread",
description_placeholders=self._get_translation_placeholders(),
)
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_otbr_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_start_otbr_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="otbr_addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
)
async def async_step_install_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the OTBR addon."""
return await self._install_addon(
get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon"
)
async def async_step_start_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure OTBR to point to the SkyConnect and run the addon."""
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
assert self._device is not None
new_addon_config = {
**addon_info.options,
"device": self._device,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
}
_LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, otbr_manager)
if not self.addon_start_task:
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._device 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."""
raise NotImplementedError
class BaseFirmwareConfigFlow(BaseFirmwareInstallFlow, ConfigFlow):
"""Base config flow for installing firmware."""
@staticmethod
@callback
@abstractmethod
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OptionsFlow:
"""Return the options flow."""
raise NotImplementedError
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm a discovery."""
return await self.async_step_pick_firmware()
class BaseFirmwareOptionsFlow(BaseFirmwareInstallFlow, OptionsFlowWithConfigEntry):
"""Zigbee and Thread options flow handlers."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate options flow."""
super().__init__(*args, **kwargs)
self._probed_firmware_type = ApplicationType(self.config_entry.data["firmware"])
# Make `context` a regular dictionary
self.context = {}
# Subclasses are expected to override `_device` and `_hardware_name`
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options flow."""
return await self.async_step_pick_firmware()
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Zigbee firmware."""
assert self._device 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._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._device is not None
for zha_entry in self.hass.config_entries.async_entries(
ZHA_DOMAIN,
include_ignore=False,
include_disabled=True,
):
if get_zha_device_path(zha_entry) == self._device:
raise AbortFlow(
"zha_still_using_stick",
description_placeholders=self._get_translation_placeholders(),
)
return await super().async_step_pick_firmware_thread(user_input)

View File

@ -1,4 +1,65 @@
{
"firmware_picker": {
"options": {
"step": {
"pick_firmware": {
"title": "Pick your firmware",
"description": "Let's get started with setting up your {model}. Do you want to use it to set up a Zigbee or Thread network?",
"menu_options": {
"pick_firmware_zigbee": "Zigbee",
"pick_firmware_thread": "Thread"
}
},
"install_zigbee_flasher_addon": {
"title": "Installing flasher",
"description": "Installing the Silicon Labs Flasher add-on."
},
"run_zigbee_flasher_addon": {
"title": "Installing Zigbee firmware",
"description": "Installing Zigbee firmware. This will take about a minute."
},
"uninstall_zigbee_flasher_addon": {
"title": "Removing flasher",
"description": "Removing the Silicon Labs Flasher add-on."
},
"zigbee_flasher_failed": {
"title": "Zigbee installation failed",
"description": "The Zigbee firmware installation process was unsuccessful. Ensure no other software is trying to communicate with the {model} and try again."
},
"confirm_zigbee": {
"title": "Zigbee setup complete",
"description": "Your {model} is now a Zigbee coordinator and will be shown as discovered by the Zigbee Home Automation integration 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": {
"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.",
"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_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."
}
}
},
"silabs_multiprotocol_hardware": {
"options": {
"step": {

View File

@ -0,0 +1,138 @@
"""Utility functions for Home Assistant SkyConnect integration."""
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
import logging
from typing import cast
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components.hassio import AddonError, AddonState, is_hassio
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,
)
from .silabs_multiprotocol_addon import (
WaitingAddonManager,
get_multiprotocol_addon_manager,
)
_LOGGER = logging.getLogger(__name__)
def get_zha_device_path(config_entry: ConfigEntry) -> str | None:
"""Get the device path from a ZHA config entry."""
return cast(str | None, config_entry.data.get("device", {}).get("path", None))
@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)
if zha_path is not None:
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]

View File

@ -4,11 +4,10 @@ from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.util import guess_firmware_type
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .util import guess_firmware_type
_LOGGER = logging.getLogger(__name__)

View File

@ -2,82 +2,46 @@
from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
import logging
from typing import Any
from typing import TYPE_CHECKING, Any, Protocol
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.components.zha.repairs.wrong_silabs_firmware import (
probe_silabs_firmware_type,
)
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
OptionsFlowWithConfigEntry,
from homeassistant.components.homeassistant_hardware import (
firmware_config_flow,
silabs_multiprotocol_addon,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow
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,
)
from .const import DOCS_WEB_FLASHER_URL, DOMAIN, HardwareVariant
from .util import get_hardware_variant, get_usb_service_info
_LOGGER = logging.getLogger(__name__)
STEP_PICK_FIRMWARE_THREAD = "pick_firmware_thread"
STEP_PICK_FIRMWARE_ZIGBEE = "pick_firmware_zigbee"
if TYPE_CHECKING:
class TranslationPlaceholderProtocol(Protocol):
"""Protocol describing `BaseFirmwareInstallFlow`'s translation placeholders."""
def _get_translation_placeholders(self) -> dict[str, str]:
return {}
else:
# Multiple inheritance with `Protocol` seems to break
TranslationPlaceholderProtocol = object
class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
"""Base flow to install firmware."""
class SkyConnectTranslationMixin(TranslationPlaceholderProtocol):
"""Translation placeholder mixin for Home Assistant SkyConnect."""
_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
context: dict[str, Any]
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"
),
**super()._get_translation_placeholders(),
"docs_web_flasher_url": DOCS_WEB_FLASHER_URL,
}
@ -85,416 +49,24 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
return placeholders
async def _async_set_addon_config(
self, config: dict, addon_manager: AddonManager
) -> None:
"""Set add-on config."""
try:
await addon_manager.async_set_addon_options(config)
except AddonError as err:
_LOGGER.error(err)
raise AbortFlow(
"addon_set_config_failed",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": addon_manager.addon_name,
},
) from err
async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo:
"""Return add-on info."""
try:
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."""
return self.async_show_menu(
step_id="pick_firmware",
menu_options=[
STEP_PICK_FIRMWARE_ZIGBEE,
STEP_PICK_FIRMWARE_THREAD,
],
description_placeholders=self._get_translation_placeholders(),
)
async def _probe_firmware_type(self) -> bool:
"""Probe the firmware currently on the device."""
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,
),
)
return self._probed_firmware_type in (
ApplicationType.EZSP,
ApplicationType.SPINEL,
ApplicationType.CPC,
)
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Zigbee firmware."""
if not await self._probe_firmware_type():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
# 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."""
if not await self._probe_firmware_type():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
# We install the OTBR addon no matter what, since it is required to use Thread
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio_thread",
description_placeholders=self._get_translation_placeholders(),
)
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_otbr_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_start_otbr_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="otbr_addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
)
async def async_step_install_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the OTBR addon."""
return await self._install_addon(
get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon"
)
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
SkyConnectTranslationMixin,
firmware_config_flow.BaseFirmwareConfigFlow,
domain=DOMAIN,
):
"""Handle a config flow for Home Assistant SkyConnect."""
VERSION = 1
MINOR_VERSION = 2
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize the config flow."""
super().__init__(*args, **kwargs)
self._usb_info: usb.UsbServiceInfo | None = None
self._hw_variant: HardwareVariant | None = None
@staticmethod
@callback
def async_get_options_flow(
@ -532,13 +104,11 @@ class HomeAssistantSkyConnectConfigFlow(
assert description is not None
self._hw_variant = HardwareVariant.from_usb_product_name(description)
return await self.async_step_confirm()
# Set parent class attributes
self._device = self._usb_info.device
self._hardware_name = self._hw_variant.full_name
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm a discovery."""
return await self.async_step_pick_firmware()
return await self.async_step_confirm()
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
@ -617,7 +187,7 @@ class HomeAssistantSkyConnectMultiPanOptionsFlowHandler(
class HomeAssistantSkyConnectOptionsFlowHandler(
BaseFirmwareInstallFlow, OptionsFlowWithConfigEntry
SkyConnectTranslationMixin, firmware_config_flow.BaseFirmwareOptionsFlow
):
"""Zigbee and Thread options flow handlers."""
@ -626,67 +196,17 @@ class HomeAssistantSkyConnectOptionsFlowHandler(
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 = {}
self._hardware_name = self._hw_variant.full_name
self._device = self._usb_info.device
# 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."""
return await self.async_step_pick_firmware()
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
for zha_entry in self.hass.config_entries.async_entries(
ZHA_DOMAIN,
include_ignore=False,
include_disabled=True,
):
if get_zha_device_path(zha_entry) == 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(

View File

@ -5,18 +5,8 @@ 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)
class VariantInfo:

View File

@ -59,44 +59,44 @@
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::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%]",
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::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%]"
"pick_firmware_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_thread%]",
"pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]"
}
},
"install_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::description%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::description%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::description%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::description%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::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%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_otbr_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::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%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::start_otbr_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::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%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::otbr_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::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%]"
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_otbr::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_otbr::description%]"
}
},
"error": {
@ -110,66 +110,66 @@
"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": "[%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."
"not_hassio_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::not_hassio_thread%]",
"otbr_addon_already_running": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_addon_already_running%]",
"zha_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::zha_still_using_stick%]",
"otbr_still_using_stick": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_still_using_stick%]"
},
"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": "[%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%]"
"install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]"
}
},
"config": {
"flow_title": "{model}",
"step": {
"pick_firmware": {
"title": "Pick your firmware",
"description": "Let's get started with setting up your {model}. Do you want to use it to set up a Zigbee or Thread network?",
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::description%]",
"menu_options": {
"pick_firmware_zigbee": "Zigbee",
"pick_firmware_thread": "Thread"
"pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]",
"pick_firmware_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_thread%]"
}
},
"install_zigbee_flasher_addon": {
"title": "Installing flasher",
"description": "Installing the Silicon Labs Flasher add-on."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "Installing Zigbee firmware",
"description": "Installing Zigbee firmware. This will take about a minute."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]"
},
"uninstall_zigbee_flasher_addon": {
"title": "Removing flasher",
"description": "Removing the Silicon Labs Flasher add-on."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "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."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "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."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]"
},
"install_otbr_addon": {
"title": "Installing OpenThread Border Router add-on",
"description": "The OpenThread Border Router (OTBR) add-on is being installed."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_otbr_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_otbr_addon::description%]"
},
"start_otbr_addon": {
"title": "Starting OpenThread Border Router add-on",
"description": "The OpenThread Border Router (OTBR) add-on is now starting."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::start_otbr_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::start_otbr_addon::description%]"
},
"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."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::otbr_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::otbr_failed::description%]"
},
"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."
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_otbr::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_otbr::description%]"
}
},
"abort": {
@ -180,16 +180,16 @@
"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."
"not_hassio_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::not_hassio_thread%]",
"otbr_addon_already_running": "[%key:component::homeassistant_hardware::firmware_picker::options::abort::otbr_addon_already_running%]"
},
"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."
"install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]"
}
}
}

View File

@ -2,33 +2,12 @@
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
import logging
from typing import cast
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 homeassistant.config_entries import ConfigEntry
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,
)
from .const import HardwareVariant
_LOGGER = logging.getLogger(__name__)
@ -48,110 +27,3 @@ def get_usb_service_info(config_entry: ConfigEntry) -> usb.UsbServiceInfo:
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["product"])
def get_zha_device_path(config_entry: ConfigEntry) -> str | None:
"""Get the device path from a ZHA config entry."""
return cast(str | None, config_entry.data.get("device", {}).get("path", None))
@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)
if zha_path is not None:
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]

View File

@ -0,0 +1,674 @@
"""Test the Home Assistant hardware firmware config flow."""
import asyncio
from collections.abc import Awaitable, Callable, Generator, Iterator
import contextlib
from typing import Any
from unittest.mock import AsyncMock, Mock, call, patch
import pytest
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components.hassio.addon_manager import AddonInfo, AddonState
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
BaseFirmwareConfigFlow,
BaseFirmwareOptionsFlow,
)
from homeassistant.components.homeassistant_hardware.util import (
get_otbr_addon_manager,
get_zigbee_flasher_addon_manager,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from tests.common import (
MockConfigEntry,
MockModule,
mock_config_flow,
mock_integration,
mock_platform,
)
TEST_DOMAIN = "test_firmware_domain"
TEST_DEVICE = "/dev/SomeDevice123"
TEST_HARDWARE_NAME = "Some Hardware Name"
class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN):
"""Config flow for `test_firmware_domain`."""
VERSION = 1
MINOR_VERSION = 2
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> OptionsFlow:
"""Return the options flow."""
return FakeFirmwareOptionsFlowHandler(config_entry)
async def async_step_hardware(
self, data: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle hardware flow."""
self._device = TEST_DEVICE
self._hardware_name = TEST_HARDWARE_NAME
return await self.async_step_confirm()
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._device is not None
assert self._hardware_name is not None
assert self._probed_firmware_type is not None
return self.async_create_entry(
title=self._hardware_name,
data={
"device": self._device,
"firmware": self._probed_firmware_type.value,
"hardware": self._hardware_name,
},
)
class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow):
"""Options flow for `test_firmware_domain`."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate options flow."""
super().__init__(*args, **kwargs)
self._device = self.config_entry.data["device"]
self._hardware_name = self.config_entry.data["hardware"]
# Regenerate the translation placeholders
self._get_translation_placeholders()
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
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={})
@pytest.fixture(autouse=True)
def mock_test_firmware_platform(
hass: HomeAssistant,
) -> Generator[None]:
"""Fixture for a test config flow."""
mock_module = MockModule(
TEST_DOMAIN, async_setup_entry=AsyncMock(return_value=True)
)
mock_integration(hass, mock_module)
mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
with mock_config_flow(TEST_DOMAIN, FakeFirmwareConfigFlow):
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
@contextlib.contextmanager
def mock_addon_info(
hass: HomeAssistant,
*,
is_hassio: bool = True,
app_type: ApplicationType = ApplicationType.EZSP,
otbr_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
),
flasher_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
),
) -> Iterator[tuple[Mock, Mock]]:
"""Mock the main addon states for the config flow."""
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_start_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_get_addon_info.return_value = flasher_addon_info
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_uninstall_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 = otbr_addon_info
with (
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio",
return_value=is_hassio,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_type",
return_value=app_type,
),
):
yield mock_otbr_manager, mock_flasher_manager
async def test_config_flow_zigbee(hass: HomeAssistant) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now configuring the addon and running it
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now uninstalling the addon
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "uninstall_zigbee_flasher_addon"
assert result["progress_action"] == "uninstall_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# We are finally done with the addon
assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()]
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
# Ensure a ZHA discovery flow has been created
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
zha_flow = flows[0]
assert zha_flow["handler"] == "zha"
assert zha_flow["context"]["source"] == "hardware"
assert zha_flow["step_id"] == "confirm"
async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) -> None:
"""Test the config flow, skip installing the addon if necessary."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
flasher_addon_info=AddonInfo(
available=True,
hostname=None,
options={
"device": "",
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.2.3",
),
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we skip installation, instead we directly run it
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
# Uninstall the addon
await hass.async_block_till_done(wait_background_tasks=True)
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"
async def test_config_flow_thread(hass: HomeAssistant) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option
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"
assert result["description_placeholders"]["firmware_type"] == "ezsp"
assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME
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": TEST_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": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -> None:
"""Test the Thread config flow, addon is already installed."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_RUNNING,
update_available=False,
version=None,
),
) as (mock_otbr_manager, mock_flasher_manager):
# 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": TEST_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
async def test_config_flow_zigbee_not_hassio(hass: HomeAssistant) -> None:
"""Test when the stick is used with a non-hassio setup."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
is_hassio=False,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
# Ensure a ZHA discovery flow has been created
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
zha_flow = flows[0]
assert zha_flow["handler"] == "zha"
assert zha_flow["context"]["source"] == "hardware"
assert zha_flow["step_id"] == "confirm"
async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
"""Test the options flow, migrating Zigbee to Thread."""
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"firmware": "ezsp",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
},
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"] == "ezsp"
assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_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": TEST_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"
async def test_options_flow_thread_to_zigbee(hass: HomeAssistant) -> None:
"""Test the options flow, migrating Thread to Zigbee."""
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"firmware": "spinel",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
},
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"] == TEST_HARDWARE_NAME
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now configuring the addon and running it
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now uninstalling the addon
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "uninstall_zigbee_flasher_addon"
assert result["progress_action"] == "uninstall_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# We are finally done with the addon
assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()]
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
# 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"

View File

@ -1,38 +1,43 @@
"""Test the Home Assistant SkyConnect config flow failure cases."""
"""Test the Home Assistant hardware firmware config flow failure cases."""
from unittest.mock import AsyncMock
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 (
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
)
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .test_config_flow import USB_DATA_ZBT1, delayed_side_effect, mock_addon_info
from .test_config_flow import (
TEST_DEVICE,
TEST_DOMAIN,
TEST_HARDWARE_NAME,
delayed_side_effect,
mock_addon_info,
mock_test_firmware_platform, # noqa: F401
)
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("usb_data", "model", "next_step"),
"next_step",
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1", STEP_PICK_FIRMWARE_ZIGBEE),
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1", STEP_PICK_FIRMWARE_THREAD),
STEP_PICK_FIRMWARE_ZIGBEE,
STEP_PICK_FIRMWARE_THREAD,
],
)
async def test_config_flow_cannot_probe_firmware(
usb_data: usb.UsbServiceInfo, model: str, next_step: str, hass: HomeAssistant
next_step: str, hass: HomeAssistant
) -> None:
"""Test failure case when firmware cannot be probed."""
@ -42,7 +47,7 @@ async def test_config_flow_cannot_probe_firmware(
) as (mock_otbr_manager, mock_flasher_manager):
# Start the flow
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
result = await hass.config_entries.flow.async_configure(
@ -54,18 +59,12 @@ async def test_config_flow_cannot_probe_firmware(
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
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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -85,18 +84,12 @@ async def test_config_flow_zigbee_not_hassio_wrong_firmware(
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
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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -125,18 +118,10 @@ async def test_config_flow_zigbee_flasher_addon_already_running(
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:
async def test_config_flow_zigbee_flasher_addon_info_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -166,18 +151,12 @@ async def test_config_flow_zigbee_flasher_addon_info_fails(
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
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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -202,18 +181,12 @@ async def test_config_flow_zigbee_flasher_addon_install_fails(
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
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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -242,18 +215,10 @@ async def test_config_flow_zigbee_flasher_addon_set_config_fails(
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:
async def test_config_flow_zigbee_flasher_run_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -279,18 +244,10 @@ async def test_config_flow_zigbee_flasher_run_fails(
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:
async def test_config_flow_zigbee_flasher_uninstall_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -319,18 +276,10 @@ async def test_config_flow_zigbee_flasher_uninstall_fails(
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:
async def test_config_flow_thread_not_hassio(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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -350,18 +299,10 @@ async def test_config_flow_thread_not_hassio(
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:
async def test_config_flow_thread_addon_info_fails(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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -382,18 +323,10 @@ async def test_config_flow_thread_addon_info_fails(
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:
async def test_config_flow_thread_addon_already_running(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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -425,18 +358,10 @@ async def test_config_flow_thread_addon_already_running(
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:
async def test_config_flow_thread_addon_install_fails(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
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -460,18 +385,10 @@ async def test_config_flow_thread_addon_install_fails(
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:
async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon cannot be configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -495,18 +412,10 @@ async def test_config_flow_thread_addon_set_config_fails(
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:
async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -531,18 +440,10 @@ async def test_config_flow_thread_flasher_run_fails(
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:
async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
@ -572,27 +473,16 @@ async def test_config_flow_thread_flasher_uninstall_fails(
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
hass: HomeAssistant,
) -> None:
"""Test the options flow migration failure, ZHA using the stick."""
config_entry = MockConfigEntry(
domain="homeassistant_sky_connect",
domain=TEST_DOMAIN,
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,
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
},
version=1,
minor_version=2,
@ -604,7 +494,7 @@ async def test_options_flow_zigbee_to_thread_zha_configured(
# Set up ZHA as well
zha_config_entry = MockConfigEntry(
domain="zha",
data={"device": {"path": usb_data.device}},
data={"device": {"path": TEST_DEVICE}},
)
zha_config_entry.add_to_hass(hass)
@ -620,27 +510,16 @@ async def test_options_flow_zigbee_to_thread_zha_configured(
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
hass: HomeAssistant,
) -> None:
"""Test the options flow migration failure, OTBR still using the stick."""
config_entry = MockConfigEntry(
domain="homeassistant_sky_connect",
domain=TEST_DOMAIN,
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,
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
},
version=1,
minor_version=2,
@ -658,7 +537,7 @@ async def test_options_flow_thread_to_zigbee_otbr_configured(
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={"device": usb_data.device},
options={"device": TEST_DEVICE},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",

View File

@ -0,0 +1,158 @@
"""Test hardware 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_hardware.util import (
FirmwareGuess,
get_zha_device_path,
guess_firmware_type,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
ZHA_CONFIG_ENTRY = MockConfigEntry(
domain="zha",
unique_id="some_unique_id",
data={
"device": {
"path": "socket://1.2.3.4:5678",
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
},
version=4,
)
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"]
)
def test_get_zha_device_path_ignored_discovery() -> None:
"""Test extracting the ZHA device path from an ignored ZHA discovery."""
config_entry = MockConfigEntry(
domain="zha",
unique_id="some_unique_id",
data={},
version=4,
)
assert get_zha_device_path(config_entry) is None
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_hardware.util.is_hassio",
return_value=True,
),
patch(
"homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager",
return_value=mock_otbr_addon_manager,
),
patch(
"homeassistant.components.homeassistant_hardware.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"
)

View File

@ -1,30 +1,20 @@
"""Test the Home Assistant SkyConnect config flow."""
import asyncio
from collections.abc import Awaitable, Callable, Iterator
import contextlib
from typing import Any
from unittest.mock import AsyncMock, Mock, call, patch
from unittest.mock import Mock, patch
import pytest
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components import usb
from homeassistant.components.hassio.addon_manager import AddonInfo, AddonState
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
STEP_PICK_FIRMWARE_ZIGBEE,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
CONF_DISABLE_MULTI_PAN,
get_flasher_addon_manager,
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.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
@ -49,86 +39,6 @@ USB_DATA_ZBT1 = usb.UsbServiceInfo(
)
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
@contextlib.contextmanager
def mock_addon_info(
hass: HomeAssistant,
*,
is_hassio: bool = True,
app_type: ApplicationType = ApplicationType.EZSP,
otbr_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
),
flasher_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
),
) -> Iterator[tuple[Mock, Mock]]:
"""Mock the main addon states for the config flow."""
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_start_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_get_addon_info.return_value = flasher_addon_info
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_uninstall_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 = otbr_addon_info
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.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=is_hassio,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=app_type,
),
):
yield mock_otbr_manager, mock_flasher_manager
@pytest.mark.parametrize(
("usb_data", "model"),
[
@ -136,7 +46,7 @@ def mock_addon_info(
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee(
async def test_config_flow(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test the config flow for SkyConnect."""
@ -146,453 +56,42 @@ async def test_config_flow_zigbee(
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
await hass.async_block_till_done(wait_background_tasks=True)
# 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,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now uninstalling the addon
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "uninstall_zigbee_flasher_addon"
assert result["progress_action"] == "uninstall_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# We are finally done with the addon
assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()]
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
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_config_flow_zigbee_skip_step_if_installed(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""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
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
flasher_addon_info=AddonInfo(
available=True,
hostname=None,
options={
"device": "",
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.2.3",
),
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we skip installation, instead we directly run it
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": 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
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option
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"
assert result["description_placeholders"]["firmware_type"] == "ezsp"
assert result["description_placeholders"]["model"] == model
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 mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_RUNNING,
update_available=False,
version=None,
),
) as (mock_otbr_manager, mock_flasher_manager):
# 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 mock_addon_info(
hass,
is_hassio=False,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"device": 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)
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"] == "ezsp"
assert result["description_placeholders"]["model"] == model
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.options.async_configure(
async def mock_async_step_pick_firmware_zigbee(self, data):
return await self.async_step_confirm_zigbee(user_input={})
with patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow.async_step_pick_firmware_zigbee",
autospec=True,
side_effect=mock_async_step_pick_firmware_zigbee,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
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_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"
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(
@ -602,10 +101,10 @@ async def test_options_flow_zigbee_to_thread(
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_options_flow_thread_to_zigbee(
async def test_options_flow(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test the options flow for SkyConnect, migrating Thread to Zigbee."""
"""Test the options flow for SkyConnect."""
config_entry = MockConfigEntry(
domain="homeassistant_sky_connect",
data={
@ -632,62 +131,32 @@ async def test_options_flow_thread_to_zigbee(
assert result["description_placeholders"]["firmware_type"] == "spinel"
assert result["description_placeholders"]["model"] == model
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
async def mock_async_step_pick_firmware_zigbee(self, data):
return await self.async_step_confirm_zigbee(user_input={})
with patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow.async_step_pick_firmware_zigbee",
autospec=True,
side_effect=mock_async_step_pick_firmware_zigbee,
):
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
assert result["result"] is True
# The firmware type has been updated
assert config_entry.data["firmware"] == "ezsp"
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,
}
@pytest.mark.parametrize(

View File

@ -4,8 +4,8 @@ from unittest.mock import patch
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components.homeassistant_hardware.util import FirmwareGuess
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.components.homeassistant_sky_connect.util import FirmwareGuess
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry

View File

@ -1,24 +1,14 @@
"""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
@ -52,20 +42,6 @@ CONNECT_ZBT1_CONFIG_ENTRY = MockConfigEntry(
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."""
@ -85,131 +61,3 @@ def test_get_hardware_variant() -> None:
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"]
)
def test_get_zha_device_path_ignored_discovery() -> None:
"""Test extracting the ZHA device path from an ignored ZHA discovery."""
config_entry = MockConfigEntry(
domain="zha",
unique_id="some_unique_id",
data={},
version=4,
)
assert get_zha_device_path(config_entry) is None
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"
)