github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Expose the SkyConnect integration with a firmware config/options flow (#115363) 

Co-authored-by: Stefan Agner <stefan@agner.ch>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
Co-authored-by: Erik <erik@montnemery.com>
This commit is contained in:
puddly 2024-04-24 11:06:24 -04:00 committed by GitHub
parent e47e62cbbf
commit 380f192c93
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 2943 additions and 762 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
homeassistant/components/homeassistant_sky_connect/__init__.py
View File

@ -2,87 +2,62 @@
from __future__ import annotations
from homeassistant.components import usb
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
check_multi_pan_addon,
get_zigbee_socket,
multi_pan_addon_using_device,
)
from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import discovery_flow
import logging
from .const import DOMAIN
from .util import get_hardware_variant, get_usb_service_info
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .util import guess_firmware_type
async def _async_usb_scan_done(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Finish Home Assistant SkyConnect config entry setup."""
matcher = usb.USBCallbackMatcher(
domain=DOMAIN,
vid=entry.data["vid"].upper(),
pid=entry.data["pid"].upper(),
serial_number=entry.data["serial_number"].lower(),
manufacturer=entry.data["manufacturer"].lower(),
description=entry.data["description"].lower(),
)
if not usb.async_is_plugged_in(hass, matcher):
# The USB dongle is not plugged in, remove the config entry
hass.async_create_task(
hass.config_entries.async_remove(entry.entry_id), eager_start=True
)
return
usb_dev = entry.data["device"]
# The call to get_serial_by_id can be removed in HA Core 2024.1
dev_path = await hass.async_add_executor_job(usb.get_serial_by_id, usb_dev)
if not await multi_pan_addon_using_device(hass, dev_path):
usb_info = get_usb_service_info(entry)
await hass.config_entries.flow.async_init(
"zha",
context={"source": "usb"},
data=usb_info,
)
return
hw_variant = get_hardware_variant(entry)
hw_discovery_data = {
"name": f"{hw_variant.short_name} Multiprotocol",
"port": {
"path": get_zigbee_socket(),
},
"radio_type": "ezsp",
}
discovery_flow.async_create_flow(
hass,
"zha",
context={"source": SOURCE_HARDWARE},
data=hw_discovery_data,
)
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a Home Assistant SkyConnect config entry."""
try:
await check_multi_pan_addon(hass)
except HomeAssistantError as err:
raise ConfigEntryNotReady from err
@callback
def async_usb_scan_done() -> None:
"""Handle usb discovery started."""
hass.async_create_task(_async_usb_scan_done(hass, entry), eager_start=True)
unsub_usb = usb.async_register_initial_scan_callback(hass, async_usb_scan_done)
entry.async_on_unload(unsub_usb)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
return True
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Migrate old entry."""
_LOGGER.debug(
"Migrating from version %s:%s", config_entry.version, config_entry.minor_version
)
if config_entry.version == 1:
if config_entry.minor_version == 1:
# Add-on startup with type service get started before Core, always (e.g. the
# Multi-Protocol add-on). Probing the firmware would interfere with the add-on,
# so we can't safely probe here. Instead, we must make an educated guess!
firmware_guess = await guess_firmware_type(
hass, config_entry.data["device"]
)
new_data = {**config_entry.data}
new_data["firmware"] = firmware_guess.firmware_type.value
# Copy `description` to `product`
new_data["product"] = new_data["description"]
hass.config_entries.async_update_entry(
config_entry,
data=new_data,
version=1,
minor_version=2,
)
_LOGGER.debug(
"Migration to version %s.%s successful",
config_entry.version,
config_entry.minor_version,
)
return True
# This means the user has downgraded from a future version
return False

630
homeassistant/components/homeassistant_sky_connect/config_flow.py
View File

@ -2,29 +2,498 @@
from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
import logging
from typing import Any
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components import usb
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
AddonManager,
AddonState,
is_hassio,
)
from homeassistant.components.homeassistant_hardware import silabs_multiprotocol_addon
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.components.zha.repairs.wrong_silabs_firmware import (
probe_silabs_firmware_type,
)
from homeassistant.config_entries import (
ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
OptionsFlowWithConfigEntry,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow
from .const import DOMAIN, HardwareVariant
from .util import get_hardware_variant, get_usb_service_info
from .const import DOCS_WEB_FLASHER_URL, DOMAIN, ZHA_DOMAIN, HardwareVariant
from .util import (
get_hardware_variant,
get_otbr_addon_manager,
get_usb_service_info,
get_zha_device_path,
get_zigbee_flasher_addon_manager,
)
_LOGGER = logging.getLogger(__name__)
STEP_PICK_FIRMWARE_THREAD = "pick_firmware_thread"
STEP_PICK_FIRMWARE_ZIGBEE = "pick_firmware_zigbee"
class HomeAssistantSkyConnectConfigFlow(ConfigFlow, domain=DOMAIN):
class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
"""Base flow to install firmware."""
_failed_addon_name: str
_failed_addon_reason: str
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate base flow."""
super().__init__(*args, **kwargs)
self._usb_info: usb.UsbServiceInfo | None = None
self._hw_variant: HardwareVariant | None = None
self._probed_firmware_type: ApplicationType | None = None
self.addon_install_task: asyncio.Task | None = None
self.addon_start_task: asyncio.Task | None = None
self.addon_uninstall_task: asyncio.Task | None = None
def _get_translation_placeholders(self) -> dict[str, str]:
"""Shared translation placeholders."""
placeholders = {
"model": (
self._hw_variant.full_name
if self._hw_variant is not None
else "unknown"
),
"firmware_type": (
self._probed_firmware_type.value
if self._probed_firmware_type is not None
else "unknown"
),
"docs_web_flasher_url": DOCS_WEB_FLASHER_URL,
}
self.context["title_placeholders"] = placeholders
return placeholders
async def _async_set_addon_config(
self, config: dict, addon_manager: AddonManager
) -> None:
"""Set add-on config."""
try:
await addon_manager.async_set_addon_options(config)
except AddonError as err:
_LOGGER.error(err)
raise AbortFlow(
"addon_set_config_failed",
description_placeholders=self._get_translation_placeholders(),
) from err
async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo:
"""Return add-on info."""
try:
addon_info = await addon_manager.async_get_addon_info()
except AddonError as err:
_LOGGER.error(err)
raise AbortFlow(
"addon_info_failed",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": addon_manager.addon_name,
},
) from err
return addon_info
async def async_step_pick_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Thread or Zigbee firmware."""
assert self._usb_info is not None
self._probed_firmware_type = await probe_silabs_firmware_type(
self._usb_info.device,
probe_methods=(
# We probe in order of frequency: Zigbee, Thread, then multi-PAN
ApplicationType.GECKO_BOOTLOADER,
ApplicationType.EZSP,
ApplicationType.SPINEL,
ApplicationType.CPC,
),
)
if self._probed_firmware_type not in (
ApplicationType.EZSP,
ApplicationType.SPINEL,
ApplicationType.CPC,
):
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
return self.async_show_menu(
step_id="pick_firmware",
menu_options=[
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
],
description_placeholders=self._get_translation_placeholders(),
)
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Zigbee firmware."""
# Allow the stick to be used with ZHA without flashing
if self._probed_firmware_type == ApplicationType.EZSP:
return await self.async_step_confirm_zigbee()
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio",
description_placeholders=self._get_translation_placeholders(),
)
# Only flash new firmware if we need to
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(fw_flasher_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_zigbee_flasher_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_run_zigbee_flasher_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
)
async def async_step_install_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the Zigbee flasher addon."""
return await self._install_addon(
get_zigbee_flasher_addon_manager(self.hass),
"install_zigbee_flasher_addon",
"run_zigbee_flasher_addon",
)
async def _install_addon(
self,
addon_manager: silabs_multiprotocol_addon.WaitingAddonManager,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult:
"""Show progress dialog for installing an addon."""
addon_info = await self._async_get_addon_info(addon_manager)
_LOGGER.debug("Flasher addon state: %s", addon_info)
if not self.addon_install_task:
self.addon_install_task = self.hass.async_create_task(
addon_manager.async_install_addon_waiting(),
"Addon install",
)
if not self.addon_install_task.done():
return self.async_show_progress(
step_id=step_id,
progress_action="install_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": addon_manager.addon_name,
},
progress_task=self.addon_install_task,
)
try:
await self.addon_install_task
except AddonError as err:
_LOGGER.error(err)
self._failed_addon_name = addon_manager.addon_name
self._failed_addon_reason = "addon_install_failed"
return self.async_show_progress_done(next_step_id="addon_operation_failed")
finally:
self.addon_install_task = None
return self.async_show_progress_done(next_step_id=next_step_id)
async def async_step_addon_operation_failed(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Abort when add-on installation or start failed."""
return self.async_abort(
reason=self._failed_addon_reason,
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": self._failed_addon_name,
},
)
async def async_step_run_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure the flasher addon to point to the SkyConnect and run it."""
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(fw_flasher_manager)
assert self._usb_info is not None
new_addon_config = {
**addon_info.options,
"device": self._usb_info.device,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
_LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, fw_flasher_manager)
if not self.addon_start_task:
async def start_and_wait_until_done() -> None:
await fw_flasher_manager.async_start_addon_waiting()
# Now that the addon is running, wait for it to finish
await fw_flasher_manager.async_wait_until_addon_state(
AddonState.NOT_RUNNING
)
self.addon_start_task = self.hass.async_create_task(
start_and_wait_until_done()
)
if not self.addon_start_task.done():
return self.async_show_progress(
step_id="run_zigbee_flasher_addon",
progress_action="run_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_start_task,
)
try:
await self.addon_start_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
self._failed_addon_name = fw_flasher_manager.addon_name
self._failed_addon_reason = "addon_start_failed"
return self.async_show_progress_done(next_step_id="addon_operation_failed")
finally:
self.addon_start_task = None
return self.async_show_progress_done(
next_step_id="uninstall_zigbee_flasher_addon"
)
async def async_step_uninstall_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Uninstall the flasher addon."""
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
if not self.addon_uninstall_task:
_LOGGER.debug("Uninstalling flasher addon")
self.addon_uninstall_task = self.hass.async_create_task(
fw_flasher_manager.async_uninstall_addon_waiting()
)
if not self.addon_uninstall_task.done():
return self.async_show_progress(
step_id="uninstall_zigbee_flasher_addon",
progress_action="uninstall_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_uninstall_task,
)
try:
await self.addon_uninstall_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
# The uninstall failing isn't critical so we can just continue
finally:
self.addon_uninstall_task = None
return self.async_show_progress_done(next_step_id="confirm_zigbee")
async def async_step_confirm_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm Zigbee setup."""
assert self._usb_info is not None
assert self._hw_variant is not None
self._probed_firmware_type = ApplicationType.EZSP
if user_input is not None:
await self.hass.config_entries.flow.async_init(
ZHA_DOMAIN,
context={"source": "hardware"},
data={
"name": self._hw_variant.full_name,
"port": {
"path": self._usb_info.device,
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
},
)
return self._async_flow_finished()
return self.async_show_form(
step_id="confirm_zigbee",
description_placeholders=self._get_translation_placeholders(),
)
async def async_step_pick_firmware_thread(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Thread firmware."""
# We install the OTBR addon no matter what, since it is required to use Thread
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio_thread",
description_placeholders=self._get_translation_placeholders(),
)
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_otbr_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_start_otbr_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="otbr_addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
)
async def async_step_install_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the OTBR addon."""
return await self._install_addon(
get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon"
)
async def async_step_start_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure OTBR to point to the SkyConnect and run the addon."""
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
assert self._usb_info is not None
new_addon_config = {
**addon_info.options,
"device": self._usb_info.device,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
}
_LOGGER.debug("Reconfiguring OTBR addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, otbr_manager)
if not self.addon_start_task:
self.addon_start_task = self.hass.async_create_task(
otbr_manager.async_start_addon_waiting()
)
if not self.addon_start_task.done():
return self.async_show_progress(
step_id="start_otbr_addon",
progress_action="start_otbr_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
progress_task=self.addon_start_task,
)
try:
await self.addon_start_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
self._failed_addon_name = otbr_manager.addon_name
self._failed_addon_reason = "addon_start_failed"
return self.async_show_progress_done(next_step_id="addon_operation_failed")
finally:
self.addon_start_task = None
return self.async_show_progress_done(next_step_id="confirm_otbr")
async def async_step_confirm_otbr(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm OTBR setup."""
assert self._usb_info is not None
assert self._hw_variant is not None
self._probed_firmware_type = ApplicationType.SPINEL
if user_input is not None:
# OTBR discovery is done automatically via hassio
return self._async_flow_finished()
return self.async_show_form(
step_id="confirm_otbr",
description_placeholders=self._get_translation_placeholders(),
)
@abstractmethod
def _async_flow_finished(self) -> ConfigFlowResult:
"""Finish the flow."""
# This should be implemented by a subclass
raise NotImplementedError
class HomeAssistantSkyConnectConfigFlow(
BaseFirmwareInstallFlow, ConfigFlow, domain=DOMAIN
):
"""Handle a config flow for Home Assistant SkyConnect."""
VERSION = 1
MINOR_VERSION = 2
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> HomeAssistantSkyConnectOptionsFlow:
) -> OptionsFlow:
"""Return the options flow."""
return HomeAssistantSkyConnectOptionsFlow(config_entry)
firmware_type = ApplicationType(config_entry.data["firmware"])
if firmware_type is ApplicationType.CPC:
return HomeAssistantSkyConnectMultiPanOptionsFlowHandler(config_entry)
return HomeAssistantSkyConnectOptionsFlowHandler(config_entry)
async def async_step_usb(
self, discovery_info: usb.UsbServiceInfo
@ -37,27 +506,62 @@ class HomeAssistantSkyConnectConfigFlow(ConfigFlow, domain=DOMAIN):
manufacturer = discovery_info.manufacturer
description = discovery_info.description
unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}"
if await self.async_set_unique_id(unique_id):
self._abort_if_unique_id_configured(updates={"device": device})
discovery_info.device = await self.hass.async_add_executor_job(
usb.get_serial_by_id, discovery_info.device
)
self._usb_info = discovery_info
assert description is not None
hw_variant = HardwareVariant.from_usb_product_name(description)
self._hw_variant = HardwareVariant.from_usb_product_name(description)
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm a discovery."""
self._set_confirm_only()
# Without confirmation, discovery can automatically progress into parts of the
# config flow logic that interacts with hardware.
if user_input is not None:
return await self.async_step_pick_firmware()
return self.async_show_form(
step_id="confirm",
description_placeholders=self._get_translation_placeholders(),
)
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._usb_info is not None
assert self._hw_variant is not None
assert self._probed_firmware_type is not None
return self.async_create_entry(
title=hw_variant.full_name,
title=self._hw_variant.full_name,
data={
"device": device,
"vid": vid,
"pid": pid,
"serial_number": serial_number,
"manufacturer": manufacturer,
"description": description,
"vid": self._usb_info.vid,
"pid": self._usb_info.pid,
"serial_number": self._usb_info.serial_number,
"manufacturer": self._usb_info.manufacturer,
"description": self._usb_info.description, # For backwards compatibility
"product": self._usb_info.description,
"device": self._usb_info.device,
"firmware": self._probed_firmware_type.value,
},
)
class HomeAssistantSkyConnectOptionsFlow(silabs_multiprotocol_addon.OptionsFlowHandler):
"""Handle an option flow for Home Assistant SkyConnect."""
class HomeAssistantSkyConnectMultiPanOptionsFlowHandler(
silabs_multiprotocol_addon.OptionsFlowHandler
):
"""Multi-PAN options flow for Home Assistant SkyConnect."""
async def _async_serial_port_settings(
self,
@ -92,3 +596,97 @@ class HomeAssistantSkyConnectOptionsFlow(silabs_multiprotocol_addon.OptionsFlowH
def _hardware_name(self) -> str:
"""Return the name of the hardware."""
return self._hw_variant.full_name
class HomeAssistantSkyConnectOptionsFlowHandler(
BaseFirmwareInstallFlow, OptionsFlowWithConfigEntry
):
"""Zigbee and Thread options flow handlers."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Instantiate options flow."""
super().__init__(*args, **kwargs)
self._usb_info = get_usb_service_info(self.config_entry)
self._probed_firmware_type = ApplicationType(self.config_entry.data["firmware"])
self._hw_variant = HardwareVariant.from_usb_product_name(
self.config_entry.data["product"]
)
# Make `context` a regular dictionary
self.context = {}
# Regenerate the translation placeholders
self._get_translation_placeholders()
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options flow."""
# Don't probe the running firmware, we load it from the config entry
return self.async_show_menu(
step_id="pick_firmware",
menu_options=[
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
],
description_placeholders=self._get_translation_placeholders(),
)
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Zigbee firmware."""
assert self._usb_info is not None
if is_hassio(self.hass):
otbr_manager = get_otbr_addon_manager(self.hass)
otbr_addon_info = await self._async_get_addon_info(otbr_manager)
if (
otbr_addon_info.state != AddonState.NOT_INSTALLED
and otbr_addon_info.options.get("device") == self._usb_info.device
):
raise AbortFlow(
"otbr_still_using_stick",
description_placeholders=self._get_translation_placeholders(),
)
return await super().async_step_pick_firmware_zigbee(user_input)
async def async_step_pick_firmware_thread(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Thread firmware."""
assert self._usb_info is not None
zha_entries = self.hass.config_entries.async_entries(
ZHA_DOMAIN,
include_ignore=False,
include_disabled=True,
)
if zha_entries and get_zha_device_path(zha_entries[0]) == self._usb_info.device:
raise AbortFlow(
"zha_still_using_stick",
description_placeholders=self._get_translation_placeholders(),
)
return await super().async_step_pick_firmware_thread(user_input)
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._usb_info is not None
assert self._hw_variant is not None
assert self._probed_firmware_type is not None
self.hass.config_entries.async_update_entry(
entry=self.config_entry,
data={
**self.config_entry.data,
"firmware": self._probed_firmware_type.value,
},
options=self.config_entry.options,
)
return self.async_create_entry(title="", data={})

11
homeassistant/components/homeassistant_sky_connect/const.py
View File

@ -5,6 +5,17 @@ import enum
from typing import Self
DOMAIN = "homeassistant_sky_connect"
ZHA_DOMAIN = "zha"
DOCS_WEB_FLASHER_URL = "https://skyconnect.home-assistant.io/firmware-update/"
OTBR_ADDON_NAME = "OpenThread Border Router"
OTBR_ADDON_MANAGER_DATA = "openthread_border_router"
OTBR_ADDON_SLUG = "core_openthread_border_router"
ZIGBEE_FLASHER_ADDON_NAME = "Silicon Labs Flasher"
ZIGBEE_FLASHER_ADDON_MANAGER_DATA = "silabs_flasher"
ZIGBEE_FLASHER_ADDON_SLUG = "core_silabs_flasher"
@dataclasses.dataclass(frozen=True)

2
homeassistant/components/homeassistant_sky_connect/hardware.py
View File

@ -25,7 +25,7 @@ def async_info(hass: HomeAssistant) -> list[HardwareInfo]:
pid=entry.data["pid"],
serial_number=entry.data["serial_number"],
manufacturer=entry.data["manufacturer"],
description=entry.data["description"],
description=entry.data["product"],
),
name=get_hardware_variant(entry).full_name,
url=DOCUMENTATION_URL,

2
homeassistant/components/homeassistant_sky_connect/manifest.json
View File

@ -5,7 +5,7 @@
"config_flow": true,
"dependencies": ["hardware", "usb", "homeassistant_hardware"],
"documentation": "https://www.home-assistant.io/integrations/homeassistant_sky_connect",
"integration_type": "hardware",
"integration_type": "device",
"usb": [
{
"vid": "10C4",

128
homeassistant/components/homeassistant_sky_connect/strings.json
View File

@ -57,6 +57,50 @@
"start_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]"
},
"confirm": {
"title": "[%key:component::homeassistant_sky_connect::config::step::confirm::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::confirm::description%]"
},
"pick_firmware": {
"title": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::description%]",
"menu_options": {
"pick_firmware_thread": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::menu_options::pick_firmware_thread%]",
"pick_firmware_zigbee": "[%key:component::homeassistant_sky_connect::config::step::pick_firmware::menu_options::pick_firmware_zigbee%]"
}
},
"install_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::run_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::confirm_zigbee::description%]"
},
"install_otbr_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::install_otbr_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::install_otbr_addon::description%]"
},
"start_otbr_addon": {
"title": "[%key:component::homeassistant_sky_connect::config::step::start_otbr_addon::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::start_otbr_addon::description%]"
},
"otbr_failed": {
"title": "[%key:component::homeassistant_sky_connect::config::step::otbr_failed::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::otbr_failed::description%]"
},
"confirm_otbr": {
"title": "[%key:component::homeassistant_sky_connect::config::step::confirm_otbr::title%]",
"description": "[%key:component::homeassistant_sky_connect::config::step::confirm_otbr::description%]"
}
},
"error": {
@ -68,12 +112,92 @@
"addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]",
"addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]",
"addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]",
"zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]",
"not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]",
"zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]"
"not_hassio_thread": "[%key:component::homeassistant_sky_connect::config::abort::not_hassio_thread%]",
"otbr_addon_already_running": "[%key:component::homeassistant_sky_connect::config::abort::otbr_addon_already_running%]",
"zha_still_using_stick": "This {model} is in use by the Zigbee Home Automation integration. Please migrate your Zigbee network to another adapter or delete the integration and try again.",
"otbr_still_using_stick": "This {model} is in use by the OpenThread Border Router add-on. If you use the Thread network, make sure you have alternative border routers. Uninstall the add-on and try again."
},
"progress": {
"install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]"
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"install_zigbee_flasher_addon": "[%key:component::homeassistant_sky_connect::config::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_sky_connect::config::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_sky_connect::config::progress::uninstall_zigbee_flasher_addon%]"
}
},
"config": {
"flow_title": "{model}",
"step": {
"confirm": {
"title": "Set up the {model}",
"description": "The {model} can be used as either a Thread border router or a Zigbee coordinator. In the next step, you will choose which firmware will be configured."
},
"pick_firmware": {
"title": "Pick your firmware",
"description": "The {model} can be used as a Thread border router or a Zigbee coordinator.",
"menu_options": {
"pick_firmware_thread": "Use as a Thread border router",
"pick_firmware_zigbee": "Use as a Zigbee coordinator"
}
},
"install_zigbee_flasher_addon": {
"title": "Installing flasher",
"description": "Installing the Silicon Labs Flasher add-on."
},
"run_zigbee_flasher_addon": {
"title": "Installing Zigbee firmware",
"description": "Installing Zigbee firmware. This will take about a minute."
},
"uninstall_zigbee_flasher_addon": {
"title": "Removing flasher",
"description": "Removing the Silicon Labs Flasher add-on."
},
"zigbee_flasher_failed": {
"title": "Zigbee installation failed",
"description": "The Zigbee firmware installation process was unsuccessful. Ensure no other software is trying to communicate with the {model} and try again."
},
"confirm_zigbee": {
"title": "Zigbee setup complete",
"description": "Your {model} is now a Zigbee coordinator and will be shown as discovered by the Zigbee Home Automation integration once you exit."
},
"install_otbr_addon": {
"title": "Installing OpenThread Border Router add-on",
"description": "The OpenThread Border Router (OTBR) add-on is being installed."
},
"start_otbr_addon": {
"title": "Starting OpenThread Border Router add-on",
"description": "The OpenThread Border Router (OTBR) add-on is now starting."
},
"otbr_failed": {
"title": "Failed to setup OpenThread Border Router",
"description": "The OpenThread Border Router add-on installation was unsuccessful. Ensure no other software is trying to communicate with the {model}, you have access to the internet and can install other add-ons, and try again. Check the Supervisor logs if the problem persists."
},
"confirm_otbr": {
"title": "OpenThread Border Router setup complete",
"description": "Your {model} is now an OpenThread Border Router and will show up in the Thread integration once you exit."
}
},
"abort": {
"addon_info_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_info_failed%]",
"addon_install_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_install_failed%]",
"addon_already_running": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_already_running%]",
"addon_set_config_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_set_config_failed%]",
"addon_start_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::addon_start_failed%]",
"zha_migration_failed": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::zha_migration_failed%]",
"not_hassio": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::abort::not_hassio%]",
"not_hassio_thread": "The OpenThread Border Router addon can only be installed with Home Assistant OS. If you would like to use the {model} as an Thread border router, please flash the firmware manually using the [web flasher]({docs_web_flasher_url}) and set up OpenThread Border Router to communicate with it.",
"otbr_addon_already_running": "The OpenThread Border Router add-on is already running, it cannot be installed again."
},
"progress": {
"install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"install_zigbee_flasher_addon": "The Silicon Labs Flasher addon is installed, this may take a few minutes.",
"run_zigbee_flasher_addon": "Please wait while Zigbee firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes.",
"uninstall_zigbee_flasher_addon": "The Silicon Labs Flasher addon is being removed."
}
}
}

140
homeassistant/components/homeassistant_sky_connect/util.py
View File

@ -2,10 +2,35 @@
from __future__ import annotations
from homeassistant.components import usb
from homeassistant.config_entries import ConfigEntry
from collections import defaultdict
from dataclasses import dataclass
import logging
from typing import cast
from .const import HardwareVariant
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components import usb
from homeassistant.components.hassio import AddonError, AddonState, is_hassio
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
WaitingAddonManager,
get_multiprotocol_addon_manager,
)
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.singleton import singleton
from .const import (
OTBR_ADDON_MANAGER_DATA,
OTBR_ADDON_NAME,
OTBR_ADDON_SLUG,
ZHA_DOMAIN,
ZIGBEE_FLASHER_ADDON_MANAGER_DATA,
ZIGBEE_FLASHER_ADDON_NAME,
ZIGBEE_FLASHER_ADDON_SLUG,
HardwareVariant,
)
_LOGGER = logging.getLogger(__name__)
def get_usb_service_info(config_entry: ConfigEntry) -> usb.UsbServiceInfo:
@ -16,10 +41,115 @@ def get_usb_service_info(config_entry: ConfigEntry) -> usb.UsbServiceInfo:
pid=config_entry.data["pid"],
serial_number=config_entry.data["serial_number"],
manufacturer=config_entry.data["manufacturer"],
description=config_entry.data["description"],
description=config_entry.data["product"],
)
def get_hardware_variant(config_entry: ConfigEntry) -> HardwareVariant:
"""Get the hardware variant from the config entry."""
return HardwareVariant.from_usb_product_name(config_entry.data["description"])
return HardwareVariant.from_usb_product_name(config_entry.data["product"])
def get_zha_device_path(config_entry: ConfigEntry) -> str:
"""Get the device path from a ZHA config entry."""
return cast(str, config_entry.data["device"]["path"])
@singleton(OTBR_ADDON_MANAGER_DATA)
@callback
def get_otbr_addon_manager(hass: HomeAssistant) -> WaitingAddonManager:
"""Get the OTBR add-on manager."""
return WaitingAddonManager(
hass,
_LOGGER,
OTBR_ADDON_NAME,
OTBR_ADDON_SLUG,
)
@singleton(ZIGBEE_FLASHER_ADDON_MANAGER_DATA)
@callback
def get_zigbee_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager:
"""Get the flasher add-on manager."""
return WaitingAddonManager(
hass,
_LOGGER,
ZIGBEE_FLASHER_ADDON_NAME,
ZIGBEE_FLASHER_ADDON_SLUG,
)
@dataclass(slots=True, kw_only=True)
class FirmwareGuess:
"""Firmware guess."""
is_running: bool
firmware_type: ApplicationType
source: str
async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> FirmwareGuess:
"""Guess the firmware type based on installed addons and other integrations."""
device_guesses: defaultdict[str | None, list[FirmwareGuess]] = defaultdict(list)
for zha_config_entry in hass.config_entries.async_entries(ZHA_DOMAIN):
zha_path = get_zha_device_path(zha_config_entry)
device_guesses[zha_path].append(
FirmwareGuess(
is_running=(zha_config_entry.state == ConfigEntryState.LOADED),
firmware_type=ApplicationType.EZSP,
source="zha",
)
)
if is_hassio(hass):
otbr_addon_manager = get_otbr_addon_manager(hass)
try:
otbr_addon_info = await otbr_addon_manager.async_get_addon_info()
except AddonError:
pass
else:
if otbr_addon_info.state != AddonState.NOT_INSTALLED:
otbr_path = otbr_addon_info.options.get("device")
device_guesses[otbr_path].append(
FirmwareGuess(
is_running=(otbr_addon_info.state == AddonState.RUNNING),
firmware_type=ApplicationType.SPINEL,
source="otbr",
)
)
multipan_addon_manager = await get_multiprotocol_addon_manager(hass)
try:
multipan_addon_info = await multipan_addon_manager.async_get_addon_info()
except AddonError:
pass
else:
if multipan_addon_info.state != AddonState.NOT_INSTALLED:
multipan_path = multipan_addon_info.options.get("device")
device_guesses[multipan_path].append(
FirmwareGuess(
is_running=(multipan_addon_info.state == AddonState.RUNNING),
firmware_type=ApplicationType.CPC,
source="multiprotocol",
)
)
# Fall back to EZSP if we can't guess the firmware type
if device_path not in device_guesses:
return FirmwareGuess(
is_running=False, firmware_type=ApplicationType.EZSP, source="unknown"
)
# Prioritizes guesses that were pulled from a running addon or integration but keep
# the sort order we defined above
guesses = sorted(
device_guesses[device_path],
key=lambda guess: guess.is_running,
)
assert guesses
return guesses[-1]

9
homeassistant/components/zha/repairs/wrong_silabs_firmware.py
View File

@ -74,9 +74,14 @@ def _detect_radio_hardware(hass: HomeAssistant, device: str) -> HardwareType:
return HardwareType.OTHER
async def probe_silabs_firmware_type(device: str) -> ApplicationType | None:
async def probe_silabs_firmware_type(
device: str, *, probe_methods: ApplicationType | None = None
) -> ApplicationType | None:
"""Probe the running firmware on a Silabs device."""
flasher = Flasher(device=device)
flasher = Flasher(
device=device,
**({"probe_methods": probe_methods} if probe_methods else {}),
)
try:
await flasher.probe_app_type()

5
homeassistant/generated/integrations.json
View File

@ -2565,6 +2565,11 @@
"integration_type": "virtual",
"supported_by": "netatmo"
},
"homeassistant_sky_connect": {
"name": "Home Assistant SkyConnect",
"integration_type": "device",
"config_flow": true
},
"homematic": {
"name": "Homematic",
"integrations": {

1
script/hassfest/dependencies.py
View File

@ -157,6 +157,7 @@ IGNORE_VIOLATIONS = {
("zha", "homeassistant_hardware"),
("zha", "homeassistant_sky_connect"),
("zha", "homeassistant_yellow"),
("homeassistant_sky_connect", "zha"),
# This should become a helper method that integrations can submit data to
("websocket_api", "lovelace"),
("websocket_api", "shopping_list"),

1096
tests/components/homeassistant_sky_connect/test_config_flow.py
View File

File diff suppressed because it is too large Load Diff

920
tests/components/homeassistant_sky_connect/test_config_flow_failures.py Normal file
View File

@ -0,0 +1,920 @@
"""Test the Home Assistant SkyConnect config flow failure cases."""
from unittest.mock import AsyncMock, Mock, patch
import pytest
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components import usb
from homeassistant.components.hassio.addon_manager import (
AddonError,
AddonInfo,
AddonState,
)
from homeassistant.components.homeassistant_sky_connect.config_flow import (
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
)
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.components.homeassistant_sky_connect.util import (
get_otbr_addon_manager,
get_zigbee_flasher_addon_manager,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .test_config_flow import USB_DATA_ZBT1, delayed_side_effect
from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_cannot_probe_firmware(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when firmware cannot be probed."""
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=None,
):
# Start the flow
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
# Probing fails
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "unsupported_firmware"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_not_hassio_wrong_firmware(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test when the stick is used with a non-hassio setup but the firmware is bad."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=False,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "not_hassio"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_addon_already_running(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon is already running."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot get addon info
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_already_running"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_addon_info_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.side_effect = AddonError()
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot get addon info
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_info_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_addon_install_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=AddonError()
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot install addon
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_install_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_addon_set_config_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_set_addon_options = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_set_config_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_run_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_start_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_zigbee_flasher_uninstall_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_start_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=AddonError()
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
# Uninstall failure isn't critical
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_not_hassio(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test when the stick is used with a non-hassio setup and Thread is selected."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=False,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "not_hassio_thread"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_addon_info_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.side_effect = AddonError()
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
# Cannot get addon info
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_info_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_addon_already_running(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when the Thread addon is already running."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
# Cannot install addon
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "otbr_addon_already_running"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_addon_install_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
# Cannot install addon
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_install_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_addon_set_config_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon cannot be configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_set_config_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_flasher_run_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_start_addon_waiting = AsyncMock(side_effect=AddonError())
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_start_failed"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_config_flow_thread_flasher_uninstall_fails(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": "usb"}, data=usb_data
)
with patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.probe_silabs_firmware_type",
return_value=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_start_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_otbr_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=AddonError()
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
# Uninstall failure isn't critical
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_options_flow_zigbee_to_thread_zha_configured(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test the options flow migration failure, ZHA using the stick."""
config_entry = MockConfigEntry(
domain="homeassistant_sky_connect",
data={
"firmware": "ezsp",
"device": usb_data.device,
"manufacturer": usb_data.manufacturer,
"pid": usb_data.pid,
"description": usb_data.description,
"product": usb_data.description,
"serial_number": usb_data.serial_number,
"vid": usb_data.vid,
},
version=1,
minor_version=2,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Set up ZHA as well
zha_config_entry = MockConfigEntry(
domain="zha",
data={"device": {"path": usb_data.device}},
)
zha_config_entry.add_to_hass(hass)
# Confirm options flow
result = await hass.config_entries.options.async_init(config_entry.entry_id)
# Pick Thread
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "zha_still_using_stick"
@pytest.mark.parametrize(
("usb_data", "model"),
[
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
],
)
async def test_options_flow_thread_to_zigbee_otbr_configured(
usb_data: usb.UsbServiceInfo, model: str, hass: HomeAssistant
) -> None:
"""Test the options flow migration failure, OTBR still using the stick."""
config_entry = MockConfigEntry(
domain="homeassistant_sky_connect",
data={
"firmware": "spinel",
"device": usb_data.device,
"manufacturer": usb_data.manufacturer,
"pid": usb_data.pid,
"description": usb_data.description,
"product": usb_data.description,
"serial_number": usb_data.serial_number,
"vid": usb_data.vid,
},
version=1,
minor_version=2,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
# Confirm options flow
result = await hass.config_entries.options.async_init(config_entry.entry_id)
# Pick Zigbee
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": usb_data.device},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.config_flow.is_hassio",
return_value=True,
),
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "otbr_still_using_stick"

22
tests/components/homeassistant_sky_connect/test_hardware.py
View File

@ -1,7 +1,5 @@
"""Test the Home Assistant SkyConnect hardware platform."""
from unittest.mock import patch
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.core import EVENT_HOMEASSISTANT_STARTED, HomeAssistant
from homeassistant.setup import async_setup_component
@ -15,7 +13,8 @@ CONFIG_ENTRY_DATA = {
"pid": "EA60",
"serial_number": "9e2adbd75b8beb119fe564a0f320645d",
"manufacturer": "Nabu Casa",
"description": "SkyConnect v1.0",
"product": "SkyConnect v1.0",
"firmware": "ezsp",
}
CONFIG_ENTRY_DATA_2 = {
@ -24,7 +23,8 @@ CONFIG_ENTRY_DATA_2 = {
"pid": "EA60",
"serial_number": "9e2adbd75b8beb119fe564a0f320645d",
"manufacturer": "Nabu Casa",
"description": "Home Assistant Connect ZBT-1",
"product": "Home Assistant Connect ZBT-1",
"firmware": "ezsp",
}
@ -42,22 +42,24 @@ async def test_hardware_info(
options={},
title="Home Assistant SkyConnect",
unique_id="unique_1",
version=1,
minor_version=2,
)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
config_entry_2 = MockConfigEntry(
data=CONFIG_ENTRY_DATA_2,
domain=DOMAIN,
options={},
title="Home Assistant Connect ZBT-1",
unique_id="unique_2",
version=1,
minor_version=2,
)
config_entry_2.add_to_hass(hass)
with patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert await hass.config_entries.async_setup(config_entry_2.entry_id)
client = await hass_ws_client(hass)

393
tests/components/homeassistant_sky_connect/test_init.py
View File

@ -1,377 +1,56 @@
"""Test the Home Assistant SkyConnect integration."""
from collections.abc import Generator
from typing import Any
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import patch
import pytest
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components import zha
from homeassistant.components.hassio.handler import HassioAPIError
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import EVENT_HOMEASSISTANT_STARTED, HomeAssistant
from homeassistant.setup import async_setup_component
from homeassistant.components.homeassistant_sky_connect.util import FirmwareGuess
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
CONFIG_ENTRY_DATA = {
"device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
"vid": "10C4",
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"description": "SkyConnect v1.0",
}
async def test_config_entry_migration_v2(hass: HomeAssistant) -> None:
"""Test migrating config entries from v1 to v2 format."""
@pytest.fixture(autouse=True)
def disable_usb_probing() -> Generator[None, None, None]:
"""Disallow touching of system USB devices during unit tests."""
with patch("homeassistant.components.usb.comports", return_value=[]):
yield
@pytest.fixture
def mock_zha_config_flow_setup() -> Generator[None, None, None]:
"""Mock the radio connection and probing of the ZHA config flow."""
def mock_probe(config: dict[str, Any]) -> None:
# The radio probing will return the correct baudrate
return {**config, "baudrate": 115200}
mock_connect_app = MagicMock()
mock_connect_app.__aenter__.return_value.backups.backups = []
with (
patch(
"bellows.zigbee.application.ControllerApplication.probe",
side_effect=mock_probe,
),
patch(
"homeassistant.components.zha.radio_manager.ZhaRadioManager.connect_zigpy_app",
return_value=mock_connect_app,
),
):
yield
@pytest.mark.parametrize(
("onboarded", "num_entries", "num_flows"), [(False, 1, 0), (True, 0, 1)]
)
async def test_setup_entry(
mock_zha_config_flow_setup,
hass: HomeAssistant,
addon_store_info,
onboarded,
num_entries,
num_flows,
) -> None:
"""Test setup of a config entry, including setup of zha."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
) as mock_is_plugged_in,
patch(
"homeassistant.components.onboarding.async_is_onboarded",
return_value=onboarded,
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(mock_is_plugged_in.mock_calls) == 1
matcher = mock_is_plugged_in.mock_calls[0].args[1]
assert matcher["vid"].isupper()
assert matcher["pid"].isupper()
assert matcher["serial_number"].islower()
assert matcher["manufacturer"].islower()
assert matcher["description"].islower()
# Finish setting up ZHA
if num_entries > 0:
zha_flows = hass.config_entries.flow.async_progress_by_handler("zha")
assert len(zha_flows) == 1
assert zha_flows[0]["step_id"] == "choose_formation_strategy"
await hass.config_entries.flow.async_configure(
zha_flows[0]["flow_id"],
user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS},
)
await hass.async_block_till_done()
assert len(hass.config_entries.flow.async_progress_by_handler("zha")) == num_flows
assert len(hass.config_entries.async_entries("zha")) == num_entries
async def test_setup_zha(
mock_zha_config_flow_setup, hass: HomeAssistant, addon_store_info
) -> None:
"""Test zha gets the right config."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
) as mock_is_plugged_in,
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(mock_is_plugged_in.mock_calls) == 1
zha_flows = hass.config_entries.flow.async_progress_by_handler("zha")
assert len(zha_flows) == 1
assert zha_flows[0]["step_id"] == "choose_formation_strategy"
# Finish setting up ZHA
await hass.config_entries.flow.async_configure(
zha_flows[0]["flow_id"],
user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS},
)
await hass.async_block_till_done()
config_entry = hass.config_entries.async_entries("zha")[0]
assert config_entry.data == {
"device": {
"baudrate": 115200,
"flow_control": None,
"path": CONFIG_ENTRY_DATA["device"],
unique_id="some_unique_id",
data={
"device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
"vid": "10C4",
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"description": "SkyConnect v1.0",
},
"radio_type": "ezsp",
}
assert config_entry.options == {}
assert config_entry.title == CONFIG_ENTRY_DATA["description"]
async def test_setup_zha_multipan(
hass: HomeAssistant, addon_info, addon_running
) -> None:
"""Test zha gets the right config."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
addon_info.return_value["options"]["device"] = CONFIG_ENTRY_DATA["device"]
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
version=1,
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
) as mock_is_plugged_in,
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
),
patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(mock_is_plugged_in.mock_calls) == 1
# Finish setting up ZHA
zha_flows = hass.config_entries.flow.async_progress_by_handler("zha")
assert len(zha_flows) == 1
assert zha_flows[0]["step_id"] == "choose_formation_strategy"
await hass.config_entries.flow.async_configure(
zha_flows[0]["flow_id"],
user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS},
)
await hass.async_block_till_done()
config_entry = hass.config_entries.async_entries("zha")[0]
assert config_entry.data == {
"device": {
"baudrate": 115200,
"flow_control": None,
"path": "socket://core-silabs-multiprotocol:9999",
},
"radio_type": "ezsp",
}
assert config_entry.options == {}
assert config_entry.title == "SkyConnect Multiprotocol"
async def test_setup_zha_multipan_other_device(
mock_zha_config_flow_setup, hass: HomeAssistant, addon_info, addon_running
) -> None:
"""Test zha gets the right config."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
addon_info.return_value["options"]["device"] = "/dev/not_our_sky_connect"
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant Yellow",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
) as mock_is_plugged_in,
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
),
patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
),
):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert len(mock_is_plugged_in.mock_calls) == 1
# Finish setting up ZHA
zha_flows = hass.config_entries.flow.async_progress_by_handler("zha")
assert len(zha_flows) == 1
assert zha_flows[0]["step_id"] == "choose_formation_strategy"
await hass.config_entries.flow.async_configure(
zha_flows[0]["flow_id"],
user_input={"next_step_id": zha.config_flow.FORMATION_REUSE_SETTINGS},
)
await hass.async_block_till_done()
config_entry = hass.config_entries.async_entries("zha")[0]
assert config_entry.data == {
"device": {
"baudrate": 115200,
"flow_control": None,
"path": CONFIG_ENTRY_DATA["device"],
},
"radio_type": "ezsp",
}
assert config_entry.options == {}
assert config_entry.title == CONFIG_ENTRY_DATA["description"]
async def test_setup_entry_wait_usb(hass: HomeAssistant) -> None:
"""Test setup of a config entry when the dongle is not plugged in."""
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=False,
) as mock_is_plugged_in:
"homeassistant.components.homeassistant_sky_connect.guess_firmware_type",
return_value=FirmwareGuess(
is_running=True,
firmware_type=ApplicationType.SPINEL,
source="otbr",
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
assert config_entry.state is ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
# USB discovery starts, config entry should be removed
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
assert len(mock_is_plugged_in.mock_calls) == 1
assert len(hass.config_entries.async_entries(DOMAIN)) == 0
assert config_entry.version == 1
assert config_entry.minor_version == 2
assert config_entry.data == {
"description": "SkyConnect v1.0",
"device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
"vid": "10C4",
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"product": "SkyConnect v1.0", # `description` has been copied to `product`
"firmware": "spinel", # new key
}
async def test_setup_entry_addon_info_fails(
hass: HomeAssistant, addon_store_info
) -> None:
"""Test setup of a config entry when fetching addon info fails."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
addon_store_info.side_effect = HassioAPIError("Boom")
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
),
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
),
patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
),
):
assert not await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_setup_entry_addon_not_running(
hass: HomeAssistant, addon_installed, start_addon
) -> None:
"""Test the addon is started if it is not running."""
assert await async_setup_component(hass, "usb", {})
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
# Setup the config entry
config_entry = MockConfigEntry(
data=CONFIG_ENTRY_DATA,
domain=DOMAIN,
options={},
title="Home Assistant SkyConnect",
)
config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_sky_connect.usb.async_is_plugged_in",
return_value=True,
),
patch(
"homeassistant.components.onboarding.async_is_onboarded", return_value=False
),
patch(
"homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon.is_hassio",
side_effect=Mock(return_value=True),
),
):
assert not await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
start_addon.assert_called_once()
await hass.config_entries.async_unload(config_entry.entry_id)

