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
This commit is contained in:
puddly
2025-03-14 19:28:02 -04:00
committed by GitHub
parent b07c28126a
commit 537302ce56
22 changed files with 1916 additions and 51 deletions

View File

@ -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

View File

@ -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"
]
}

View File

@ -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()

View File

@ -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:

View File

@ -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,

View File

@ -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,
)

View File

@ -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:

View File

@ -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)

View File

@ -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,

View File

@ -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,
},
)

View File

@ -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"
)

View File

@ -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)

5
requirements_all.txt generated
View File

@ -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

View File

@ -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()

View File

@ -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

View File

@ -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`."""

View File

@ -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",
)

View File

@ -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,

View File

@ -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)

View File

@ -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

View File

@ -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",
}

View File

@ -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