1
0
mirror of https://github.com/home-assistant/core.git synced 2026-02-15 07:36:16 +00:00

Beta firmware update switch for Connect integrations (#155370)

This commit is contained in:
puddly
2025-10-29 09:45:56 -04:00
committed by GitHub
parent 80912045d7
commit aa5b970102
28 changed files with 1124 additions and 139 deletions

View File

@@ -2,20 +2,35 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
import os.path
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.usb import USBDevice, async_register_port_event_callback
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import DEVICE, DOMAIN
from .const import DEVICE, DOMAIN, NABU_CASA_FIRMWARE_RELEASES_URL
_LOGGER = logging.getLogger(__name__)
type HomeAssistantConnectZBT2ConfigEntry = ConfigEntry[HomeAssistantConnectZBT2Data]
@dataclass
class HomeAssistantConnectZBT2Data:
"""Runtime data definition."""
coordinator: FirmwareUpdateCoordinator
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@@ -49,7 +64,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantConnectZBT2ConfigEntry
) -> bool:
"""Set up a Home Assistant Connect ZBT-2 config entry."""
# Postpone loading the config entry if the device is missing
@@ -60,12 +77,23 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
translation_key="device_disconnected",
)
await hass.config_entries.async_forward_entry_setups(entry, ["update"])
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantConnectZBT2Data(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantConnectZBT2ConfigEntry
) -> bool:
"""Unload a config entry."""
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])

View File

@@ -3,7 +3,7 @@
DOMAIN = "homeassistant_connect_zbt2"
NABU_CASA_FIRMWARE_RELEASES_URL = (
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
)
FIRMWARE = "firmware"

View File

@@ -0,0 +1,9 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -90,6 +90,13 @@
}
}
},
"entity": {
"switch": {
"beta_firmware": {
"name": "Beta firmware updates"
}
}
},
"exceptions": {
"device_disconnected": {
"message": "The device is not plugged in"

View File

@@ -0,0 +1,54 @@
"""Home Assistant Connect ZBT-2 switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantConnectZBT2ConfigEntry
from .const import DOMAIN, HARDWARE_NAME, SERIAL_NUMBER
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant Connect ZBT-2."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant Connect ZBT-2 beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
serial_number = self._config_entry.data[SERIAL_NUMBER]
self._attr_unique_id = f"{serial_number}_beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
name=f"{HARDWARE_NAME} ({serial_number})",
model=HARDWARE_NAME,
manufacturer="Nabu Casa",
serial_number=serial_number,
)

View File

@@ -4,8 +4,6 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -19,22 +17,14 @@ from homeassistant.components.homeassistant_hardware.util import (
ResetTarget,
)
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 device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
HARDWARE_NAME,
NABU_CASA_FIRMWARE_RELEASES_URL,
SERIAL_NUMBER,
)
from . import HomeAssistantConnectZBT2ConfigEntry
from .const import DOMAIN, FIRMWARE, FIRMWARE_VERSION, HARDWARE_NAME, SERIAL_NUMBER
_LOGGER = logging.getLogger(__name__)
@@ -91,8 +81,7 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -111,12 +100,7 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=config_entry.data["device"],
config_entry=config_entry,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
update_coordinator=config_entry.runtime_data.coordinator,
entity_description=entity_description,
)
@@ -126,11 +110,7 @@ def _async_create_update_entity(
"""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
)
]
[_async_create_update_entity(hass, config_entry, async_add_entities)]
)
entity.async_on_remove(
@@ -142,14 +122,11 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
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
)
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
async_add_entities([entity])
@@ -162,7 +139,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: ConfigEntry,
config_entry: HomeAssistantConnectZBT2ConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

View File

@@ -8,6 +8,6 @@
"integration_type": "system",
"requirements": [
"universal-silabs-flasher==0.0.35",
"ha-silabs-firmware-client==0.2.0"
"ha-silabs-firmware-client==0.3.0"
]
}

View File