203
tests/components/homeassistant_sky_connect/test_util.py Normal file
View File

@ -0,0 +1,203 @@
"""Test SkyConnect utilities."""
from unittest.mock import AsyncMock, patch
from universal_silabs_flasher.const import ApplicationType
from homeassistant.components.hassio import AddonError, AddonInfo, AddonState
from homeassistant.components.homeassistant_sky_connect.const import (
DOMAIN,
HardwareVariant,
)
from homeassistant.components.homeassistant_sky_connect.util import (
FirmwareGuess,
get_hardware_variant,
get_usb_service_info,
get_zha_device_path,
guess_firmware_type,
)
from homeassistant.components.usb import UsbServiceInfo
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
SKYCONNECT_CONFIG_ENTRY = MockConfigEntry(
domain=DOMAIN,
unique_id="some_unique_id",
data={
"device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
"vid": "10C4",
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"product": "SkyConnect v1.0",
"firmware": "ezsp",
},
version=2,
)
CONNECT_ZBT1_CONFIG_ENTRY = MockConfigEntry(
domain=DOMAIN,
unique_id="some_unique_id",
data={
"device": "/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_9e2adbd75b8beb119fe564a0f320645d-if00-port0",
"vid": "10C4",
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"product": "Home Assistant Connect ZBT-1",
"firmware": "ezsp",
},
version=2,
)
ZHA_CONFIG_ENTRY = MockConfigEntry(
domain="zha",
unique_id="some_unique_id",
data={
"device": {
"path": "/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_3c0ed67c628beb11b1cd64a0f320645d-if00-port0",
"baudrate": 115200,
"flow_control": None,
},
"radio_type": "ezsp",
},
version=4,
)
def test_get_usb_service_info() -> None:
"""Test `get_usb_service_info` conversion."""
assert get_usb_service_info(SKYCONNECT_CONFIG_ENTRY) == UsbServiceInfo(
device=SKYCONNECT_CONFIG_ENTRY.data["device"],
vid=SKYCONNECT_CONFIG_ENTRY.data["vid"],
pid=SKYCONNECT_CONFIG_ENTRY.data["pid"],
serial_number=SKYCONNECT_CONFIG_ENTRY.data["serial_number"],
manufacturer=SKYCONNECT_CONFIG_ENTRY.data["manufacturer"],
description=SKYCONNECT_CONFIG_ENTRY.data["product"],
)
def test_get_hardware_variant() -> None:
"""Test `get_hardware_variant` extraction."""
assert get_hardware_variant(SKYCONNECT_CONFIG_ENTRY) == HardwareVariant.SKYCONNECT
assert (
get_hardware_variant(CONNECT_ZBT1_CONFIG_ENTRY) == HardwareVariant.CONNECT_ZBT1
)
def test_get_zha_device_path() -> None:
"""Test extracting the ZHA device path from its config entry."""
assert (
get_zha_device_path(ZHA_CONFIG_ENTRY) == ZHA_CONFIG_ENTRY.data["device"]["path"]
)
async def test_guess_firmware_type_unknown(hass: HomeAssistant) -> None:
"""Test guessing the firmware type."""
assert (await guess_firmware_type(hass, "/dev/missing")) == FirmwareGuess(
is_running=False, firmware_type=ApplicationType.EZSP, source="unknown"
)
async def test_guess_firmware_type(hass: HomeAssistant) -> None:
"""Test guessing the firmware."""
path = ZHA_CONFIG_ENTRY.data["device"]["path"]
ZHA_CONFIG_ENTRY.add_to_hass(hass)
ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.NOT_LOADED)
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=False, firmware_type=ApplicationType.EZSP, source="zha"
)
# When ZHA is running, we indicate as such when guessing
ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.LOADED)
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
)
mock_otbr_addon_manager = AsyncMock()
mock_multipan_addon_manager = AsyncMock()
with (
patch(
"homeassistant.components.homeassistant_sky_connect.util.is_hassio",
return_value=True,
),
patch(
"homeassistant.components.homeassistant_sky_connect.util.get_otbr_addon_manager",
return_value=mock_otbr_addon_manager,
),
patch(
"homeassistant.components.homeassistant_sky_connect.util.get_multiprotocol_addon_manager",
return_value=mock_multipan_addon_manager,
),
):
mock_otbr_addon_manager.async_get_addon_info.side_effect = AddonError()
mock_multipan_addon_manager.async_get_addon_info.side_effect = AddonError()
# Hassio errors are ignored and we still go with ZHA
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
)
mock_otbr_addon_manager.async_get_addon_info.side_effect = None
mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": "/some/other/device"},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
# We will prefer ZHA, as it is running (and actually pointing to the device)
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
)
mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": path},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.0.0",
)
# We will still prefer ZHA, as it is the one actually running
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.EZSP, source="zha"
)
mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": path},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
# Finally, ZHA loses out to OTBR
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.SPINEL, source="otbr"
)
mock_multipan_addon_manager.async_get_addon_info.side_effect = None
mock_multipan_addon_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": path},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
)
# Which will lose out to multi-PAN
assert (await guess_firmware_type(hass, path)) == FirmwareGuess(
is_running=True, firmware_type=ApplicationType.CPC, source="multiprotocol"
)

