From 6f28902a4fbc886528bf00a660b3eecc17b2fa86 Mon Sep 17 00:00:00 2001 From: Mike Degatano Date: Mon, 4 May 2026 14:24:33 -0400 Subject: [PATCH] Refactor hassio coordinators to use typed dataclasses instead of dicts (#168847) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- .../components/hassio/binary_sensor.py | 75 ++-- homeassistant/components/hassio/const.py | 10 +- .../components/hassio/coordinator.py | 332 ++++++++++++------ .../components/hassio/diagnostics.py | 6 +- homeassistant/components/hassio/entity.py | 128 +++---- homeassistant/components/hassio/sensor.py | 150 +++++--- homeassistant/components/hassio/switch.py | 20 +- homeassistant/components/hassio/update.py | 56 ++- tests/components/hassio/common.py | 1 + tests/components/hassio/test_init.py | 207 +++++++++++ tests/components/http/test_ban.py | 7 + 11 files changed, 678 insertions(+), 314 deletions(-) diff --git a/homeassistant/components/hassio/binary_sensor.py b/homeassistant/components/hassio/binary_sensor.py index 111e053f3fb..4c4819169b5 100644 --- a/homeassistant/components/hassio/binary_sensor.py +++ b/homeassistant/components/hassio/binary_sensor.py @@ -1,8 +1,9 @@ """Binary sensor platform for Hass.io addons.""" +from collections.abc import Callable from dataclasses import dataclass -import itertools +from aiohasupervisor.models import AddonState from aiohasupervisor.models.mounts import MountState from homeassistant.components.binary_sensor import ( @@ -14,41 +15,46 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import ( - ADDONS_COORDINATOR, - ATTR_STARTED, - ATTR_STATE, - DATA_KEY_ADDONS, - DATA_KEY_MOUNTS, - MAIN_COORDINATOR, -) +from .const import ADDONS_COORDINATOR, MAIN_COORDINATOR from .entity import HassioAddonEntity, HassioMountEntity -@dataclass(frozen=True) -class HassioBinarySensorEntityDescription(BinarySensorEntityDescription): - """Hassio binary sensor entity description.""" +@dataclass(frozen=True, kw_only=True) +class HassioAddonBinarySensorEntityDescription(BinarySensorEntityDescription): + """Hass.io add-on binary sensor entity description.""" - target: str | None = None + value_fn: Callable[[HassioAddonBinarySensor], bool] + + +@dataclass(frozen=True, kw_only=True) +class HassioMountBinarySensorEntityDescription(BinarySensorEntityDescription): + """Hass.io mount binary sensor entity description.""" + + value_fn: Callable[[HassioMountBinarySensor], bool] ADDON_ENTITY_DESCRIPTIONS = ( - HassioBinarySensorEntityDescription( + HassioAddonBinarySensorEntityDescription( device_class=BinarySensorDeviceClass.RUNNING, entity_registry_enabled_default=False, - key=ATTR_STATE, + key="state", translation_key="state", - target=ATTR_STARTED, + value_fn=lambda entity: ( + entity.coordinator.data.addons[entity.addon_slug].addon.state + == AddonState.STARTED + ), ), ) MOUNT_ENTITY_DESCRIPTIONS = ( - HassioBinarySensorEntityDescription( + HassioMountBinarySensorEntityDescription( device_class=BinarySensorDeviceClass.CONNECTIVITY, entity_registry_enabled_default=False, - key=ATTR_STATE, + key="state", translation_key="mount", - target=MountState.ACTIVE.value, + value_fn=lambda entity: ( + entity.coordinator.data.mounts[entity.mount_name].state == MountState.ACTIVE + ), ), ) @@ -63,57 +69,46 @@ async def async_setup_entry( coordinator = hass.data[MAIN_COORDINATOR] async_add_entities( - itertools.chain( - [ + [ + *[ HassioAddonBinarySensor( addon=addon, coordinator=addons_coordinator, entity_description=entity_description, ) - for addon in addons_coordinator.data[DATA_KEY_ADDONS].values() + for addon in addons_coordinator.data.addons.values() for entity_description in ADDON_ENTITY_DESCRIPTIONS ], - [ + *[ HassioMountBinarySensor( mount=mount, coordinator=coordinator, entity_description=entity_description, ) - for mount in coordinator.data[DATA_KEY_MOUNTS].values() + for mount in coordinator.data.mounts.values() for entity_description in MOUNT_ENTITY_DESCRIPTIONS ], - ) + ] ) class HassioAddonBinarySensor(HassioAddonEntity, BinarySensorEntity): """Binary sensor for Hass.io add-ons.""" - entity_description: HassioBinarySensorEntityDescription + entity_description: HassioAddonBinarySensorEntityDescription @property def is_on(self) -> bool: """Return true if the binary sensor is on.""" - value = self.coordinator.data[DATA_KEY_ADDONS][self._addon_slug][ - self.entity_description.key - ] - if self.entity_description.target is None: - return value - return value == self.entity_description.target + return self.entity_description.value_fn(self) class HassioMountBinarySensor(HassioMountEntity, BinarySensorEntity): """Binary sensor for Hass.io mount.""" - entity_description: HassioBinarySensorEntityDescription + entity_description: HassioMountBinarySensorEntityDescription @property def is_on(self) -> bool: """Return true if the binary sensor is on.""" - value = getattr( - self.coordinator.data[DATA_KEY_MOUNTS][self._mount.name], - self.entity_description.key, - ) - if self.entity_description.target is None: - return value - return value == self.entity_description.target + return self.entity_description.value_fn(self) diff --git a/homeassistant/components/hassio/const.py b/homeassistant/components/hassio/const.py index 46033c7784e..6978b545766 100644 --- a/homeassistant/components/hassio/const.py +++ b/homeassistant/components/hassio/const.py @@ -8,9 +8,11 @@ from homeassistant.util.hass_dict import HassKey if TYPE_CHECKING: from aiohasupervisor.models import ( + AddonsStats, HomeAssistantInfo, HostInfo, InstalledAddon, + InstalledAddonComplete, NetworkInfo, OSInfo, RootInfo, @@ -112,8 +114,12 @@ DATA_OS_INFO: HassKey[OSInfo] = HassKey("hassio_os_info") DATA_NETWORK_INFO: HassKey[NetworkInfo] = HassKey("hassio_network_info") DATA_SUPERVISOR_INFO: HassKey[SupervisorInfo] = HassKey("hassio_supervisor_info") DATA_SUPERVISOR_STATS = "hassio_supervisor_stats" -DATA_ADDONS_INFO = "hassio_addons_info" -DATA_ADDONS_STATS = "hassio_addons_stats" +DATA_ADDONS_INFO: HassKey[dict[str, InstalledAddonComplete | None]] = HassKey( + "hassio_addons_info" +) +DATA_ADDONS_STATS: HassKey[dict[str, AddonsStats | None]] = HassKey( + "hassio_addons_stats" +) DATA_ADDONS_LIST: HassKey[list[InstalledAddon]] = HassKey("hassio_addons_list") HASSIO_MAIN_UPDATE_INTERVAL = timedelta(minutes=5) HASSIO_ADDON_UPDATE_INTERVAL = timedelta(minutes=15) diff --git a/homeassistant/components/hassio/coordinator.py b/homeassistant/components/hassio/coordinator.py index 1eee409a20f..5ca558fbc72 100644 --- a/homeassistant/components/hassio/coordinator.py +++ b/homeassistant/components/hassio/coordinator.py @@ -3,17 +3,20 @@ import asyncio from collections import defaultdict from collections.abc import Awaitable -from copy import deepcopy +from dataclasses import dataclass import logging from typing import TYPE_CHECKING, Any, cast from aiohasupervisor import SupervisorError, SupervisorNotFoundError from aiohasupervisor.models import ( + AddonsStats, AddonState, CIFSMountResponse, HomeAssistantInfo, + HomeAssistantStats, HostInfo, InstalledAddon, + InstalledAddonComplete, NetworkInfo, NFSMountResponse, OSInfo, @@ -21,10 +24,11 @@ from aiohasupervisor.models import ( RootInfo, StoreInfo, SupervisorInfo, + SupervisorStats, ) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ATTR_MANUFACTURER, ATTR_NAME +from homeassistant.const import ATTR_MANUFACTURER from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback from homeassistant.helpers import device_registry as dr from homeassistant.helpers.debounce import Debouncer @@ -34,15 +38,10 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda from .const import ( ATTR_ADDONS, - ATTR_AUTO_UPDATE, ATTR_DATA, ATTR_REPOSITORIES, - ATTR_REPOSITORY, - ATTR_SLUG, ATTR_STARTUP, ATTR_UPDATE_KEY, - ATTR_URL, - ATTR_VERSION, ATTR_WS_EVENT, CONTAINER_STATS, CORE_CONTAINER, @@ -53,12 +52,6 @@ from .const import ( DATA_CORE_STATS, DATA_HOST_INFO, DATA_INFO, - DATA_KEY_ADDONS, - DATA_KEY_CORE, - DATA_KEY_HOST, - DATA_KEY_MOUNTS, - DATA_KEY_OS, - DATA_KEY_SUPERVISOR, DATA_KEY_SUPERVISOR_ISSUES, DATA_NETWORK_INFO, DATA_OS_INFO, @@ -86,6 +79,106 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) +@dataclass +class HassioMainData: + """Data class for HassioMainDataUpdateCoordinator.""" + + core: HomeAssistantInfo + supervisor: SupervisorInfo + host: HostInfo + mounts: dict[str, CIFSMountResponse | NFSMountResponse] + os: OSInfo | None + + def to_dict(self) -> dict[str, Any]: + """Return a dictionary representation of the data.""" + return { + "core": self.core.to_dict(), + "supervisor": self.supervisor.to_dict(), + "host": self.host.to_dict(), + "mounts": {name: mount.to_dict() for name, mount in self.mounts.items()}, + "os": self.os.to_dict() if self.os is not None else None, + } + + +@dataclass +class AddonData: + """Data for a single installed addon.""" + + addon: InstalledAddon + auto_update: bool + repository: str + + +@dataclass +class HassioAddonData: + """Data class for HassioAddOnDataUpdateCoordinator.""" + + addons: dict[str, AddonData] + + def to_dict(self) -> dict[str, Any]: + """Return a dictionary representation of the data.""" + return { + "addons": { + slug: { + "addon": addon_data.addon.to_dict(), + "auto_update": addon_data.auto_update, + "repository": addon_data.repository, + } + for slug, addon_data in self.addons.items() + }, + } + + +@dataclass +class HassioStatsData: + """Data class for HassioStatsDataUpdateCoordinator.""" + + core: HomeAssistantStats | None + supervisor: SupervisorStats | None + addons: dict[str, AddonsStats | None] + + def to_dict(self) -> dict[str, Any]: + """Return a dictionary representation of the data.""" + return { + "core": self.core.to_dict() if self.core is not None else None, + "supervisor": ( + self.supervisor.to_dict() if self.supervisor is not None else None + ), + "addons": { + slug: stats.to_dict() if stats is not None else None + for slug, stats in self.addons.items() + }, + } + + +def _installed_addon_from_complete(info: InstalledAddonComplete) -> InstalledAddon: + """Build an InstalledAddon from an InstalledAddonComplete object. + + InstalledAddonComplete contains a superset of InstalledAddon fields. + This helper extracts only the fields needed for InstalledAddon so fresh + data from an addon_info call can be stored in AddonData.addon. + """ + return InstalledAddon( + advanced=info.advanced, + available=info.available, + build=info.build, + description=info.description, + homeassistant=info.homeassistant, + icon=info.icon, + logo=info.logo, + name=info.name, + repository=info.repository, + slug=info.slug, + stage=info.stage, + update_available=info.update_available, + url=info.url, + version_latest=info.version_latest, + version=info.version, + detached=info.detached, + state=info.state, + ) + + @callback def get_info(hass: HomeAssistant) -> dict[str, Any] | None: """Return generic information from Supervisor. @@ -151,7 +244,25 @@ def get_addons_info(hass: HomeAssistant) -> dict[str, dict[str, Any] | None] | N Async friendly. """ - return hass.data.get(DATA_ADDONS_INFO) + addons_info: dict[str, InstalledAddonComplete | None] | None = hass.data.get( + DATA_ADDONS_INFO + ) + if addons_info is None: + return None + # Converting these fields for compatibility as that is what was returned here. + # We'll leave it this way as long as these component APIs continue to return + # dictionaries. If/when we switch to using the aiohasupervisor models for everything + # internally and externally that will be dropped. + return { + slug: dict( + hassio_api=info.supervisor_api, + hassio_role=info.supervisor_role, + **info.to_dict(), + ) + if info is not None + else None + for slug, info in addons_info.items() + } @callback @@ -170,7 +281,11 @@ def get_addons_stats(hass: HomeAssistant) -> dict[str, dict[str, Any] | None]: Async friendly. """ - return hass.data.get(DATA_ADDONS_STATS) or {} + addons_stats: dict[str, AddonsStats | None] = hass.data.get(DATA_ADDONS_STATS) or {} + return { + slug: stats.to_dict() if stats is not None else None + for slug, stats in addons_stats.items() + } @callback @@ -179,7 +294,8 @@ def get_core_stats(hass: HomeAssistant) -> dict[str, Any]: Async friendly. """ - return hass.data.get(DATA_CORE_STATS) or {} + stats = hass.data.get(DATA_CORE_STATS) + return stats.to_dict() if stats is not None else {} @callback @@ -188,7 +304,8 @@ def get_supervisor_stats(hass: HomeAssistant) -> dict[str, Any]: Async friendly. """ - return hass.data.get(DATA_SUPERVISOR_STATS) or {} + stats = hass.data.get(DATA_SUPERVISOR_STATS) + return stats.to_dict() if stats is not None else {} @callback @@ -222,19 +339,20 @@ def get_issues_info(hass: HomeAssistant) -> SupervisorIssues | None: @callback def async_register_addons_in_dev_reg( - entry_id: str, dev_reg: dr.DeviceRegistry, addons: list[dict[str, Any]] + entry_id: str, dev_reg: dr.DeviceRegistry, addons: list[AddonData] ) -> None: """Register addons in the device registry.""" - for addon in addons: + for addon_data in addons: + addon = addon_data.addon params = DeviceInfo( - identifiers={(DOMAIN, addon[ATTR_SLUG])}, + identifiers={(DOMAIN, addon.slug)}, model=SupervisorEntityModel.ADDON, - sw_version=addon[ATTR_VERSION], - name=addon[ATTR_NAME], + sw_version=addon.version, + name=addon.name, entry_type=dr.DeviceEntryType.SERVICE, - configuration_url=f"homeassistant://hassio/addon/{addon[ATTR_SLUG]}", + configuration_url=f"homeassistant://hassio/addon/{addon.slug}", ) - if manufacturer := addon.get(ATTR_REPOSITORY) or addon.get(ATTR_URL): + if manufacturer := addon_data.repository or addon.url: params[ATTR_MANUFACTURER] = manufacturer dev_reg.async_get_or_create(config_entry_id=entry_id, **params) @@ -260,14 +378,14 @@ def async_register_mounts_in_dev_reg( @callback def async_register_os_in_dev_reg( - entry_id: str, dev_reg: dr.DeviceRegistry, os_dict: dict[str, Any] + entry_id: str, dev_reg: dr.DeviceRegistry, os_info: OSInfo ) -> None: """Register OS in the device registry.""" params = DeviceInfo( identifiers={(DOMAIN, "OS")}, manufacturer="Home Assistant", model=SupervisorEntityModel.OS, - sw_version=os_dict[ATTR_VERSION], + sw_version=os_info.version, name="Home Assistant Operating System", entry_type=dr.DeviceEntryType.SERVICE, ) @@ -294,14 +412,14 @@ def async_register_host_in_dev_reg( def async_register_core_in_dev_reg( entry_id: str, dev_reg: dr.DeviceRegistry, - core_dict: dict[str, Any], + core_info: HomeAssistantInfo, ) -> None: - """Register OS in the device registry.""" + """Register core in the device registry.""" params = DeviceInfo( identifiers={(DOMAIN, "core")}, manufacturer="Home Assistant", model=SupervisorEntityModel.CORE, - sw_version=core_dict[ATTR_VERSION], + sw_version=core_info.version, name="Home Assistant Core", entry_type=dr.DeviceEntryType.SERVICE, ) @@ -312,14 +430,14 @@ def async_register_core_in_dev_reg( def async_register_supervisor_in_dev_reg( entry_id: str, dev_reg: dr.DeviceRegistry, - supervisor_dict: dict[str, Any], + supervisor_info: SupervisorInfo, ) -> None: - """Register OS in the device registry.""" + """Register supervisor in the device registry.""" params = DeviceInfo( identifiers={(DOMAIN, "supervisor")}, manufacturer="Home Assistant", model=SupervisorEntityModel.SUPERVISOR, - sw_version=supervisor_dict[ATTR_VERSION], + sw_version=supervisor_info.version, name="Home Assistant Supervisor", entry_type=dr.DeviceEntryType.SERVICE, ) @@ -336,7 +454,7 @@ def async_remove_devices_from_dev_reg( dev_reg.async_remove_device(dev.id) -class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): +class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[HassioStatsData]): """Class to retrieve Hass.io container stats.""" config_entry: ConfigEntry @@ -358,18 +476,18 @@ class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): lambda: defaultdict(set) ) - async def _async_update_data(self) -> dict[str, Any]: + async def _async_update_data(self) -> HassioStatsData: """Update stats data via library.""" try: await self._fetch_stats() except SupervisorError as err: raise UpdateFailed(f"Error on Supervisor API: {err}") from err - new_data: dict[str, Any] = {} - new_data[DATA_KEY_CORE] = get_core_stats(self.hass) - new_data[DATA_KEY_SUPERVISOR] = get_supervisor_stats(self.hass) - new_data[DATA_KEY_ADDONS] = get_addons_stats(self.hass) - return new_data + return HassioStatsData( + core=self.hass.data.get(DATA_CORE_STATS), + supervisor=self.hass.data.get(DATA_SUPERVISOR_STATS), + addons=self.hass.data.get(DATA_ADDONS_STATS) or {}, + ) async def _fetch_stats(self) -> None: """Fetch container stats for subscribed entities.""" @@ -387,7 +505,7 @@ class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): if updates: api_results: list[ResponseData] = await asyncio.gather(*updates.values()) for key, result in zip(updates, api_results, strict=True): - data[key] = result.to_dict() + data[key] = result # Fetch addon stats addons_list: list[InstalledAddon] = self.hass.data.get(DATA_ADDONS_LIST) or [] @@ -397,7 +515,9 @@ class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): if addon.state in {AddonState.STARTED, AddonState.STARTUP} } - addons_stats: dict[str, Any] = data.setdefault(DATA_ADDONS_STATS, {}) + addons_stats: dict[str, AddonsStats | None] = data.setdefault( + DATA_ADDONS_STATS, {} + ) # Clean up cache for stopped/removed addons for slug in addons_stats.keys() - started_addons: @@ -415,14 +535,14 @@ class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): ) addons_stats.update(addon_stats_results) - async def _update_addon_stats(self, slug: str) -> tuple[str, dict[str, Any] | None]: + async def _update_addon_stats(self, slug: str) -> tuple[str, AddonsStats | None]: """Update single addon stats.""" try: stats = await self.supervisor_client.addons.addon_stats(slug) except SupervisorError as err: _LOGGER.warning("Could not fetch stats for %s: %s", slug, err) return (slug, None) - return (slug, stats.to_dict()) + return (slug, stats) @callback def async_enable_container_updates( @@ -445,7 +565,7 @@ class HassioStatsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): return _remove -class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): +class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[HassioAddonData]): """Class to retrieve Hass.io Add-on status.""" config_entry: ConfigEntry @@ -476,7 +596,7 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): self.supervisor_client = get_supervisor_client(hass) self.jobs = jobs - async def _async_update_data(self) -> dict[str, Any]: + async def _async_update_data(self) -> HassioAddonData: """Update data via library.""" is_first_update = not self.data client = self.supervisor_client @@ -487,7 +607,7 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # Fetch addon info for all addons on first update, or only # for addons with subscribed entities on subsequent updates. - addon_info_results = dict( + addon_info_results: dict[str, InstalledAddonComplete | None] = dict( await asyncio.gather( *[ self._update_addon_info(slug) @@ -503,39 +623,35 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): self.hass.data[DATA_ADDONS_LIST] = installed_addons # Update addon info cache in hass.data - addon_info_cache: dict[str, Any] = self.hass.data.setdefault( - DATA_ADDONS_INFO, {} - ) + addon_info_cache = self.hass.data.setdefault(DATA_ADDONS_INFO, {}) for slug in addon_info_cache.keys() - all_addons: del addon_info_cache[slug] addon_info_cache.update(addon_info_results) - # Build clean coordinator data + # Build repository name lookup from store data store = self.hass.data.get(DATA_STORE) - if store: - repositories = {repo.slug: repo.name for repo in store.repositories} - else: - repositories = {} + repositories: dict[str, str] = ( + {repo.slug: repo.name for repo in store.repositories} if store else {} + ) - addons_list_dicts = [addon.to_dict() for addon in installed_addons] - new_data: dict[str, Any] = {} - new_data[DATA_KEY_ADDONS] = { - (slug := addon[ATTR_SLUG]): { - **addon, - ATTR_AUTO_UPDATE: (addon_info_cache.get(slug) or {}).get( - ATTR_AUTO_UPDATE, False - ), - ATTR_REPOSITORY: repositories.get( - repo_slug := addon.get(ATTR_REPOSITORY, ""), repo_slug - ), - } - for addon in addons_list_dicts - } + # Build clean coordinator data + new_addons: dict[str, AddonData] = {} + for addon in installed_addons: + addon_info = addon_info_cache.get(addon.slug) + auto_update = addon_info.auto_update if addon_info is not None else False + repo_slug = addon.repository + repository = repositories.get(repo_slug, repo_slug) + new_addons[addon.slug] = AddonData( + addon=addon, + auto_update=auto_update, + repository=repository, + ) + new_data = HassioAddonData(addons=new_addons) # If this is the initial refresh, register all addons if is_first_update: async_register_addons_in_dev_reg( - self.entry_id, self.dev_reg, new_data[DATA_KEY_ADDONS].values() + self.entry_id, self.dev_reg, list(new_data.addons.values()) ) # Remove add-ons that are no longer installed from device registry @@ -546,19 +662,16 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): ) if device.model == SupervisorEntityModel.ADDON } - if stale_addons := supervisor_addon_devices - set(new_data[DATA_KEY_ADDONS]): + if stale_addons := supervisor_addon_devices - set(new_data.addons): async_remove_devices_from_dev_reg(self.dev_reg, stale_addons) # If there are new add-ons, we should reload the config entry so we can - # create new devices and entities. We can return an empty dict because + # create new devices and entities. We can return the new data because # coordinator will be recreated. - if self.data and ( - set(new_data[DATA_KEY_ADDONS]) - set(self.data[DATA_KEY_ADDONS]) - ): + if self.data and (set(new_data.addons) - set(self.data.addons)): self.hass.async_create_task( self.hass.config_entries.async_reload(self.entry_id) ) - return {} return new_data @@ -569,18 +682,16 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): except SupervisorNotFoundError: return None - async def _update_addon_info(self, slug: str) -> tuple[str, dict[str, Any] | None]: + async def _update_addon_info( + self, slug: str + ) -> tuple[str, InstalledAddonComplete | None]: """Return the info for an addon.""" try: info = await self.supervisor_client.addons.addon_info(slug) except SupervisorError as err: _LOGGER.warning("Could not fetch info for %s: %s", slug, err) return (slug, None) - # Translate to legacy hassio names for compatibility - info_dict = info.to_dict() - info_dict["hassio_api"] = info_dict.pop("supervisor_api") - info_dict["hassio_role"] = info_dict.pop("supervisor_role") - return (slug, info_dict) + return (slug, info) @callback def async_enable_addon_info_updates( @@ -627,16 +738,26 @@ class HassioAddOnDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): """Force refresh of addon info data for a specific addon.""" try: slug, info = await self._update_addon_info(addon_slug) - if info is not None and DATA_KEY_ADDONS in self.data: - if slug in self.data[DATA_KEY_ADDONS]: - data = deepcopy(self.data) - data[DATA_KEY_ADDONS][slug].update(info) - self.async_set_updated_data(data) except SupervisorError as err: _LOGGER.warning("Could not refresh info for %s: %s", addon_slug, err) + return + + if info is not None and self.data and slug in self.data.addons: + updated = AddonData( + addon=_installed_addon_from_complete(info), + auto_update=info.auto_update, + repository=self.data.addons[slug].repository, + ) + self.async_set_updated_data( + HassioAddonData(addons={**self.data.addons, slug: updated}) + ) + + # Update addon info cache in hass.data + addon_info_cache = self.hass.data.setdefault(DATA_ADDONS_INFO, {}) + addon_info_cache[slug] = info -class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): +class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[HassioMainData]): """Class to retrieve Hass.io status.""" config_entry: ConfigEntry @@ -679,7 +800,7 @@ class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): ): self.config_entry.async_create_task(self.hass, self.async_request_refresh()) - async def _async_update_data(self) -> dict[str, Any]: + async def _async_update_data(self) -> HassioMainData: """Update data via library.""" is_first_update = not self.data client = self.supervisor_client @@ -722,13 +843,13 @@ class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): raise UpdateFailed(f"Error on Supervisor API: {err}") from err # Build clean coordinator data - new_data: dict[str, Any] = {} - new_data[DATA_KEY_CORE] = core_info.to_dict() - new_data[DATA_KEY_SUPERVISOR] = supervisor_info.to_dict() - new_data[DATA_KEY_HOST] = host_info.to_dict() - new_data[DATA_KEY_MOUNTS] = {mount.name: mount for mount in mounts_info.mounts} - if self.is_hass_os: - new_data[DATA_KEY_OS] = os_info.to_dict() + new_data = HassioMainData( + core=core_info, + supervisor=supervisor_info, + host=host_info, + mounts={mount.name: mount for mount in mounts_info.mounts}, + os=os_info if self.is_hass_os else None, + ) # Update hass.data for legacy accessor functions self.hass.data[DATA_INFO] = info @@ -742,19 +863,15 @@ class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # If this is the initial refresh, register all main components if is_first_update: async_register_mounts_in_dev_reg( - self.entry_id, self.dev_reg, new_data[DATA_KEY_MOUNTS].values() - ) - async_register_core_in_dev_reg( - self.entry_id, self.dev_reg, new_data[DATA_KEY_CORE] + self.entry_id, self.dev_reg, list(new_data.mounts.values()) ) + async_register_core_in_dev_reg(self.entry_id, self.dev_reg, new_data.core) async_register_supervisor_in_dev_reg( - self.entry_id, self.dev_reg, new_data[DATA_KEY_SUPERVISOR] + self.entry_id, self.dev_reg, new_data.supervisor ) async_register_host_in_dev_reg(self.entry_id, self.dev_reg) if self.is_hass_os: - async_register_os_in_dev_reg( - self.entry_id, self.dev_reg, new_data[DATA_KEY_OS] - ) + async_register_os_in_dev_reg(self.entry_id, self.dev_reg, os_info) # Remove mounts that no longer exists from device registry supervisor_mount_devices = { @@ -764,7 +881,7 @@ class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): ) if device.model == SupervisorEntityModel.MOUNT } - if stale_mounts := supervisor_mount_devices - set(new_data[DATA_KEY_MOUNTS]): + if stale_mounts := supervisor_mount_devices - set(new_data.mounts): async_remove_devices_from_dev_reg( self.dev_reg, {f"mount_{stale_mount}" for stale_mount in stale_mounts} ) @@ -776,15 +893,12 @@ class HassioMainDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): self.dev_reg.async_remove_device(dev.id) # If there are new mounts, we should reload the config entry so we can - # create new devices and entities. We can return an empty dict because + # create new devices and entities. We can return the new data because # coordinator will be recreated. - if self.data and ( - set(new_data[DATA_KEY_MOUNTS]) - set(self.data.get(DATA_KEY_MOUNTS, {})) - ): + if self.data and (set(new_data.mounts) - set(self.data.mounts)): self.hass.async_create_task( self.hass.config_entries.async_reload(self.entry_id) ) - return {} return new_data diff --git a/homeassistant/components/hassio/diagnostics.py b/homeassistant/components/hassio/diagnostics.py index 704f6770b77..a3166d15888 100644 --- a/homeassistant/components/hassio/diagnostics.py +++ b/homeassistant/components/hassio/diagnostics.py @@ -56,8 +56,8 @@ async def async_get_config_entry_diagnostics( devices.append({"device": asdict(device), "entities": entities}) return { - "coordinator_data": coordinator.data, - "addons_coordinator_data": addons_coordinator.data, - "stats_coordinator_data": stats_coordinator.data, + "coordinator_data": coordinator.data.to_dict(), + "addons_coordinator_data": addons_coordinator.data.to_dict(), + "stats_coordinator_data": stats_coordinator.data.to_dict(), "devices": devices, } diff --git a/homeassistant/components/hassio/entity.py b/homeassistant/components/hassio/entity.py index b3551cd494c..616862ed65e 100644 --- a/homeassistant/components/hassio/entity.py +++ b/homeassistant/components/hassio/entity.py @@ -1,27 +1,20 @@ """Base for Hass.io entities.""" -from typing import Any +from collections.abc import Callable -from aiohasupervisor.models.mounts import CIFSMountResponse, NFSMountResponse +from aiohasupervisor.models import CIFSMountResponse, HostInfo, NFSMountResponse, OSInfo +from aiohasupervisor.models.base import ContainerStats from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity import EntityDescription from homeassistant.helpers.update_coordinator import CoordinatorEntity -from .const import ( - ATTR_SLUG, - CONTAINER_STATS, - DATA_KEY_ADDONS, - DATA_KEY_CORE, - DATA_KEY_HOST, - DATA_KEY_MOUNTS, - DATA_KEY_OS, - DATA_KEY_SUPERVISOR, - DOMAIN, -) +from .const import CONTAINER_STATS, DOMAIN from .coordinator import ( + AddonData, HassioAddOnDataUpdateCoordinator, HassioMainDataUpdateCoordinator, + HassioStatsData, HassioStatsDataUpdateCoordinator, ) @@ -37,7 +30,7 @@ class HassioStatsEntity(CoordinatorEntity[HassioStatsDataUpdateCoordinator]): entity_description: EntityDescription, *, container_id: str, - data_key: str, + stats_fn: Callable[[HassioStatsData], ContainerStats | None], device_id: str, unique_id_prefix: str, ) -> None: @@ -45,27 +38,25 @@ class HassioStatsEntity(CoordinatorEntity[HassioStatsDataUpdateCoordinator]): super().__init__(coordinator) self.entity_description = entity_description self._container_id = container_id - self._data_key = data_key + self._stats_fn = stats_fn self._attr_unique_id = f"{unique_id_prefix}_{entity_description.key}" self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, device_id)}) + @property + def _stats(self) -> ContainerStats | None: + """Return the stats object for this entity's container.""" + return self._stats_fn(self.coordinator.data) + + @property + def stats(self) -> ContainerStats: + """Return the stats object, asserting it is available.""" + assert self._stats is not None + return self._stats + @property def available(self) -> bool: """Return True if entity is available.""" - if self._data_key == DATA_KEY_ADDONS: - return ( - super().available - and DATA_KEY_ADDONS in self.coordinator.data - and self.entity_description.key - in ( - self.coordinator.data[DATA_KEY_ADDONS].get(self._container_id) or {} - ) - ) - return ( - super().available - and self._data_key in self.coordinator.data - and self.entity_description.key in self.coordinator.data[self._data_key] - ) + return super().available and self._stats is not None async def async_added_to_hass(self) -> None: """Subscribe to stats updates.""" @@ -92,24 +83,31 @@ class HassioAddonEntity(CoordinatorEntity[HassioAddOnDataUpdateCoordinator]): self, coordinator: HassioAddOnDataUpdateCoordinator, entity_description: EntityDescription, - addon: dict[str, Any], + addon: AddonData, ) -> None: """Initialize base entity.""" super().__init__(coordinator) self.entity_description = entity_description - self._addon_slug = addon[ATTR_SLUG] - self._attr_unique_id = f"{addon[ATTR_SLUG]}_{entity_description.key}" - self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, addon[ATTR_SLUG])}) + self._addon_slug = addon.addon.slug + self._attr_unique_id = f"{addon.addon.slug}_{entity_description.key}" + self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, addon.addon.slug)}) + + @property + def addon_slug(self) -> str: + """Return the add-on slug.""" + return self._addon_slug + + @property + def addon_data(self) -> AddonData: + """Return the add-on data, asserting it is available.""" + data = self.coordinator.data + assert self._addon_slug in data.addons + return data.addons[self._addon_slug] @property def available(self) -> bool: """Return True if entity is available.""" - return ( - super().available - and DATA_KEY_ADDONS in self.coordinator.data - and self.entity_description.key - in self.coordinator.data[DATA_KEY_ADDONS].get(self._addon_slug, {}) - ) + return super().available and self._addon_slug in self.coordinator.data.addons async def async_added_to_hass(self) -> None: """Subscribe to addon info updates.""" @@ -140,11 +138,13 @@ class HassioOSEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): @property def available(self) -> bool: """Return True if entity is available.""" - return ( - super().available - and DATA_KEY_OS in self.coordinator.data - and self.entity_description.key in self.coordinator.data[DATA_KEY_OS] - ) + return super().available and self.coordinator.data.os is not None + + @property + def os(self) -> OSInfo: + """Return the OS info object, asserting it is available.""" + assert self.coordinator.data.os is not None + return self.coordinator.data.os class HassioHostEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): @@ -164,13 +164,10 @@ class HassioHostEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, "host")}) @property - def available(self) -> bool: - """Return True if entity is available.""" - return ( - super().available - and DATA_KEY_HOST in self.coordinator.data - and self.entity_description.key in self.coordinator.data[DATA_KEY_HOST] - ) + def host(self) -> HostInfo: + """Return the host info, asserting it is available.""" + assert self.coordinator.data.host is not None + return self.coordinator.data.host class HassioSupervisorEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): @@ -189,16 +186,6 @@ class HassioSupervisorEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]) self._attr_unique_id = f"home_assistant_supervisor_{entity_description.key}" self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, "supervisor")}) - @property - def available(self) -> bool: - """Return True if entity is available.""" - return ( - super().available - and DATA_KEY_SUPERVISOR in self.coordinator.data - and self.entity_description.key - in self.coordinator.data[DATA_KEY_SUPERVISOR] - ) - class HassioCoreEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): """Base Entity for Core.""" @@ -216,15 +203,6 @@ class HassioCoreEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): self._attr_unique_id = f"home_assistant_core_{entity_description.key}" self._attr_device_info = DeviceInfo(identifiers={(DOMAIN, "core")}) - @property - def available(self) -> bool: - """Return True if entity is available.""" - return ( - super().available - and DATA_KEY_CORE in self.coordinator.data - and self.entity_description.key in self.coordinator.data[DATA_KEY_CORE] - ) - class HassioMountEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): """Base Entity for Mount.""" @@ -248,10 +226,12 @@ class HassioMountEntity(CoordinatorEntity[HassioMainDataUpdateCoordinator]): ) self._mount = mount + @property + def mount_name(self) -> str: + """Return the mount name.""" + return self._mount.name + @property def available(self) -> bool: """Return True if entity is available.""" - return ( - super().available - and self._mount.name in self.coordinator.data[DATA_KEY_MOUNTS] - ) + return super().available and self.mount_name in self.coordinator.data.mounts diff --git a/homeassistant/components/hassio/sensor.py b/homeassistant/components/hassio/sensor.py index 6b5014fc76e..8acc4880388 100644 --- a/homeassistant/components/hassio/sensor.py +++ b/homeassistant/components/hassio/sensor.py @@ -1,5 +1,10 @@ """Sensor platform for Hass.io addons.""" +from collections.abc import Callable +from dataclasses import dataclass + +from aiohasupervisor.models.base import ContainerStats + from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, @@ -15,19 +20,12 @@ from .const import ( ADDONS_COORDINATOR, ATTR_CPU_PERCENT, ATTR_MEMORY_PERCENT, - ATTR_SLUG, - ATTR_VERSION, - ATTR_VERSION_LATEST, CORE_CONTAINER, - DATA_KEY_ADDONS, - DATA_KEY_CORE, - DATA_KEY_HOST, - DATA_KEY_OS, - DATA_KEY_SUPERVISOR, MAIN_COORDINATOR, STATS_COORDINATOR, SUPERVISOR_CONTAINER, ) +from .coordinator import HassioStatsData from .entity import ( HassioAddonEntity, HassioHostEntity, @@ -35,74 +33,125 @@ from .entity import ( HassioStatsEntity, ) -COMMON_ENTITY_DESCRIPTIONS = ( - SensorEntityDescription( + +@dataclass(frozen=True, kw_only=True) +class HassioAddonSensorEntityDescription(SensorEntityDescription): + """Hass.io add-on sensor entity description.""" + + value_fn: Callable[[HassioAddonSensor], str | None] + + +@dataclass(frozen=True, kw_only=True) +class HassioStatsSensorEntityDescription(SensorEntityDescription): + """Hass.io stats sensor entity description.""" + + value_fn: Callable[[HassioStatsSensor], float] + + +@dataclass(frozen=True, kw_only=True) +class HassioOSSensorEntityDescription(SensorEntityDescription): + """Hass.io OS sensor entity description.""" + + value_fn: Callable[[HassioOSSensor], str | None] + + +@dataclass(frozen=True, kw_only=True) +class HassioHostSensorEntityDescription(SensorEntityDescription): + """Hass.io host sensor entity description.""" + + value_fn: Callable[[HostSensor], str | float | None] + + +ADDON_ENTITY_DESCRIPTIONS = ( + HassioAddonSensorEntityDescription( entity_registry_enabled_default=False, - key=ATTR_VERSION, + key="version", translation_key="version", + value_fn=lambda entity: entity.addon_data.addon.version, ), - SensorEntityDescription( + HassioAddonSensorEntityDescription( entity_registry_enabled_default=False, - key=ATTR_VERSION_LATEST, + key="version_latest", translation_key="version_latest", + value_fn=lambda entity: entity.addon_data.addon.version_latest, ), ) STATS_ENTITY_DESCRIPTIONS = ( - SensorEntityDescription( + HassioStatsSensorEntityDescription( entity_registry_enabled_default=False, key=ATTR_CPU_PERCENT, translation_key="cpu_percent", native_unit_of_measurement=PERCENTAGE, state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda entity: entity.stats.cpu_percent, ), - SensorEntityDescription( + HassioStatsSensorEntityDescription( entity_registry_enabled_default=False, key=ATTR_MEMORY_PERCENT, translation_key="memory_percent", native_unit_of_measurement=PERCENTAGE, state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda entity: entity.stats.memory_percent, ), ) -OS_ENTITY_DESCRIPTIONS = COMMON_ENTITY_DESCRIPTIONS +OS_ENTITY_DESCRIPTIONS = ( + HassioOSSensorEntityDescription( + entity_registry_enabled_default=False, + key="version", + translation_key="version", + value_fn=lambda entity: entity.os.version, + ), + HassioOSSensorEntityDescription( + entity_registry_enabled_default=False, + key="version_latest", + translation_key="version_latest", + value_fn=lambda entity: entity.os.version_latest, + ), +) HOST_ENTITY_DESCRIPTIONS = ( - SensorEntityDescription( + HassioHostSensorEntityDescription( entity_registry_enabled_default=False, key="agent_version", translation_key="agent_version", entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda entity: entity.host.agent_version, ), - SensorEntityDescription( + HassioHostSensorEntityDescription( entity_registry_enabled_default=False, key="apparmor_version", translation_key="apparmor_version", entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda entity: entity.host.apparmor_version, ), - SensorEntityDescription( + HassioHostSensorEntityDescription( entity_registry_enabled_default=False, key="disk_total", translation_key="disk_total", native_unit_of_measurement=UnitOfInformation.GIGABYTES, device_class=SensorDeviceClass.DATA_SIZE, entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda entity: entity.host.disk_total, ), - SensorEntityDescription( + HassioHostSensorEntityDescription( entity_registry_enabled_default=False, key="disk_used", translation_key="disk_used", native_unit_of_measurement=UnitOfInformation.GIGABYTES, device_class=SensorDeviceClass.DATA_SIZE, entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda entity: entity.host.disk_used, ), - SensorEntityDescription( + HassioHostSensorEntityDescription( entity_registry_enabled_default=False, key="disk_free", translation_key="disk_free", native_unit_of_measurement=UnitOfInformation.GIGABYTES, device_class=SensorDeviceClass.DATA_SIZE, entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda entity: entity.host.disk_free, ), ) @@ -126,21 +175,32 @@ async def async_setup_entry( coordinator=addons_coordinator, entity_description=entity_description, ) - for addon in addons_coordinator.data[DATA_KEY_ADDONS].values() - for entity_description in COMMON_ENTITY_DESCRIPTIONS + for addon in addons_coordinator.data.addons.values() + for entity_description in ADDON_ENTITY_DESCRIPTIONS ) # Add-on stats sensors (cpu_percent, memory_percent) + def stats_fn_factory( + addon_slug: str, + ) -> Callable[[HassioStatsData], ContainerStats | None]: + """Return a stats_fn for the given add-on slug.""" + + def stats_fn(data: HassioStatsData) -> ContainerStats | None: + """Return the stats for the given add-on.""" + return data.addons.get(addon_slug) + + return stats_fn + entities.extend( HassioStatsSensor( coordinator=stats_coordinator, entity_description=entity_description, - container_id=addon[ATTR_SLUG], - data_key=DATA_KEY_ADDONS, - device_id=addon[ATTR_SLUG], - unique_id_prefix=addon[ATTR_SLUG], + container_id=addon.addon.slug, + stats_fn=stats_fn_factory(addon.addon.slug), + device_id=addon.addon.slug, + unique_id_prefix=addon.addon.slug, ) - for addon in addons_coordinator.data[DATA_KEY_ADDONS].values() + for addon in addons_coordinator.data.addons.values() for entity_description in STATS_ENTITY_DESCRIPTIONS ) @@ -150,7 +210,7 @@ async def async_setup_entry( coordinator=stats_coordinator, entity_description=entity_description, container_id=CORE_CONTAINER, - data_key=DATA_KEY_CORE, + stats_fn=lambda data: data.core, device_id="core", unique_id_prefix="home_assistant_core", ) @@ -163,7 +223,7 @@ async def async_setup_entry( coordinator=stats_coordinator, entity_description=entity_description, container_id=SUPERVISOR_CONTAINER, - data_key=DATA_KEY_SUPERVISOR, + stats_fn=lambda data: data.supervisor, device_id="supervisor", unique_id_prefix="home_assistant_supervisor", ) @@ -195,40 +255,42 @@ async def async_setup_entry( class HassioAddonSensor(HassioAddonEntity, SensorEntity): """Sensor to track a Hass.io add-on attribute.""" + entity_description: HassioAddonSensorEntityDescription + @property - def native_value(self) -> str: + def native_value(self) -> str | None: """Return native value of entity.""" - return self.coordinator.data[DATA_KEY_ADDONS][self._addon_slug][ - self.entity_description.key - ] + return self.entity_description.value_fn(self) class HassioStatsSensor(HassioStatsEntity, SensorEntity): """Sensor to track container stats.""" + entity_description: HassioStatsSensorEntityDescription + @property - def native_value(self) -> str: + def native_value(self) -> float: """Return native value of entity.""" - if self._data_key == DATA_KEY_ADDONS: - return self.coordinator.data[DATA_KEY_ADDONS][self._container_id][ - self.entity_description.key - ] - return self.coordinator.data[self._data_key][self.entity_description.key] + return self.entity_description.value_fn(self) class HassioOSSensor(HassioOSEntity, SensorEntity): """Sensor to track a Hass.io OS attribute.""" + entity_description: HassioOSSensorEntityDescription + @property - def native_value(self) -> str: + def native_value(self) -> str | None: """Return native value of entity.""" - return self.coordinator.data[DATA_KEY_OS][self.entity_description.key] + return self.entity_description.value_fn(self) class HostSensor(HassioHostEntity, SensorEntity): """Sensor to track a host attribute.""" + entity_description: HassioHostSensorEntityDescription + @property - def native_value(self) -> str: + def native_value(self) -> str | float | None: """Return native value of entity.""" - return self.coordinator.data[DATA_KEY_HOST][self.entity_description.key] + return self.entity_description.value_fn(self) diff --git a/homeassistant/components/hassio/switch.py b/homeassistant/components/hassio/switch.py index 9bf9374a532..9454b917e33 100644 --- a/homeassistant/components/hassio/switch.py +++ b/homeassistant/components/hassio/switch.py @@ -4,15 +4,15 @@ import logging from typing import Any from aiohasupervisor import SupervisorError +from aiohasupervisor.models import AddonState from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ATTR_ICON from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import ADDONS_COORDINATOR, ATTR_STARTED, ATTR_STATE, DATA_KEY_ADDONS +from .const import ADDONS_COORDINATOR from .entity import HassioAddonEntity from .handler import get_supervisor_client @@ -20,7 +20,7 @@ _LOGGER = logging.getLogger(__name__) ENTITY_DESCRIPTION = SwitchEntityDescription( - key=ATTR_STATE, + key="state", name=None, icon="mdi:puzzle", entity_registry_enabled_default=False, @@ -41,7 +41,7 @@ async def async_setup_entry( coordinator=coordinator, entity_description=ENTITY_DESCRIPTION, ) - for addon in coordinator.data[DATA_KEY_ADDONS].values() + for addon in coordinator.data.addons.values() ) @@ -49,19 +49,19 @@ class HassioAddonSwitch(HassioAddonEntity, SwitchEntity): """Switch for Hass.io add-ons.""" @property - def is_on(self) -> bool | None: + def is_on(self) -> bool: """Return true if the add-on is on.""" - addon_data = self.coordinator.data[DATA_KEY_ADDONS].get(self._addon_slug, {}) - state = addon_data.get(self.entity_description.key) - return state == ATTR_STARTED + return ( + self.coordinator.data.addons[self._addon_slug].addon.state + == AddonState.STARTED + ) @property def entity_picture(self) -> str | None: """Return the icon of the add-on if any.""" if not self.available: return None - addon_data = self.coordinator.data[DATA_KEY_ADDONS].get(self._addon_slug, {}) - if addon_data.get(ATTR_ICON): + if self.coordinator.data.addons[self._addon_slug].addon.icon: return f"/api/hassio/addons/{self._addon_slug}/icon" return None diff --git a/homeassistant/components/hassio/update.py b/homeassistant/components/hassio/update.py index 44d36db7439..7005f1ac324 100644 --- a/homeassistant/components/hassio/update.py +++ b/homeassistant/components/hassio/update.py @@ -13,22 +13,12 @@ from homeassistant.components.update import ( UpdateEntityFeature, ) from homeassistant.config_entries import ConfigEntry -from homeassistant.const import ATTR_ICON, ATTR_NAME from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import ( - ADDONS_COORDINATOR, - ATTR_AUTO_UPDATE, - ATTR_VERSION, - ATTR_VERSION_LATEST, - DATA_KEY_ADDONS, - DATA_KEY_CORE, - DATA_KEY_OS, - DATA_KEY_SUPERVISOR, - MAIN_COORDINATOR, -) +from .const import ADDONS_COORDINATOR, ATTR_VERSION_LATEST, MAIN_COORDINATOR +from .coordinator import AddonData from .entity import ( HassioAddonEntity, HassioCoreEntity, @@ -78,7 +68,7 @@ async def async_setup_entry( coordinator=addons_coordinator, entity_description=ENTITY_DESCRIPTION, ) - for addon in addons_coordinator.data[DATA_KEY_ADDONS].values() + for addon in addons_coordinator.data.addons.values() ) async_add_entities(entities) @@ -108,29 +98,29 @@ class SupervisorAddonUpdateEntity(HassioAddonEntity, UpdateEntity): _version_before_update: str | None = None @property - def _addon_data(self) -> dict: + def _addon_data(self) -> AddonData: """Return the add-on data.""" - return self.coordinator.data[DATA_KEY_ADDONS][self._addon_slug] + return self.coordinator.data.addons[self._addon_slug] @property def auto_update(self) -> bool: """Return true if auto-update is enabled for the add-on.""" - return self._addon_data[ATTR_AUTO_UPDATE] + return self._addon_data.auto_update @property def title(self) -> str | None: """Return the title of the update.""" - return self._addon_data[ATTR_NAME] + return self._addon_data.addon.name @property def latest_version(self) -> str | None: """Latest version available for install.""" - return self._addon_data[ATTR_VERSION_LATEST] + return self._addon_data.addon.version_latest @property def installed_version(self) -> str | None: """Version installed and in use.""" - return self._addon_data[ATTR_VERSION] + return self._addon_data.addon.version @property def in_progress(self) -> bool | None: @@ -144,7 +134,7 @@ class SupervisorAddonUpdateEntity(HassioAddonEntity, UpdateEntity): """Return the icon of the add-on if any.""" if not self.available: return None - if self._addon_data[ATTR_ICON]: + if self._addon_data.addon.icon: return f"/api/hassio/addons/{self._addon_slug}/icon" return None @@ -236,14 +226,16 @@ class SupervisorOSUpdateEntity(HassioOSEntity, UpdateEntity): _attr_title = "Home Assistant Operating System" @property - def latest_version(self) -> str: + def latest_version(self) -> str | None: """Return the latest version.""" - return self.coordinator.data[DATA_KEY_OS][ATTR_VERSION_LATEST] + assert self.coordinator.data.os is not None + return self.coordinator.data.os.version_latest @property - def installed_version(self) -> str: + def installed_version(self) -> str | None: """Return the installed version.""" - return self.coordinator.data[DATA_KEY_OS][ATTR_VERSION] + assert self.coordinator.data.os is not None + return self.coordinator.data.os.version @property def entity_picture(self) -> str | None: @@ -293,19 +285,19 @@ class SupervisorSupervisorUpdateEntity(HassioSupervisorEntity, UpdateEntity): return self._attr_in_progress @property - def latest_version(self) -> str: + def latest_version(self) -> str | None: """Return the latest version.""" - return self.coordinator.data[DATA_KEY_SUPERVISOR][ATTR_VERSION_LATEST] + return self.coordinator.data.supervisor.version_latest @property def installed_version(self) -> str: """Return the installed version.""" - return self.coordinator.data[DATA_KEY_SUPERVISOR][ATTR_VERSION] + return self.coordinator.data.supervisor.version @property def auto_update(self) -> bool: """Return true if auto-update is enabled for supervisor.""" - return self.coordinator.data[DATA_KEY_SUPERVISOR][ATTR_AUTO_UPDATE] + return self.coordinator.data.supervisor.auto_update @property def release_url(self) -> str | None: @@ -389,14 +381,14 @@ class SupervisorCoreUpdateEntity(HassioCoreEntity, UpdateEntity): _attr_title = "Home Assistant Core" @property - def latest_version(self) -> str: + def latest_version(self) -> str | None: """Return the latest version.""" - return self.coordinator.data[DATA_KEY_CORE][ATTR_VERSION_LATEST] + return self.coordinator.data.core.version_latest @property - def installed_version(self) -> str: + def installed_version(self) -> str | None: """Return the installed version.""" - return self.coordinator.data[DATA_KEY_CORE][ATTR_VERSION] + return self.coordinator.data.core.version @property def entity_picture(self) -> str | None: diff --git a/tests/components/hassio/common.py b/tests/components/hassio/common.py index d98d68e48aa..45f9d11e3e2 100644 --- a/tests/components/hassio/common.py +++ b/tests/components/hassio/common.py @@ -120,6 +120,7 @@ def mock_addon_info( supervisor_api=False, supervisor_role="default", icon=False, + auto_update=False, ) addon_info.name = "test" addon_info.to_dict = MethodType( diff --git a/tests/components/hassio/test_init.py b/tests/components/hassio/test_init.py index 5978de0c92b..9f633112a29 100644 --- a/tests/components/hassio/test_init.py +++ b/tests/components/hassio/test_init.py @@ -37,8 +37,18 @@ from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAI from homeassistant.components.hassio import ( ADDONS_COORDINATOR, DOMAIN, + get_addons_info, + get_addons_list, + get_addons_stats, get_core_info, + get_core_stats, + get_host_info, + get_info, + get_network_info, + get_os_info, + get_store, get_supervisor_info, + get_supervisor_stats, hostname_from_addon_slug, ) from homeassistant.components.hassio.config import STORAGE_KEY @@ -1522,3 +1532,200 @@ async def test_get_supervisor_info(hass: HomeAssistant) -> None: assert "addons" in result assert isinstance(result["addons"], list) assert all(isinstance(addon, dict) for addon in result["addons"]) + + +@pytest.mark.usefixtures("mock_all") +async def test_get_info(hass: HomeAssistant) -> None: + """Test get_info returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_info(hass) + assert isinstance(result, dict) + assert result["supervisor"] == "222" + assert result["homeassistant"] == "0.110.0" + assert result["hassos"] == "1.2.3" + + +@pytest.mark.usefixtures("mock_all") +async def test_get_host_info(hass: HomeAssistant) -> None: + """Test get_host_info returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_host_info(hass) + assert isinstance(result, dict) + assert result["chassis"] == "vm" + assert result["disk_total"] == 100.0 + assert result["kernel"] == "4.19.0-6-amd64" + + +@pytest.mark.usefixtures("mock_all") +async def test_get_store(hass: HomeAssistant) -> None: + """Test get_store returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_store(hass) + assert isinstance(result, dict) + assert "addons" in result + assert "repositories" in result + assert isinstance(result["addons"], list) + assert isinstance(result["repositories"], list) + + +@pytest.mark.usefixtures("mock_all") +async def test_get_network_info(hass: HomeAssistant) -> None: + """Test get_network_info returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_network_info(hass) + assert isinstance(result, dict) + assert result["host_internet"] is True + assert result["supervisor_internet"] is True + assert isinstance(result["interfaces"], list) + + +@pytest.mark.usefixtures("mock_all") +async def test_get_addons_info(hass: HomeAssistant) -> None: + """Test get_addons_info returns serialized dicts, not model objects.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_addons_info(hass) + assert isinstance(result, dict) + assert "test" in result + assert isinstance(result["test"], dict) + assert result["test"]["slug"] == "test" + assert result["test"]["version"] == "1.0.0" + assert result["test"]["hassio_api"] is False + assert result["test"]["supervisor_api"] is False + assert result["test"]["hassio_role"] == "default" + assert result["test"]["supervisor_role"] == "default" + + +@pytest.mark.usefixtures("mock_all") +async def test_get_addons_list(hass: HomeAssistant) -> None: + """Test get_addons_list returns a list of serialized dicts.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_addons_list(hass) + assert isinstance(result, list) + assert all(isinstance(addon, dict) for addon in result) + slugs = {addon["slug"] for addon in result} + assert "test" in slugs + assert "test2" in slugs + + +@pytest.mark.usefixtures("mock_all", "entity_registry_enabled_by_default") +async def test_get_addons_stats(hass: HomeAssistant) -> None: + """Test get_addons_stats returns serialized dicts, not model objects. + + Both test addons are STOPPED in mock_all so no addon stats are fetched; + the result is an empty dict which is the correct return type. + """ + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_addons_stats(hass) + assert isinstance(result, dict) + # All values must be plain dicts, never AddonsStats model objects + for stats in result.values(): + assert isinstance(stats, dict) + + +@pytest.mark.usefixtures("mock_all", "entity_registry_enabled_by_default") +async def test_get_core_stats(hass: HomeAssistant) -> None: + """Test get_core_stats returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + # Stats entities subscribe during setup and trigger a debounced refresh + async_fire_time_changed( + hass, dt_util.now() + timedelta(seconds=REQUEST_REFRESH_DELAY) + ) + await hass.async_block_till_done() + + result = get_core_stats(hass) + assert isinstance(result, dict) + assert result["cpu_percent"] == 0.99 + assert result["memory_percent"] == 4.59 + + +@pytest.mark.usefixtures("mock_all", "entity_registry_enabled_by_default") +async def test_get_supervisor_stats(hass: HomeAssistant) -> None: + """Test get_supervisor_stats returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + # Stats entities subscribe during setup and trigger a debounced refresh + async_fire_time_changed( + hass, dt_util.now() + timedelta(seconds=REQUEST_REFRESH_DELAY) + ) + await hass.async_block_till_done() + + result = get_supervisor_stats(hass) + assert isinstance(result, dict) + assert result["cpu_percent"] == 0.99 + assert result["memory_percent"] == 4.59 + + +@pytest.mark.usefixtures("mock_all") +async def test_get_os_info(hass: HomeAssistant) -> None: + """Test get_os_info returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_os_info(hass) + assert isinstance(result, dict) + assert result["version"] == "1.0.0" + assert result["version_latest"] == "1.0.0" + assert result["update_available"] is False + + +@pytest.mark.usefixtures("mock_all") +async def test_get_core_info(hass: HomeAssistant) -> None: + """Test get_core_info returns serialized dict with expected values.""" + with patch.dict(os.environ, MOCK_ENVIRON): + config_entry = MockConfigEntry(domain=DOMAIN, data={}, unique_id=DOMAIN) + config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(config_entry.entry_id) + await hass.async_block_till_done() + + result = get_core_info(hass) + assert isinstance(result, dict) + assert result["version"] == "1.0.0" + assert result["version_latest"] == "1.0.0" + assert result["image"] == "homeassistant" diff --git a/tests/components/http/test_ban.py b/tests/components/http/test_ban.py index ce553beb01a..ab947e5bbf9 100644 --- a/tests/components/http/test_ban.py +++ b/tests/components/http/test_ban.py @@ -251,6 +251,13 @@ async def test_ip_ban_manager_never_started( "os_info", "store_info", "supervisor_info", + "homeassistant_info", + "host_info", + "network_info", + "addons_list", + "addon_info", + "homeassistant_stats", + "supervisor_stats", "ingress_panels", ) async def test_access_from_supervisor_ip(