diff --git a/homeassistant/components/homeassistant_hardware/__init__.py b/homeassistant/components/homeassistant_hardware/__init__.py index c33dabe1ec86c7..fc2b393805e7e5 100644 --- a/homeassistant/components/homeassistant_hardware/__init__.py +++ b/homeassistant/components/homeassistant_hardware/__init__.py @@ -6,10 +6,15 @@ from homeassistant.helpers import config_validation as cv from homeassistant.helpers.typing import ConfigType -DOMAIN = "homeassistant_hardware" +from .const import DATA_COMPONENT, DOMAIN +from .helpers import HardwareInfoDispatcher + CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN) async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the component.""" + + hass.data[DATA_COMPONENT] = HardwareInfoDispatcher(hass) + return True diff --git a/homeassistant/components/homeassistant_hardware/const.py b/homeassistant/components/homeassistant_hardware/const.py index 8fddbe41b7d65c..a3c091ff7eed51 100644 --- a/homeassistant/components/homeassistant_hardware/const.py +++ b/homeassistant/components/homeassistant_hardware/const.py @@ -1,10 +1,23 @@ """Constants for the Homeassistant Hardware integration.""" +from __future__ import annotations + import logging +from typing import TYPE_CHECKING + +from homeassistant.util.hass_dict import HassKey + +if TYPE_CHECKING: + from .helpers import HardwareInfoDispatcher + LOGGER = logging.getLogger(__package__) +DOMAIN = "homeassistant_hardware" +DATA_COMPONENT: HassKey[HardwareInfoDispatcher] = HassKey(DOMAIN) + ZHA_DOMAIN = "zha" +OTBR_DOMAIN = "otbr" OTBR_ADDON_NAME = "OpenThread Border Router" OTBR_ADDON_MANAGER_DATA = "openthread_border_router" diff --git a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py index fac3d2d973565b..8d7a302e7864a9 100644 --- a/homeassistant/components/homeassistant_hardware/firmware_config_flow.py +++ b/homeassistant/components/homeassistant_hardware/firmware_config_flow.py @@ -25,12 +25,14 @@ from homeassistant.helpers.hassio import is_hassio from . import silabs_multiprotocol_addon -from .const import ZHA_DOMAIN +from .const import OTBR_DOMAIN, ZHA_DOMAIN from .util import ( ApplicationType, + OwningAddon, + OwningIntegration, get_otbr_addon_manager, - get_zha_device_path, get_zigbee_flasher_addon_manager, + guess_hardware_owners, probe_silabs_firmware_type, ) @@ -519,19 +521,15 @@ async def async_step_pick_firmware_zigbee( ) -> ConfigFlowResult: """Pick Zigbee firmware.""" assert self._device is not None + owners = await guess_hardware_owners(self.hass, self._device) - if is_hassio(self.hass): - otbr_manager = get_otbr_addon_manager(self.hass) - otbr_addon_info = await self._async_get_addon_info(otbr_manager) - - if ( - otbr_addon_info.state != AddonState.NOT_INSTALLED - and otbr_addon_info.options.get("device") == self._device - ): - raise AbortFlow( - "otbr_still_using_stick", - description_placeholders=self._get_translation_placeholders(), - ) + for info in owners: + for owner in info.owners: + if info.source == OTBR_DOMAIN and isinstance(owner, OwningAddon): + raise AbortFlow( + "otbr_still_using_stick", + description_placeholders=self._get_translation_placeholders(), + ) return await super().async_step_pick_firmware_zigbee(user_input) @@ -541,15 +539,14 @@ async def async_step_pick_firmware_thread( """Pick Thread firmware.""" assert self._device is not None - for zha_entry in self.hass.config_entries.async_entries( - ZHA_DOMAIN, - include_ignore=False, - include_disabled=True, - ): - if get_zha_device_path(zha_entry) == self._device: - raise AbortFlow( - "zha_still_using_stick", - description_placeholders=self._get_translation_placeholders(), - ) + owners = await guess_hardware_owners(self.hass, self._device) + + for info in owners: + for owner in info.owners: + if info.source == ZHA_DOMAIN and isinstance(owner, OwningIntegration): + raise AbortFlow( + "zha_still_using_stick", + description_placeholders=self._get_translation_placeholders(), + ) return await super().async_step_pick_firmware_thread(user_input) diff --git a/homeassistant/components/homeassistant_hardware/helpers.py b/homeassistant/components/homeassistant_hardware/helpers.py new file mode 100644 index 00000000000000..a9b3703ee4a23d --- /dev/null +++ b/homeassistant/components/homeassistant_hardware/helpers.py @@ -0,0 +1,143 @@ +"""Home Assistant Hardware integration helpers.""" + +from collections import defaultdict +from collections.abc import AsyncIterator, Awaitable, Callable +import logging +from typing import Protocol + +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback + +from . import DATA_COMPONENT +from .util import FirmwareInfo + +_LOGGER = logging.getLogger(__name__) + + +class SyncHardwareFirmwareInfoModule(Protocol): + """Protocol type for Home Assistant Hardware firmware info platform modules.""" + + def get_firmware_info( + self, + hass: HomeAssistant, + entry: ConfigEntry, + ) -> FirmwareInfo | None: + """Return radio firmware information for the config entry, synchronously.""" + + +class AsyncHardwareFirmwareInfoModule(Protocol): + """Protocol type for Home Assistant Hardware firmware info platform modules.""" + + async def async_get_firmware_info( + self, + hass: HomeAssistant, + entry: ConfigEntry, + ) -> FirmwareInfo | None: + """Return radio firmware information for the config entry, asynchronously.""" + + +type HardwareFirmwareInfoModule = ( + SyncHardwareFirmwareInfoModule | AsyncHardwareFirmwareInfoModule +) + + +class HardwareInfoDispatcher: + """Central dispatcher for hardware/firmware information.""" + + def __init__(self, hass: HomeAssistant) -> None: + """Initialize the dispatcher.""" + self.hass = hass + self._providers: dict[str, HardwareFirmwareInfoModule] = {} + self._notification_callbacks: defaultdict[ + str, set[Callable[[FirmwareInfo], None]] + ] = defaultdict(set) + + def register_firmware_info_provider( + self, domain: str, platform: HardwareFirmwareInfoModule + ) -> None: + """Register a firmware info provider.""" + if domain in self._providers: + raise ValueError( + f"Domain {domain} is already registered as a firmware info provider" + ) + + # There is no need to handle "unregistration" because integrations cannot be + # wholly removed at runtime + self._providers[domain] = platform + _LOGGER.debug( + "Registered firmware info provider from domain %r: %s", domain, platform + ) + + def register_firmware_info_callback( + self, device: str, callback: Callable[[FirmwareInfo], None] + ) -> CALLBACK_TYPE: + """Register a firmware info notification callback.""" + self._notification_callbacks[device].add(callback) + + @hass_callback + def async_remove_callback() -> None: + self._notification_callbacks[device].discard(callback) + + return async_remove_callback + + async def notify_firmware_info( + self, domain: str, firmware_info: FirmwareInfo + ) -> None: + """Notify the dispatcher of new firmware information.""" + _LOGGER.debug( + "Received firmware info notification from %r: %s", domain, firmware_info + ) + + for callback in self._notification_callbacks.get(firmware_info.device, []): + try: + callback(firmware_info) + except Exception: + _LOGGER.exception( + "Error while notifying firmware info listener %s", callback + ) + + async def iter_firmware_info(self) -> AsyncIterator[FirmwareInfo]: + """Iterate over all firmware information for all hardware.""" + for domain, fw_info_module in self._providers.items(): + for config_entry in self.hass.config_entries.async_entries(domain): + try: + if hasattr(fw_info_module, "get_firmware_info"): + fw_info = fw_info_module.get_firmware_info( + self.hass, config_entry + ) + else: + fw_info = await fw_info_module.async_get_firmware_info( + self.hass, config_entry + ) + except Exception: + _LOGGER.exception( + "Error while getting firmware info from %r", fw_info_module + ) + continue + + if fw_info is not None: + yield fw_info + + +@hass_callback +def async_register_firmware_info_provider( + hass: HomeAssistant, domain: str, platform: HardwareFirmwareInfoModule +) -> None: + """Register a firmware info provider.""" + return hass.data[DATA_COMPONENT].register_firmware_info_provider(domain, platform) + + +@hass_callback +def async_register_firmware_info_callback( + hass: HomeAssistant, device: str, callback: Callable[[FirmwareInfo], None] +) -> CALLBACK_TYPE: + """Register a firmware info provider.""" + return hass.data[DATA_COMPONENT].register_firmware_info_callback(device, callback) + + +@hass_callback +def async_notify_firmware_info( + hass: HomeAssistant, domain: str, firmware_info: FirmwareInfo +) -> Awaitable[None]: + """Notify the dispatcher of new firmware information.""" + return hass.data[DATA_COMPONENT].notify_firmware_info(domain, firmware_info) diff --git a/homeassistant/components/homeassistant_hardware/util.py b/homeassistant/components/homeassistant_hardware/util.py index 3fd5bc60037286..53cbcbae5d409e 100644 --- a/homeassistant/components/homeassistant_hardware/util.py +++ b/homeassistant/components/homeassistant_hardware/util.py @@ -2,27 +2,27 @@ from __future__ import annotations +import asyncio from collections import defaultdict from collections.abc import Iterable from dataclasses import dataclass from enum import StrEnum import logging -from typing import cast from universal_silabs_flasher.const import ApplicationType as FlasherApplicationType from universal_silabs_flasher.flasher import Flasher from homeassistant.components.hassio import AddonError, AddonState -from homeassistant.config_entries import ConfigEntry, ConfigEntryState +from homeassistant.config_entries import ConfigEntryState from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.hassio import is_hassio from homeassistant.helpers.singleton import singleton +from . import DATA_COMPONENT from .const import ( OTBR_ADDON_MANAGER_DATA, OTBR_ADDON_NAME, OTBR_ADDON_SLUG, - ZHA_DOMAIN, ZIGBEE_FLASHER_ADDON_MANAGER_DATA, ZIGBEE_FLASHER_ADDON_NAME, ZIGBEE_FLASHER_ADDON_SLUG, @@ -55,11 +55,6 @@ def as_flasher_application_type(self) -> FlasherApplicationType: return FlasherApplicationType(self.value) -def get_zha_device_path(config_entry: ConfigEntry) -> str | None: - """Get the device path from a ZHA config entry.""" - return cast(str | None, config_entry.data.get("device", {}).get("path", None)) - - @singleton(OTBR_ADDON_MANAGER_DATA) @callback def get_otbr_addon_manager(hass: HomeAssistant) -> WaitingAddonManager: @@ -84,31 +79,80 @@ def get_zigbee_flasher_addon_manager(hass: HomeAssistant) -> WaitingAddonManager ) -@dataclass(slots=True, kw_only=True) -class FirmwareGuess: +@dataclass(kw_only=True) +class OwningAddon: + """Owning add-on.""" + + slug: str + + def _get_addon_manager(self, hass: HomeAssistant) -> WaitingAddonManager: + return WaitingAddonManager( + hass, + _LOGGER, + f"Add-on {self.slug}", + self.slug, + ) + + async def is_running(self, hass: HomeAssistant) -> bool: + """Check if the add-on is running.""" + addon_manager = self._get_addon_manager(hass) + + try: + addon_info = await addon_manager.async_get_addon_info() + except AddonError: + return False + else: + return addon_info.state == AddonState.RUNNING + + +@dataclass(kw_only=True) +class OwningIntegration: + """Owning integration.""" + + config_entry_id: str + + async def is_running(self, hass: HomeAssistant) -> bool: + """Check if the integration is running.""" + if (entry := hass.config_entries.async_get_entry(self.config_entry_id)) is None: + return False + + return entry.state in ( + ConfigEntryState.LOADED, + ConfigEntryState.SETUP_RETRY, + ConfigEntryState.SETUP_IN_PROGRESS, + ) + + +@dataclass(kw_only=True) +class FirmwareInfo: """Firmware guess.""" - is_running: bool + device: str firmware_type: ApplicationType + firmware_version: str | None + source: str + owners: list[OwningAddon | OwningIntegration] + async def is_running(self, hass: HomeAssistant) -> bool: + """Check if the firmware owner is running.""" + states = await asyncio.gather(*(o.is_running(hass) for o in self.owners)) + if not states: + return False -async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> FirmwareGuess: - """Guess the firmware type based on installed addons and other integrations.""" - device_guesses: defaultdict[str | None, list[FirmwareGuess]] = defaultdict(list) + return all(states) - for zha_config_entry in hass.config_entries.async_entries(ZHA_DOMAIN): - zha_path = get_zha_device_path(zha_config_entry) - if zha_path is not None: - device_guesses[zha_path].append( - FirmwareGuess( - is_running=(zha_config_entry.state == ConfigEntryState.LOADED), - firmware_type=ApplicationType.EZSP, - source="zha", - ) - ) +async def guess_hardware_owners( + hass: HomeAssistant, device_path: str +) -> list[FirmwareInfo]: + """Guess the firmware info based on installed addons and other integrations.""" + device_guesses: defaultdict[str, list[FirmwareInfo]] = defaultdict(list) + async for firmware_info in hass.data[DATA_COMPONENT].iter_firmware_info(): + device_guesses[firmware_info.device].append(firmware_info) + + # It may be possible for the OTBR addon to be present without the integration if is_hassio(hass): otbr_addon_manager = get_otbr_addon_manager(hass) @@ -119,14 +163,22 @@ async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> Firmware else: if otbr_addon_info.state != AddonState.NOT_INSTALLED: otbr_path = otbr_addon_info.options.get("device") - device_guesses[otbr_path].append( - FirmwareGuess( - is_running=(otbr_addon_info.state == AddonState.RUNNING), - firmware_type=ApplicationType.SPINEL, - source="otbr", + + # Only create a new entry if there are no existing OTBR ones + if otbr_path is not None and not any( + info.source == "otbr" for info in device_guesses[otbr_path] + ): + device_guesses[otbr_path].append( + FirmwareInfo( + device=otbr_path, + firmware_type=ApplicationType.SPINEL, + firmware_version=None, + source="otbr", + owners=[OwningAddon(slug=otbr_addon_manager.addon_slug)], + ) ) - ) + if is_hassio(hass): multipan_addon_manager = await get_multiprotocol_addon_manager(hass) try: @@ -136,30 +188,48 @@ async def guess_firmware_type(hass: HomeAssistant, device_path: str) -> Firmware else: if multipan_addon_info.state != AddonState.NOT_INSTALLED: multipan_path = multipan_addon_info.options.get("device") - device_guesses[multipan_path].append( - FirmwareGuess( - is_running=(multipan_addon_info.state == AddonState.RUNNING), - firmware_type=ApplicationType.CPC, - source="multiprotocol", + + if multipan_path is not None: + device_guesses[multipan_path].append( + FirmwareInfo( + device=multipan_path, + firmware_type=ApplicationType.CPC, + firmware_version=None, + source="multiprotocol", + owners=[ + OwningAddon(slug=multipan_addon_manager.addon_slug) + ], + ) ) - ) - # Fall back to EZSP if we can't guess the firmware type - if device_path not in device_guesses: - return FirmwareGuess( - is_running=False, firmware_type=ApplicationType.EZSP, source="unknown" - ) + return device_guesses.get(device_path, []) - # Prioritizes guesses that were pulled from a running addon or integration but keep - # the sort order we defined above - guesses = sorted( - device_guesses[device_path], - key=lambda guess: guess.is_running, - ) +async def guess_firmware_info(hass: HomeAssistant, device_path: str) -> FirmwareInfo: + """Guess the firmware type based on installed addons and other integrations.""" + + hardware_owners = await guess_hardware_owners(hass, device_path) + + # Fall back to EZSP if we have no way to guess + if not hardware_owners: + return FirmwareInfo( + device=device_path, + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="unknown", + owners=[], + ) + + # Prioritize guesses that are pulled from a real source + guesses = [ + (guess, sum([await owner.is_running(hass) for owner in guess.owners])) + for guess in hardware_owners + ] + guesses.sort(key=lambda p: p[1]) assert guesses - return guesses[-1] + # Pick the best one. We use a stable sort so ZHA < OTBR < multi-PAN + return guesses[-1][0] async def probe_silabs_firmware_type( diff --git a/homeassistant/components/homeassistant_sky_connect/__init__.py b/homeassistant/components/homeassistant_sky_connect/__init__.py index 43d42e4fa59e58..758f0c1e1ef33a 100644 --- a/homeassistant/components/homeassistant_sky_connect/__init__.py +++ b/homeassistant/components/homeassistant_sky_connect/__init__.py @@ -4,7 +4,7 @@ import logging -from homeassistant.components.homeassistant_hardware.util import guess_firmware_type +from homeassistant.components.homeassistant_hardware.util import guess_firmware_info from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant @@ -33,7 +33,7 @@ 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_type( + firmware_guess = await guess_firmware_info( hass, config_entry.data["device"] ) diff --git a/homeassistant/components/homeassistant_yellow/__init__.py b/homeassistant/components/homeassistant_yellow/__init__.py index dc34cc4cdc9592..b0837eeedbeba2 100644 --- a/homeassistant/components/homeassistant_yellow/__init__.py +++ b/homeassistant/components/homeassistant_yellow/__init__.py @@ -10,7 +10,7 @@ ) from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, - guess_firmware_type, + guess_firmware_info, ) from homeassistant.config_entries import SOURCE_HARDWARE, ConfigEntry from homeassistant.core import HomeAssistant @@ -75,7 +75,7 @@ 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_type(hass, RADIO_DEVICE) + firmware_guess = await guess_firmware_info(hass, RADIO_DEVICE) new_data = {**config_entry.data} new_data[FIRMWARE] = firmware_guess.firmware_type.value diff --git a/homeassistant/components/zha/__init__.py b/homeassistant/components/zha/__init__.py index 28f029b62d5d9b..e446f32cf08d94 100644 --- a/homeassistant/components/zha/__init__.py +++ b/homeassistant/components/zha/__init__.py @@ -12,6 +12,10 @@ from zigpy.config import CONF_DATABASE, CONF_DEVICE, CONF_DEVICE_PATH from zigpy.exceptions import NetworkSettingsInconsistent, TransientConnectionError +from homeassistant.components.homeassistant_hardware.helpers import ( + async_notify_firmware_info, + async_register_firmware_info_provider, +) from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( CONF_TYPE, @@ -25,7 +29,7 @@ from homeassistant.helpers.dispatcher import async_dispatcher_send from homeassistant.helpers.typing import ConfigType -from . import repairs, websocket_api +from . import homeassistant_hardware, repairs, websocket_api from .const import ( CONF_BAUDRATE, CONF_CUSTOM_QUIRKS_PATH, @@ -110,6 +114,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: ha_zha_data = HAZHAData(yaml_config=config.get(DOMAIN, {})) hass.data[DATA_ZHA] = ha_zha_data + async_register_firmware_info_provider(hass, DOMAIN, homeassistant_hardware) + return True @@ -218,6 +224,13 @@ def update_config(event: Event) -> None: hass.bus.async_listen(EVENT_CORE_CONFIG_UPDATE, update_config) ) + if fw_info := homeassistant_hardware.get_firmware_info(hass, config_entry): + await async_notify_firmware_info( + hass, + DOMAIN, + firmware_info=fw_info, + ) + await ha_zha_data.gateway_proxy.async_initialize_devices_and_entities() await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) async_dispatcher_send(hass, SIGNAL_ADD_ENTITIES) diff --git a/homeassistant/components/zha/homeassistant_hardware.py b/homeassistant/components/zha/homeassistant_hardware.py new file mode 100644 index 00000000000000..18057d3b64d200 --- /dev/null +++ b/homeassistant/components/zha/homeassistant_hardware.py @@ -0,0 +1,43 @@ +"""Home Assistant Hardware firmware utilities.""" + +from __future__ import annotations + +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, + OwningIntegration, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant, callback + +from .const import DOMAIN +from .helpers import get_zha_gateway + + +@callback +def get_firmware_info( + hass: HomeAssistant, config_entry: ConfigEntry +) -> FirmwareInfo | None: + """Return firmware information for the ZHA instance, synchronously.""" + + # We only support EZSP firmware for now + if config_entry.data.get("radio_type", None) != "ezsp": + return None + + if (device := config_entry.data.get("device", {}).get("path")) is None: + return None + + try: + gateway = get_zha_gateway(hass) + except ValueError: + firmware_version = None + else: + firmware_version = gateway.state.node_info.version + + return FirmwareInfo( + device=device, + firmware_type=ApplicationType.EZSP, + firmware_version=firmware_version, + source=DOMAIN, + owners=[OwningIntegration(config_entry_id=config_entry.entry_id)], + ) diff --git a/tests/components/homeassistant_hardware/test_config_flow.py b/tests/components/homeassistant_hardware/test_config_flow.py index 145087073af7c6..3696ea66c03a54 100644 --- a/tests/components/homeassistant_hardware/test_config_flow.py +++ b/tests/components/homeassistant_hardware/test_config_flow.py @@ -23,6 +23,7 @@ from homeassistant.config_entries import ConfigEntry, ConfigFlowResult, OptionsFlow from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResultType +from homeassistant.setup import async_setup_component from tests.common import ( MockConfigEntry, @@ -106,7 +107,7 @@ def _async_flow_finished(self) -> ConfigFlowResult: @pytest.fixture(autouse=True) -def mock_test_firmware_platform( +async def mock_test_firmware_platform( hass: HomeAssistant, ) -> Generator[None]: """Fixture for a test config flow.""" @@ -116,6 +117,8 @@ def mock_test_firmware_platform( mock_integration(hass, mock_module) mock_platform(hass, f"{TEST_DOMAIN}.config_flow") + await async_setup_component(hass, "homeassistant_hardware", {}) + with mock_config_flow(TEST_DOMAIN, FakeFirmwareConfigFlow): yield @@ -189,6 +192,10 @@ def mock_addon_info( "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_otbr_addon_manager", return_value=mock_otbr_manager, ), + patch( + "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager", + return_value=mock_otbr_manager, + ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.get_zigbee_flasher_addon_manager", return_value=mock_flasher_manager, @@ -197,6 +204,10 @@ def mock_addon_info( "homeassistant.components.homeassistant_hardware.firmware_config_flow.is_hassio", return_value=is_hassio, ), + patch( + "homeassistant.components.homeassistant_hardware.util.is_hassio", + return_value=is_hassio, + ), patch( "homeassistant.components.homeassistant_hardware.firmware_config_flow.probe_silabs_firmware_type", return_value=app_type, diff --git a/tests/components/homeassistant_hardware/test_config_flow_failures.py b/tests/components/homeassistant_hardware/test_config_flow_failures.py index f5375fb51ddd9f..c240d0198ca77b 100644 --- a/tests/components/homeassistant_hardware/test_config_flow_failures.py +++ b/tests/components/homeassistant_hardware/test_config_flow_failures.py @@ -1,6 +1,6 @@ """Test the Home Assistant hardware firmware config flow failure cases.""" -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch import pytest @@ -9,7 +9,11 @@ STEP_PICK_FIRMWARE_THREAD, STEP_PICK_FIRMWARE_ZIGBEE, ) -from homeassistant.components.homeassistant_hardware.util import ApplicationType +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, + OwningIntegration, +) from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResultType @@ -548,21 +552,28 @@ async def test_options_flow_zigbee_to_thread_zha_configured( assert await hass.config_entries.async_setup(config_entry.entry_id) - # Set up ZHA as well - zha_config_entry = MockConfigEntry( - domain="zha", - data={"device": {"path": TEST_DEVICE}}, - ) - zha_config_entry.add_to_hass(hass) - - # Confirm options flow - result = await hass.config_entries.options.async_init(config_entry.entry_id) + # Pretend ZHA is using the stick + with patch( + "homeassistant.components.homeassistant_hardware.firmware_config_flow.guess_hardware_owners", + return_value=[ + FirmwareInfo( + device=TEST_DEVICE, + firmware_type=ApplicationType.EZSP, + firmware_version="1.2.3.4", + source="zha", + owners=[OwningIntegration(config_entry_id="some_config_entry_id")], + ) + ], + ): + # Confirm options flow + result = await hass.config_entries.options.async_init(config_entry.entry_id) + + # Pick Thread + result = await hass.config_entries.options.async_configure( + result["flow_id"], + user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, + ) - # Pick Thread - result = await hass.config_entries.options.async_configure( - result["flow_id"], - user_input={"next_step_id": STEP_PICK_FIRMWARE_THREAD}, - ) assert result["type"] == FlowResultType.ABORT assert result["reason"] == "zha_still_using_stick" diff --git a/tests/components/homeassistant_hardware/test_helpers.py b/tests/components/homeassistant_hardware/test_helpers.py new file mode 100644 index 00000000000000..183995be7ce3d1 --- /dev/null +++ b/tests/components/homeassistant_hardware/test_helpers.py @@ -0,0 +1,185 @@ +"""Test hardware helpers.""" + +import logging +from unittest.mock import AsyncMock, MagicMock, Mock, call + +import pytest + +from homeassistant.components.homeassistant_hardware.const import DATA_COMPONENT +from homeassistant.components.homeassistant_hardware.helpers import ( + async_notify_firmware_info, + async_register_firmware_info_callback, + async_register_firmware_info_provider, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, +) +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry + +FIRMWARE_INFO_EZSP = FirmwareInfo( + device="/dev/serial/by-id/device1", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="zha", + owners=[AsyncMock(is_running=AsyncMock(return_value=True))], +) + +FIRMWARE_INFO_SPINEL = FirmwareInfo( + device="/dev/serial/by-id/device2", + firmware_type=ApplicationType.SPINEL, + firmware_version=None, + source="otbr", + owners=[AsyncMock(is_running=AsyncMock(return_value=True))], +) + + +async def test_dispatcher_registration(hass: HomeAssistant) -> None: + """Test HardwareInfoDispatcher registration.""" + + await async_setup_component(hass, "homeassistant_hardware", {}) + + # Mock provider 1 with a synchronous method to pull firmware info + provider1_config_entry = MockConfigEntry( + domain="zha", + unique_id="some_unique_id1", + data={}, + ) + provider1_config_entry.add_to_hass(hass) + provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + provider1_firmware = MagicMock(spec=["get_firmware_info"]) + provider1_firmware.get_firmware_info = MagicMock(return_value=FIRMWARE_INFO_EZSP) + async_register_firmware_info_provider(hass, "zha", provider1_firmware) + + # Mock provider 2 with an asynchronous method to pull firmware info + provider2_config_entry = MockConfigEntry( + domain="otbr", + unique_id="some_unique_id2", + data={}, + ) + provider2_config_entry.add_to_hass(hass) + provider2_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + provider2_firmware = MagicMock(spec=["async_get_firmware_info"]) + provider2_firmware.async_get_firmware_info = AsyncMock( + return_value=FIRMWARE_INFO_SPINEL + ) + async_register_firmware_info_provider(hass, "otbr", provider2_firmware) + + # Double registration won't work + with pytest.raises(ValueError, match="Domain zha is already registered"): + async_register_firmware_info_provider(hass, "zha", provider1_firmware) + + # We can iterate over the results + info = [i async for i in hass.data[DATA_COMPONENT].iter_firmware_info()] + assert info == [ + FIRMWARE_INFO_EZSP, + FIRMWARE_INFO_SPINEL, + ] + + callback1 = Mock() + cancel1 = async_register_firmware_info_callback( + hass, "/dev/serial/by-id/device1", callback1 + ) + + callback2 = Mock() + cancel2 = async_register_firmware_info_callback( + hass, "/dev/serial/by-id/device2", callback2 + ) + + # And receive notification callbacks + await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP) + await async_notify_firmware_info(hass, "otbr", firmware_info=FIRMWARE_INFO_SPINEL) + await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP) + cancel1() + await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP) + await async_notify_firmware_info(hass, "otbr", firmware_info=FIRMWARE_INFO_SPINEL) + cancel2() + + assert callback1.mock_calls == [ + call(FIRMWARE_INFO_EZSP), + call(FIRMWARE_INFO_EZSP), + ] + + assert callback2.mock_calls == [ + call(FIRMWARE_INFO_SPINEL), + call(FIRMWARE_INFO_SPINEL), + ] + + +async def test_dispatcher_iter_error_handling( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test HardwareInfoDispatcher ignoring errors from firmware info providers.""" + + await async_setup_component(hass, "homeassistant_hardware", {}) + + provider1_config_entry = MockConfigEntry( + domain="zha", + unique_id="some_unique_id1", + data={}, + ) + provider1_config_entry.add_to_hass(hass) + provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + provider1_firmware = MagicMock(spec=["get_firmware_info"]) + provider1_firmware.get_firmware_info = MagicMock(side_effect=Exception("Boom!")) + async_register_firmware_info_provider(hass, "zha", provider1_firmware) + + provider2_config_entry = MockConfigEntry( + domain="otbr", + unique_id="some_unique_id2", + data={}, + ) + provider2_config_entry.add_to_hass(hass) + provider2_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + provider2_firmware = MagicMock(spec=["async_get_firmware_info"]) + provider2_firmware.async_get_firmware_info = AsyncMock( + return_value=FIRMWARE_INFO_SPINEL + ) + async_register_firmware_info_provider(hass, "otbr", provider2_firmware) + + with caplog.at_level(logging.ERROR): + info = [i async for i in hass.data[DATA_COMPONENT].iter_firmware_info()] + + assert info == [FIRMWARE_INFO_SPINEL] + assert "Error while getting firmware info from" in caplog.text + + +async def test_dispatcher_callback_error_handling( + hass: HomeAssistant, caplog: pytest.LogCaptureFixture +) -> None: + """Test HardwareInfoDispatcher ignoring errors from firmware info callbacks.""" + + await async_setup_component(hass, "homeassistant_hardware", {}) + provider1_config_entry = MockConfigEntry( + domain="zha", + unique_id="some_unique_id1", + data={}, + ) + provider1_config_entry.add_to_hass(hass) + provider1_config_entry.mock_state(hass, ConfigEntryState.LOADED) + + provider1_firmware = MagicMock(spec=["get_firmware_info"]) + provider1_firmware.get_firmware_info = MagicMock(return_value=FIRMWARE_INFO_EZSP) + async_register_firmware_info_provider(hass, "zha", provider1_firmware) + + callback1 = Mock(side_effect=Exception("Some error")) + async_register_firmware_info_callback(hass, "/dev/serial/by-id/device1", callback1) + + callback2 = Mock() + async_register_firmware_info_callback(hass, "/dev/serial/by-id/device1", callback2) + + with caplog.at_level(logging.ERROR): + await async_notify_firmware_info(hass, "zha", firmware_info=FIRMWARE_INFO_EZSP) + + assert "Error while notifying firmware info listener" in caplog.text + + assert callback1.mock_calls == [call(FIRMWARE_INFO_EZSP)] + assert callback2.mock_calls == [call(FIRMWARE_INFO_EZSP)] diff --git a/tests/components/homeassistant_hardware/test_util.py b/tests/components/homeassistant_hardware/test_util.py index 3f019a0409cbef..047de3e452cf22 100644 --- a/tests/components/homeassistant_hardware/test_util.py +++ b/tests/components/homeassistant_hardware/test_util.py @@ -1,18 +1,21 @@ """Test hardware utilities.""" -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from homeassistant.components.hassio import AddonError, AddonInfo, AddonState +from homeassistant.components.homeassistant_hardware.helpers import ( + async_register_firmware_info_provider, +) from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, - FirmwareGuess, - FlasherApplicationType, - get_zha_device_path, - guess_firmware_type, - probe_silabs_firmware_type, + FirmwareInfo, + OwningAddon, + OwningIntegration, + guess_firmware_info, ) -from homeassistant.config_entries import ConfigEntryState +from homeassistant.config_entries import ConfigEntry, ConfigEntryState from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component from tests.common import MockConfigEntry @@ -21,7 +24,21 @@ unique_id="some_unique_id", data={ "device": { - "path": "socket://1.2.3.4:5678", + "path": "/dev/ttyUSB1", + "baudrate": 115200, + "flow_control": None, + }, + "radio_type": "ezsp", + }, + version=4, +) + +ZHA_CONFIG_ENTRY2 = MockConfigEntry( + domain="zha", + unique_id="some_other_unique_id", + data={ + "device": { + "path": "/dev/ttyUSB2", "baudrate": 115200, "flow_control": None, }, @@ -31,153 +48,202 @@ ) -def test_get_zha_device_path() -> None: - """Test extracting the ZHA device path from its config entry.""" - assert ( - get_zha_device_path(ZHA_CONFIG_ENTRY) == ZHA_CONFIG_ENTRY.data["device"]["path"] - ) +async def test_guess_firmware_info_unknown(hass: HomeAssistant) -> None: + """Test guessing the firmware type.""" + await async_setup_component(hass, "homeassistant_hardware", {}) -def test_get_zha_device_path_ignored_discovery() -> None: - """Test extracting the ZHA device path from an ignored ZHA discovery.""" - config_entry = MockConfigEntry( - domain="zha", - unique_id="some_unique_id", - data={}, - version=4, + assert (await guess_firmware_info(hass, "/dev/missing")) == FirmwareInfo( + device="/dev/missing", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="unknown", + owners=[], ) - assert get_zha_device_path(config_entry) is None +async def test_guess_firmware_info_integrations(hass: HomeAssistant) -> None: + """Test guessing the firmware via OTBR and ZHA.""" -async def test_guess_firmware_type_unknown(hass: HomeAssistant) -> None: - """Test guessing the firmware type.""" + await async_setup_component(hass, "homeassistant_hardware", {}) - assert (await guess_firmware_type(hass, "/dev/missing")) == FirmwareGuess( - is_running=False, firmware_type=ApplicationType.EZSP, source="unknown" - ) + # One instance of ZHA and two OTBRs + zha = MockConfigEntry(domain="zha", unique_id="some_unique_id_1") + zha.add_to_hass(hass) + otbr1 = MockConfigEntry(domain="otbr", unique_id="some_unique_id_2") + otbr1.add_to_hass(hass) -async def test_guess_firmware_type(hass: HomeAssistant) -> None: - """Test guessing the firmware.""" - path = ZHA_CONFIG_ENTRY.data["device"]["path"] + otbr2 = MockConfigEntry(domain="otbr", unique_id="some_unique_id_3") + otbr2.add_to_hass(hass) - ZHA_CONFIG_ENTRY.add_to_hass(hass) + # First ZHA is running with the stick + zha_firmware_info = FirmwareInfo( + device="/dev/serial/by-id/device1", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="zha", + owners=[AsyncMock(is_running=AsyncMock(return_value=True))], + ) - ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.NOT_LOADED) - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=False, firmware_type=ApplicationType.EZSP, source="zha" + # First OTBR: neither the addon or the integration are loaded + otbr_firmware_info1 = FirmwareInfo( + device="/dev/serial/by-id/device1", + firmware_type=ApplicationType.SPINEL, + firmware_version=None, + source="otbr", + owners=[ + AsyncMock(is_running=AsyncMock(return_value=False)), + AsyncMock(is_running=AsyncMock(return_value=False)), + ], ) - # When ZHA is running, we indicate as such when guessing - ZHA_CONFIG_ENTRY.mock_state(hass, ConfigEntryState.LOADED) - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.EZSP, source="zha" + # Second OTBR: fully running but is with an unrelated device + otbr_firmware_info2 = FirmwareInfo( + device="/dev/serial/by-id/device2", # An unrelated device + firmware_type=ApplicationType.SPINEL, + firmware_version=None, + source="otbr", + owners=[ + AsyncMock(is_running=AsyncMock(return_value=True)), + AsyncMock(is_running=AsyncMock(return_value=True)), + ], ) - mock_otbr_addon_manager = AsyncMock() - mock_multipan_addon_manager = AsyncMock() - - with ( - patch( - "homeassistant.components.homeassistant_hardware.util.is_hassio", - return_value=True, - ), - patch( - "homeassistant.components.homeassistant_hardware.util.get_otbr_addon_manager", - return_value=mock_otbr_addon_manager, - ), - patch( - "homeassistant.components.homeassistant_hardware.util.get_multiprotocol_addon_manager", - return_value=mock_multipan_addon_manager, - ), - ): - mock_otbr_addon_manager.async_get_addon_info.side_effect = AddonError() - mock_multipan_addon_manager.async_get_addon_info.side_effect = AddonError() - - # Hassio errors are ignored and we still go with ZHA - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.EZSP, source="zha" - ) + mock_zha_hardware_info = MagicMock(spec=["get_firmware_info"]) + mock_zha_hardware_info.get_firmware_info = MagicMock(return_value=zha_firmware_info) + async_register_firmware_info_provider(hass, "zha", mock_zha_hardware_info) + + async def mock_otbr_async_get_firmware_info( + hass: HomeAssistant, config_entry: ConfigEntry + ) -> FirmwareInfo | None: + return { + otbr1.entry_id: otbr_firmware_info1, + otbr2.entry_id: otbr_firmware_info2, + }.get(config_entry.entry_id) + + mock_otbr_hardware_info = MagicMock(spec=["async_get_firmware_info"]) + mock_otbr_hardware_info.async_get_firmware_info = AsyncMock( + side_effect=mock_otbr_async_get_firmware_info + ) + async_register_firmware_info_provider(hass, "otbr", mock_otbr_hardware_info) - mock_otbr_addon_manager.async_get_addon_info.side_effect = None - mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={"device": "/some/other/device"}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ) + # ZHA wins for the first stick, since it's actually running + assert ( + await guess_firmware_info(hass, "/dev/serial/by-id/device1") + ) == zha_firmware_info - # We will prefer ZHA, as it is running (and actually pointing to the device) - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.EZSP, source="zha" - ) + # Second stick is communicating exclusively with the second OTBR + assert ( + await guess_firmware_info(hass, "/dev/serial/by-id/device2") + ) == otbr_firmware_info2 - mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={"device": path}, - state=AddonState.NOT_RUNNING, - update_available=False, - version="1.0.0", - ) + # If we stop ZHA, OTBR will take priority + zha_firmware_info.owners[0].is_running.return_value = False + otbr_firmware_info1.owners[0].is_running.return_value = True + assert ( + await guess_firmware_info(hass, "/dev/serial/by-id/device1") + ) == otbr_firmware_info1 - # We will still prefer ZHA, as it is the one actually running - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.EZSP, source="zha" - ) - mock_otbr_addon_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={"device": path}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", - ) +async def test_owning_addon(hass: HomeAssistant) -> None: + """Test `OwningAddon`.""" - # Finally, ZHA loses out to OTBR - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.SPINEL, source="otbr" + owning_addon = OwningAddon(slug="some-addon-slug") + + # Explicitly running + with patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager" + ) as mock_manager: + mock_manager.return_value.async_get_addon_info = AsyncMock( + return_value=AddonInfo( + available=True, + hostname="core_some_addon_slug", + options={}, + state=AddonState.RUNNING, + update_available=False, + version="1.0.0", + ) ) + assert (await owning_addon.is_running(hass)) is True - mock_multipan_addon_manager.async_get_addon_info.side_effect = None - mock_multipan_addon_manager.async_get_addon_info.return_value = AddonInfo( - available=True, - hostname=None, - options={"device": path}, - state=AddonState.RUNNING, - update_available=False, - version="1.0.0", + # Explicitly not running + with patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager" + ) as mock_manager: + mock_manager.return_value.async_get_addon_info = AsyncMock( + return_value=AddonInfo( + available=True, + hostname="core_some_addon_slug", + options={}, + state=AddonState.NOT_RUNNING, + update_available=False, + version="1.0.0", + ) ) + assert (await owning_addon.is_running(hass)) is False - # Which will lose out to multi-PAN - assert (await guess_firmware_type(hass, path)) == FirmwareGuess( - is_running=True, firmware_type=ApplicationType.CPC, source="multiprotocol" + # Failed to get status + with patch( + "homeassistant.components.homeassistant_hardware.util.WaitingAddonManager" + ) as mock_manager: + mock_manager.return_value.async_get_addon_info = AsyncMock( + side_effect=AddonError() ) + assert (await owning_addon.is_running(hass)) is False -async def test_probe_silabs_firmware_type() -> None: - """Test probing Silabs firmware type.""" +async def test_owning_integration(hass: HomeAssistant) -> None: + """Test `OwningIntegration`.""" + config_entry = MockConfigEntry(domain="mock_domain", unique_id="some_unique_id") + config_entry.add_to_hass(hass) - with patch( - "homeassistant.components.homeassistant_hardware.util.Flasher.probe_app_type", - side_effect=RuntimeError, - ): - assert (await probe_silabs_firmware_type("/dev/ttyUSB0")) is None + owning_integration = OwningIntegration(config_entry_id=config_entry.entry_id) - with patch( - "homeassistant.components.homeassistant_hardware.util.Flasher.probe_app_type", - side_effect=lambda self: setattr(self, "app_type", FlasherApplicationType.EZSP), - autospec=True, - ) as mock_probe_app_type: - # The application type constant is converted back and forth transparently - result = await probe_silabs_firmware_type( - "/dev/ttyUSB0", probe_methods=[ApplicationType.EZSP] - ) - assert result is ApplicationType.EZSP + # Explicitly running + config_entry.mock_state(hass, ConfigEntryState.LOADED) + assert (await owning_integration.is_running(hass)) is True + + # Explicitly not running + config_entry.mock_state(hass, ConfigEntryState.NOT_LOADED) + assert (await owning_integration.is_running(hass)) is False + + # Missing config entry + owning_integration2 = OwningIntegration(config_entry_id="some_nonexistenct_id") + assert (await owning_integration2.is_running(hass)) is False + + +async def test_firmware_info(hass: HomeAssistant) -> None: + """Test `FirmwareInfo`.""" + + owner1 = AsyncMock() + owner2 = AsyncMock() + + firmware_info = FirmwareInfo( + device="/dev/ttyUSB1", + firmware_type=ApplicationType.EZSP, + firmware_version="1.0.0", + source="zha", + owners=[owner1, owner2], + ) + + # Both running + owner1.is_running.return_value = True + owner2.is_running.return_value = True + assert (await firmware_info.is_running(hass)) is True + + # Only one running + owner1.is_running.return_value = True + owner2.is_running.return_value = False + assert (await firmware_info.is_running(hass)) is False + + # No owners + firmware_info2 = FirmwareInfo( + device="/dev/ttyUSB1", + firmware_type=ApplicationType.EZSP, + firmware_version="1.0.0", + source="zha", + owners=[], + ) - flasher = mock_probe_app_type.mock_calls[0].args[0] - assert flasher._probe_methods == [FlasherApplicationType.EZSP] + assert (await firmware_info2.is_running(hass)) is False diff --git a/tests/components/homeassistant_sky_connect/test_init.py b/tests/components/homeassistant_sky_connect/test_init.py index 15eeb205537087..8e90039a4fcc5c 100644 --- a/tests/components/homeassistant_sky_connect/test_init.py +++ b/tests/components/homeassistant_sky_connect/test_init.py @@ -4,7 +4,7 @@ from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, - FirmwareGuess, + FirmwareInfo, ) from homeassistant.components.homeassistant_sky_connect.const import DOMAIN from homeassistant.core import HomeAssistant @@ -32,11 +32,13 @@ async def test_config_entry_migration_v2(hass: HomeAssistant) -> None: config_entry.add_to_hass(hass) with patch( - "homeassistant.components.homeassistant_sky_connect.guess_firmware_type", - return_value=FirmwareGuess( - is_running=True, + "homeassistant.components.homeassistant_sky_connect.guess_firmware_info", + return_value=FirmwareInfo( + device="/dev/serial/by-id/usb-Nabu_Casa_SkyConnect_v1.0_9e2adbd75b8beb119fe564a0f320645d-if00-port0", + firmware_version=None, firmware_type=ApplicationType.SPINEL, source="otbr", + owners=[], ), ): await hass.config_entries.async_setup(config_entry.entry_id) diff --git a/tests/components/homeassistant_yellow/test_init.py b/tests/components/homeassistant_yellow/test_init.py index 5d534dad1e7876..57d63c7441e2bf 100644 --- a/tests/components/homeassistant_yellow/test_init.py +++ b/tests/components/homeassistant_yellow/test_init.py @@ -8,7 +8,7 @@ from homeassistant.components.hassio import DOMAIN as HASSIO_DOMAIN from homeassistant.components.homeassistant_hardware.util import ( ApplicationType, - FirmwareGuess, + FirmwareInfo, ) from homeassistant.components.homeassistant_yellow.const import DOMAIN from homeassistant.config_entries import ConfigEntryState @@ -49,11 +49,13 @@ async def test_setup_entry( return_value=onboarded, ), patch( - "homeassistant.components.homeassistant_yellow.guess_firmware_type", - return_value=FirmwareGuess( # Nothing is setup - is_running=False, + "homeassistant.components.homeassistant_yellow.guess_firmware_info", + return_value=FirmwareInfo( # Nothing is setup + device="/dev/ttyAMA1", + firmware_version=None, firmware_type=ApplicationType.EZSP, source="unknown", + owners=[], ), ), ): diff --git a/tests/components/zha/conftest.py b/tests/components/zha/conftest.py index 78d335469b8658..96a61a6628b0ee 100644 --- a/tests/components/zha/conftest.py +++ b/tests/components/zha/conftest.py @@ -155,6 +155,7 @@ async def zigpy_app_controller(): app.state.node_info.ieee = zigpy.types.EUI64.convert("00:15:8d:00:02:32:4f:32") app.state.node_info.manufacturer = "Coordinator Manufacturer" app.state.node_info.model = "Coordinator Model" + app.state.node_info.version = "7.1.4.0 build 389" app.state.network_info.pan_id = 0x1234 app.state.network_info.extended_pan_id = app.state.node_info.ieee app.state.network_info.channel = 15 diff --git a/tests/components/zha/snapshots/test_diagnostics.ambr b/tests/components/zha/snapshots/test_diagnostics.ambr index f46a06e84b8ffc..c9a5e80b1c9e41 100644 --- a/tests/components/zha/snapshots/test_diagnostics.ambr +++ b/tests/components/zha/snapshots/test_diagnostics.ambr @@ -75,7 +75,7 @@ 'manufacturer': 'Coordinator Manufacturer', 'model': 'Coordinator Model', 'nwk': 0, - 'version': None, + 'version': '7.1.4.0 build 389', }), }), 'config': dict({ diff --git a/tests/components/zha/test_homeassistant_hardware.py b/tests/components/zha/test_homeassistant_hardware.py new file mode 100644 index 00000000000000..722855211822da --- /dev/null +++ b/tests/components/zha/test_homeassistant_hardware.py @@ -0,0 +1,120 @@ +"""Test Home Assistant Hardware platform for ZHA.""" + +from unittest.mock import MagicMock, patch + +import pytest +from zigpy.application import ControllerApplication + +from homeassistant.components.homeassistant_hardware.helpers import ( + async_register_firmware_info_callback, +) +from homeassistant.components.homeassistant_hardware.util import ( + ApplicationType, + FirmwareInfo, + OwningIntegration, +) +from homeassistant.components.zha.homeassistant_hardware import get_firmware_info +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant +from homeassistant.setup import async_setup_component + +from tests.common import MockConfigEntry + + +async def test_get_firmware_info_normal(hass: HomeAssistant) -> None: + """Test `get_firmware_info`.""" + + zha = MockConfigEntry( + domain="zha", + unique_id="some_unique_id", + data={ + "device": { + "path": "/dev/ttyUSB1", + "baudrate": 115200, + "flow_control": None, + }, + "radio_type": "ezsp", + }, + version=4, + ) + zha.add_to_hass(hass) + zha.mock_state(hass, ConfigEntryState.LOADED) + + # With ZHA running + with patch( + "homeassistant.components.zha.homeassistant_hardware.get_zha_gateway" + ) as mock_get_zha_gateway: + mock_get_zha_gateway.return_value.state.node_info.version = "1.2.3.4" + fw_info_running = get_firmware_info(hass, zha) + + assert fw_info_running == FirmwareInfo( + device="/dev/ttyUSB1", + firmware_type=ApplicationType.EZSP, + firmware_version="1.2.3.4", + source="zha", + owners=[OwningIntegration(config_entry_id=zha.entry_id)], + ) + assert await fw_info_running.is_running(hass) is True + + # With ZHA not running + zha.mock_state(hass, ConfigEntryState.NOT_LOADED) + fw_info_not_running = get_firmware_info(hass, zha) + + assert fw_info_not_running == FirmwareInfo( + device="/dev/ttyUSB1", + firmware_type=ApplicationType.EZSP, + firmware_version=None, + source="zha", + owners=[OwningIntegration(config_entry_id=zha.entry_id)], + ) + assert await fw_info_not_running.is_running(hass) is False + + +@pytest.mark.parametrize( + "data", + [ + # Missing data + {}, + # Bad radio type + {"device": {"path": "/dev/ttyUSB1"}, "radio_type": "znp"}, + ], +) +async def test_get_firmware_info_errors( + hass: HomeAssistant, data: dict[str, str | int | None] +) -> None: + """Test `get_firmware_info` with config entry data format errors.""" + zha = MockConfigEntry( + domain="zha", + unique_id="some_unique_id", + data=data, + version=4, + ) + zha.add_to_hass(hass) + + assert (get_firmware_info(hass, zha)) is None + + +async def test_hardware_firmware_info_provider_notification( + hass: HomeAssistant, + config_entry: MockConfigEntry, + mock_zigpy_connect: ControllerApplication, +) -> None: + """Test that the ZHA gateway provides hardware and firmware information.""" + config_entry.add_to_hass(hass) + + await async_setup_component(hass, "homeassistant_hardware", {}) + + callback = MagicMock() + async_register_firmware_info_callback(hass, "/dev/ttyUSB0", callback) + + await hass.config_entries.async_setup(config_entry.entry_id) + + callback.assert_called_once_with( + FirmwareInfo( + device="/dev/ttyUSB0", + firmware_type=ApplicationType.EZSP, + firmware_version="7.1.4.0 build 389", + source="zha", + owners=[OwningIntegration(config_entry_id=config_entry.entry_id)], + ) + )