@@ -0,0 +1,64 @@
"""Home Assistant Hardware base beta firmware switch entity."""
from __future__ import annotations
import logging
from typing import Any
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.helpers.restore_state import RestoreEntity
from .coordinator import FirmwareUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class BaseBetaFirmwareSwitch(SwitchEntity, RestoreEntity):
"""Base switch to enable beta firmware updates."""
_attr_has_entity_name = True
_attr_entity_category = EntityCategory.CONFIG
_attr_entity_registry_enabled_default = False
_attr_translation_key = "beta_firmware"
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: ConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
self._coordinator = coordinator
self._config_entry = config_entry
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added to hass."""
await super().async_added_to_hass()
# Restore the last state
last_state = await self.async_get_last_state()
if last_state is not None:
self._attr_is_on = last_state.state == "on"
else:
self._attr_is_on = False
# Apply the restored state to the coordinator
await self._update_coordinator_prerelease()
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn on beta firmware updates."""
self._attr_is_on = True
self.async_write_ha_state()
await self._update_coordinator_prerelease()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn off beta firmware updates."""
self._attr_is_on = False
self.async_write_ha_state()
await self._update_coordinator_prerelease()
async def _update_coordinator_prerelease(self) -> None:
"""Update the coordinator with the current prerelease setting."""
self._coordinator.client.update_prerelease(bool(self._attr_is_on))
await self._coordinator.async_refresh()

View File

@@ -2,9 +2,13 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
import os.path
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.util import guess_firmware_info
from homeassistant.components.usb import (
USBDevice,
@@ -15,6 +19,7 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import (
@@ -24,6 +29,7 @@ from .const import (
FIRMWARE,
FIRMWARE_VERSION,
MANUFACTURER,
NABU_CASA_FIRMWARE_RELEASES_URL,
PID,
PRODUCT,
SERIAL_NUMBER,
@@ -32,6 +38,16 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
type HomeAssistantSkyConnectConfigEntry = ConfigEntry[HomeAssistantSkyConnectData]
@dataclass
class HomeAssistantSkyConnectData:
"""Runtime data definition."""
coordinator: FirmwareUpdateCoordinator
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
@@ -65,7 +81,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
"""Set up a Home Assistant SkyConnect config entry."""
# Postpone loading the config entry if the device is missing
@@ -76,18 +94,31 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
translation_key="device_disconnected",
)
await hass.config_entries.async_forward_entry_setups(entry, ["update"])
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantSkyConnectData(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
"""Unload a config entry."""
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
async def async_migrate_entry(
hass: HomeAssistant, config_entry: HomeAssistantSkyConnectConfigEntry
) -> bool:
"""Migrate old entry."""
_LOGGER.debug(

View File

@@ -8,7 +8,7 @@ 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"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
)
FIRMWARE = "firmware"

View File

@@ -0,0 +1,9 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -90,6 +90,13 @@
}
}
},
"entity": {
"switch": {
"beta_firmware": {
"name": "Beta firmware updates"
}
}
},
"exceptions": {
"device_disconnected": {
"message": "The device is not plugged in"

View File

@@ -0,0 +1,57 @@
"""Home Assistant SkyConnect switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantSkyConnectConfigEntry
from .const import DOMAIN, PRODUCT, SERIAL_NUMBER, HardwareVariant
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantSkyConnectConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant SkyConnect."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant SkyConnect beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantSkyConnectConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
variant = HardwareVariant.from_usb_product_name(
self._config_entry.data[PRODUCT]
)
serial_number = self._config_entry.data[SERIAL_NUMBER]
self._attr_unique_id = f"{serial_number}_beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, serial_number)},
name=f"{variant.full_name} ({serial_number[:8]})",
model=variant.full_name,
manufacturer="Nabu Casa",
serial_number=serial_number,
)

View File

