jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-19 11:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Flash ZBT-1 and Yellow firmwares from Core instead of using addons (#145019)

Browse Source 
* Make `async_flash_firmware` a public helper

* [ZBT-1] Implement flashing for Zigbee and Thread within the config flow

* WIP: Begin fixing unit tests

* WIP: Unit tests, pass 2

* WIP: pass 3

* Fix hardware unit tests

* Have the individual hardware integrations depend on the firmware flasher

* Break out firmware filter into its own helper

* Mirror to Yellow

* Simplify

* Simplify

* Revert "Have the individual hardware integrations depend on the firmware flasher"

This reverts commit 096f4297dc3b0a0529ed6288c8baea92fcfbfb11.

* Move `async_flash_silabs_firmware` into `util`

* Fix existing unit tests

* Unconditionally upgrade Zigbee firmware during installation

* Fix failing error case unit tests

* Fix remaining failing unit tests

* Increase test coverage

* 100% test coverage

* Remove old translation strings

* Add new translation strings

* Do not probe OTBR firmware when completing the flow

* More translation strings

* Probe OTBR firmware info before starting the addon
This commit is contained in:
puddly 2025-06-24 16:21:02 -04:00 committed by GitHub
parent f735331699
commit 9e7c7ec97e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 1084 additions and 1179 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

440
homeassistant/components/homeassistant_hardware/firmware_config_flow.py
View File

@ -7,6 +7,8 @@ import asyncio
import logging
from typing import Any
from ha_silabs_firmware_client import FirmwareUpdateClient
from homeassistant.components.hassio import (
AddonError,
AddonInfo,
@ -22,17 +24,17 @@ from homeassistant.config_entries import (
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.hassio import is_hassio
from . import silabs_multiprotocol_addon
from .const import OTBR_DOMAIN, ZHA_DOMAIN
from .util import (
ApplicationType,
FirmwareInfo,
OwningAddon,
OwningIntegration,
async_flash_silabs_firmware,
get_otbr_addon_manager,
get_zigbee_flasher_addon_manager,
guess_firmware_info,
guess_hardware_owners,
probe_silabs_firmware_info,
@ -61,6 +63,7 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
self.addon_install_task: asyncio.Task | None = None
self.addon_start_task: asyncio.Task | None = None
self.addon_uninstall_task: asyncio.Task | None = None
self.firmware_install_task: asyncio.Task | None = None
def _get_translation_placeholders(self) -> dict[str, str]:
"""Shared translation placeholders."""
@ -77,22 +80,6 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
return placeholders
async def _async_set_addon_config(
self, config: dict, addon_manager: AddonManager
) -> None:
"""Set add-on config."""
try:
await addon_manager.async_set_addon_options(config)
except AddonError as err:
_LOGGER.error(err)
raise AbortFlow(
"addon_set_config_failed",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": addon_manager.addon_name,
},
) from err
async def _async_get_addon_info(self, addon_manager: AddonManager) -> AddonInfo:
"""Return add-on info."""
try:
@ -150,6 +137,54 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
)
)
async def _install_firmware_step(
self,
fw_update_url: str,
fw_type: str,
firmware_name: str,
expected_installed_firmware_type: ApplicationType,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult:
assert self._device is not None
if not self.firmware_install_task:
session = async_get_clientsession(self.hass)
client = FirmwareUpdateClient(fw_update_url, session)
manifest = await client.async_update_data()
fw_meta = next(
fw for fw in manifest.firmwares if fw.filename.startswith(fw_type)
)
fw_data = await client.async_fetch_firmware(fw_meta)
self.firmware_install_task = self.hass.async_create_task(
async_flash_silabs_firmware(
hass=self.hass,
device=self._device,
fw_data=fw_data,
expected_installed_firmware_type=expected_installed_firmware_type,
bootloader_reset_type=None,
progress_callback=lambda offset, total: self.async_update_progress(
offset / total
),
),
f"Flash {firmware_name} firmware",
)
if not self.firmware_install_task.done():
return self.async_show_progress(
step_id=step_id,
progress_action="install_firmware",
description_placeholders={
**self._get_translation_placeholders(),
"firmware_name": firmware_name,
},
progress_task=self.firmware_install_task,
)
return self.async_show_progress_done(next_step_id=next_step_id)
async def async_step_pick_firmware_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -160,68 +195,133 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
description_placeholders=self._get_translation_placeholders(),
)
# Allow the stick to be used with ZHA without flashing
if (
self._probed_firmware_info is not None
and self._probed_firmware_info.firmware_type == ApplicationType.EZSP
):
return await self.async_step_confirm_zigbee()
return await self.async_step_install_zigbee_firmware()
if not is_hassio(self.hass):
async def async_step_install_zigbee_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Zigbee firmware."""
raise NotImplementedError
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="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",
reason=self._failed_addon_reason,
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
"addon_name": self._failed_addon_name,
},
)
async def async_step_install_zigbee_flasher_addon(
async def async_step_confirm_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the Zigbee flasher addon."""
return await self._install_addon(
get_zigbee_flasher_addon_manager(self.hass),
"install_zigbee_flasher_addon",
"run_zigbee_flasher_addon",
"""Confirm Zigbee setup."""
assert self._device is not None
assert self._hardware_name is not None
if user_input is None:
return self.async_show_form(
step_id="confirm_zigbee",
description_placeholders=self._get_translation_placeholders(),
)
async def _install_addon(
self,
addon_manager: silabs_multiprotocol_addon.WaitingAddonManager,
step_id: str,
next_step_id: str,
if not await self._probe_firmware_info(probe_methods=(ApplicationType.EZSP,)):
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
await self.hass.config_entries.flow.async_init(
ZHA_DOMAIN,
context={"source": "hardware"},
data={
"name": self._hardware_name,
"port": {
"path": self._device,
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
},
)
return self._async_flow_finished()
async def _ensure_thread_addon_setup(self) -> ConfigFlowResult | None:
"""Ensure the OTBR addon is set up and not running."""
# We install the OTBR addon no matter what, since it is required to use Thread
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio_thread",
description_placeholders=self._get_translation_placeholders(),
)
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_otbr_addon()
if addon_info.state == AddonState.RUNNING:
# We only fail setup if we have an instance of OTBR running *and* it's
# pointing to different hardware
if addon_info.options["device"] != self._device:
return self.async_abort(
reason="otbr_addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
)
# Otherwise, stop the addon before continuing to flash firmware
await otbr_manager.async_stop_addon()
return None
async def async_step_pick_firmware_thread(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing an addon."""
"""Pick Thread firmware."""
if not await self._probe_firmware_info():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
if result := await self._ensure_thread_addon_setup():
return result
return await self.async_step_install_thread_firmware()
async def async_step_install_thread_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Thread firmware."""
raise NotImplementedError
async def async_step_install_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the OTBR addon."""
addon_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(addon_manager)
_LOGGER.debug("Flasher addon state: %s", addon_info)
_LOGGER.debug("OTBR addon info: %s", addon_info)
if not self.addon_install_task:
self.addon_install_task = self.hass.async_create_task(
addon_manager.async_install_addon_waiting(),
"Addon install",
"OTBR addon install",
)
if not self.addon_install_task.done():
return self.async_show_progress(
step_id=step_id,
step_id="install_otbr_addon",
progress_action="install_addon",
description_placeholders={
**self._get_translation_placeholders(),
@ -240,193 +340,25 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
finally:
self.addon_install_task = None
return self.async_show_progress_done(next_step_id=next_step_id)
async def async_step_addon_operation_failed(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Abort when add-on installation or start failed."""
return self.async_abort(
reason=self._failed_addon_reason,
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": self._failed_addon_name,
},
)
async def async_step_run_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure the flasher addon to point to the SkyConnect and run it."""
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(fw_flasher_manager)
assert self._device is not None
new_addon_config = {
**addon_info.options,
"device": self._device,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
_LOGGER.debug("Reconfiguring flasher addon with %s", new_addon_config)
await self._async_set_addon_config(new_addon_config, fw_flasher_manager)
if not self.addon_start_task:
async def start_and_wait_until_done() -> None:
await fw_flasher_manager.async_start_addon_waiting()
# Now that the addon is running, wait for it to finish
await fw_flasher_manager.async_wait_until_addon_state(
AddonState.NOT_RUNNING
)
self.addon_start_task = self.hass.async_create_task(
start_and_wait_until_done()
)
if not self.addon_start_task.done():
return self.async_show_progress(
step_id="run_zigbee_flasher_addon",
progress_action="run_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_start_task,
)
try:
await self.addon_start_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
self._failed_addon_name = fw_flasher_manager.addon_name
self._failed_addon_reason = "addon_start_failed"
return self.async_show_progress_done(next_step_id="addon_operation_failed")
finally:
self.addon_start_task = None
return self.async_show_progress_done(
next_step_id="uninstall_zigbee_flasher_addon"
)
async def async_step_uninstall_zigbee_flasher_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Uninstall the flasher addon."""
fw_flasher_manager = get_zigbee_flasher_addon_manager(self.hass)
if not self.addon_uninstall_task:
_LOGGER.debug("Uninstalling flasher addon")
self.addon_uninstall_task = self.hass.async_create_task(
fw_flasher_manager.async_uninstall_addon_waiting()
)
if not self.addon_uninstall_task.done():
return self.async_show_progress(
step_id="uninstall_zigbee_flasher_addon",
progress_action="uninstall_zigbee_flasher_addon",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": fw_flasher_manager.addon_name,
},
progress_task=self.addon_uninstall_task,
)
try:
await self.addon_uninstall_task
except (AddonError, AbortFlow) as err:
_LOGGER.error(err)
# The uninstall failing isn't critical so we can just continue
finally:
self.addon_uninstall_task = None
return self.async_show_progress_done(next_step_id="confirm_zigbee")
async def async_step_confirm_zigbee(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Confirm Zigbee setup."""
assert self._device is not None
assert self._hardware_name is not None
if not await self._probe_firmware_info(probe_methods=(ApplicationType.EZSP,)):
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
if user_input is not None:
await self.hass.config_entries.flow.async_init(
ZHA_DOMAIN,
context={"source": "hardware"},
data={
"name": self._hardware_name,
"port": {
"path": self._device,
"baudrate": 115200,
"flow_control": "hardware",
},
"radio_type": "ezsp",
},
)
return self._async_flow_finished()
return self.async_show_form(
step_id="confirm_zigbee",
description_placeholders=self._get_translation_placeholders(),
)
async def async_step_pick_firmware_thread(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Pick Thread firmware."""
if not await self._probe_firmware_info():
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
# We install the OTBR addon no matter what, since it is required to use Thread
if not is_hassio(self.hass):
return self.async_abort(
reason="not_hassio_thread",
description_placeholders=self._get_translation_placeholders(),
)
otbr_manager = get_otbr_addon_manager(self.hass)
addon_info = await self._async_get_addon_info(otbr_manager)
if addon_info.state == AddonState.NOT_INSTALLED:
return await self.async_step_install_otbr_addon()
if addon_info.state == AddonState.NOT_RUNNING:
return await self.async_step_start_otbr_addon()
# If the addon is already installed and running, fail
return self.async_abort(
reason="otbr_addon_already_running",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
)
async def async_step_install_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Show progress dialog for installing the OTBR addon."""
return await self._install_addon(
get_otbr_addon_manager(self.hass), "install_otbr_addon", "start_otbr_addon"
)
return self.async_show_progress_done(next_step_id="install_thread_firmware")
async def async_step_start_otbr_addon(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Configure OTBR to point to the SkyConnect and run the addon."""
otbr_manager = get_otbr_addon_manager(self.hass)
if not self.addon_start_task:
# Before we start the addon, confirm that the correct firmware is running
# and populate `self._probed_firmware_info` with the correct information
if not await self._probe_firmware_info(
probe_methods=(ApplicationType.SPINEL,)
):
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
addon_info = await self._async_get_addon_info(otbr_manager)
assert self._device is not None
@ -435,13 +367,23 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
"device": self._device,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False,
}
_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:
try:
await otbr_manager.async_set_addon_options(new_addon_config)
except AddonError as err:
_LOGGER.error(err)
raise AbortFlow(
"addon_set_config_failed",
description_placeholders={
**self._get_translation_placeholders(),
"addon_name": otbr_manager.addon_name,
},
) from err
self.addon_start_task = self.hass.async_create_task(
otbr_manager.async_start_addon_waiting()
)
@ -475,21 +417,15 @@ class BaseFirmwareInstallFlow(ConfigEntryBaseFlow, ABC):
"""Confirm OTBR setup."""
assert self._device is not None
if not await self._probe_firmware_info(probe_methods=(ApplicationType.SPINEL,)):
return self.async_abort(
reason="unsupported_firmware",
description_placeholders=self._get_translation_placeholders(),
)
if user_input is not None:
# OTBR discovery is done automatically via hassio
return self._async_flow_finished()
if user_input is None:
return self.async_show_form(
step_id="confirm_otbr",
description_placeholders=self._get_translation_placeholders(),
)
# OTBR discovery is done automatically via hassio
return self._async_flow_finished()
@abstractmethod
def _async_flow_finished(self) -> ConfigFlowResult:
"""Finish the flow."""

30
homeassistant/components/homeassistant_hardware/strings.json
View File

@ -10,22 +10,6 @@
"pick_firmware_thread": "Thread"
}
},
"install_zigbee_flasher_addon": {
"title": "Installing flasher",
"description": "Installing the Silicon Labs Flasher add-on."
},
"run_zigbee_flasher_addon": {
"title": "Installing Zigbee firmware",
"description": "Installing Zigbee firmware. This will take about a minute."
},
"uninstall_zigbee_flasher_addon": {
"title": "Removing flasher",
"description": "Removing the Silicon Labs Flasher add-on."
},
"zigbee_flasher_failed": {
"title": "Zigbee installation failed",
"description": "The Zigbee firmware installation process was unsuccessful. Ensure no other software is trying to communicate with the {model} and try again."
},
"confirm_zigbee": {
"title": "Zigbee setup complete",
"description": "Your {model} is now a Zigbee coordinator and will be shown as discovered by the Zigbee Home Automation integration."
@ -55,9 +39,7 @@
"unsupported_firmware": "The radio firmware on your {model} could not be determined. Make sure that no other integration or add-on is currently trying to communicate with the device. If you are running Home Assistant OS in a virtual machine or in Docker, please make sure that permissions are set correctly for the device."
},
"progress": {
"install_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is installed, this may take a few minutes.",
"run_zigbee_flasher_addon": "Please wait while Zigbee firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes.",
"uninstall_zigbee_flasher_addon": "The Silicon Labs Flasher add-on is being removed."
"install_firmware": "Please wait while {firmware_name} firmware is installed to your {model}, this will take a few minutes. Do not make any changes to your hardware or software until this finishes."
}
}
},
@ -110,16 +92,6 @@
"data": {
"disable_multi_pan": "Disable multiprotocol support"
}
},
"install_flasher_addon": {
"title": "The Silicon Labs Flasher add-on installation has started"
},
"configure_flasher_addon": {
"title": "The Silicon Labs Flasher add-on installation has started"
},
"start_flasher_addon": {
"title": "Installing firmware",
"description": "Zigbee firmware is now being installed. This will take a few minutes."
}
},
"error": {

80
homeassistant/components/homeassistant_hardware/update.py
View File

@ -2,15 +2,12 @@
from __future__ import annotations
from collections.abc import AsyncIterator, Callable
from contextlib import AsyncExitStack, asynccontextmanager
from collections.abc import Callable
from dataclasses import dataclass
import logging
from typing import Any, cast
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
from universal_silabs_flasher.firmware import parse_firmware_image
from universal_silabs_flasher.flasher import Flasher
from yarl import URL
from homeassistant.components.update import (
@ -20,18 +17,12 @@ from homeassistant.components.update import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.restore_state import ExtraStoredData
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import FirmwareUpdateCoordinator
from .helpers import async_register_firmware_info_callback
from .util import (
ApplicationType,
FirmwareInfo,
guess_firmware_info,
probe_silabs_firmware_info,
)
from .util import ApplicationType, FirmwareInfo, async_flash_silabs_firmware
_LOGGER = logging.getLogger(__name__)
@ -249,19 +240,11 @@ class BaseFirmwareUpdateEntity(
self._attr_update_percentage = round((offset * 100) / total_size)
self.async_write_ha_state()
@asynccontextmanager
async def _temporarily_stop_hardware_owners(
self, device: str
) -> AsyncIterator[None]:
"""Temporarily stop addons and integrations communicating with the device."""
firmware_info = await guess_firmware_info(self.hass, device)
_LOGGER.debug("Identified firmware info: %s", firmware_info)
async with AsyncExitStack() as stack:
for owner in firmware_info.owners:
await stack.enter_async_context(owner.temporarily_stop(self.hass))
yield
# Switch to an indeterminate progress bar after installation is complete, since
# we probe the firmware after flashing
if offset == total_size:
self._attr_update_percentage = None
self.async_write_ha_state()
async def async_install(
self, version: str | None, backup: bool, **kwargs: Any
@ -278,49 +261,18 @@ class BaseFirmwareUpdateEntity(
fw_data = await self.coordinator.client.async_fetch_firmware(
self._latest_firmware
)
fw_image = await self.hass.async_add_executor_job(parse_firmware_image, fw_data)
device = self._current_device
flasher = Flasher(
device=device,
probe_methods=(
ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(),
ApplicationType.EZSP.as_flasher_application_type(),
ApplicationType.SPINEL.as_flasher_application_type(),
ApplicationType.CPC.as_flasher_application_type(),
),
bootloader_reset=self.bootloader_reset_type,
)
async with self._temporarily_stop_hardware_owners(device):
try:
try:
# Enter the bootloader with indeterminate progress
await flasher.enter_bootloader()
# Flash the firmware, with progress
await flasher.flash_firmware(
fw_image, progress_callback=self._update_progress
firmware_info = await async_flash_silabs_firmware(
hass=self.hass,
device=self._current_device,
fw_data=fw_data,
expected_installed_firmware_type=self.entity_description.expected_firmware_type,
bootloader_reset_type=self.bootloader_reset_type,
progress_callback=self._update_progress,
)
except Exception as err:
raise HomeAssistantError("Failed to flash firmware") from err
# Probe the running application type with indeterminate progress
self._attr_update_percentage = None
self.async_write_ha_state()
firmware_info = await probe_silabs_firmware_info(
device,
probe_methods=(self.entity_description.expected_firmware_type,),
)
if firmware_info is None:
raise HomeAssistantError(
"Failed to probe the firmware after flashing"
)
self._firmware_info_callback(firmware_info)
finally:
self._attr_in_progress = False
self.async_write_ha_state()
self._firmware_info_callback(firmware_info)

55
homeassistant/components/homeassistant_hardware/util.py
View File

@ -4,18 +4,20 @@ from __future__ import annotations
import asyncio
from collections import defaultdict
from collections.abc import AsyncIterator, Iterable
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator, Callable, Iterable
from contextlib import AsyncExitStack, asynccontextmanager
from dataclasses import dataclass
from enum import StrEnum
import logging
from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
from universal_silabs_flasher.firmware import parse_firmware_image
from universal_silabs_flasher.flasher import Flasher
from homeassistant.components.hassio import AddonError, AddonManager, AddonState
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.hassio import is_hassio
from homeassistant.helpers.singleton import singleton
@ -333,3 +335,52 @@ async def probe_silabs_firmware_type(
return None
return fw_info.firmware_type
async def async_flash_silabs_firmware(
hass: HomeAssistant,
device: str,
fw_data: bytes,
expected_installed_firmware_type: ApplicationType,
bootloader_reset_type: str | None = None,
progress_callback: Callable[[int, int], None] | None = None,
) -> FirmwareInfo:
"""Flash firmware to the SiLabs device."""
firmware_info = await guess_firmware_info(hass, device)
_LOGGER.debug("Identified firmware info: %s", firmware_info)
fw_image = await hass.async_add_executor_job(parse_firmware_image, fw_data)
flasher = Flasher(
device=device,
probe_methods=(
ApplicationType.GECKO_BOOTLOADER.as_flasher_application_type(),
ApplicationType.EZSP.as_flasher_application_type(),
ApplicationType.SPINEL.as_flasher_application_type(),
ApplicationType.CPC.as_flasher_application_type(),
),
bootloader_reset=bootloader_reset_type,
)
async with AsyncExitStack() as stack:
for owner in firmware_info.owners:
await stack.enter_async_context(owner.temporarily_stop(hass))
try:
# Enter the bootloader with indeterminate progress
await flasher.enter_bootloader()
# Flash the firmware, with progress
await flasher.flash_firmware(fw_image, progress_callback=progress_callback)
except Exception as err:
raise HomeAssistantError("Failed to flash firmware") from err
probed_firmware_info = await probe_silabs_firmware_info(
device,
probe_methods=(expected_installed_firmware_type,),
)
if probed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
return probed_firmware_info

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

@ -32,6 +32,7 @@ from .const import (
FIRMWARE,
FIRMWARE_VERSION,
MANUFACTURER,
NABU_CASA_FIRMWARE_RELEASES_URL,
PID,
PRODUCT,
SERIAL_NUMBER,
@ -45,19 +46,29 @@ _LOGGER = logging.getLogger(__name__)
if TYPE_CHECKING:
class TranslationPlaceholderProtocol(Protocol):
"""Protocol describing `BaseFirmwareInstallFlow`'s translation placeholders."""
class FirmwareInstallFlowProtocol(Protocol):
"""Protocol describing `BaseFirmwareInstallFlow` for a mixin."""
def _get_translation_placeholders(self) -> dict[str, str]:
return {}
async def _install_firmware_step(
self,
fw_update_url: str,
fw_type: str,
firmware_name: str,
expected_installed_firmware_type: ApplicationType,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult: ...
else:
# Multiple inheritance with `Protocol` seems to break
TranslationPlaceholderProtocol = object
FirmwareInstallFlowProtocol = object
class SkyConnectTranslationMixin(ConfigEntryBaseFlow, TranslationPlaceholderProtocol):
"""Translation placeholder mixin for Home Assistant SkyConnect."""
class SkyConnectFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol):
"""Mixin for Home Assistant SkyConnect firmware methods."""
context: ConfigFlowContext
@ -72,9 +83,35 @@ class SkyConnectTranslationMixin(ConfigEntryBaseFlow, TranslationPlaceholderProt
return placeholders
async def async_step_install_zigbee_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Zigbee firmware."""
return await self._install_firmware_step(
fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
fw_type="skyconnect_zigbee_ncp",
firmware_name="Zigbee",
expected_installed_firmware_type=ApplicationType.EZSP,
step_id="install_zigbee_firmware",
next_step_id="confirm_zigbee",
)
async def async_step_install_thread_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Thread firmware."""
return await self._install_firmware_step(
fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
fw_type="skyconnect_openthread_rcp",
firmware_name="OpenThread",
expected_installed_firmware_type=ApplicationType.SPINEL,
step_id="install_thread_firmware",
next_step_id="start_otbr_addon",
)
class HomeAssistantSkyConnectConfigFlow(
SkyConnectTranslationMixin,
SkyConnectFirmwareMixin,
firmware_config_flow.BaseFirmwareConfigFlow,
domain=DOMAIN,
):
@ -207,7 +244,7 @@ class HomeAssistantSkyConnectMultiPanOptionsFlowHandler(
class HomeAssistantSkyConnectOptionsFlowHandler(
SkyConnectTranslationMixin, firmware_config_flow.BaseFirmwareOptionsFlow
SkyConnectFirmwareMixin, firmware_config_flow.BaseFirmwareOptionsFlow
):
"""Zigbee and Thread options flow handlers."""

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

@ -48,16 +48,6 @@
"disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]"
}
},
"install_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]"
},
"configure_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]"
},
"start_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]"
},
"pick_firmware": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::description%]",
@ -66,18 +56,6 @@
"pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]"
}
},
"install_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]"
@ -120,9 +98,7 @@
"install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]"
"install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]"
}
},
"config": {
@ -136,22 +112,6 @@
"pick_firmware_thread": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_thread%]"
}
},
"install_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]"
},
"uninstall_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::uninstall_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]"
@ -191,9 +151,7 @@
"install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]"
"install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]"
}
},
"exceptions": {

61
homeassistant/components/homeassistant_yellow/config_flow.py
View File

@ -5,7 +5,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
import asyncio
import logging
from typing import Any, final
from typing import TYPE_CHECKING, Any, Protocol, final
import aiohttp
import voluptuous as vol
@ -31,6 +31,7 @@ from homeassistant.components.homeassistant_hardware.util import (
from homeassistant.config_entries import (
SOURCE_HARDWARE,
ConfigEntry,
ConfigEntryBaseFlow,
ConfigFlowResult,
OptionsFlow,
)
@ -41,6 +42,7 @@ from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
NABU_CASA_FIRMWARE_RELEASES_URL,
RADIO_DEVICE,
ZHA_DOMAIN,
ZHA_HW_DISCOVERY_DATA,
@ -57,8 +59,59 @@ STEP_HW_SETTINGS_SCHEMA = vol.Schema(
}
)
if TYPE_CHECKING:
class HomeAssistantYellowConfigFlow(BaseFirmwareConfigFlow, domain=DOMAIN):
class FirmwareInstallFlowProtocol(Protocol):
"""Protocol describing `BaseFirmwareInstallFlow` for a mixin."""
async def _install_firmware_step(
self,
fw_update_url: str,
fw_type: str,
firmware_name: str,
expected_installed_firmware_type: ApplicationType,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult: ...
else:
# Multiple inheritance with `Protocol` seems to break
FirmwareInstallFlowProtocol = object
class YellowFirmwareMixin(ConfigEntryBaseFlow, FirmwareInstallFlowProtocol):
"""Mixin for Home Assistant Yellow firmware methods."""
async def async_step_install_zigbee_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Zigbee firmware."""
return await self._install_firmware_step(
fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
fw_type="yellow_zigbee_ncp",
firmware_name="Zigbee",
expected_installed_firmware_type=ApplicationType.EZSP,
step_id="install_zigbee_firmware",
next_step_id="confirm_zigbee",
)
async def async_step_install_thread_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Thread firmware."""
return await self._install_firmware_step(
fw_update_url=NABU_CASA_FIRMWARE_RELEASES_URL,
fw_type="yellow_openthread_rcp",
firmware_name="OpenThread",
expected_installed_firmware_type=ApplicationType.SPINEL,
step_id="install_thread_firmware",
next_step_id="start_otbr_addon",
)
class HomeAssistantYellowConfigFlow(
YellowFirmwareMixin, BaseFirmwareConfigFlow, domain=DOMAIN
):
"""Handle a config flow for Home Assistant Yellow."""
VERSION = 1
@ -275,7 +328,9 @@ class HomeAssistantYellowMultiPanOptionsFlowHandler(
class HomeAssistantYellowOptionsFlowHandler(
BaseHomeAssistantYellowOptionsFlow, BaseFirmwareOptionsFlow
YellowFirmwareMixin,
BaseHomeAssistantYellowOptionsFlow,
BaseFirmwareOptionsFlow,
):
"""Handle a firmware options flow for Home Assistant Yellow."""

26
homeassistant/components/homeassistant_yellow/strings.json
View File

@ -71,16 +71,6 @@
"disable_multi_pan": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::uninstall_addon::data::disable_multi_pan%]"
}
},
"install_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::install_flasher_addon::title%]"
},
"configure_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::configure_flasher_addon::title%]"
},
"start_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::step::start_flasher_addon::description%]"
},
"pick_firmware": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::description%]",
@ -89,18 +79,6 @@
"pick_firmware_zigbee": "[%key:component::homeassistant_hardware::firmware_picker::options::step::pick_firmware::menu_options::pick_firmware_zigbee%]"
}
},
"install_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::install_zigbee_flasher_addon::description%]"
},
"run_zigbee_flasher_addon": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::run_zigbee_flasher_addon::description%]"
},
"zigbee_flasher_failed": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::zigbee_flasher_failed::description%]"
},
"confirm_zigbee": {
"title": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::title%]",
"description": "[%key:component::homeassistant_hardware::firmware_picker::options::step::confirm_zigbee::description%]"
@ -145,9 +123,7 @@
"install_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::install_addon%]",
"start_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"start_otbr_addon": "[%key:component::homeassistant_hardware::silabs_multiprotocol_hardware::options::progress::start_addon%]",
"install_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_zigbee_flasher_addon%]",
"run_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::run_zigbee_flasher_addon%]",
"uninstall_zigbee_flasher_addon": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::uninstall_zigbee_flasher_addon%]"
"install_firmware": "[%key:component::homeassistant_hardware::firmware_picker::options::progress::install_firmware%]"
}
},
"entity": {

541
tests/components/homeassistant_hardware/test_config_flow.py
View File

@ -4,9 +4,15 @@ import asyncio
from collections.abc import Awaitable, Callable, Generator, Iterator
import contextlib
from typing import Any
from unittest.mock import AsyncMock, Mock, call, patch
from unittest.mock import AsyncMock, MagicMock, Mock, call, patch
from ha_silabs_firmware_client import (
FirmwareManifest,
FirmwareMetadata,
FirmwareUpdateClient,
)
import pytest
from yarl import URL
from homeassistant.components.hassio import AddonInfo, AddonState
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
@ -19,12 +25,13 @@ from homeassistant.components.homeassistant_hardware.util import (
ApplicationType,
FirmwareInfo,
get_otbr_addon_manager,
get_zigbee_flasher_addon_manager,
)
from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from homeassistant.util.dt import utcnow
from tests.common import (
MockConfigEntry,
@ -37,6 +44,7 @@ from tests.common import (
TEST_DOMAIN = "test_firmware_domain"
TEST_DEVICE = "/dev/SomeDevice123"
TEST_HARDWARE_NAME = "Some Hardware Name"
TEST_RELEASES_URL = URL("http://invalid/releases")
class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN):
@ -62,6 +70,32 @@ class FakeFirmwareConfigFlow(BaseFirmwareConfigFlow, domain=TEST_DOMAIN):
return await self.async_step_confirm()
async def async_step_install_zigbee_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Zigbee firmware."""
return await self._install_firmware_step(
fw_update_url=TEST_RELEASES_URL,
fw_type="fake_zigbee_ncp",
firmware_name="Zigbee",
expected_installed_firmware_type=ApplicationType.EZSP,
step_id="install_zigbee_firmware",
next_step_id="confirm_zigbee",
)
async def async_step_install_thread_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Thread firmware."""
return await self._install_firmware_step(
fw_update_url=TEST_RELEASES_URL,
fw_type="fake_openthread_rcp",
firmware_name="Thread",
expected_installed_firmware_type=ApplicationType.SPINEL,
step_id="install_thread_firmware",
next_step_id="start_otbr_addon",
)
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._device is not None
@ -99,6 +133,18 @@ class FakeFirmwareOptionsFlowHandler(BaseFirmwareOptionsFlow):
# Regenerate the translation placeholders
self._get_translation_placeholders()
async def async_step_install_zigbee_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Zigbee firmware."""
return await self.async_step_confirm_zigbee()
async def async_step_install_thread_firmware(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Install Thread firmware."""
return await self.async_step_start_otbr_addon()
def _async_flow_finished(self) -> ConfigFlowResult:
"""Create the config entry."""
assert self._probed_firmware_info is not None
@ -146,12 +192,22 @@ def delayed_side_effect() -> Callable[..., Awaitable[None]]:
return side_effect
def create_mock_owner() -> Mock:
"""Mock for OwningAddon / OwningIntegration."""
owner = Mock()
owner.is_running = AsyncMock(return_value=True)
owner.temporarily_stop = MagicMock()
owner.temporarily_stop.return_value.__aenter__.return_value = AsyncMock()
return owner
@contextlib.contextmanager
def mock_addon_info(
def mock_firmware_info(
hass: HomeAssistant,
*,
is_hassio: bool = True,
app_type: ApplicationType | None = ApplicationType.EZSP,
probe_app_type: ApplicationType | None = ApplicationType.EZSP,
otbr_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
@ -160,29 +216,9 @@ def mock_addon_info(
update_available=False,
version=None,
),
flasher_addon_info: AddonInfo = AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.NOT_INSTALLED,
update_available=False,
version=None,
),
flash_app_type: ApplicationType = ApplicationType.EZSP,
) -> Iterator[tuple[Mock, Mock]]:
"""Mock the main addon states for the config flow."""
mock_flasher_manager = Mock(spec_set=get_zigbee_flasher_addon_manager(hass))
mock_flasher_manager.addon_name = "Silicon Labs Flasher"
mock_flasher_manager.async_start_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_get_addon_info.return_value = flasher_addon_info
mock_otbr_manager = Mock(spec_set=get_otbr_addon_manager(hass))
mock_otbr_manager.addon_name = "OpenThread Border Router"
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
@ -196,17 +232,73 @@ def mock_addon_info(
)
mock_otbr_manager.async_get_addon_info.return_value = otbr_addon_info
if app_type is None:
firmware_info_result = None
mock_update_client = AsyncMock(spec_set=FirmwareUpdateClient)
mock_update_client.async_update_data.return_value = FirmwareManifest(
url=TEST_RELEASES_URL,
html_url=TEST_RELEASES_URL / "html",
created_at=utcnow(),
firmwares=[
FirmwareMetadata(
filename="fake_openthread_rcp_7.4.4.0_variant.gbl",
checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
size=123,
release_notes="Some release notes",
metadata={},
url=TEST_RELEASES_URL / "fake_openthread_rcp_7.4.4.0_variant.gbl",
),
FirmwareMetadata(
filename="fake_zigbee_ncp_7.4.4.0_variant.gbl",
checksum="sha256:1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
size=123,
release_notes="Some release notes",
metadata={},
url=TEST_RELEASES_URL / "fake_zigbee_ncp_7.4.4.0_variant.gbl",
),
],
)
if probe_app_type is None:
probed_firmware_info = None
else:
firmware_info_result = FirmwareInfo(
probed_firmware_info = FirmwareInfo(
device="/dev/ttyUSB0", # Not used
firmware_type=app_type,
firmware_type=probe_app_type,
firmware_version=None,
owners=[],
source="probe",
)
if flash_app_type is None:
flashed_firmware_info = None
else:
flashed_firmware_info = FirmwareInfo(
device=TEST_DEVICE,
firmware_type=flash_app_type,
firmware_version="7.4.4.0",
owners=[create_mock_owner()],
source="probe",
)
async def mock_flash_firmware(
hass: HomeAssistant,
device: str,
fw_data: bytes,
expected_installed_firmware_type: ApplicationType,
bootloader_reset_type: str | None = None,
progress_callback: Callable[[int, int], None] | None = None,
) -> FirmwareInfo:
await asyncio.sleep(0)
progress_callback(0, 100)
await asyncio.sleep(0)
progress_callback(50, 100)
await asyncio.sleep(0)
progress_callback(100, 100)
if flashed_firmware_info is None:
raise HomeAssistantError("Failed to probe the firmware after flashing")
return flashed_firmware_info
with (
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager",
@ -216,10 +308,6 @@ def mock_addon_info(
"homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager",
return_value=mock_otbr_manager,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.get_zigbee_flasher_addon_manager",
return_value=mock_flasher_manager,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio",
return_value=is_hassio,
@ -229,81 +317,85 @@ def mock_addon_info(
return_value=is_hassio,
),
patch(
# We probe once before installation and once after
"homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
return_value=firmware_info_result,
side_effect=(probed_firmware_info, flashed_firmware_info),
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.FirmwareUpdateClient",
return_value=mock_update_client,
),
patch(
"homeassistant.components.homeassistant_hardware.util.parse_firmware_image"
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.async_flash_silabs_firmware",
side_effect=mock_flash_firmware,
),
):
yield mock_otbr_manager, mock_flasher_manager
yield mock_otbr_manager
async def consume_progress_flow(
hass: HomeAssistant,
flow_id: str,
valid_step_ids: tuple[str],
) -> ConfigFlowResult:
"""Consume a progress flow until it is done."""
while True:
result = await hass.config_entries.flow.async_configure(flow_id)
flow_id = result["flow_id"]
if result["type"] != FlowResultType.SHOW_PROGRESS:
break
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] in valid_step_ids
await asyncio.sleep(0.1)
return result
async def test_config_flow_zigbee(hass: HomeAssistant) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
assert init_result["type"] is FlowResultType.MENU
assert init_result["step_id"] == "pick_firmware"
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
probe_app_type=ApplicationType.SPINEL,
flash_app_type=ApplicationType.EZSP,
):
# Pick the menu option: we are flashing the firmware
pick_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
await hass.async_block_till_done(wait_background_tasks=True)
assert pick_result["type"] is FlowResultType.SHOW_PROGRESS
assert pick_result["progress_action"] == "install_firmware"
assert pick_result["step_id"] == "install_zigbee_firmware"
# Progress the flow, we are now configuring the addon and running it
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now uninstalling the addon
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "uninstall_zigbee_flasher_addon"
assert result["progress_action"] == "uninstall_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# We are finally done with the addon
assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()]
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
with mock_addon_info(
confirm_result = await consume_progress_flow(
hass,
app_type=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
flow_id=pick_result["flow_id"],
valid_step_ids=("install_zigbee_firmware",),
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert confirm_result["type"] is FlowResultType.FORM
assert confirm_result["step_id"] == "confirm_zigbee"
create_result = await hass.config_entries.flow.async_configure(
confirm_result["flow_id"], user_input={}
)
assert create_result["type"] is FlowResultType.CREATE_ENTRY
config_entry = create_result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"device": TEST_DEVICE,
@ -328,52 +420,20 @@ async def test_config_flow_zigbee_skip_step_if_installed(hass: HomeAssistant) ->
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
flasher_addon_info=AddonInfo(
available=True,
hostname=None,
options={
"device": "",
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.2.3",
),
) as (mock_otbr_manager, mock_flasher_manager):
with mock_firmware_info(hass, probe_app_type=ApplicationType.SPINEL):
# Pick the menu option: we skip installation, instead we directly run it
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert result["description_placeholders"]["firmware_type"] == "spinel"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
# Uninstall the addon
await hass.async_block_till_done(wait_background_tasks=True)
# Confirm
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Done
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
probe_app_type=ApplicationType.EZSP,
):
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
@ -409,28 +469,29 @@ async def test_config_flow_auto_confirm_if_running(hass: HomeAssistant) -> None:
async def test_config_flow_thread(hass: HomeAssistant) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
assert init_result["type"] is FlowResultType.MENU
assert init_result["step_id"] == "pick_firmware"
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
flash_app_type=ApplicationType.SPINEL,
) as mock_otbr_manager:
# Pick the menu option
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
pick_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_otbr_addon"
assert result["description_placeholders"]["firmware_type"] == "ezsp"
assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME
assert pick_result["type"] is FlowResultType.SHOW_PROGRESS
assert pick_result["progress_action"] == "install_addon"
assert pick_result["step_id"] == "install_otbr_addon"
assert pick_result["description_placeholders"]["firmware_type"] == "ezsp"
assert pick_result["description_placeholders"]["model"] == TEST_HARDWARE_NAME
await hass.async_block_till_done(wait_background_tasks=True)
@ -441,19 +502,37 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None:
"device": "",
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False,
},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.2.3",
)
# Progress the flow, it is now configuring the addon and running it
result = await hass.config_entries.flow.async_configure(result["flow_id"])
# Progress the flow, it is now installing firmware
confirm_otbr_result = await consume_progress_flow(
hass,
flow_id=pick_result["flow_id"],
valid_step_ids=(
"pick_firmware_thread",
"install_otbr_addon",
"install_thread_firmware",
"start_otbr_addon",
),
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_otbr_addon"
assert result["progress_action"] == "start_otbr_addon"
# Installation will conclude with the config entry being created
create_result = await hass.config_entries.flow.async_configure(
confirm_otbr_result["flow_id"], user_input={}
)
assert create_result["type"] is FlowResultType.CREATE_ENTRY
config_entry = create_result["result"]
assert config_entry.data == {
"firmware": "spinel",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
assert mock_otbr_manager.async_set_addon_options.mock_calls == [
call(
@ -461,44 +540,22 @@ async def test_config_flow_thread(hass: HomeAssistant) -> None:
"device": TEST_DEVICE,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# The addon is now running
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "spinel",
"device": TEST_DEVICE,
"hardware": TEST_HARDWARE_NAME,
}
async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -> None:
"""Test the Thread config flow, addon is already installed."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
probe_app_type=ApplicationType.EZSP,
flash_app_type=ApplicationType.SPINEL,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
@ -507,82 +564,51 @@ async def test_config_flow_thread_addon_already_installed(hass: HomeAssistant) -
update_available=False,
version=None,
),
) as (mock_otbr_manager, mock_flasher_manager):
) as mock_otbr_manager:
# Pick the menu option
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
pick_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "start_otbr_addon"
assert result["progress_action"] == "start_otbr_addon"
# Progress
confirm_otbr_result = await consume_progress_flow(
hass,
flow_id=pick_result["flow_id"],
valid_step_ids=(
"pick_firmware_thread",
"install_thread_firmware",
"start_otbr_addon",
),
)
# We're now waiting to confirm OTBR
assert confirm_otbr_result["type"] is FlowResultType.FORM
assert confirm_otbr_result["step_id"] == "confirm_otbr"
# The addon has been installed
assert mock_otbr_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False, # And firmware flashing is disabled
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# The addon is now running
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
# Finally, create the config entry
create_result = await hass.config_entries.flow.async_configure(
confirm_otbr_result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
async def test_config_flow_zigbee_not_hassio(hass: HomeAssistant) -> None:
"""Test when the stick is used with a non-hassio setup."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
is_hassio=False,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "ezsp",
assert create_result["type"] is FlowResultType.CREATE_ENTRY
assert create_result["result"].data == {
"firmware": "spinel",
"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"
@pytest.mark.usefixtures("addon_store_info")
async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
@ -601,10 +627,11 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
assert await hass.config_entries.async_setup(config_entry.entry_id)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
flash_app_type=ApplicationType.SPINEL,
) as mock_otbr_manager:
# First step is confirmation
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] is FlowResultType.MENU
@ -630,7 +657,7 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
"device": "",
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False,
},
state=AddonState.NOT_RUNNING,
update_available=False,
@ -650,7 +677,7 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
"device": TEST_DEVICE,
"baudrate": 460800,
"flow_control": True,
"autoflash_firmware": True,
"autoflash_firmware": False,
}
)
]
@ -662,10 +689,6 @@ async def test_options_flow_zigbee_to_thread(hass: HomeAssistant) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
):
# We are now done
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={}
@ -700,57 +723,23 @@ async def test_options_flow_thread_to_zigbee(hass: HomeAssistant) -> None:
assert result["description_placeholders"]["firmware_type"] == "spinel"
assert result["description_placeholders"]["model"] == TEST_HARDWARE_NAME
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.SPINEL,
):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["progress_action"] == "install_addon"
assert result["step_id"] == "install_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now configuring the addon and running it
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "run_zigbee_flasher_addon"
assert result["progress_action"] == "run_zigbee_flasher_addon"
assert mock_flasher_manager.async_set_addon_options.mock_calls == [
call(
{
"device": TEST_DEVICE,
"baudrate": 115200,
"bootloader_baudrate": 115200,
"flow_control": True,
}
)
]
await hass.async_block_till_done(wait_background_tasks=True)
# Progress the flow, we are now uninstalling the addon
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.SHOW_PROGRESS
assert result["step_id"] == "uninstall_zigbee_flasher_addon"
assert result["progress_action"] == "uninstall_zigbee_flasher_addon"
await hass.async_block_till_done(wait_background_tasks=True)
# We are finally done with the addon
assert mock_flasher_manager.async_uninstall_addon_waiting.mock_calls == [call()]
result = await hass.config_entries.options.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
probe_app_type=ApplicationType.EZSP,
):
# We are now done
result = await hass.config_entries.options.async_configure(

490
tests/components/homeassistant_hardware/test_config_flow_failures.py
View File

@ -21,8 +21,8 @@ from .test_config_flow import (
TEST_DEVICE,
TEST_DOMAIN,
TEST_HARDWARE_NAME,
delayed_side_effect,
mock_addon_info,
consume_progress_flow,
mock_firmware_info,
mock_test_firmware_platform, # noqa: F401
)
@ -51,10 +51,10 @@ async def test_config_flow_cannot_probe_firmware(
) -> None:
"""Test failure case when firmware cannot be probed."""
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=None,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=None,
):
# Start the flow
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
@ -69,283 +69,6 @@ async def test_config_flow_cannot_probe_firmware(
assert result["reason"] == "unsupported_firmware"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_not_hassio_wrong_firmware(
hass: HomeAssistant,
) -> None:
"""Test when the stick is used with a non-hassio setup but the firmware is bad."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
is_hassio=False,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "not_hassio"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_flasher_addon_already_running(
hass: HomeAssistant,
) -> None:
"""Test failure case when flasher addon is already running."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
flasher_addon_info=AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
),
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot get addon info
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_already_running"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_flasher_addon_info_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
flasher_addon_info=AddonInfo(
available=True,
hostname=None,
options={},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
),
) as (mock_otbr_manager, mock_flasher_manager):
mock_flasher_manager.async_get_addon_info.side_effect = AddonError()
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot get addon info
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_info_failed"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_flasher_addon_install_fails(
hass: HomeAssistant,
) -> None:
"""Test failure case when flasher addon cannot be installed."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=AddonError()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
# Cannot install addon
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_install_failed"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_flasher_addon_set_config_fails(
hass: HomeAssistant,
) -> None:
"""Test failure case when flasher addon cannot be configured."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
mock_flasher_manager.async_install_addon_waiting = AsyncMock(
side_effect=delayed_side_effect()
)
mock_flasher_manager.async_set_addon_options = AsyncMock(
side_effect=AddonError()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_set_config_failed"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_flasher_run_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
mock_flasher_manager.async_start_addon_waiting = AsyncMock(
side_effect=AddonError()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_start_failed"
async def test_config_flow_zigbee_flasher_uninstall_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
hass,
app_type=ApplicationType.SPINEL,
) as (mock_otbr_manager, mock_flasher_manager):
mock_flasher_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=AddonError()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
# Uninstall failure isn't critical
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_zigbee_confirmation_fails(hass: HomeAssistant) -> None:
"""Test the config flow failing due to Zigbee firmware not being detected."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
assert result["type"] is FlowResultType.MENU
assert result["step_id"] == "pick_firmware"
with mock_addon_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
# Pick the menu option: we are now installing the addon
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_zigbee"
with mock_addon_info(
hass,
app_type=None, # Probing fails
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unsupported_firmware"
@pytest.mark.parametrize(
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
@ -356,11 +79,11 @@ async def test_config_flow_thread_not_hassio(hass: HomeAssistant) -> None:
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
is_hassio=False,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
@ -383,10 +106,10 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None:
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
) as mock_otbr_manager:
mock_otbr_manager.async_get_addon_info.side_effect = AddonError()
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
@ -405,24 +128,26 @@ async def test_config_flow_thread_addon_info_fails(hass: HomeAssistant) -> None:
"ignore_translations_for_mock_domains",
["test_firmware_domain"],
)
async def test_config_flow_thread_addon_already_running(hass: HomeAssistant) -> None:
async def test_config_flow_thread_addon_already_configured(hass: HomeAssistant) -> None:
"""Test failure case when the Thread addon is already running."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
probe_app_type=ApplicationType.EZSP,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={},
options={
"device": TEST_DEVICE + "2", # A different device
},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
),
) as (mock_otbr_manager, mock_flasher_manager):
) as mock_otbr_manager:
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=AddonError()
)
@ -450,10 +175,10 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
) as mock_otbr_manager:
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=AddonError()
)
@ -477,29 +202,51 @@ async def test_config_flow_thread_addon_install_fails(hass: HomeAssistant) -> No
)
async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon cannot be configured."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
) as mock_otbr_manager:
async def install_addon() -> None:
mock_otbr_manager.async_get_addon_info.return_value = AddonInfo(
available=True,
hostname=None,
options={"device": TEST_DEVICE},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.0.0",
)
mock_otbr_manager.async_install_addon_waiting = AsyncMock(
side_effect=install_addon
)
mock_otbr_manager.async_set_addon_options = AsyncMock(side_effect=AddonError())
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
confirm_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
pick_thread_result = await hass.config_entries.flow.async_configure(
confirm_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_set_config_failed"
pick_thread_progress_result = await consume_progress_flow(
hass,
flow_id=pick_thread_result["flow_id"],
valid_step_ids=(
"pick_firmware_thread",
"install_thread_firmware",
"start_otbr_addon",
),
)
assert pick_thread_progress_result["type"] == FlowResultType.ABORT
assert pick_thread_progress_result["reason"] == "addon_set_config_failed"
@pytest.mark.parametrize(
@ -508,63 +255,45 @@ async def test_config_flow_thread_addon_set_config_fails(hass: HomeAssistant) ->
)
async def test_config_flow_thread_flasher_run_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon fails to run."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
probe_app_type=ApplicationType.EZSP,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={"device": TEST_DEVICE},
state=AddonState.NOT_RUNNING,
update_available=False,
version="1.0.0",
),
) as mock_otbr_manager:
mock_otbr_manager.async_start_addon_waiting = AsyncMock(
side_effect=AddonError()
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
confirm_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
pick_thread_result = await hass.config_entries.flow.async_configure(
confirm_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "addon_start_failed"
async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -> None:
"""Test failure case when flasher addon uninstall fails."""
result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
pick_thread_progress_result = await consume_progress_flow(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
mock_otbr_manager.async_uninstall_addon_waiting = AsyncMock(
side_effect=AddonError()
flow_id=pick_thread_result["flow_id"],
valid_step_ids=(
"pick_firmware_thread",
"install_thread_firmware",
"start_otbr_addon",
),
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
# Uninstall failure isn't critical
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
assert pick_thread_progress_result["type"] == FlowResultType.ABORT
assert pick_thread_progress_result["reason"] == "addon_start_failed"
@pytest.mark.parametrize(
@ -573,40 +302,43 @@ async def test_config_flow_thread_flasher_uninstall_fails(hass: HomeAssistant) -
)
async def test_config_flow_thread_confirmation_fails(hass: HomeAssistant) -> None:
"""Test the config flow failing due to OpenThread firmware not being detected."""
result = await hass.config_entries.flow.async_init(
init_result = await hass.config_entries.flow.async_init(
TEST_DOMAIN, context={"source": "hardware"}
)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.EZSP,
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
probe_app_type=ApplicationType.EZSP,
flash_app_type=None,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
options={"device": TEST_DEVICE},
state=AddonState.RUNNING,
update_available=False,
version="1.0.0",
),
):
confirm_result = await hass.config_entries.flow.async_configure(
init_result["flow_id"], user_input={}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
pick_thread_result = await hass.config_entries.flow.async_configure(
confirm_result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done(wait_background_tasks=True)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "confirm_otbr"
with mock_addon_info(
pick_thread_progress_result = await consume_progress_flow(
hass,
app_type=None, # Probing fails
) as (mock_otbr_manager, mock_flasher_manager):
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
flow_id=pick_thread_result["flow_id"],
valid_step_ids=(
"pick_firmware_thread",
"install_thread_firmware",
"start_otbr_addon",
),
)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "unsupported_firmware"
assert pick_thread_progress_result["type"] is FlowResultType.ABORT
assert pick_thread_progress_result["reason"] == "unsupported_firmware"
@pytest.mark.parametrize(
@ -683,9 +415,9 @@ async def test_options_flow_thread_to_zigbee_otbr_configured(
# Confirm options flow
result = await hass.config_entries.options.async_init(config_entry.entry_id)
with mock_addon_info(
with mock_firmware_info(
hass,
app_type=ApplicationType.SPINEL,
probe_app_type=ApplicationType.SPINEL,
otbr_addon_info=AddonInfo(
available=True,
hostname=None,
@ -694,7 +426,7 @@ async def test_options_flow_thread_to_zigbee_otbr_configured(
update_available=False,
version="1.0.0",
),
) as (mock_otbr_manager, mock_flasher_manager):
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},

71
tests/components/homeassistant_hardware/test_update.py
View File

@ -3,10 +3,10 @@
from __future__ import annotations
import asyncio
from collections.abc import AsyncGenerator
from collections.abc import AsyncGenerator, Callable
import dataclasses
import logging
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import Mock, patch
import aiohttp
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
@ -355,10 +355,14 @@ async def test_update_entity_installation(
"https://example.org/release_notes"
)
mock_firmware = Mock()
mock_flasher = AsyncMock()
async def mock_flash_firmware(fw_image, progress_callback):
async def mock_flash_firmware(
hass: HomeAssistant,
device: str,
fw_data: bytes,
expected_installed_firmware_type: ApplicationType,
bootloader_reset_type: str | None = None,
progress_callback: Callable[[int, int], None] | None = None,
) -> FirmwareInfo:
await asyncio.sleep(0)
progress_callback(0, 100)
await asyncio.sleep(0)
@ -366,31 +370,20 @@ async def test_update_entity_installation(
await asyncio.sleep(0)
progress_callback(100, 100)
mock_flasher.flash_firmware = mock_flash_firmware
# When we install it, the other integration is reloaded
with (
patch(
"homeassistant.components.homeassistant_hardware.update.parse_firmware_image",
return_value=mock_firmware,
),
patch(
"homeassistant.components.homeassistant_hardware.update.Flasher",
return_value=mock_flasher,
),
patch(
"homeassistant.components.homeassistant_hardware.update.probe_silabs_firmware_info",
return_value=FirmwareInfo(
return FirmwareInfo(
device=TEST_DEVICE,
firmware_type=ApplicationType.EZSP,
firmware_version="7.4.4.0 build 0",
owners=[],
source="probe",
)
# When we install it, the other integration is reloaded
with (
patch(
"homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware",
side_effect=mock_flash_firmware,
),
),
patch.object(
owning_config_entry, "async_unload", wraps=owning_config_entry.async_unload
) as owning_config_entry_unload,
):
state_changes: list[Event[EventStateChangedData]] = async_capture_events(
hass, EVENT_STATE_CHANGED
@ -423,9 +416,6 @@ async def test_update_entity_installation(
assert state_changes[6].data["new_state"].attributes["update_percentage"] is None
assert state_changes[6].data["new_state"].attributes["in_progress"] is False
# The owning integration was unloaded and is again running
assert len(owning_config_entry_unload.mock_calls) == 1
# After the firmware update, the entity has the new version and the correct state
state_after_install = hass.states.get(TEST_UPDATE_ENTITY_ID)
assert state_after_install is not None
@ -456,19 +446,10 @@ async def test_update_entity_installation_failure(
assert state_before_install.attributes["installed_version"] == "7.3.1.0"
assert state_before_install.attributes["latest_version"] == "7.4.4.0"
mock_flasher = AsyncMock()
mock_flasher.flash_firmware.side_effect = RuntimeError(
"Something broke during flashing!"
)
with (
patch(
"homeassistant.components.homeassistant_hardware.update.parse_firmware_image",
return_value=Mock(),
),
patch(
"homeassistant.components.homeassistant_hardware.update.Flasher",
return_value=mock_flasher,
"homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware",
side_effect=HomeAssistantError("Failed to flash firmware"),
),
pytest.raises(HomeAssistantError, match="Failed to flash firmware"),
):
@ -511,16 +492,10 @@ async def test_update_entity_installation_probe_failure(
with (
patch(
"homeassistant.components.homeassistant_hardware.update.parse_firmware_image",
return_value=Mock(),
"homeassistant.components.homeassistant_hardware.update.async_flash_silabs_firmware",
side_effect=HomeAssistantError(
"Failed to probe the firmware after flashing"
),
patch(
"homeassistant.components.homeassistant_hardware.update.Flasher",
return_value=AsyncMock(),
),
patch(
"homeassistant.components.homeassistant_hardware.update.probe_silabs_firmware_info",
return_value=None,
),
pytest.raises(
HomeAssistantError, match="Failed to probe the firmware after flashing"

207
tests/components/homeassistant_hardware/test_util.py
View File

@ -1,10 +1,13 @@
"""Test hardware utilities."""
from unittest.mock import AsyncMock, MagicMock, patch
import asyncio
from collections.abc import Callable
from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch
import pytest
from universal_silabs_flasher.common import Version as FlasherVersion
from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType
from universal_silabs_flasher.firmware import GBLImage
from homeassistant.components.hassio import (
AddonError,
@ -20,6 +23,7 @@ from homeassistant.components.homeassistant_hardware.util import (
FirmwareInfo,
OwningAddon,
OwningIntegration,
async_flash_silabs_firmware,
get_otbr_addon_firmware_info,
guess_firmware_info,
probe_silabs_firmware_info,
@ -27,8 +31,11 @@ from homeassistant.components.homeassistant_hardware.util import (
)
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.setup import async_setup_component
from .test_config_flow import create_mock_owner
from tests.common import MockConfigEntry
ZHA_CONFIG_ENTRY = MockConfigEntry(
@ -526,3 +533,201 @@ async def test_probe_silabs_firmware_type(
):
result = await probe_silabs_firmware_type("/dev/ttyUSB0")
assert result == expected
async def test_async_flash_silabs_firmware(hass: HomeAssistant) -> None:
"""Test async_flash_silabs_firmware."""
owner1 = create_mock_owner()
owner2 = create_mock_owner()
progress_callback = Mock()
async def mock_flash_firmware(
fw_image: GBLImage, progress_callback: Callable[[int, int], None]
) -> None:
"""Mock flash firmware function."""
await asyncio.sleep(0)
progress_callback(0, 100)
await asyncio.sleep(0)
progress_callback(50, 100)
await asyncio.sleep(0)
progress_callback(100, 100)
await asyncio.sleep(0)
mock_flasher = Mock()
mock_flasher.enter_bootloader = AsyncMock()
mock_flasher.flash_firmware = AsyncMock(side_effect=mock_flash_firmware)
expected_firmware_info = FirmwareInfo(
device="/dev/ttyUSB0",
firmware_type=ApplicationType.SPINEL,
firmware_version=None,
source="probe",
owners=[],
)
with (
patch(
"homeassistant.components.homeassistant_hardware.util.guess_firmware_info",
return_value=FirmwareInfo(
device="/dev/ttyUSB0",
firmware_type=ApplicationType.EZSP,
firmware_version=None,
source="unknown",
owners=[owner1, owner2],
),
),
patch(
"homeassistant.components.homeassistant_hardware.util.Flasher",
return_value=mock_flasher,
),
patch(
"homeassistant.components.homeassistant_hardware.util.parse_firmware_image"
),
patch(
"homeassistant.components.homeassistant_hardware.util.probe_silabs_firmware_info",
return_value=expected_firmware_info,
),
):
after_flash_info = await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_type=None,
progress_callback=progress_callback,
)
assert progress_callback.mock_calls == [call(0, 100), call(50, 100), call(100, 100)]
assert after_flash_info == expected_firmware_info
# Both owning integrations/addons are stopped and restarted
assert owner1.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, None, None, None),
]
assert owner2.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, None, None, None),
]
async def test_async_flash_silabs_firmware_flash_failure(hass: HomeAssistant) -> None:
"""Test async_flash_silabs_firmware flash failure."""
owner1 = create_mock_owner()
owner2 = create_mock_owner()
mock_flasher = Mock()
mock_flasher.enter_bootloader = AsyncMock()
mock_flasher.flash_firmware = AsyncMock(side_effect=RuntimeError("Failure!"))
with (
patch(
"homeassistant.components.homeassistant_hardware.util.guess_firmware_info",
return_value=FirmwareInfo(
device="/dev/ttyUSB0",
firmware_type=ApplicationType.EZSP,
firmware_version=None,
source="unknown",
owners=[owner1, owner2],
),
),
patch(
"homeassistant.components.homeassistant_hardware.util.Flasher",
return_value=mock_flasher,
),
patch(
"homeassistant.components.homeassistant_hardware.util.parse_firmware_image"
),
pytest.raises(HomeAssistantError, match="Failed to flash firmware") as exc,
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_type=None,
)
# Both owning integrations/addons are stopped and restarted
assert owner1.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, HomeAssistantError, exc.value, ANY),
]
assert owner2.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, HomeAssistantError, exc.value, ANY),
]
async def test_async_flash_silabs_firmware_probe_failure(hass: HomeAssistant) -> None:
"""Test async_flash_silabs_firmware probe failure."""
owner1 = create_mock_owner()
owner2 = create_mock_owner()
mock_flasher = Mock()
mock_flasher.enter_bootloader = AsyncMock()
mock_flasher.flash_firmware = AsyncMock()
with (
patch(
"homeassistant.components.homeassistant_hardware.util.guess_firmware_info",
return_value=FirmwareInfo(
device="/dev/ttyUSB0",
firmware_type=ApplicationType.EZSP,
firmware_version=None,
source="unknown",
owners=[owner1, owner2],
),
),
patch(
"homeassistant.components.homeassistant_hardware.util.Flasher",
return_value=mock_flasher,
),
patch(
"homeassistant.components.homeassistant_hardware.util.parse_firmware_image"
),
patch(
"homeassistant.components.homeassistant_hardware.util.probe_silabs_firmware_info",
return_value=None,
),
pytest.raises(
HomeAssistantError, match="Failed to probe the firmware after flashing"
),
):
await async_flash_silabs_firmware(
hass=hass,
device="/dev/ttyUSB0",
fw_data=b"firmware contents",
expected_installed_firmware_type=ApplicationType.SPINEL,
bootloader_reset_type=None,
)
# Both owning integrations/addons are stopped and restarted
assert owner1.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, None, None, None),
]
assert owner2.temporarily_stop.mock_calls == [
call(hass),
# pylint: disable-next=unnecessary-dunder-call
call().__aenter__(ANY),
# pylint: disable-next=unnecessary-dunder-call
call().__aexit__(ANY, None, None, None),
]

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

@ -6,6 +6,7 @@ import pytest
from homeassistant.components.hassio import AddonInfo, AddonState
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
@ -18,6 +19,7 @@ from homeassistant.components.homeassistant_hardware.util import (
FirmwareInfo,
)
from homeassistant.components.homeassistant_sky_connect.const import DOMAIN
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.service_info.usb import UsbServiceInfo
@ -28,14 +30,31 @@ from tests.common import MockConfigEntry
@pytest.mark.parametrize(
("usb_data", "model"),
("step", "usb_data", "model", "fw_type", "fw_version"),
[
(USB_DATA_SKY, "Home Assistant SkyConnect"),
(USB_DATA_ZBT1, "Home Assistant Connect ZBT-1"),
(
STEP_PICK_FIRMWARE_ZIGBEE,
USB_DATA_SKY,
"Home Assistant SkyConnect",
ApplicationType.EZSP,
"7.4.4.0 build 0",
),
(
STEP_PICK_FIRMWARE_THREAD,
USB_DATA_ZBT1,
"Home Assistant Connect ZBT-1",
ApplicationType.SPINEL,
"2.4.4.0",
),
],
)
async def test_config_flow(
usb_data: UsbServiceInfo, model: str, hass: HomeAssistant
step: str,
usb_data: UsbServiceInfo,
model: str,
fw_type: ApplicationType,
fw_version: str,
hass: HomeAssistant,
) -> None:
"""Test the config flow for SkyConnect."""
result = await hass.config_entries.flow.async_init(
@ -46,21 +65,36 @@ async def test_config_flow(
assert result["step_id"] == "pick_firmware"
assert result["description_placeholders"]["model"] == model
async def mock_async_step_pick_firmware_zigbee(self, data):
return await self.async_step_confirm_zigbee(user_input={})
async def mock_install_firmware_step(
self,
fw_update_url: str,
fw_type: str,
firmware_name: str,
expected_installed_firmware_type: ApplicationType,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult:
if next_step_id == "start_otbr_addon":
next_step_id = "confirm_otbr"
return await getattr(self, f"async_step_{next_step_id}")(user_input={})
with (
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow.async_step_pick_firmware_zigbee",
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup",
return_value=None,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._install_firmware_step",
autospec=True,
side_effect=mock_async_step_pick_firmware_zigbee,
side_effect=mock_install_firmware_step,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
return_value=FirmwareInfo(
device=usb_data.device,
firmware_type=ApplicationType.EZSP,
firmware_version="7.4.4.0 build 0",
firmware_type=fw_type,
firmware_version=fw_version,
owners=[],
source="probe",
),
@ -68,15 +102,15 @@ async def test_config_flow(
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
user_input={"next_step_id": step},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
config_entry = result["result"]
assert config_entry.data == {
"firmware": "ezsp",
"firmware_version": "7.4.4.0 build 0",
"firmware": fw_type.value,
"firmware_version": fw_version,
"device": usb_data.device,
"manufacturer": usb_data.manufacturer,
"pid": usb_data.pid,
@ -86,13 +120,17 @@ async def test_config_flow(
"vid": usb_data.vid,
}
# Ensure a ZHA discovery flow has been created
flows = hass.config_entries.flow.async_progress()
if step == STEP_PICK_FIRMWARE_ZIGBEE:
# Ensure a ZHA discovery flow has been created
assert len(flows) == 1
zha_flow = flows[0]
assert zha_flow["handler"] == "zha"
assert zha_flow["context"]["source"] == "hardware"
assert zha_flow["step_id"] == "confirm"
else:
assert len(flows) == 0
@pytest.mark.parametrize(

45
tests/components/homeassistant_yellow/test_config_flow.py
View File

@ -11,6 +11,7 @@ from homeassistant.components.hassio import (
AddonState,
)
from homeassistant.components.homeassistant_hardware.firmware_config_flow import (
STEP_PICK_FIRMWARE_THREAD,
STEP_PICK_FIRMWARE_ZIGBEE,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
@ -23,6 +24,7 @@ from homeassistant.components.homeassistant_hardware.util import (
FirmwareInfo,
)
from homeassistant.components.homeassistant_yellow.const import DOMAIN, RADIO_DEVICE
from homeassistant.config_entries import ConfigFlowResult
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.setup import async_setup_component
@ -305,7 +307,16 @@ async def test_option_flow_led_settings_fail_2(
assert result["reason"] == "write_hw_settings_error"
async def test_firmware_options_flow(hass: HomeAssistant) -> None:
@pytest.mark.parametrize(
("step", "fw_type", "fw_version"),
[
(STEP_PICK_FIRMWARE_ZIGBEE, ApplicationType.EZSP, "7.4.4.0 build 0"),
(STEP_PICK_FIRMWARE_THREAD, ApplicationType.SPINEL, "2.4.4.0"),
],
)
async def test_firmware_options_flow(
step: str, fw_type: ApplicationType, fw_version: str, hass: HomeAssistant
) -> None:
"""Test the firmware options flow for Yellow."""
mock_integration(hass, MockModule("hassio"))
await async_setup_component(hass, HASSIO_DOMAIN, {})
@ -339,18 +350,36 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None:
async def mock_async_step_pick_firmware_zigbee(self, data):
return await self.async_step_confirm_zigbee(user_input={})
async def mock_install_firmware_step(
self,
fw_update_url: str,
fw_type: str,
firmware_name: str,
expected_installed_firmware_type: ApplicationType,
step_id: str,
next_step_id: str,
) -> ConfigFlowResult:
if next_step_id == "start_otbr_addon":
next_step_id = "confirm_otbr"
return await getattr(self, f"async_step_{next_step_id}")(user_input={})
with (
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareOptionsFlow.async_step_pick_firmware_zigbee",
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareConfigFlow._ensure_thread_addon_setup",
return_value=None,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.BaseFirmwareInstallFlow._install_firmware_step",
autospec=True,
side_effect=mock_async_step_pick_firmware_zigbee,
side_effect=mock_install_firmware_step,
),
patch(
"homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_info",
return_value=FirmwareInfo(
device=RADIO_DEVICE,
firmware_type=ApplicationType.EZSP,
firmware_version="7.4.4.0 build 0",
firmware_type=fw_type,
firmware_version=fw_version,
owners=[],
source="probe",
),
@ -358,15 +387,15 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None:
):
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={"next_step_id": STEP_PICK_FIRMWARE_ZIGBEE},
user_input={"next_step_id": step},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"] is True
assert config_entry.data == {
"firmware": "ezsp",
"firmware_version": "7.4.4.0 build 0",
"firmware": fw_type.value,
"firmware_version": fw_version,
}