From 537302ce56e0b27429bf69e29d85e2855e23e271 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Fri, 14 Mar 2025 19:28:02 -0400 Subject: [PATCH] ZBT-1 and Yellow firmware update entities for Zigbee/Thread (#138505) * Initial implementation of hardware update model * Fixes * WIP: change the `homeassistant_sky_connect` integration type * More fixes * WIP * Display firmware info in the device page * Make progress more responsive * WIP: Yellow * Abstract the bootloader reset type * Clean up comments * Make the Yellow integration non-hardware * Use the correct radio device for Yellow * Avoid hardcoding strings * Use `FIRMWARE_VERSION` within config flows * Fix up unit tests * Revert integration type changes * Rewrite hardware ownership context manager name, for clarity * Move manifest parsing logic into a new package Pass the correct type to the firmware API library * Create and delete entities instead of mutating the entity description * Move entity replacement into a `async_setup_entry` callback * Change update entity category from "diagnostic" to "config" * Have the client library handle firmware fetching * Switch from dispatcher to `async_on_state_change` * Remove unnecessary type annotation on base update entity * Simplify state recomputation * Remove device registry code, since the devices will not be visible * Further simplify state computation * Give the device-less update entity a more descriptive name * Limit state changes to integer increments when sending firmware update progress * Re-raise `HomeAssistantError` if there is a problem during flashing * Remove unnecessary state write during entity creation * Rename `_maybe_recompute_state` to `_update_attributes` * Bump the flasher to 0.0.30 * Add some tests * Ensure the update entity has a sensible name * Initial ZBT-1 unit tests * Replace `_update_config_entry_after_install` with a more explicit `_firmware_info_callback` override * Write the firmware version to the config entry as well * Test the hardware update platform independently * Add unit tests to the Yellow and ZBT-1 integrations * Load firmware info from the config entry when creating the update entity * Test entity state restoration * Test the reloading of integrations marked as "owning" * Test installation failure cases * Test firmware type change callback failure case * Address review comments --- .../homeassistant_hardware/coordinator.py | 47 ++ .../homeassistant_hardware/manifest.json | 5 +- .../homeassistant_hardware/update.py | 331 +++++++++ .../components/homeassistant_hardware/util.py | 42 +- .../homeassistant_sky_connect/__init__.py | 25 +- .../homeassistant_sky_connect/config_flow.py | 53 +- .../homeassistant_sky_connect/const.py | 14 + .../homeassistant_sky_connect/update.py | 169 +++++ .../homeassistant_yellow/__init__.py | 16 +- .../homeassistant_yellow/config_flow.py | 12 +- .../components/homeassistant_yellow/const.py | 8 + .../components/homeassistant_yellow/update.py | 172 +++++ requirements_all.txt | 5 +- .../test_coordinator.py | 55 ++ .../homeassistant_hardware/test_update.py | 637 ++++++++++++++++++ .../homeassistant_hardware/test_util.py | 148 ++++ .../homeassistant_sky_connect/common.py | 21 + .../test_config_flow.py | 26 +- .../homeassistant_sky_connect/test_init.py | 3 +- .../homeassistant_sky_connect/test_update.py | 86 +++ .../homeassistant_yellow/test_config_flow.py | 3 +- .../homeassistant_yellow/test_update.py | 89 +++ 22 files changed, 1916 insertions(+), 51 deletions(-) create mode 100644 homeassistant/components/homeassistant_hardware/coordinator.py create mode 100644 homeassistant/components/homeassistant_hardware/update.py create mode 100644 homeassistant/components/homeassistant_sky_connect/update.py create mode 100644 homeassistant/components/homeassistant_yellow/update.py create mode 100644 tests/components/homeassistant_hardware/test_coordinator.py create mode 100644 tests/components/homeassistant_hardware/test_update.py create mode 100644 tests/components/homeassistant_sky_connect/common.py create mode 100644 tests/components/homeassistant_sky_connect/test_update.py create mode 100644 tests/components/homeassistant_yellow/test_update.py diff --git a/homeassistant/components/homeassistant_hardware/coordinator.py b/homeassistant/components/homeassistant_hardware/coordinator.py new file mode 100644 index 00000000000..9eb900b13fd --- /dev/null +++ b/homeassistant/components/homeassistant_hardware/coordinator.py @@ -0,0 +1,47 @@ +"""Home Assistant hardware firmware update coordinator.""" + +from __future__ import annotations + +from datetime import timedelta +import logging + +from aiohttp import ClientSession +from ha_silabs_firmware_client import ( + FirmwareManifest, + FirmwareUpdateClient, + ManifestMissing, +) + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +_LOGGER = logging.getLogger(__name__) + + +FIRMWARE_REFRESH_INTERVAL = timedelta(hours=8) + + +class FirmwareUpdateCoordinator(DataUpdateCoordinator[FirmwareManifest]): + """Coordinator to manage firmware updates.""" + + def __init__(self, hass: HomeAssistant, session: ClientSession, url: str) -> None: + """Initialize the firmware update coordinator.""" + super().__init__( + hass, + _LOGGER, + name="firmware update coordinator", + update_interval=FIRMWARE_REFRESH_INTERVAL, + always_update=False, + ) + self.hass = hass + self.session = session + + self.client = FirmwareUpdateClient(url, session) + + async def _async_update_data(self) -> FirmwareManifest: + try: + return await self.client.async_update_data() + except ManifestMissing as err: + raise UpdateFailed( + "GitHub release assets haven't been uploaded yet" + ) from err diff --git a/homeassistant/components/homeassistant_hardware/manifest.json b/homeassistant/components/homeassistant_hardware/manifest.json index 8f59ab61600..f3a02185b83 100644 --- a/homeassistant/components/homeassistant_hardware/manifest.json +++ b/homeassistant/components/homeassistant_hardware/manifest.json @@ -5,5 +5,8 @@ "codeowners": ["@home-assistant/core"], "documentation": "https://www.home-assistant.io/integrations/homeassistant_hardware", "integration_type": "system", - "requirements": ["universal-silabs-flasher==0.0.29"] + "requirements": [ + "universal-silabs-flasher==0.0.30", + "ha-silabs-firmware-client==0.2.0" + ] } diff --git a/homeassistant/components/homeassistant_hardware/update.py b/homeassistant/components/homeassistant_hardware/update.py new file mode 100644 index 00000000000..e835286238f --- /dev/null +++ b/homeassistant/components/homeassistant_hardware/update.py @@ -0,0 +1,331 @@ +"""Home Assistant Hardware base firmware update entity.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Callable +from contextlib import AsyncExitStack, asynccontextmanager +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 ( + UpdateEntity, + UpdateEntityDescription, + UpdateEntityFeature, +) +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, +) + +_LOGGER = logging.getLogger(__name__) + +type FirmwareChangeCallbackType = Callable[ + [ApplicationType | None, ApplicationType | None], None +] + + +@dataclass(kw_only=True, frozen=True) +class FirmwareUpdateEntityDescription(UpdateEntityDescription): + """Describes Home Assistant Hardware firmware update entity.""" + + version_parser: Callable[[str], str] + fw_type: str | None + version_key: str | None + expected_firmware_type: ApplicationType | None + firmware_name: str | None + + +@dataclass +class FirmwareUpdateExtraStoredData(ExtraStoredData): + """Extra stored data for Home Assistant Hardware firmware update entity.""" + + firmware_manifest: FirmwareManifest | None = None + + def as_dict(self) -> dict[str, Any]: + """Return a dict representation of the extra data.""" + return { + "firmware_manifest": ( + self.firmware_manifest.as_dict() + if self.firmware_manifest is not None + else None + ) + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> FirmwareUpdateExtraStoredData: + """Initialize the extra data from a dict.""" + if data["firmware_manifest"] is None: + return cls(firmware_manifest=None) + + return cls( + FirmwareManifest.from_json( + data["firmware_manifest"], + # This data is not technically part of the manifest and is loaded externally + url=URL(data["firmware_manifest"]["url"]), + html_url=URL(data["firmware_manifest"]["html_url"]), + ) + ) + + +class BaseFirmwareUpdateEntity( + CoordinatorEntity[FirmwareUpdateCoordinator], UpdateEntity +): + """Base Home Assistant Hardware firmware update entity.""" + + # Subclasses provide the mapping between firmware types and entity descriptions + entity_description: FirmwareUpdateEntityDescription + bootloader_reset_type: str | None = None + + _attr_supported_features = ( + UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS + ) + # Until this entity can be associated with a device, we must manually name it + _attr_has_entity_name = False + + def __init__( + self, + device: str, + config_entry: ConfigEntry, + update_coordinator: FirmwareUpdateCoordinator, + entity_description: FirmwareUpdateEntityDescription, + ) -> None: + """Initialize the Hardware firmware update entity.""" + super().__init__(update_coordinator) + + self.entity_description = entity_description + self._current_device = device + self._config_entry = config_entry + self._current_firmware_info: FirmwareInfo | None = None + self._firmware_type_change_callbacks: set[FirmwareChangeCallbackType] = set() + + self._latest_manifest: FirmwareManifest | None = None + self._latest_firmware: FirmwareMetadata | None = None + + def add_firmware_type_changed_callback( + self, + change_callback: FirmwareChangeCallbackType, + ) -> CALLBACK_TYPE: + """Add a callback for when the firmware type changes.""" + self._firmware_type_change_callbacks.add(change_callback) + + @callback + def remove_callback() -> None: + self._firmware_type_change_callbacks.discard(change_callback) + + return remove_callback + + async def async_added_to_hass(self) -> None: + """Handle entity which will be added.""" + await super().async_added_to_hass() + + self.async_on_remove( + async_register_firmware_info_callback( + self.hass, + self._current_device, + self._firmware_info_callback, + ) + ) + + self.async_on_remove( + self._config_entry.async_on_state_change(self._on_config_entry_change) + ) + + if (extra_data := await self.async_get_last_extra_data()) and ( + hardware_extra_data := FirmwareUpdateExtraStoredData.from_dict( + extra_data.as_dict() + ) + ): + self._latest_manifest = hardware_extra_data.firmware_manifest + + self._update_attributes() + + @property + def extra_restore_state_data(self) -> FirmwareUpdateExtraStoredData: + """Return state data to be restored.""" + return FirmwareUpdateExtraStoredData(firmware_manifest=self._latest_manifest) + + @callback + def _on_config_entry_change(self) -> None: + """Handle config entry changes.""" + self._update_attributes() + self.async_write_ha_state() + + @callback + def _firmware_info_callback(self, firmware_info: FirmwareInfo) -> None: + """Handle updated firmware info being pushed by an integration.""" + self._current_firmware_info = firmware_info + + # If the firmware type does not change, we can just update the attributes + if ( + self._current_firmware_info.firmware_type + == self.entity_description.expected_firmware_type + ): + self._update_attributes() + self.async_write_ha_state() + return + + # Otherwise, fire the firmware type change callbacks. They are expected to + # replace the entity so there is no purpose in firing other callbacks. + for change_callback in self._firmware_type_change_callbacks.copy(): + try: + change_callback( + self.entity_description.expected_firmware_type, + self._current_firmware_info.firmware_type, + ) + except Exception: # noqa: BLE001 + _LOGGER.warning( + "Failed to call firmware type changed callback", exc_info=True + ) + + def _update_attributes(self) -> None: + """Recompute the attributes of the entity.""" + + # This entity is not currently associated with a device so we must manually + # give it a name + self._attr_name = f"{self._config_entry.title} Update" + self._attr_title = self.entity_description.firmware_name or "unknown" + + if ( + self._current_firmware_info is None + or self._current_firmware_info.firmware_version is None + ): + self._attr_installed_version = None + else: + self._attr_installed_version = self.entity_description.version_parser( + self._current_firmware_info.firmware_version + ) + + self._latest_firmware = None + self._attr_latest_version = None + self._attr_release_summary = None + self._attr_release_url = None + + if ( + self._latest_manifest is None + or self.entity_description.fw_type is None + or self.entity_description.version_key is None + ): + return + + try: + self._latest_firmware = next( + f + for f in self._latest_manifest.firmwares + if f.filename.startswith(self.entity_description.fw_type) + ) + except StopIteration: + pass + else: + version = cast( + str, self._latest_firmware.metadata[self.entity_description.version_key] + ) + self._attr_latest_version = self.entity_description.version_parser(version) + self._attr_release_summary = self._latest_firmware.release_notes + self._attr_release_url = str(self._latest_manifest.html_url) + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + self._latest_manifest = self.coordinator.data + self._update_attributes() + self.async_write_ha_state() + + def _update_progress(self, offset: int, total_size: int) -> None: + """Handle update progress.""" + + # Firmware updates in ~30s so we still get responsive update progress even + # without decimal places + 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 + + async def async_install( + self, version: str | None, backup: bool, **kwargs: Any + ) -> None: + """Install an update.""" + assert self._latest_firmware is not None + assert self.entity_description.expected_firmware_type is not None + + # Start off by setting the progress bar to an indeterminate state + self._attr_in_progress = True + self._attr_update_percentage = None + self.async_write_ha_state() + + 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 + ) + 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() diff --git a/homeassistant/components/homeassistant_hardware/util.py b/homeassistant/components/homeassistant_hardware/util.py index 1afb786369e..64f363e4f23 100644 --- a/homeassistant/components/homeassistant_hardware/util.py +++ b/homeassistant/components/homeassistant_hardware/util.py @@ -4,7 +4,8 @@ from __future__ import annotations import asyncio from collections import defaultdict -from collections.abc import Iterable +from collections.abc import AsyncIterator, Iterable +from contextlib import asynccontextmanager from dataclasses import dataclass from enum import StrEnum import logging @@ -105,6 +106,28 @@ class OwningAddon: else: return addon_info.state == AddonState.RUNNING + @asynccontextmanager + async def temporarily_stop(self, hass: HomeAssistant) -> AsyncIterator[None]: + """Temporarily stop the add-on, restarting it after completion.""" + addon_manager = self._get_addon_manager(hass) + + try: + addon_info = await addon_manager.async_get_addon_info() + except AddonError: + yield + return + + if addon_info.state != AddonState.RUNNING: + yield + return + + try: + await addon_manager.async_stop_addon() + await addon_manager.async_wait_until_addon_state(AddonState.NOT_RUNNING) + yield + finally: + await addon_manager.async_start_addon_waiting() + @dataclass(kw_only=True) class OwningIntegration: @@ -123,6 +146,23 @@ class OwningIntegration: ConfigEntryState.SETUP_IN_PROGRESS, ) + @asynccontextmanager + async def temporarily_stop(self, hass: HomeAssistant) -> AsyncIterator[None]: + """Temporarily stop the integration, restarting it after completion.""" + if (entry := hass.config_entries.async_get_entry(self.config_entry_id)) is None: + yield + return + + if entry.state != ConfigEntryState.LOADED: + yield + return + + try: + await hass.config_entries.async_unload(entry.entry_id) + yield + finally: + await hass.config_entries.async_setup(entry.entry_id) + @dataclass(kw_only=True) class FirmwareInfo: diff --git a/homeassistant/components/homeassistant_sky_connect/__init__.py b/homeassistant/components/homeassistant_sky_connect/__init__.py index 758f0c1e1ef..b3af47df61d 100644 --- a/homeassistant/components/homeassistant_sky_connect/__init__.py +++ b/homeassistant/components/homeassistant_sky_connect/__init__.py @@ -8,11 +8,16 @@ from homeassistant.components.homeassistant_hardware.util import guess_firmware_ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant +from .const import DESCRIPTION, DEVICE, FIRMWARE, FIRMWARE_VERSION, PRODUCT + _LOGGER = logging.getLogger(__name__) async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up a Home Assistant SkyConnect config entry.""" + + await hass.config_entries.async_forward_entry_setups(entry, ["update"]) + return True @@ -33,15 +38,13 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> # Add-on startup with type service get started before Core, always (e.g. the # Multi-Protocol add-on). Probing the firmware would interfere with the add-on, # so we can't safely probe here. Instead, we must make an educated guess! - firmware_guess = await guess_firmware_info( - hass, config_entry.data["device"] - ) + firmware_guess = await guess_firmware_info(hass, config_entry.data[DEVICE]) new_data = {**config_entry.data} - new_data["firmware"] = firmware_guess.firmware_type.value + new_data[FIRMWARE] = firmware_guess.firmware_type.value # Copy `description` to `product` - new_data["product"] = new_data["description"] + new_data[PRODUCT] = new_data[DESCRIPTION] hass.config_entries.async_update_entry( config_entry, @@ -50,6 +53,18 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> minor_version=2, ) + if config_entry.minor_version == 2: + # Add a `firmware_version` key + hass.config_entries.async_update_entry( + config_entry, + data={ + **config_entry.data, + FIRMWARE_VERSION: None, + }, + version=1, + minor_version=3, + ) + _LOGGER.debug( "Migration to version %s.%s successful", config_entry.version, diff --git a/homeassistant/components/homeassistant_sky_connect/config_flow.py b/homeassistant/components/homeassistant_sky_connect/config_flow.py index d8446c2d3f9..d28d74a681c 100644 --- a/homeassistant/components/homeassistant_sky_connect/config_flow.py +++ b/homeassistant/components/homeassistant_sky_connect/config_flow.py @@ -24,7 +24,20 @@ from homeassistant.config_entries import ( from homeassistant.core import callback from homeassistant.helpers.service_info.usb import UsbServiceInfo -from .const import DOCS_WEB_FLASHER_URL, DOMAIN, HardwareVariant +from .const import ( + DESCRIPTION, + DEVICE, + DOCS_WEB_FLASHER_URL, + DOMAIN, + FIRMWARE, + FIRMWARE_VERSION, + MANUFACTURER, + PID, + PRODUCT, + SERIAL_NUMBER, + VID, + HardwareVariant, +) from .util import get_hardware_variant, get_usb_service_info _LOGGER = logging.getLogger(__name__) @@ -37,6 +50,7 @@ if TYPE_CHECKING: def _get_translation_placeholders(self) -> dict[str, str]: return {} + else: # Multiple inheritance with `Protocol` seems to break TranslationPlaceholderProtocol = object @@ -67,7 +81,7 @@ class HomeAssistantSkyConnectConfigFlow( """Handle a config flow for Home Assistant SkyConnect.""" VERSION = 1 - MINOR_VERSION = 2 + MINOR_VERSION = 3 def __init__(self, *args: Any, **kwargs: Any) -> None: """Initialize the config flow.""" @@ -82,7 +96,7 @@ class HomeAssistantSkyConnectConfigFlow( config_entry: ConfigEntry, ) -> OptionsFlow: """Return the options flow.""" - firmware_type = ApplicationType(config_entry.data["firmware"]) + firmware_type = ApplicationType(config_entry.data[FIRMWARE]) if firmware_type is ApplicationType.CPC: return HomeAssistantSkyConnectMultiPanOptionsFlowHandler(config_entry) @@ -100,7 +114,7 @@ class HomeAssistantSkyConnectConfigFlow( unique_id = f"{vid}:{pid}_{serial_number}_{manufacturer}_{description}" if await self.async_set_unique_id(unique_id): - self._abort_if_unique_id_configured(updates={"device": device}) + self._abort_if_unique_id_configured(updates={DEVICE: device}) discovery_info.device = await self.hass.async_add_executor_job( usb.get_serial_by_id, discovery_info.device @@ -126,14 +140,15 @@ class HomeAssistantSkyConnectConfigFlow( return self.async_create_entry( title=self._hw_variant.full_name, data={ - "vid": self._usb_info.vid, - "pid": self._usb_info.pid, - "serial_number": self._usb_info.serial_number, - "manufacturer": self._usb_info.manufacturer, - "description": self._usb_info.description, # For backwards compatibility - "product": self._usb_info.description, - "device": self._usb_info.device, - "firmware": self._probed_firmware_info.firmware_type.value, + VID: self._usb_info.vid, + PID: self._usb_info.pid, + SERIAL_NUMBER: self._usb_info.serial_number, + MANUFACTURER: self._usb_info.manufacturer, + DESCRIPTION: self._usb_info.description, # For backwards compatibility + PRODUCT: self._usb_info.description, + DEVICE: self._usb_info.device, + FIRMWARE: self._probed_firmware_info.firmware_type.value, + FIRMWARE_VERSION: self._probed_firmware_info.firmware_version, }, ) @@ -148,7 +163,7 @@ class HomeAssistantSkyConnectMultiPanOptionsFlowHandler( ) -> silabs_multiprotocol_addon.SerialPortSettings: """Return the radio serial port settings.""" return silabs_multiprotocol_addon.SerialPortSettings( - device=self.config_entry.data["device"], + device=self.config_entry.data[DEVICE], baudrate="115200", flow_control=True, ) @@ -182,7 +197,8 @@ class HomeAssistantSkyConnectMultiPanOptionsFlowHandler( entry=self.config_entry, data={ **self.config_entry.data, - "firmware": ApplicationType.EZSP.value, + FIRMWARE: ApplicationType.EZSP.value, + FIRMWARE_VERSION: None, }, options=self.config_entry.options, ) @@ -201,15 +217,15 @@ class HomeAssistantSkyConnectOptionsFlowHandler( self._usb_info = get_usb_service_info(self.config_entry) self._hw_variant = HardwareVariant.from_usb_product_name( - self.config_entry.data["product"] + self.config_entry.data[PRODUCT] ) self._hardware_name = self._hw_variant.full_name self._device = self._usb_info.device self._probed_firmware_info = FirmwareInfo( device=self._device, - firmware_type=ApplicationType(self.config_entry.data["firmware"]), - firmware_version=None, + firmware_type=ApplicationType(self.config_entry.data[FIRMWARE]), + firmware_version=self.config_entry.data[FIRMWARE_VERSION], source="guess", owners=[], ) @@ -225,7 +241,8 @@ class HomeAssistantSkyConnectOptionsFlowHandler( entry=self.config_entry, data={ **self.config_entry.data, - "firmware": self._probed_firmware_info.firmware_type.value, + FIRMWARE: self._probed_firmware_info.firmware_type.value, + FIRMWARE_VERSION: self._probed_firmware_info.firmware_version, }, options=self.config_entry.options, ) diff --git a/homeassistant/components/homeassistant_sky_connect/const.py b/homeassistant/components/homeassistant_sky_connect/const.py index cae0b98a25b..70ff047366d 100644 --- a/homeassistant/components/homeassistant_sky_connect/const.py +++ b/homeassistant/components/homeassistant_sky_connect/const.py @@ -7,6 +7,20 @@ from typing import Self DOMAIN = "homeassistant_sky_connect" DOCS_WEB_FLASHER_URL = "https://skyconnect.home-assistant.io/firmware-update/" +NABU_CASA_FIRMWARE_RELEASES_URL = ( + "https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest" +) + +FIRMWARE = "firmware" +FIRMWARE_VERSION = "firmware_version" +SERIAL_NUMBER = "serial_number" +MANUFACTURER = "manufacturer" +PRODUCT = "product" +DESCRIPTION = "description" +PID = "pid" +VID = "vid" +DEVICE = "device" + @dataclasses.dataclass(frozen=True) class VariantInfo: diff --git a/homeassistant/components/homeassistant_sky_connect/update.py b/homeassistant/components/homeassistant_sky_connect/update.py new file mode 100644 index 00000000000..43e3f1ca255 --- /dev/null +++ b/homeassistant/components/homeassistant_sky_connect/update.py @@ -0,0 +1,169 @@ +"""Home Assistant SkyConnect firmware update entity.""" + +from __future__ import annotations + +import logging + +import aiohttp + +from homeassistant.components.homeassistant_hardware.coordinator import ( + FirmwareUpdateCoordinator, +) +from homeassistant.components.homeassistant_hardware.update import ( + BaseFirmwareUpdateEntity, + FirmwareUpdateEntityDescription, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, +) +from homeassistant.components.update import UpdateDeviceClass +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import EntityCategory +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .const import FIRMWARE, FIRMWARE_VERSION, NABU_CASA_FIRMWARE_RELEASES_URL + +_LOGGER = logging.getLogger(__name__) + + +FIRMWARE_ENTITY_DESCRIPTIONS: dict[ + ApplicationType | None, FirmwareUpdateEntityDescription +] = { + ApplicationType.EZSP: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split(" ", 1)[0], + fw_type="skyconnect_zigbee_ncp", + version_key="ezsp_version", + expected_firmware_type=ApplicationType.EZSP, + firmware_name="EmberZNet", + ), + ApplicationType.SPINEL: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split("/", 1)[1].split("_", 1)[0], + fw_type="skyconnect_openthread_rcp", + version_key="ot_rcp_version", + expected_firmware_type=ApplicationType.SPINEL, + firmware_name="OpenThread RCP", + ), + None: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw, + fw_type=None, + version_key=None, + expected_firmware_type=None, + firmware_name=None, + ), +} + + +def _async_create_update_entity( + hass: HomeAssistant, + config_entry: ConfigEntry, + session: aiohttp.ClientSession, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> FirmwareUpdateEntity: + """Create an update entity that handles firmware type changes.""" + firmware_type = config_entry.data[FIRMWARE] + entity_description = FIRMWARE_ENTITY_DESCRIPTIONS[ + ApplicationType(firmware_type) if firmware_type is not None else None + ] + + entity = FirmwareUpdateEntity( + device=config_entry.data["device"], + config_entry=config_entry, + update_coordinator=FirmwareUpdateCoordinator( + hass, + session, + NABU_CASA_FIRMWARE_RELEASES_URL, + ), + entity_description=entity_description, + ) + + def firmware_type_changed( + old_type: ApplicationType | None, new_type: ApplicationType | None + ) -> None: + """Replace the current entity when the firmware type changes.""" + er.async_get(hass).async_remove(entity.entity_id) + async_add_entities( + [ + _async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + ] + ) + + entity.async_on_remove( + entity.add_firmware_type_changed_callback(firmware_type_changed) + ) + + return entity + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the firmware update config entry.""" + session = async_get_clientsession(hass) + entity = _async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + + async_add_entities([entity]) + + +class FirmwareUpdateEntity(BaseFirmwareUpdateEntity): + """SkyConnect firmware update entity.""" + + bootloader_reset_type = None + + def __init__( + self, + device: str, + config_entry: ConfigEntry, + update_coordinator: FirmwareUpdateCoordinator, + entity_description: FirmwareUpdateEntityDescription, + ) -> None: + """Initialize the SkyConnect firmware update entity.""" + super().__init__(device, config_entry, update_coordinator, entity_description) + + self._attr_unique_id = ( + f"{self._config_entry.data['serial_number']}_{self.entity_description.key}" + ) + + # Use the cached firmware info if it exists + if self._config_entry.data[FIRMWARE] is not None: + self._current_firmware_info = FirmwareInfo( + device=device, + firmware_type=ApplicationType(self._config_entry.data[FIRMWARE]), + firmware_version=self._config_entry.data[FIRMWARE_VERSION], + owners=[], + source="homeassistant_sky_connect", + ) + + @callback + def _firmware_info_callback(self, firmware_info: FirmwareInfo) -> None: + """Handle updated firmware info being pushed by an integration.""" + self.hass.config_entries.async_update_entry( + self._config_entry, + data={ + **self._config_entry.data, + FIRMWARE: firmware_info.firmware_type, + FIRMWARE_VERSION: firmware_info.firmware_version, + }, + ) + super()._firmware_info_callback(firmware_info) diff --git a/homeassistant/components/homeassistant_yellow/__init__.py b/homeassistant/components/homeassistant_yellow/__init__.py index b0837eeedbe..06f908ab61e 100644 --- a/homeassistant/components/homeassistant_yellow/__init__.py +++ b/homeassistant/components/homeassistant_yellow/__init__.py @@ -18,7 +18,7 @@ from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError from homeassistant.helpers import discovery_flow from homeassistant.helpers.hassio import is_hassio -from .const import FIRMWARE, RADIO_DEVICE, ZHA_HW_DISCOVERY_DATA +from .const import FIRMWARE, FIRMWARE_VERSION, RADIO_DEVICE, ZHA_HW_DISCOVERY_DATA _LOGGER = logging.getLogger(__name__) @@ -55,6 +55,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: data=ZHA_HW_DISCOVERY_DATA, ) + await hass.config_entries.async_forward_entry_setups(entry, ["update"]) + return True @@ -87,6 +89,18 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> minor_version=2, ) + if config_entry.minor_version == 2: + # Add a `firmware_version` key + hass.config_entries.async_update_entry( + config_entry, + data={ + **config_entry.data, + FIRMWARE_VERSION: None, + }, + version=1, + minor_version=3, + ) + _LOGGER.debug( "Migration to version %s.%s successful", config_entry.version, diff --git a/homeassistant/components/homeassistant_yellow/config_flow.py b/homeassistant/components/homeassistant_yellow/config_flow.py index b916c6e46ca..5472c346e94 100644 --- a/homeassistant/components/homeassistant_yellow/config_flow.py +++ b/homeassistant/components/homeassistant_yellow/config_flow.py @@ -37,7 +37,14 @@ from homeassistant.config_entries import ( from homeassistant.core import HomeAssistant, async_get_hass, callback from homeassistant.helpers import discovery_flow, selector -from .const import DOMAIN, FIRMWARE, RADIO_DEVICE, ZHA_DOMAIN, ZHA_HW_DISCOVERY_DATA +from .const import ( + DOMAIN, + FIRMWARE, + FIRMWARE_VERSION, + RADIO_DEVICE, + ZHA_DOMAIN, + ZHA_HW_DISCOVERY_DATA, +) from .hardware import BOARD_NAME _LOGGER = logging.getLogger(__name__) @@ -55,7 +62,7 @@ class HomeAssistantYellowConfigFlow(BaseFirmwareConfigFlow, domain=DOMAIN): """Handle a config flow for Home Assistant Yellow.""" VERSION = 1 - MINOR_VERSION = 2 + MINOR_VERSION = 3 def __init__(self, *args: Any, **kwargs: Any) -> None: """Instantiate config flow.""" @@ -310,6 +317,7 @@ class HomeAssistantYellowOptionsFlowHandler( data={ **self.config_entry.data, FIRMWARE: self._probed_firmware_info.firmware_type.value, + FIRMWARE_VERSION: self._probed_firmware_info.firmware_version, }, ) diff --git a/homeassistant/components/homeassistant_yellow/const.py b/homeassistant/components/homeassistant_yellow/const.py index 79753ae9b9e..b98b1133d01 100644 --- a/homeassistant/components/homeassistant_yellow/const.py +++ b/homeassistant/components/homeassistant_yellow/const.py @@ -2,7 +2,10 @@ DOMAIN = "homeassistant_yellow" +RADIO_MODEL = "Home Assistant Yellow" +RADIO_MANUFACTURER = "Nabu Casa" RADIO_DEVICE = "/dev/ttyAMA1" + ZHA_HW_DISCOVERY_DATA = { "name": "Yellow", "port": { @@ -14,4 +17,9 @@ ZHA_HW_DISCOVERY_DATA = { } FIRMWARE = "firmware" +FIRMWARE_VERSION = "firmware_version" ZHA_DOMAIN = "zha" + +NABU_CASA_FIRMWARE_RELEASES_URL = ( + "https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest" +) diff --git a/homeassistant/components/homeassistant_yellow/update.py b/homeassistant/components/homeassistant_yellow/update.py new file mode 100644 index 00000000000..88d4f2912d3 --- /dev/null +++ b/homeassistant/components/homeassistant_yellow/update.py @@ -0,0 +1,172 @@ +"""Home Assistant Yellow firmware update entity.""" + +from __future__ import annotations + +import logging + +import aiohttp + +from homeassistant.components.homeassistant_hardware.coordinator import ( + FirmwareUpdateCoordinator, +) +from homeassistant.components.homeassistant_hardware.update import ( + BaseFirmwareUpdateEntity, + FirmwareUpdateEntityDescription, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, +) +from homeassistant.components.update import UpdateDeviceClass +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import EntityCategory +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback + +from .const import ( + FIRMWARE, + FIRMWARE_VERSION, + NABU_CASA_FIRMWARE_RELEASES_URL, + RADIO_DEVICE, +) + +_LOGGER = logging.getLogger(__name__) + + +FIRMWARE_ENTITY_DESCRIPTIONS: dict[ + ApplicationType | None, FirmwareUpdateEntityDescription +] = { + ApplicationType.EZSP: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split(" ", 1)[0], + fw_type="yellow_zigbee_ncp", + version_key="ezsp_version", + expected_firmware_type=ApplicationType.EZSP, + firmware_name="EmberZNet", + ), + ApplicationType.SPINEL: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split("/", 1)[1].split("_", 1)[0], + fw_type="yellow_openthread_rcp", + version_key="ot_rcp_version", + expected_firmware_type=ApplicationType.SPINEL, + firmware_name="OpenThread RCP", + ), + None: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw, + fw_type=None, + version_key=None, + expected_firmware_type=None, + firmware_name=None, + ), +} + + +def _async_create_update_entity( + hass: HomeAssistant, + config_entry: ConfigEntry, + session: aiohttp.ClientSession, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> FirmwareUpdateEntity: + """Create an update entity that handles firmware type changes.""" + firmware_type = config_entry.data[FIRMWARE] + entity_description = FIRMWARE_ENTITY_DESCRIPTIONS[ + ApplicationType(firmware_type) if firmware_type is not None else None + ] + + entity = FirmwareUpdateEntity( + device=RADIO_DEVICE, + config_entry=config_entry, + update_coordinator=FirmwareUpdateCoordinator( + hass, + session, + NABU_CASA_FIRMWARE_RELEASES_URL, + ), + entity_description=entity_description, + ) + + def firmware_type_changed( + old_type: ApplicationType | None, new_type: ApplicationType | None + ) -> None: + """Replace the current entity when the firmware type changes.""" + er.async_get(hass).async_remove(entity.entity_id) + async_add_entities( + [ + _async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + ] + ) + + entity.async_on_remove( + entity.add_firmware_type_changed_callback(firmware_type_changed) + ) + + return entity + + +async def async_setup_entry( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the firmware update config entry.""" + session = async_get_clientsession(hass) + entity = _async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + + async_add_entities([entity]) + + +class FirmwareUpdateEntity(BaseFirmwareUpdateEntity): + """Yellow firmware update entity.""" + + bootloader_reset_type = "yellow" # Triggers a GPIO reset + + def __init__( + self, + device: str, + config_entry: ConfigEntry, + update_coordinator: FirmwareUpdateCoordinator, + entity_description: FirmwareUpdateEntityDescription, + ) -> None: + """Initialize the Yellow firmware update entity.""" + super().__init__(device, config_entry, update_coordinator, entity_description) + + self._attr_unique_id = self.entity_description.key + + # Use the cached firmware info if it exists + if self._config_entry.data[FIRMWARE] is not None: + self._current_firmware_info = FirmwareInfo( + device=device, + firmware_type=ApplicationType(self._config_entry.data[FIRMWARE]), + firmware_version=self._config_entry.data[FIRMWARE_VERSION], + owners=[], + source="homeassistant_yellow", + ) + + @callback + def _firmware_info_callback(self, firmware_info: FirmwareInfo) -> None: + """Handle updated firmware info being pushed by an integration.""" + self.hass.config_entries.async_update_entry( + self._config_entry, + data={ + **self._config_entry.data, + FIRMWARE: firmware_info.firmware_type, + FIRMWARE_VERSION: firmware_info.firmware_version, + }, + ) + super()._firmware_info_callback(firmware_info) diff --git a/requirements_all.txt b/requirements_all.txt index 916bdcee3c4..f02cdb56fc9 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -1105,6 +1105,9 @@ ha-iotawattpy==0.1.2 # homeassistant.components.philips_js ha-philipsjs==3.2.2 +# homeassistant.components.homeassistant_hardware +ha-silabs-firmware-client==0.2.0 + # homeassistant.components.habitica habiticalib==0.3.7 @@ -2974,7 +2977,7 @@ unifi_ap==0.0.2 unifiled==0.11 # homeassistant.components.homeassistant_hardware -universal-silabs-flasher==0.0.29 +universal-silabs-flasher==0.0.30 # homeassistant.components.upb upb-lib==0.6.1 diff --git a/tests/components/homeassistant_hardware/test_coordinator.py b/tests/components/homeassistant_hardware/test_coordinator.py new file mode 100644 index 00000000000..9c57aac6811 --- /dev/null +++ b/tests/components/homeassistant_hardware/test_coordinator.py @@ -0,0 +1,55 @@ +"""Test firmware update coordinator for Home Assistant Hardware.""" + +from unittest.mock import AsyncMock, Mock, call, patch + +from ha_silabs_firmware_client import FirmwareManifest, ManifestMissing +import pytest +from yarl import URL + +from homeassistant.components.homeassistant_hardware.coordinator import ( + FirmwareUpdateCoordinator, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.util import dt as dt_util + + +async def test_firmware_update_coordinator_fetching( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test the firmware update coordinator loads manifests.""" + session = async_get_clientsession(hass) + + manifest = FirmwareManifest( + url=URL("https://example.org/firmware"), + html_url=URL("https://example.org/release_notes"), + created_at=dt_util.utcnow(), + firmwares=(), + ) + + mock_client = Mock() + mock_client.async_update_data = AsyncMock(side_effect=[ManifestMissing(), manifest]) + + with patch( + "homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient", + return_value=mock_client, + ): + coordinator = FirmwareUpdateCoordinator( + hass, session, "https://example.org/firmware" + ) + + listener = Mock() + coordinator.async_add_listener(listener) + + # The first update will fail + await coordinator.async_refresh() + assert listener.mock_calls == [call()] + assert coordinator.data is None + assert "GitHub release assets haven't been uploaded yet" in caplog.text + + # The second will succeed + await coordinator.async_refresh() + assert listener.mock_calls == [call(), call()] + assert coordinator.data == manifest + + await coordinator.async_shutdown() diff --git a/tests/components/homeassistant_hardware/test_update.py b/tests/components/homeassistant_hardware/test_update.py new file mode 100644 index 00000000000..0c351141e12 --- /dev/null +++ b/tests/components/homeassistant_hardware/test_update.py @@ -0,0 +1,637 @@ +"""Test Home Assistant Hardware firmware update entity.""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncGenerator +import dataclasses +import logging +from unittest.mock import AsyncMock, Mock, patch + +import aiohttp +from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata +import pytest +from yarl import URL + +from homeassistant.components.homeassistant_hardware.coordinator import ( + FirmwareUpdateCoordinator, +) +from homeassistant.components.homeassistant_hardware.helpers import ( + async_notify_firmware_info, + async_register_firmware_info_provider, +) +from homeassistant.components.homeassistant_hardware.update import ( + BaseFirmwareUpdateEntity, + FirmwareUpdateEntityDescription, + FirmwareUpdateExtraStoredData, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, + OwningIntegration, +) +from homeassistant.components.update import UpdateDeviceClass +from homeassistant.config_entries import ConfigEntry, ConfigEntryState, ConfigFlow +from homeassistant.const import EVENT_STATE_CHANGED, EntityCategory +from homeassistant.core import ( + Event, + EventStateChangedData, + HomeAssistant, + HomeAssistantError, + State, + callback, +) +from homeassistant.helpers import entity_registry as er +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.setup import async_setup_component +from homeassistant.util import dt as dt_util + +from tests.common import ( + MockConfigEntry, + MockModule, + MockPlatform, + async_capture_events, + mock_config_flow, + mock_integration, + mock_platform, + mock_restore_cache_with_extra_data, +) + +TEST_DOMAIN = "test" +TEST_DEVICE = "/dev/serial/by-id/some-unique-serial-device-12345" +TEST_FIRMWARE_RELEASES_URL = "https://example.org/firmware" +TEST_UPDATE_ENTITY_ID = "update.test_firmware" +TEST_MANIFEST = FirmwareManifest( + url=URL("https://example.org/firmware"), + html_url=URL("https://example.org/release_notes"), + created_at=dt_util.utcnow(), + firmwares=( + FirmwareMetadata( + filename="skyconnect_zigbee_ncp_test.gbl", + checksum="aaa", + size=123, + release_notes="Some release notes go here", + metadata={ + "baudrate": 115200, + "ezsp_version": "7.4.4.0", + "fw_type": "zigbee_ncp", + "fw_variant": None, + "metadata_version": 2, + "sdk_version": "4.4.4", + }, + url=URL("https://example.org/firmwares/skyconnect_zigbee_ncp_test.gbl"), + ), + ), +) + + +TEST_FIRMWARE_ENTITY_DESCRIPTIONS: dict[ + ApplicationType | None, FirmwareUpdateEntityDescription +] = { + ApplicationType.EZSP: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split(" ", 1)[0], + fw_type="skyconnect_zigbee_ncp", + version_key="ezsp_version", + expected_firmware_type=ApplicationType.EZSP, + firmware_name="EmberZNet", + ), + ApplicationType.SPINEL: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw.split("/", 1)[1].split("_", 1)[0], + fw_type="skyconnect_openthread_rcp", + version_key="ot_rcp_version", + expected_firmware_type=ApplicationType.SPINEL, + firmware_name="OpenThread RCP", + ), + None: FirmwareUpdateEntityDescription( + key="firmware", + display_precision=0, + device_class=UpdateDeviceClass.FIRMWARE, + entity_category=EntityCategory.CONFIG, + version_parser=lambda fw: fw, + fw_type=None, + version_key=None, + expected_firmware_type=None, + firmware_name=None, + ), +} + + +def _mock_async_create_update_entity( + hass: HomeAssistant, + config_entry: ConfigEntry, + session: aiohttp.ClientSession, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> MockFirmwareUpdateEntity: + """Create an update entity that handles firmware type changes.""" + firmware_type = config_entry.data["firmware"] + entity_description = TEST_FIRMWARE_ENTITY_DESCRIPTIONS[ + ApplicationType(firmware_type) if firmware_type is not None else None + ] + + entity = MockFirmwareUpdateEntity( + device=config_entry.data["device"], + config_entry=config_entry, + update_coordinator=FirmwareUpdateCoordinator( + hass, + session, + TEST_FIRMWARE_RELEASES_URL, + ), + entity_description=entity_description, + ) + + def firmware_type_changed( + old_type: ApplicationType | None, new_type: ApplicationType | None + ) -> None: + """Replace the current entity when the firmware type changes.""" + er.async_get(hass).async_remove(entity.entity_id) + async_add_entities( + [ + _mock_async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + ] + ) + + entity.async_on_remove( + entity.add_firmware_type_changed_callback(firmware_type_changed) + ) + + return entity + + +async def mock_async_setup_entry( + hass: HomeAssistant, config_entry: ConfigEntry +) -> bool: + """Set up test config entry.""" + await hass.config_entries.async_forward_entry_setups(config_entry, ["update"]) + return True + + +async def mock_async_setup_update_entities( + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the firmware update config entry.""" + session = async_get_clientsession(hass) + entity = _mock_async_create_update_entity( + hass, config_entry, session, async_add_entities + ) + + async_add_entities([entity]) + + +class MockFirmwareUpdateEntity(BaseFirmwareUpdateEntity): + """Mock SkyConnect firmware update entity.""" + + bootloader_reset_type = None + + def __init__( + self, + device: str, + config_entry: ConfigEntry, + update_coordinator: FirmwareUpdateCoordinator, + entity_description: FirmwareUpdateEntityDescription, + ) -> None: + """Initialize the mock SkyConnect firmware update entity.""" + super().__init__(device, config_entry, update_coordinator, entity_description) + self._attr_unique_id = self.entity_description.key + + # Use the cached firmware info if it exists + if self._config_entry.data["firmware"] is not None: + self._current_firmware_info = FirmwareInfo( + device=device, + firmware_type=ApplicationType(self._config_entry.data["firmware"]), + firmware_version=self._config_entry.data["firmware_version"], + owners=[], + source=TEST_DOMAIN, + ) + + @callback + def _firmware_info_callback(self, firmware_info: FirmwareInfo) -> None: + """Handle updated firmware info being pushed by an integration.""" + super()._firmware_info_callback(firmware_info) + + self.hass.config_entries.async_update_entry( + self._config_entry, + data={ + **self._config_entry.data, + "firmware": firmware_info.firmware_type, + "firmware_version": firmware_info.firmware_version, + }, + ) + + +@pytest.fixture(name="update_config_entry") +async def mock_update_config_entry( + hass: HomeAssistant, +) -> AsyncGenerator[ConfigEntry]: + """Set up a mock Home Assistant Hardware firmware update entity.""" + await async_setup_component(hass, "homeassistant", {}) + await async_setup_component(hass, "homeassistant_hardware", {}) + + mock_integration( + hass, + MockModule( + TEST_DOMAIN, + async_setup_entry=mock_async_setup_entry, + ), + built_in=False, + ) + mock_platform(hass, "test.config_flow") + mock_platform( + hass, + "test.update", + MockPlatform(async_setup_entry=mock_async_setup_update_entities), + ) + + # Set up a mock integration using the hardware update entity + config_entry = MockConfigEntry( + domain=TEST_DOMAIN, + data={ + "device": TEST_DEVICE, + "firmware": "ezsp", + "firmware_version": "7.3.1.0 build 0", + }, + ) + config_entry.add_to_hass(hass) + + with ( + patch( + "homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient", + autospec=True, + ) as mock_update_client, + mock_config_flow(TEST_DOMAIN, ConfigFlow), + ): + mock_update_client.return_value.async_update_data.return_value = TEST_MANIFEST + yield config_entry + + +async def test_update_entity_installation( + hass: HomeAssistant, update_config_entry: ConfigEntry +) -> None: + """Test the Hardware firmware update entity installation.""" + + assert await hass.config_entries.async_setup(update_config_entry.entry_id) + await hass.async_block_till_done() + + # Set up another integration communicating with the device + owning_config_entry = MockConfigEntry( + domain="another_integration", + data={ + "device": { + "path": TEST_DEVICE, + "flow_control": "hardware", + "baudrate": 115200, + }, + "radio_type": "ezsp", + }, + version=4, + ) + owning_config_entry.add_to_hass(hass) + owning_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + # The integration provides firmware info + mock_hw_module = Mock() + mock_hw_module.get_firmware_info = lambda hass, config_entry: FirmwareInfo( + device=TEST_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="7.3.1.0 build 0", + owners=[OwningIntegration(config_entry_id=config_entry.entry_id)], + source="another_integration", + ) + + async_register_firmware_info_provider(hass, "another_integration", mock_hw_module) + + # Pretend the other integration loaded and notified hardware of the running firmware + await async_notify_firmware_info( + hass, + "another_integration", + mock_hw_module.get_firmware_info(hass, owning_config_entry), + ) + + state_before_update = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_before_update is not None + assert state_before_update.state == "unknown" + assert state_before_update.attributes["title"] == "EmberZNet" + assert state_before_update.attributes["installed_version"] == "7.3.1.0" + assert state_before_update.attributes["latest_version"] is None + + # When we check for an update, one will be shown + await hass.services.async_call( + "homeassistant", + "update_entity", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + state_after_update = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_after_update is not None + assert state_after_update.state == "on" + assert state_after_update.attributes["title"] == "EmberZNet" + assert state_after_update.attributes["installed_version"] == "7.3.1.0" + assert state_after_update.attributes["latest_version"] == "7.4.4.0" + assert state_after_update.attributes["release_summary"] == ( + "Some release notes go here" + ) + assert state_after_update.attributes["release_url"] == ( + "https://example.org/release_notes" + ) + + mock_firmware = Mock() + mock_flasher = AsyncMock() + + async def mock_flash_firmware(fw_image, progress_callback): + await asyncio.sleep(0) + progress_callback(0, 100) + await asyncio.sleep(0) + progress_callback(50, 100) + 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( + device=TEST_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="7.4.4.0 build 0", + owners=[], + source="probe", + ), + ), + patch.object( + owning_config_entry, "async_unload", wraps=owning_config_entry.async_unload + ) as owning_config_entry_unload, + ): + state_changes: list[Event[EventStateChangedData]] = async_capture_events( + hass, EVENT_STATE_CHANGED + ) + await hass.services.async_call( + "update", + "install", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + + # Progress events are emitted during the installation + assert len(state_changes) == 7 + + # Indeterminate progress first + assert state_changes[0].data["new_state"].attributes["in_progress"] is True + assert state_changes[0].data["new_state"].attributes["update_percentage"] is None + + # Then the update starts + assert state_changes[1].data["new_state"].attributes["update_percentage"] == 0 + assert state_changes[2].data["new_state"].attributes["update_percentage"] == 50 + assert state_changes[3].data["new_state"].attributes["update_percentage"] == 100 + + # Once it is done, we probe the firmware + assert state_changes[4].data["new_state"].attributes["in_progress"] is True + assert state_changes[4].data["new_state"].attributes["update_percentage"] is None + + # Finally, the update finishes + assert state_changes[5].data["new_state"].attributes["update_percentage"] is None + 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 + assert state_after_install.state == "off" + assert state_after_install.attributes["title"] == "EmberZNet" + assert state_after_install.attributes["installed_version"] == "7.4.4.0" + assert state_after_install.attributes["latest_version"] == "7.4.4.0" + + +async def test_update_entity_installation_failure( + hass: HomeAssistant, update_config_entry: ConfigEntry +) -> None: + """Test installation failing during flashing.""" + assert await hass.config_entries.async_setup(update_config_entry.entry_id) + await hass.async_block_till_done() + + await hass.services.async_call( + "homeassistant", + "update_entity", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + + state_before_install = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_before_install is not None + assert state_before_install.state == "on" + assert state_before_install.attributes["title"] == "EmberZNet" + 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, + ), + pytest.raises(HomeAssistantError, match="Failed to flash firmware"), + ): + await hass.services.async_call( + "update", + "install", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + + # After the firmware update fails, we can still try again + state_after_install = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_after_install is not None + assert state_after_install.state == "on" + assert state_after_install.attributes["title"] == "EmberZNet" + assert state_after_install.attributes["installed_version"] == "7.3.1.0" + assert state_after_install.attributes["latest_version"] == "7.4.4.0" + + +async def test_update_entity_installation_probe_failure( + hass: HomeAssistant, update_config_entry: ConfigEntry +) -> None: + """Test installation failing during post-flashing probing.""" + assert await hass.config_entries.async_setup(update_config_entry.entry_id) + await hass.async_block_till_done() + + await hass.services.async_call( + "homeassistant", + "update_entity", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + + state_before_install = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_before_install is not None + assert state_before_install.state == "on" + assert state_before_install.attributes["title"] == "EmberZNet" + assert state_before_install.attributes["installed_version"] == "7.3.1.0" + assert state_before_install.attributes["latest_version"] == "7.4.4.0" + + with ( + patch( + "homeassistant.components.homeassistant_hardware.update.parse_firmware_image", + return_value=Mock(), + ), + patch( + "homeassistant.components.homeassistant_hardware.update.Flasher", + return_value=AsyncMock(), + ), + patch( + "homeassistant.components.homeassistant_hardware.update.probe_silabs_firmware_info", + return_value=None, + ), + pytest.raises( + HomeAssistantError, match="Failed to probe the firmware after flashing" + ), + ): + await hass.services.async_call( + "update", + "install", + {"entity_id": TEST_UPDATE_ENTITY_ID}, + blocking=True, + ) + + # After the firmware update fails, we can still try again + state_after_install = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state_after_install is not None + assert state_after_install.state == "on" + assert state_after_install.attributes["title"] == "EmberZNet" + assert state_after_install.attributes["installed_version"] == "7.3.1.0" + assert state_after_install.attributes["latest_version"] == "7.4.4.0" + + +async def test_update_entity_state_restoration( + hass: HomeAssistant, update_config_entry: ConfigEntry +) -> None: + """Test the Hardware firmware update entity state restoration.""" + + mock_restore_cache_with_extra_data( + hass, + [ + ( + State(TEST_UPDATE_ENTITY_ID, "on"), + FirmwareUpdateExtraStoredData( + firmware_manifest=TEST_MANIFEST + ).as_dict(), + ) + ], + ) + + assert await hass.config_entries.async_setup(update_config_entry.entry_id) + await hass.async_block_till_done() + + # The state is correctly restored + state = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state is not None + assert state.state == "on" + assert state.attributes["title"] == "EmberZNet" + assert state.attributes["installed_version"] == "7.3.1.0" + assert state.attributes["latest_version"] == "7.4.4.0" + assert state.attributes["release_summary"] == ("Some release notes go here") + assert state.attributes["release_url"] == ("https://example.org/release_notes") + + +async def test_update_entity_firmware_missing_from_manifest( + hass: HomeAssistant, update_config_entry: ConfigEntry +) -> None: + """Test the Hardware firmware update entity handles missing firmware.""" + + mock_restore_cache_with_extra_data( + hass, + [ + ( + State(TEST_UPDATE_ENTITY_ID, "on"), + # Ensure the manifest does not contain our expected firmware type + FirmwareUpdateExtraStoredData( + firmware_manifest=dataclasses.replace(TEST_MANIFEST, firmwares=()) + ).as_dict(), + ) + ], + ) + + assert await hass.config_entries.async_setup(update_config_entry.entry_id) + await hass.async_block_till_done() + + # The state is restored, accounting for the missing firmware + state = hass.states.get(TEST_UPDATE_ENTITY_ID) + assert state is not None + assert state.state == "unknown" + assert state.attributes["title"] == "EmberZNet" + assert state.attributes["installed_version"] == "7.3.1.0" + assert state.attributes["latest_version"] is None + assert state.attributes["release_summary"] is None + assert state.attributes["release_url"] is None + + +async def test_update_entity_graceful_firmware_type_callback_errors( + hass: HomeAssistant, + update_config_entry: ConfigEntry, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test firmware update entity handling of firmware type callback errors.""" + + session = async_get_clientsession(hass) + update_entity = MockFirmwareUpdateEntity( + device=TEST_DEVICE, + config_entry=update_config_entry, + update_coordinator=FirmwareUpdateCoordinator( + hass, + session, + TEST_FIRMWARE_RELEASES_URL, + ), + entity_description=TEST_FIRMWARE_ENTITY_DESCRIPTIONS[ApplicationType.EZSP], + ) + update_entity.hass = hass + await update_entity.async_added_to_hass() + + callback = Mock(side_effect=RuntimeError("Callback failed")) + unregister_callback = update_entity.add_firmware_type_changed_callback(callback) + + with caplog.at_level(logging.WARNING): + await async_notify_firmware_info( + hass, + "some_integration", + FirmwareInfo( + device=TEST_DEVICE, + firmware_type=ApplicationType.SPINEL, + firmware_version="SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4; EFR32; Oct 21 2024 14:40:57", + owners=[], + source="probe", + ), + ) + + unregister_callback() + assert "Failed to call firmware type changed callback" in caplog.text diff --git a/tests/components/homeassistant_hardware/test_util.py b/tests/components/homeassistant_hardware/test_util.py index b467380c431..1b7bfe4a8ac 100644 --- a/tests/components/homeassistant_hardware/test_util.py +++ b/tests/components/homeassistant_hardware/test_util.py @@ -205,6 +205,93 @@ async def test_owning_addon(hass: HomeAssistant) -> None: assert (await owning_addon.is_running(hass)) is False +async def test_owning_addon_temporarily_stop_info_error(hass: HomeAssistant) -> None: + """Test `OwningAddon` temporarily stopping with an info error.""" + + owning_addon = OwningAddon(slug="some-addon-slug") + mock_manager = AsyncMock() + mock_manager.async_get_addon_info.side_effect = AddonError() + + with patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager", + return_value=mock_manager, + ): + async with owning_addon.temporarily_stop(hass): + pass + + # We never restart it + assert len(mock_manager.async_get_addon_info.mock_calls) == 1 + assert len(mock_manager.async_stop_addon.mock_calls) == 0 + assert len(mock_manager.async_wait_until_addon_state.mock_calls) == 0 + assert len(mock_manager.async_start_addon_waiting.mock_calls) == 0 + + +async def test_owning_addon_temporarily_stop_not_running(hass: HomeAssistant) -> None: + """Test `OwningAddon` temporarily stopping when the addon is not running.""" + + owning_addon = OwningAddon(slug="some-addon-slug") + + mock_manager = AsyncMock() + mock_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname="core_some_addon_slug", + options={}, + state=AddonState.NOT_RUNNING, + update_available=False, + version="1.0.0", + ) + + with patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager", + return_value=mock_manager, + ): + async with owning_addon.temporarily_stop(hass): + pass + + # We never restart it + assert len(mock_manager.async_get_addon_info.mock_calls) == 1 + assert len(mock_manager.async_stop_addon.mock_calls) == 0 + assert len(mock_manager.async_wait_until_addon_state.mock_calls) == 0 + assert len(mock_manager.async_start_addon_waiting.mock_calls) == 0 + + +async def test_owning_addon_temporarily_stop(hass: HomeAssistant) -> None: + """Test `OwningAddon` temporarily stopping when the addon is running.""" + + owning_addon = OwningAddon(slug="some-addon-slug") + + mock_manager = AsyncMock() + mock_manager.async_get_addon_info.return_value = AddonInfo( + available=True, + hostname="core_some_addon_slug", + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) + + mock_manager.async_stop_addon = AsyncMock() + mock_manager.async_wait_until_addon_state = AsyncMock() + mock_manager.async_start_addon_waiting = AsyncMock() + + # The error is propagated but it doesn't affect restarting the addon + with ( + patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager", + return_value=mock_manager, + ), + pytest.raises(RuntimeError), + ): + async with owning_addon.temporarily_stop(hass): + raise RuntimeError("Some error") + + # We restart it + assert len(mock_manager.async_get_addon_info.mock_calls) == 1 + assert len(mock_manager.async_stop_addon.mock_calls) == 1 + assert len(mock_manager.async_wait_until_addon_state.mock_calls) == 1 + assert len(mock_manager.async_start_addon_waiting.mock_calls) == 1 + + async def test_owning_integration(hass: HomeAssistant) -> None: """Test `OwningIntegration`.""" config_entry = MockConfigEntry(domain="mock_domain", unique_id="some_unique_id") @@ -225,6 +312,67 @@ async def test_owning_integration(hass: HomeAssistant) -> None: assert (await owning_integration2.is_running(hass)) is False +async def test_owning_integration_temporarily_stop_missing_entry( + hass: HomeAssistant, +) -> None: + """Test temporarily stopping the integration when the config entry doesn't exist.""" + missing_integration = OwningIntegration(config_entry_id="missing_entry_id") + + with ( + patch.object(hass.config_entries, "async_unload") as mock_unload, + patch.object(hass.config_entries, "async_setup") as mock_setup, + ): + async with missing_integration.temporarily_stop(hass): + pass + + # Because there's no matching entry, no unload or setup calls are made + assert len(mock_unload.mock_calls) == 0 + assert len(mock_setup.mock_calls) == 0 + + +async def test_owning_integration_temporarily_stop_not_loaded( + hass: HomeAssistant, +) -> None: + """Test temporarily stopping the integration when the config entry is not loaded.""" + entry = MockConfigEntry(domain="test_domain") + entry.add_to_hass(hass) + entry.mock_state(hass, ConfigEntryState.NOT_LOADED) + + integration = OwningIntegration(config_entry_id=entry.entry_id) + + with ( + patch.object(hass.config_entries, "async_unload") as mock_unload, + patch.object(hass.config_entries, "async_setup") as mock_setup, + ): + async with integration.temporarily_stop(hass): + pass + + # Since the entry was not loaded, we never unload or re-setup + assert len(mock_unload.mock_calls) == 0 + assert len(mock_setup.mock_calls) == 0 + + +async def test_owning_integration_temporarily_stop_loaded(hass: HomeAssistant) -> None: + """Test temporarily stopping the integration when the config entry is loaded.""" + entry = MockConfigEntry(domain="test_domain") + entry.add_to_hass(hass) + entry.mock_state(hass, ConfigEntryState.LOADED) + + integration = OwningIntegration(config_entry_id=entry.entry_id) + + with ( + patch.object(hass.config_entries, "async_unload") as mock_unload, + patch.object(hass.config_entries, "async_setup") as mock_setup, + pytest.raises(RuntimeError), + ): + async with integration.temporarily_stop(hass): + raise RuntimeError("Some error during the temporary stop") + + # We expect one unload followed by one setup call + mock_unload.assert_called_once_with(entry.entry_id) + mock_setup.assert_called_once_with(entry.entry_id) + + async def test_firmware_info(hass: HomeAssistant) -> None: """Test `FirmwareInfo`.""" diff --git a/tests/components/homeassistant_sky_connect/common.py b/tests/components/homeassistant_sky_connect/common.py new file mode 100644 index 00000000000..335fd6d2e12 --- /dev/null +++ b/tests/components/homeassistant_sky_connect/common.py @@ -0,0 +1,21 @@ +"""Common constants for the SkyConnect integration tests.""" + +from homeassistant.helpers.service_info.usb import UsbServiceInfo + +USB_DATA_SKY = UsbServiceInfo( + device="/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + vid="10C4", + pid="EA60", + serial_number="9e2adbd75b8beb119fe564a0f320645d", + manufacturer="Nabu Casa", + description="SkyConnect v1.0", +) + +USB_DATA_ZBT1 = UsbServiceInfo( + device="/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + vid="10C4", + pid="EA60", + serial_number="9e2adbd75b8beb119fe564a0f320645d", + manufacturer="Nabu Casa", + description="Home Assistant Connect ZBT-1", +) diff --git a/tests/components/homeassistant_sky_connect/test_config_flow.py b/tests/components/homeassistant_sky_connect/test_config_flow.py index d8542002ae8..44a5e0029c3 100644 --- a/tests/components/homeassistant_sky_connect/test_config_flow.py +++ b/tests/components/homeassistant_sky_connect/test_config_flow.py @@ -22,26 +22,10 @@ from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType from homeassistant.helpers.service_info.usb import UsbServiceInfo +from .common import USB_DATA_SKY, USB_DATA_ZBT1 + from tests.common import MockConfigEntry -USB_DATA_SKY = UsbServiceInfo( - device="/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", - vid="10C4", - pid="EA60", - serial_number="9e2adbd75b8beb119fe564a0f320645d", - manufacturer="Nabu Casa", - description="SkyConnect v1.0", -) - -USB_DATA_ZBT1 = UsbServiceInfo( - device="/dev/serial/by-id/usb-Nabu_Casa_Home_Assistant_Connect_ZBT-1_9e2adbd75b8beb119fe564a0f320645d-if00-port0", - vid="10C4", - pid="EA60", - serial_number="9e2adbd75b8beb119fe564a0f320645d", - manufacturer="Nabu Casa", - description="Home Assistant Connect ZBT-1", -) - @pytest.mark.parametrize( ("usb_data", "model"), @@ -76,7 +60,7 @@ async def test_config_flow( return_value=FirmwareInfo( device=usb_data.device, firmware_type=ApplicationType.EZSP, - firmware_version=None, + firmware_version="7.4.4.0 build 0", owners=[], source="probe", ), @@ -92,6 +76,7 @@ async def test_config_flow( config_entry = result["result"] assert config_entry.data == { "firmware": "ezsp", + "firmware_version": "7.4.4.0 build 0", "device": usb_data.device, "manufacturer": usb_data.manufacturer, "pid": usb_data.pid, @@ -161,7 +146,7 @@ async def test_options_flow( return_value=FirmwareInfo( device=usb_data.device, firmware_type=ApplicationType.EZSP, - firmware_version=None, + firmware_version="7.4.4.0 build 0", owners=[], source="probe", ), @@ -177,6 +162,7 @@ async def test_options_flow( assert config_entry.data == { "firmware": "ezsp", + "firmware_version": "7.4.4.0 build 0", "device": usb_data.device, "manufacturer": usb_data.manufacturer, "pid": usb_data.pid, diff --git a/tests/components/homeassistant_sky_connect/test_init.py b/tests/components/homeassistant_sky_connect/test_init.py index 8e90039a4fc..c467a9e0d60 100644 --- a/tests/components/homeassistant_sky_connect/test_init.py +++ b/tests/components/homeassistant_sky_connect/test_init.py @@ -44,7 +44,7 @@ async def test_config_entry_migration_v2(hass: HomeAssistant) -> None: await hass.config_entries.async_setup(config_entry.entry_id) assert config_entry.version == 1 - assert config_entry.minor_version == 2 + assert config_entry.minor_version == 3 assert config_entry.data == { "description": "SkyConnect v1.0", "device": "/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", @@ -54,6 +54,7 @@ async def test_config_entry_migration_v2(hass: HomeAssistant) -> None: "manufacturer": "Nabu Casa", "product": "SkyConnect v1.0", # `description` has been copied to `product` "firmware": "spinel", # new key + "firmware_version": None, # new key } await hass.config_entries.async_unload(config_entry.entry_id) diff --git a/tests/components/homeassistant_sky_connect/test_update.py b/tests/components/homeassistant_sky_connect/test_update.py new file mode 100644 index 00000000000..9fb7528987e --- /dev/null +++ b/tests/components/homeassistant_sky_connect/test_update.py @@ -0,0 +1,86 @@ +"""Test SkyConnect firmware update entity.""" + +from homeassistant.components.homeassistant_hardware.helpers import ( + async_notify_firmware_info, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, +) +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from .common import USB_DATA_ZBT1 + +from tests.common import MockConfigEntry + +UPDATE_ENTITY_ID = ( + "update.homeassistant_sky_connect_9e2adbd75b8beb119fe564a0f320645d_firmware" +) + + +async def test_zbt1_update_entity(hass: HomeAssistant) -> None: + """Test the ZBT-1 firmware update entity.""" + await async_setup_component(hass, "homeassistant", {}) + + # Set up the ZBT-1 integration + zbt1_config_entry = MockConfigEntry( + domain="homeassistant_sky_connect", + data={ + "firmware": "ezsp", + "firmware_version": "7.3.1.0 build 0", + "device": USB_DATA_ZBT1.device, + "manufacturer": USB_DATA_ZBT1.manufacturer, + "pid": USB_DATA_ZBT1.pid, + "product": USB_DATA_ZBT1.description, + "serial_number": USB_DATA_ZBT1.serial_number, + "vid": USB_DATA_ZBT1.vid, + }, + version=1, + minor_version=3, + ) + zbt1_config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(zbt1_config_entry.entry_id) + await hass.async_block_till_done() + + # Pretend ZHA loaded and notified hardware of the running firmware + await async_notify_firmware_info( + hass, + "zha", + FirmwareInfo( + device=USB_DATA_ZBT1.device, + firmware_type=ApplicationType.EZSP, + firmware_version="7.3.1.0 build 0", + owners=[], + source="zha", + ), + ) + await hass.async_block_till_done() + + state_ezsp = hass.states.get(UPDATE_ENTITY_ID) + assert state_ezsp.state == "unknown" + assert state_ezsp.attributes["title"] == "EmberZNet" + assert state_ezsp.attributes["installed_version"] == "7.3.1.0" + assert state_ezsp.attributes["latest_version"] is None + + # Now, have OTBR push some info + await async_notify_firmware_info( + hass, + "otbr", + FirmwareInfo( + device=USB_DATA_ZBT1.device, + firmware_type=ApplicationType.SPINEL, + firmware_version="SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4; EFR32; Oct 21 2024 14:40:57", + owners=[], + source="otbr", + ), + ) + await hass.async_block_till_done() + + # After the firmware update, the entity has the new version and the correct state + state_spinel = hass.states.get(UPDATE_ENTITY_ID) + assert state_spinel.state == "unknown" + assert state_spinel.attributes["title"] == "OpenThread RCP" + assert state_spinel.attributes["installed_version"] == "2.4.4.0" + assert state_spinel.attributes["latest_version"] is None diff --git a/tests/components/homeassistant_yellow/test_config_flow.py b/tests/components/homeassistant_yellow/test_config_flow.py index 78fd45c6b5b..46fec0a1f30 100644 --- a/tests/components/homeassistant_yellow/test_config_flow.py +++ b/tests/components/homeassistant_yellow/test_config_flow.py @@ -350,7 +350,7 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None: return_value=FirmwareInfo( device=RADIO_DEVICE, firmware_type=ApplicationType.EZSP, - firmware_version=None, + firmware_version="7.4.4.0 build 0", owners=[], source="probe", ), @@ -366,6 +366,7 @@ async def test_firmware_options_flow(hass: HomeAssistant) -> None: assert config_entry.data == { "firmware": "ezsp", + "firmware_version": "7.4.4.0 build 0", } diff --git a/tests/components/homeassistant_yellow/test_update.py b/tests/components/homeassistant_yellow/test_update.py new file mode 100644 index 00000000000..269ff2afc49 --- /dev/null +++ b/tests/components/homeassistant_yellow/test_update.py @@ -0,0 +1,89 @@ +"""Test Yellow firmware update entity.""" + +from unittest.mock import patch + +from homeassistant.components.homeassistant_hardware.helpers import ( + async_notify_firmware_info, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, +) +from homeassistant.components.homeassistant_yellow.const import RADIO_DEVICE +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry + +UPDATE_ENTITY_ID = "update.homeassistant_yellow_firmware" + + +async def test_yellow_update_entity(hass: HomeAssistant) -> None: + """Test the Yellow firmware update entity.""" + await async_setup_component(hass, "homeassistant", {}) + + # Set up the Yellow integration + yellow_config_entry = MockConfigEntry( + domain="homeassistant_yellow", + data={ + "firmware": "ezsp", + "firmware_version": "7.3.1.0 build 0", + "device": RADIO_DEVICE, + }, + version=1, + minor_version=3, + ) + yellow_config_entry.add_to_hass(hass) + + with ( + patch( + "homeassistant.components.homeassistant_yellow.is_hassio", return_value=True + ), + patch( + "homeassistant.components.homeassistant_yellow.get_os_info", + return_value={"board": "yellow"}, + ), + ): + assert await hass.config_entries.async_setup(yellow_config_entry.entry_id) + await hass.async_block_till_done() + + # Pretend ZHA loaded and notified hardware of the running firmware + await async_notify_firmware_info( + hass, + "zha", + FirmwareInfo( + device=RADIO_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="7.3.1.0 build 0", + owners=[], + source="zha", + ), + ) + await hass.async_block_till_done() + + state_ezsp = hass.states.get(UPDATE_ENTITY_ID) + assert state_ezsp.state == "unknown" + assert state_ezsp.attributes["title"] == "EmberZNet" + assert state_ezsp.attributes["installed_version"] == "7.3.1.0" + assert state_ezsp.attributes["latest_version"] is None + + # Now, have OTBR push some info + await async_notify_firmware_info( + hass, + "otbr", + FirmwareInfo( + device=RADIO_DEVICE, + firmware_type=ApplicationType.SPINEL, + firmware_version="SL-OPENTHREAD/2.4.4.0_GitHub-7074a43e4; EFR32; Oct 21 2024 14:40:57", + owners=[], + source="otbr", + ), + ) + await hass.async_block_till_done() + + # After the firmware update, the entity has the new version and the correct state + state_spinel = hass.states.get(UPDATE_ENTITY_ID) + assert state_spinel.state == "unknown" + assert state_spinel.attributes["title"] == "OpenThread RCP" + assert state_spinel.attributes["installed_version"] == "2.4.4.0" + assert state_spinel.attributes["latest_version"] is None