@@ -4,8 +4,6 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -18,19 +16,17 @@ from homeassistant.components.homeassistant_hardware.util import (
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 device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantSkyConnectConfigEntry
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
NABU_CASA_FIRMWARE_RELEASES_URL,
PRODUCT,
SERIAL_NUMBER,
HardwareVariant,
@@ -102,8 +98,7 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
config_entry: HomeAssistantSkyConnectConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -122,12 +117,7 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=config_entry.data["device"],
config_entry=config_entry,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
update_coordinator=config_entry.runtime_data.coordinator,
entity_description=entity_description,
)
@@ -137,11 +127,7 @@ def _async_create_update_entity(
"""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
)
]
[_async_create_update_entity(hass, config_entry, async_add_entities)]
)
entity.async_on_remove(
@@ -153,14 +139,11 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: HomeAssistantSkyConnectConfigEntry,
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
)
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
async_add_entities([entity])
@@ -174,7 +157,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: ConfigEntry,
config_entry: HomeAssistantSkyConnectConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

View File

@@ -2,9 +2,13 @@
from __future__ import annotations
from dataclasses import dataclass
import logging
from homeassistant.components.hassio import get_os_info
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.silabs_multiprotocol_addon import (
check_multi_pan_addon,
)
@@ -16,14 +20,34 @@ from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady, HomeAssistantError
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.hassio import is_hassio
from .const import FIRMWARE, FIRMWARE_VERSION, RADIO_DEVICE, ZHA_HW_DISCOVERY_DATA
from .const import (
FIRMWARE,
FIRMWARE_VERSION,
NABU_CASA_FIRMWARE_RELEASES_URL,
RADIO_DEVICE,
ZHA_HW_DISCOVERY_DATA,
)
_LOGGER = logging.getLogger(__name__)
type HomeAssistantYellowConfigEntry = ConfigEntry[HomeAssistantYellowData]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@dataclass
class HomeAssistantYellowData:
"""Runtime data definition."""
coordinator: (
FirmwareUpdateCoordinator # Type from homeassistant_hardware.coordinator
)
async def async_setup_entry(
hass: HomeAssistant, entry: HomeAssistantYellowConfigEntry
) -> bool:
"""Set up a Home Assistant Yellow config entry."""
if not is_hassio(hass):
# Not running under supervisor, Home Assistant may have been migrated
@@ -56,18 +80,31 @@ 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"])
# Create and store the firmware update coordinator in runtime_data
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
)
entry.runtime_data = HomeAssistantYellowData(coordinator)
await hass.config_entries.async_forward_entry_setups(entry, ["switch", "update"])
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(
hass: HomeAssistant, entry: HomeAssistantYellowConfigEntry
) -> bool:
"""Unload a config entry."""
await hass.config_entries.async_unload_platforms(entry, ["update"])
return True
return await hass.config_entries.async_unload_platforms(entry, ["switch", "update"])
async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
async def async_migrate_entry(
hass: HomeAssistant, config_entry: HomeAssistantYellowConfigEntry
) -> bool:
"""Migrate old entry."""
_LOGGER.debug(

View File

@@ -22,5 +22,5 @@ FIRMWARE_VERSION = "firmware_version"
ZHA_DOMAIN = "zha"
NABU_CASA_FIRMWARE_RELEASES_URL = (
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases/latest"
"https://api.github.com/repos/NabuCasa/silabs-firmware-builder/releases"
)

View File

@@ -0,0 +1,9 @@
{
"entity": {
"switch": {
"beta_firmware": {
"default": "mdi:test-tube"
}
}
}
}

View File

@@ -1,5 +1,10 @@
{
"entity": {
"switch": {
"beta_firmware": {
"name": "Radio beta firmware updates"
}
},
"update": {
"radio_firmware": {
"name": "Radio firmware"

View File

@@ -0,0 +1,50 @@
"""Home Assistant Yellow switch entities."""
from __future__ import annotations
import logging
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import HomeAssistantYellowConfigEntry
from .const import DOMAIN, MANUFACTURER, MODEL
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: HomeAssistantYellowConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the switch platform for Home Assistant Yellow."""
async_add_entities(
[BetaFirmwareSwitch(config_entry.runtime_data.coordinator, config_entry)]
)
class BetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Home Assistant Yellow beta firmware switch."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: HomeAssistantYellowConfigEntry,
) -> None:
"""Initialize the beta firmware switch."""
super().__init__(coordinator, config_entry)
self._attr_unique_id = "beta_firmware"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, "yellow")},
name=MODEL,
model=MODEL,
manufacturer=MANUFACTURER,
)

View File