16
tests/components/usb/__init__.py
View File

@ -26,3 +26,19 @@ electro_lama_device = USBDevice(
manufacturer=None,
description="USB2.0-Serial",
)
skyconnect_macos_correct = USBDevice(
device="/dev/cu.SLAB_USBtoUART",
vid="10C4",
pid="EA60",
serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d",
manufacturer="Nabu Casa",
description="SkyConnect v1.0",
)
skyconnect_macos_incorrect = USBDevice(
device="/dev/cu.usbserial-2110",
vid="10C4",
pid="EA60",
serial_number="9ab1da1ea4b3ed11956f4eaca7669f5d",
manufacturer="Nabu Casa",
description="SkyConnect v1.0",
)

10
tests/components/zha/test_repairs.py
View File

@ -12,7 +12,7 @@ from zigpy.application import ControllerApplication
import zigpy.backups
from zigpy.exceptions import NetworkSettingsInconsistent
from homeassistant.components.homeassistant_sky_connect import (
from homeassistant.components.homeassistant_sky_connect.const import (
DOMAIN as SKYCONNECT_DOMAIN,
)
from homeassistant.components.repairs import DOMAIN as REPAIRS_DOMAIN
@ -59,8 +59,10 @@ def test_detect_radio_hardware(hass: HomeAssistant) -> None:
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"description": "SkyConnect v1.0",
"product": "SkyConnect v1.0",
"firmware": "ezsp",
},
version=2,
domain=SKYCONNECT_DOMAIN,
options={},
title="Home Assistant SkyConnect",
@ -74,8 +76,10 @@ def test_detect_radio_hardware(hass: HomeAssistant) -> None:
"pid": "EA60",
"serial_number": "3c0ed67c628beb11b1cd64a0f320645d",
"manufacturer": "Nabu Casa",
"description": "Home Assistant Connect ZBT-1",
"product": "Home Assistant Connect ZBT-1",
"firmware": "ezsp",
},
version=2,
domain=SKYCONNECT_DOMAIN,
options={},
title="Home Assistant Connect ZBT-1",