@@ -4,8 +4,6 @@ from __future__ import annotations
import logging
import aiohttp
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
@@ -19,23 +17,14 @@ from homeassistant.components.homeassistant_hardware.util import (
ResetTarget,
)
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 device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import (
DOMAIN,
FIRMWARE,
FIRMWARE_VERSION,
MANUFACTURER,
MODEL,
NABU_CASA_FIRMWARE_RELEASES_URL,
RADIO_DEVICE,
)
from . import HomeAssistantYellowConfigEntry
from .const import DOMAIN, FIRMWARE, FIRMWARE_VERSION, MANUFACTURER, MODEL, RADIO_DEVICE
_LOGGER = logging.getLogger(__name__)
@@ -108,8 +97,7 @@ FIRMWARE_ENTITY_DESCRIPTIONS: dict[
def _async_create_update_entity(
hass: HomeAssistant,
config_entry: ConfigEntry,
session: aiohttp.ClientSession,
config_entry: HomeAssistantYellowConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> FirmwareUpdateEntity:
"""Create an update entity that handles firmware type changes."""
@@ -128,12 +116,7 @@ def _async_create_update_entity(
entity = FirmwareUpdateEntity(
device=RADIO_DEVICE,
config_entry=config_entry,
update_coordinator=FirmwareUpdateCoordinator(
hass,
config_entry,
session,
NABU_CASA_FIRMWARE_RELEASES_URL,
),
update_coordinator=config_entry.runtime_data.coordinator,
entity_description=entity_description,
)
@@ -143,11 +126,7 @@ def _async_create_update_entity(
"""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
)
]
[_async_create_update_entity(hass, config_entry, async_add_entities)]
)
entity.async_on_remove(
@@ -159,14 +138,11 @@ def _async_create_update_entity(
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
config_entry: HomeAssistantYellowConfigEntry,
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
)
entity = _async_create_update_entity(hass, config_entry, async_add_entities)
async_add_entities([entity])
@@ -179,7 +155,7 @@ class FirmwareUpdateEntity(BaseFirmwareUpdateEntity):
def __init__(
self,
device: str,
config_entry: ConfigEntry,
config_entry: HomeAssistantYellowConfigEntry,
update_coordinator: FirmwareUpdateCoordinator,
entity_description: FirmwareUpdateEntityDescription,
) -> None:

2
requirements_all.txt generated
View File

@@ -1151,7 +1151,7 @@ ha-iotawattpy==0.1.2
ha-philipsjs==3.2.4
# homeassistant.components.homeassistant_hardware
ha-silabs-firmware-client==0.2.0
ha-silabs-firmware-client==0.3.0
# homeassistant.components.habitica
habiticalib==0.4.6

View File

@@ -1012,7 +1012,7 @@ ha-iotawattpy==0.1.2
ha-philipsjs==3.2.4
# homeassistant.components.homeassistant_hardware
ha-silabs-firmware-client==0.2.0
ha-silabs-firmware-client==0.3.0
# homeassistant.components.habitica
habiticalib==0.4.6

View File

@@ -0,0 +1,136 @@
"""Test Connect ZBT-2 beta firmware switch entity."""
from unittest.mock import Mock, call, patch
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
import pytest
from yarl import URL
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .common import USB_DATA_ZBT2
from tests.common import MockConfigEntry, mock_restore_cache
SWITCH_ENTITY_ID = (
"switch.home_assistant_connect_zbt_2_80b54eefae18_beta_firmware_updates"
)
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"),
),
),
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("service", "target_state", "expected_prerelease"),
[
(SERVICE_TURN_ON, STATE_ON, True),
(SERVICE_TURN_OFF, STATE_OFF, False),
],
)
async def test_switch_turn_on_off(
hass: HomeAssistant,
service: str,
target_state: str,
expected_prerelease: bool,
) -> None:
"""Test turning switch on/off updates state and coordinator."""
await async_setup_component(hass, "homeassistant", {})
# Start with opposite state
mock_restore_cache(
hass,
[
State(
SWITCH_ENTITY_ID,
STATE_ON if service == SERVICE_TURN_OFF else STATE_OFF,
)
],
)
# Set up the ZBT-2 integration
zbt2_config_entry = MockConfigEntry(
title="Home Assistant Connect ZBT-2",
domain="homeassistant_connect_zbt2",
data={
"firmware": "ezsp",
"firmware_version": "7.3.1.0 build 0",
"device": USB_DATA_ZBT2.device,
"manufacturer": USB_DATA_ZBT2.manufacturer,
"pid": USB_DATA_ZBT2.pid,
"product": USB_DATA_ZBT2.description,
"serial_number": USB_DATA_ZBT2.serial_number,
"vid": USB_DATA_ZBT2.vid,
},
version=1,
minor_version=1,
)
zbt2_config_entry.add_to_hass(hass)
with (
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateCoordinator.async_refresh"
) as mock_refresh,
):
mock_client.return_value.async_update_data.return_value = TEST_MANIFEST
mock_client.return_value.update_prerelease = Mock()
assert await hass.config_entries.async_setup(zbt2_config_entry.entry_id)
await hass.async_block_till_done()
# Reset mocks after setup
mock_client.return_value.update_prerelease.reset_mock()
mock_refresh.reset_mock()
# Call the service
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
blocking=True,
)
# Verify state changed
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == target_state
# Verify coordinator methods were called
assert mock_client.return_value.update_prerelease.mock_calls == [
call(expected_prerelease)
]
assert len(mock_refresh.mock_calls) == 1

View File

@@ -0,0 +1,31 @@
"""Common test constants for homeassistant_hardware tests."""
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
from yarl import URL
from homeassistant.util import dt as dt_util
TEST_DOMAIN = "test"
TEST_FIRMWARE_RELEASES_URL = "https://example.org/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"),
),
),
)

View File

@@ -0,0 +1,271 @@
"""Test Home Assistant Hardware beta firmware switch entity."""
from __future__ import annotations
from collections.abc import AsyncGenerator
from unittest.mock import Mock, call, patch
import pytest
from homeassistant.components.homeassistant_hardware.coordinator import (
FirmwareUpdateCoordinator,
)
from homeassistant.components.homeassistant_hardware.switch import (
BaseBetaFirmwareSwitch,
)
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import ConfigEntry, ConfigFlow
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
EntityCategory,
Platform,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.setup import async_setup_component
from .common import TEST_DOMAIN, TEST_FIRMWARE_RELEASES_URL, TEST_MANIFEST
from tests.common import (
MockConfigEntry,
MockModule,
MockPlatform,
mock_config_flow,
mock_integration,
mock_platform,
mock_restore_cache,
)
TEST_DEVICE = "/dev/serial/by-id/test-device-12345"
TEST_SWITCH_ENTITY_ID = "switch.mock_device_beta_firmware_updates"
class MockBetaFirmwareSwitch(BaseBetaFirmwareSwitch):
"""Mock beta firmware switch for testing."""
def __init__(
self,
coordinator: FirmwareUpdateCoordinator,
config_entry: ConfigEntry,
) -> None:
"""Initialize the mock beta firmware switch."""
super().__init__(coordinator, config_entry)
self._attr_unique_id = "beta_firmware"
self._attr_name = "Beta firmware updates"
self._attr_device_info = DeviceInfo(
identifiers={(TEST_DOMAIN, "test_device")},
name="Mock Device",
model="Mock Model",
manufacturer="Mock Manufacturer",
)
def _mock_async_create_switch_entity(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> MockBetaFirmwareSwitch:
"""Create a mock switch entity."""
session = async_get_clientsession(hass)
coordinator = FirmwareUpdateCoordinator(
hass,
config_entry,
session,
TEST_FIRMWARE_RELEASES_URL,
)
entity = MockBetaFirmwareSwitch(coordinator, config_entry)
async_add_entities([entity])
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, [Platform.SWITCH]
)
return True
async def mock_async_setup_switch_entities(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the beta firmware switch config entry."""
_mock_async_create_switch_entity(hass, config_entry, async_add_entities)
@pytest.fixture(name="mock_firmware_client")
def mock_firmware_client_fixture():
"""Create a mock firmware update client."""
with patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient",
autospec=True,
) as mock_client:
mock_client.return_value.async_update_data.return_value = TEST_MANIFEST
mock_client.return_value.update_prerelease = Mock()
yield mock_client.return_value
@pytest.fixture(name="switch_config_entry")
async def mock_switch_config_entry(
hass: HomeAssistant,
mock_firmware_client,
) -> AsyncGenerator[ConfigEntry]:
"""Set up a mock config entry for testing."""
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.switch",
MockPlatform(async_setup_entry=mock_async_setup_switch_entities),
)
# Set up a mock integration using the hardware switch entity
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"device": TEST_DEVICE,
},
)
config_entry.add_to_hass(hass)
with mock_config_flow(TEST_DOMAIN, ConfigFlow):
yield config_entry
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_switch_default_off_state(
hass: HomeAssistant,
switch_config_entry: ConfigEntry,
mock_firmware_client,
) -> None:
"""Test switch defaults to off when no previous state."""
assert await hass.config_entries.async_setup(switch_config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(TEST_SWITCH_ENTITY_ID)
assert state is not None
assert state.state == STATE_OFF
# Verify coordinator was called with False during setup
assert mock_firmware_client.update_prerelease.mock_calls == [call(False)]
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("initial_state", "expected_state", "expected_prerelease"),
[
(STATE_ON, STATE_ON, True),
(STATE_OFF, STATE_OFF, False),
],
)
async def test_switch_restore_state(
hass: HomeAssistant,
switch_config_entry: ConfigEntry,
mock_firmware_client,
initial_state: str,
expected_state: str,
expected_prerelease: bool,
) -> None:
"""Test switch restores previous state and has correct entity attributes."""
mock_restore_cache(hass, [State(TEST_SWITCH_ENTITY_ID, initial_state)])
assert await hass.config_entries.async_setup(switch_config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(TEST_SWITCH_ENTITY_ID)
assert state is not None
assert state.state == expected_state
assert state.attributes.get("friendly_name") == "Mock Device Beta firmware updates"
# Verify coordinator was called with correct value during setup
assert mock_firmware_client.update_prerelease.mock_calls == [
call(expected_prerelease)
]
# Verify entity registry attributes
entity_registry = er.async_get(hass)
entity_entry = entity_registry.async_get(TEST_SWITCH_ENTITY_ID)
assert entity_entry is not None
assert entity_entry.entity_category == EntityCategory.CONFIG
assert entity_entry.translation_key == "beta_firmware"
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("service", "target_state", "expected_prerelease"),
[
(SERVICE_TURN_ON, STATE_ON, True),
(SERVICE_TURN_OFF, STATE_OFF, False),
],
)
async def test_switch_turn_on_off(
hass: HomeAssistant,
switch_config_entry: ConfigEntry,
mock_firmware_client,
service: str,
target_state: str,
expected_prerelease: bool,
) -> None:
"""Test turning switch on/off updates state and coordinator."""
# Start with opposite state
mock_restore_cache(
hass,
[
State(
TEST_SWITCH_ENTITY_ID,
STATE_ON if service == SERVICE_TURN_OFF else STATE_OFF,
)
],
)
# Track async_refresh calls
with patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateCoordinator.async_refresh"
) as mock_refresh:
assert await hass.config_entries.async_setup(switch_config_entry.entry_id)
await hass.async_block_till_done()
# Reset mocks after setup
mock_firmware_client.update_prerelease.reset_mock()
mock_refresh.reset_mock()
# Call the service
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: TEST_SWITCH_ENTITY_ID},
blocking=True,
)
# Verify state changed
state = hass.states.get(TEST_SWITCH_ENTITY_ID)
assert state is not None
assert state.state == target_state
# Verify coordinator methods were called
assert mock_firmware_client.update_prerelease.mock_calls == [
call(expected_prerelease)
]
assert len(mock_refresh.mock_calls) == 1

View File

@@ -9,9 +9,7 @@ import logging
from unittest.mock import 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,
@@ -47,7 +45,8 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .common import TEST_DOMAIN, TEST_FIRMWARE_RELEASES_URL, TEST_MANIFEST
from tests.common import (
MockConfigEntry,
@@ -60,32 +59,8 @@ from tests.common import (
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.mock_name_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[

View File

@@ -0,0 +1,134 @@
"""Test SkyConnect beta firmware switch entity."""
from unittest.mock import Mock, call, patch
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
import pytest
from yarl import URL
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from .common import USB_DATA_ZBT1
from tests.common import MockConfigEntry, mock_restore_cache
SWITCH_ENTITY_ID = "switch.home_assistant_connect_zbt_1_9e2adbd7_beta_firmware_updates"
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"),
),
),
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("service", "target_state", "expected_prerelease"),
[
(SERVICE_TURN_ON, STATE_ON, True),
(SERVICE_TURN_OFF, STATE_OFF, False),
],
)
async def test_switch_turn_on_off(
hass: HomeAssistant,
service: str,
target_state: str,
expected_prerelease: bool,
) -> None:
"""Test turning switch on/off updates state and coordinator."""
await async_setup_component(hass, "homeassistant", {})
# Start with opposite state
mock_restore_cache(
hass,
[
State(
SWITCH_ENTITY_ID,
STATE_ON if service == SERVICE_TURN_OFF else STATE_OFF,
)
],
)
# Set up the ZBT-1 integration
zbt1_config_entry = MockConfigEntry(
title="Home Assistant Connect ZBT-1",
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)
with (
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateCoordinator.async_refresh"
) as mock_refresh,
):
mock_client.return_value.async_update_data.return_value = TEST_MANIFEST
mock_client.return_value.update_prerelease = Mock()
assert await hass.config_entries.async_setup(zbt1_config_entry.entry_id)
await hass.async_block_till_done()
# Reset mocks after setup
mock_client.return_value.update_prerelease.reset_mock()
mock_refresh.reset_mock()
# Call the service
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
blocking=True,
)
# Verify state changed
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == target_state
# Verify coordinator methods were called
assert mock_client.return_value.update_prerelease.mock_calls == [
call(expected_prerelease)
]
assert len(mock_refresh.mock_calls) == 1

View File

@@ -0,0 +1,135 @@
"""Test Yellow beta firmware switch entity."""
from unittest.mock import Mock, call, patch
from ha_silabs_firmware_client import FirmwareManifest, FirmwareMetadata
import pytest
from yarl import URL
from homeassistant.components.homeassistant_yellow.const import RADIO_DEVICE
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from tests.common import MockConfigEntry, mock_restore_cache
SWITCH_ENTITY_ID = "switch.home_assistant_yellow_radio_beta_firmware_updates"
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"),
),
),
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.parametrize(
("service", "target_state", "expected_prerelease"),
[
(SERVICE_TURN_ON, STATE_ON, True),
(SERVICE_TURN_OFF, STATE_OFF, False),
],
)
async def test_switch_turn_on_off(
hass: HomeAssistant,
service: str,
target_state: str,
expected_prerelease: bool,
) -> None:
"""Test turning switch on/off updates state and coordinator."""
await async_setup_component(hass, "homeassistant", {})
# Start with opposite state
mock_restore_cache(
hass,
[
State(
SWITCH_ENTITY_ID,
STATE_ON if service == SERVICE_TURN_OFF else STATE_OFF,
)
],
)
# Set up the Yellow integration
yellow_config_entry = MockConfigEntry(
title="Home Assistant Yellow",
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"},
),
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateClient",
autospec=True,
) as mock_client,
patch(
"homeassistant.components.homeassistant_hardware.coordinator.FirmwareUpdateCoordinator.async_refresh"
) as mock_refresh,
):
mock_client.return_value.async_update_data.return_value = TEST_MANIFEST
mock_client.return_value.update_prerelease = Mock()
assert await hass.config_entries.async_setup(yellow_config_entry.entry_id)
await hass.async_block_till_done()
# Reset mocks after setup
mock_client.return_value.update_prerelease.reset_mock()
mock_refresh.reset_mock()
# Call the service
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: SWITCH_ENTITY_ID},
blocking=True,
)
# Verify state changed
state = hass.states.get(SWITCH_ENTITY_ID)
assert state is not None
assert state.state == target_state
# Verify coordinator methods were called
assert mock_client.return_value.update_prerelease.mock_calls == [
call(expected_prerelease)
]
assert len(mock_refresh.mock_calls) == 1