diff --git a/homeassistant/components/tplink_omada/__init__.py b/homeassistant/components/tplink_omada/__init__.py index 470c56af2ade..299db284767e 100644 --- a/homeassistant/components/tplink_omada/__init__.py +++ b/homeassistant/components/tplink_omada/__init__.py @@ -75,12 +75,17 @@ async def async_setup_entry(hass: HomeAssistant, entry: OmadaConfigEntry) -> boo ) from ex site_client = await client.get_site_client(OmadaSite("", entry.data[CONF_SITE])) - controller = OmadaSiteController(hass, entry, site_client) + controller = OmadaSiteController(hass, entry, client, site_client) await controller.initialize_first_refresh() entry.runtime_data = controller - _remove_old_devices(hass, entry, controller.devices_coordinator.data) + _remove_old_devices( + hass, + entry, + controller.devices_coordinator.data, + controller.controller_status_coordinator.data.mac, + ) await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) @@ -96,6 +101,7 @@ def _remove_old_devices( hass: HomeAssistant, entry: OmadaConfigEntry, omada_devices: dict[str, OmadaListDevice], + controller_mac: str, ) -> None: device_registry = dr.async_get(hass) @@ -105,7 +111,7 @@ def _remove_old_devices( mac = next( (i[1] for i in registered_device.identifiers if i[0] == DOMAIN), None ) - if mac and mac not in omada_devices: + if mac and mac != controller_mac and mac not in omada_devices: device_registry.async_remove_device(registered_device.id) diff --git a/homeassistant/components/tplink_omada/config_flow.py b/homeassistant/components/tplink_omada/config_flow.py index f29716335d99..7bacf8d53ea6 100644 --- a/homeassistant/components/tplink_omada/config_flow.py +++ b/homeassistant/components/tplink_omada/config_flow.py @@ -84,7 +84,8 @@ async def _validate_input(hass: HomeAssistant, data: dict[str, Any]) -> HubInfo: client = await create_omada_client(hass, data) controller_id = await client.login() - name = await client.get_controller_name() + controller_status = await client.get_controller_status() + name = controller_status.name or controller_status.model sites = await client.get_sites() return HubInfo(controller_id, name, sites) diff --git a/homeassistant/components/tplink_omada/controller.py b/homeassistant/components/tplink_omada/controller.py index fbb4b6712964..4060699e6157 100644 --- a/homeassistant/components/tplink_omada/controller.py +++ b/homeassistant/components/tplink_omada/controller.py @@ -3,7 +3,7 @@ from collections.abc import Awaitable, Callable from typing import TYPE_CHECKING -from tplink_omada_client import OmadaSiteClient +from tplink_omada_client import OmadaClient, OmadaSiteClient from tplink_omada_client.devices import OmadaListDevice, OmadaSwitch from homeassistant.core import HomeAssistant, callback @@ -13,6 +13,8 @@ if TYPE_CHECKING: from .coordinator import ( OmadaClientsCoordinator, + OmadaControllerStatusCoordinator, + OmadaControllerUpdateCoordinator, OmadaDevicesCoordinator, OmadaGatewayCoordinator, OmadaSwitchPortCoordinator, @@ -28,13 +30,21 @@ class OmadaSiteController: self, hass: HomeAssistant, config_entry: OmadaConfigEntry, + controller_client: OmadaClient, omada_client: OmadaSiteClient, ) -> None: """Create the controller.""" self._hass = hass self._config_entry = config_entry + self._controller_client = controller_client self._omada_client = omada_client + self._controller_status_coordinator = OmadaControllerStatusCoordinator( + hass, config_entry, controller_client + ) + self._controller_update_coordinator = OmadaControllerUpdateCoordinator( + hass, config_entry, controller_client + ) self._switch_port_coordinators: dict[str, OmadaSwitchPortCoordinator] = {} self._devices_coordinator = OmadaDevicesCoordinator( hass, config_entry, omada_client @@ -45,6 +55,8 @@ class OmadaSiteController: async def initialize_first_refresh(self) -> None: """Initialize the all coordinators, and perform first refresh.""" + await self._controller_status_coordinator.async_config_entry_first_refresh() + await self._controller_update_coordinator.async_request_refresh() await self._devices_coordinator.async_config_entry_first_refresh() devices = self._devices_coordinator.data.values() @@ -69,8 +81,7 @@ class OmadaSiteController: Args: device_filter: Function that returns True if a device should be processed. - entity_callback: Given a discovered Omada device, - creates entities for that device. + entity_callback: Given a discovered Omada device, creates entities for that device. """ # Track which devices have been processed already processed_devices: set[str] = set() @@ -101,6 +112,11 @@ class OmadaSiteController: # Call once on initial setup await _async_register_entities() + @property + def controller_client(self) -> OmadaClient: + """Get the connected client API for the Omada Controller.""" + return self._controller_client + @property def omada_client(self) -> OmadaSiteClient: """Get the connected client API for the site to manage.""" @@ -117,6 +133,16 @@ class OmadaSiteController: return self._switch_port_coordinators[switch.mac] + @property + def controller_status_coordinator(self) -> OmadaControllerStatusCoordinator: + """Get the coordinator for the Omada Controller status.""" + return self._controller_status_coordinator + + @property + def controller_update_coordinator(self) -> OmadaControllerUpdateCoordinator: + """Get the coordinator for Omada Controller firmware updates.""" + return self._controller_update_coordinator + @property def gateway_coordinator(self) -> OmadaGatewayCoordinator | None: """Gets the coordinator for site's gateway, or None if there is no gateway.""" diff --git a/homeassistant/components/tplink_omada/coordinator.py b/homeassistant/components/tplink_omada/coordinator.py index c15dfc9d96c4..3388da0811b4 100644 --- a/homeassistant/components/tplink_omada/coordinator.py +++ b/homeassistant/components/tplink_omada/coordinator.py @@ -5,7 +5,13 @@ from datetime import timedelta import logging from typing import TYPE_CHECKING, NamedTuple, override -from tplink_omada_client import OmadaSiteClient, OmadaSwitchPortDetails +from tplink_omada_client import ( + OmadaClient, + OmadaControllerStatus, + OmadaControllerUpdateInfo, + OmadaSiteClient, + OmadaSwitchPortDetails, +) from tplink_omada_client.clients import OmadaWirelessClient from tplink_omada_client.devices import ( OmadaFirmwareUpdate, @@ -29,6 +35,8 @@ POLL_SWITCH_PORT = 30 POLL_GATEWAY = 300 POLL_CLIENTS = 300 POLL_DEVICES = 300 +POLL_CONTROLLER = 300 +POLL_CONTROLLER_UPDATE = 3600 POLL_UPGRADE = 60 @@ -72,6 +80,76 @@ class OmadaCoordinator[_T](DataUpdateCoordinator[dict[str, _T]]): raise NotImplementedError("Update method not implemented") +class OmadaControllerStatusCoordinator(DataUpdateCoordinator[OmadaControllerStatus]): + """Coordinator for getting status information about the Omada Controller.""" + + config_entry: OmadaConfigEntry + + def __init__( + self, + hass: HomeAssistant, + config_entry: OmadaConfigEntry, + omada_client: OmadaClient, + ) -> None: + """Initialize the controller status coordinator.""" + super().__init__( + hass, + _LOGGER, + config_entry=config_entry, + name="Omada API Data - Controller Status", + update_interval=timedelta(seconds=POLL_CONTROLLER), + ) + self.omada_client = omada_client + + @override + async def _async_update_data(self) -> OmadaControllerStatus: + """Fetch controller status from the API.""" + try: + async with asyncio.timeout(10): + return await self.omada_client.get_controller_status() + except OmadaClientException as err: + raise UpdateFailed( + translation_domain=DOMAIN, + translation_key="api_error", + ) from err + + +class OmadaControllerUpdateCoordinator( + DataUpdateCoordinator[OmadaControllerUpdateInfo] +): + """Coordinator for controller firmware update information.""" + + config_entry: OmadaConfigEntry + + def __init__( + self, + hass: HomeAssistant, + config_entry: OmadaConfigEntry, + omada_client: OmadaClient, + ) -> None: + """Initialize the controller update coordinator.""" + super().__init__( + hass, + _LOGGER, + config_entry=config_entry, + name="Omada API Data - Controller Firmware Update", + update_interval=timedelta(seconds=POLL_CONTROLLER_UPDATE), + ) + self.omada_client = omada_client + + @override + async def _async_update_data(self) -> OmadaControllerUpdateInfo: + """Fetch controller firmware update information from the API.""" + try: + async with asyncio.timeout(10): + return await self.omada_client.check_firmware_updates() + except OmadaClientException as err: + raise UpdateFailed( + translation_domain=DOMAIN, + translation_key="api_error", + ) from err + + class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]): """Coordinator for getting details about ports on a switch.""" diff --git a/homeassistant/components/tplink_omada/entity.py b/homeassistant/components/tplink_omada/entity.py index 609582f6bc2d..23578e22a34b 100644 --- a/homeassistant/components/tplink_omada/entity.py +++ b/homeassistant/components/tplink_omada/entity.py @@ -1,14 +1,16 @@ """Base entity definitions.""" -from typing import Any +from typing import Any, override +from tplink_omada_client import OmadaControllerStatus from tplink_omada_client.devices import OmadaDevice, OmadaSwitchPortDetails +from homeassistant.core import callback from homeassistant.helpers import device_registry as dr from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN -from .coordinator import OmadaCoordinator +from .coordinator import OmadaControllerStatusCoordinator, OmadaCoordinator class OmadaDeviceEntity[_T: OmadaCoordinator[Any]](CoordinatorEntity[_T]): @@ -34,3 +36,52 @@ def get_switch_port_base_name(port: OmadaSwitchPortDetails) -> str: if port.name == f"Port{port.port}": return str(port.port) return f"{port.port} ({port.name})" + + +class OmadaControllerEntity(CoordinatorEntity[OmadaControllerStatusCoordinator]): + """Common base class for entities associated with the Omada Controller.""" + + _attr_has_entity_name = True + + def __init__(self, coordinator: OmadaControllerStatusCoordinator) -> None: + """Initialize the controller entity.""" + super().__init__(coordinator) + + controller: OmadaControllerStatus = coordinator.data + self._controller_identifier = (DOMAIN, controller.mac) + + device_name = ( + f"{controller.model} - {controller.name}" + if controller.name + else controller.model + ) + + self._attr_device_info = dr.DeviceInfo( + connections={(dr.CONNECTION_NETWORK_MAC, controller.mac)}, + identifiers={(DOMAIN, controller.mac)}, + manufacturer="TP-Link", + model=controller.model, + name=device_name, + sw_version=controller.current_version, + ) + + @callback + @override + def _handle_coordinator_update(self) -> None: + """Handle updated controller status data.""" + device_registry = dr.async_get(self.hass) + controller = self.coordinator.data + device_entry = device_registry.async_get_device_by_identifier( + self._controller_identifier, + self.coordinator.config_entry.entry_id, + ) + if ( + device_entry is not None + and device_entry.sw_version != controller.current_version + ): + device_registry.async_update_device( + device_entry.id, + sw_version=controller.current_version, + ) + + super()._handle_coordinator_update() diff --git a/homeassistant/components/tplink_omada/sensor.py b/homeassistant/components/tplink_omada/sensor.py index fbbea0268d63..3fdb42ea71d2 100644 --- a/homeassistant/components/tplink_omada/sensor.py +++ b/homeassistant/components/tplink_omada/sensor.py @@ -24,9 +24,14 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from homeassistant.helpers.typing import StateType from . import OmadaConfigEntry +from .config_flow import CONF_SITE from .const import OmadaDeviceStatus -from .coordinator import OmadaDevicesCoordinator, OmadaSwitchPortCoordinator -from .entity import OmadaDeviceEntity, get_switch_port_base_name +from .coordinator import ( + OmadaControllerStatusCoordinator, + OmadaDevicesCoordinator, + OmadaSwitchPortCoordinator, +) +from .entity import OmadaControllerEntity, OmadaDeviceEntity, get_switch_port_base_name PARALLEL_UPDATES = 0 @@ -68,6 +73,10 @@ async def async_setup_entry( """Set up sensors.""" controller = config_entry.runtime_data + async_add_entities( + [OmadaControllerStatusSensor(controller.controller_status_coordinator)] + ) + devices_coordinator = controller.devices_coordinator async def _create_device_sensor_entities( @@ -153,6 +162,22 @@ OMADA_DEVICE_SENSORS: list[OmadaDeviceSensorEntityDescription] = [ ] +class OmadaControllerStatusSensor(OmadaControllerEntity, SensorEntity): + """Status sensor for the Omada Controller.""" + + _attr_translation_key = "device_status" + _attr_device_class = SensorDeviceClass.ENUM + _attr_entity_category = EntityCategory.DIAGNOSTIC + _attr_options = [v.value for v in OmadaDeviceStatus] + _attr_native_value = OmadaDeviceStatus.CONNECTED.value + + def __init__(self, coordinator: OmadaControllerStatusCoordinator) -> None: + """Initialize the controller status sensor.""" + super().__init__(coordinator) + site_id = coordinator.config_entry.data[CONF_SITE] + self._attr_unique_id = f"{coordinator.data.mac}_{site_id}_device_status" + + class OmadaDeviceSensor(OmadaDeviceEntity[OmadaDevicesCoordinator], SensorEntity): """Sensor for property of a generic Omada device.""" diff --git a/homeassistant/components/tplink_omada/strings.json b/homeassistant/components/tplink_omada/strings.json index 36d0e1894045..1fd655c87e9d 100644 --- a/homeassistant/components/tplink_omada/strings.json +++ b/homeassistant/components/tplink_omada/strings.json @@ -99,6 +99,11 @@ "wan_connect_ipv6": { "name": "Port {port_name} Internet connected (IPv6)" } + }, + "update": { + "firmware": { + "name": "Firmware" + } } }, "exceptions": { diff --git a/homeassistant/components/tplink_omada/update.py b/homeassistant/components/tplink_omada/update.py index ede62897db45..ebb085a4b2a2 100644 --- a/homeassistant/components/tplink_omada/update.py +++ b/homeassistant/components/tplink_omada/update.py @@ -1,7 +1,8 @@ """Support for TPLink Omada device firmware updates.""" -from typing import Any, override +from typing import Any, cast, override +from tplink_omada_client import OmadaControllerUpdateInfo from tplink_omada_client.devices import OmadaListDevice from tplink_omada_client.exceptions import OmadaClientException, RequestFailed @@ -10,14 +11,20 @@ from homeassistant.components.update import ( UpdateEntity, UpdateEntityFeature, ) +from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant, callback from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import OmadaConfigEntry +from .config_flow import CONF_SITE from .const import DOMAIN -from .coordinator import OmadaFirmwareUpdateCoordinator -from .entity import OmadaDeviceEntity +from .coordinator import ( + OmadaControllerStatusCoordinator, + OmadaControllerUpdateCoordinator, + OmadaFirmwareUpdateCoordinator, +) +from .entity import OmadaControllerEntity, OmadaDeviceEntity PARALLEL_UPDATES = 0 @@ -27,7 +34,7 @@ async def async_setup_entry( config_entry: OmadaConfigEntry, async_add_entities: AddConfigEntryEntitiesCallback, ) -> None: - """Set up switches.""" + """Set up firmware updates.""" controller = config_entry.runtime_data devices = controller.devices_coordinator.data @@ -37,11 +44,153 @@ async def async_setup_entry( ) async_add_entities( - OmadaDeviceUpdate(coordinator, device) for device in devices.values() + [ + OmadaControllerUpdate( + controller.controller_status_coordinator, + controller.controller_update_coordinator, + ), + *(OmadaDeviceUpdate(coordinator, device) for device in devices.values()), + ] ) await coordinator.async_request_refresh() +class OmadaControllerUpdate(OmadaControllerEntity, UpdateEntity): + """Firmware update status for the Omada Controller.""" + + _attr_translation_key = "firmware" + _attr_device_class = UpdateDeviceClass.FIRMWARE + _attr_entity_category = EntityCategory.CONFIG + + def __init__( + self, + status_coordinator: OmadaControllerStatusCoordinator, + update_coordinator: OmadaControllerUpdateCoordinator, + ) -> None: + """Initialize the controller update entity.""" + super().__init__(status_coordinator) + self._update_coordinator = update_coordinator + self._omada_client = update_coordinator.omada_client + site_id = status_coordinator.config_entry.data[CONF_SITE] + self._attr_unique_id = f"{status_coordinator.data.mac}_{site_id}_firmware" + + self._update_attrs() + + @override + async def async_added_to_hass(self) -> None: + """Register for controller update coordinator changes.""" + await super().async_added_to_hass() + self.async_on_remove( + self._update_coordinator.async_add_listener( + self._handle_update_coordinator_update + ) + ) + + @property + @override + def available(self) -> bool: + """Return if entity is available.""" + return super().available and self._update_coordinator.last_update_success + + @callback + def _handle_update_coordinator_update(self) -> None: + """Handle updated controller firmware information.""" + self._update_attrs() + self.async_write_ha_state() + + @callback + @override + def _handle_coordinator_update(self) -> None: + """Handle updated controller status data.""" + self._update_attrs() + super()._handle_coordinator_update() + + @property + def _update_data(self) -> OmadaControllerUpdateInfo | None: + """Return controller update data when the optional refresh succeeded.""" + return cast(OmadaControllerUpdateInfo | None, self._update_coordinator.data) + + def _update_attrs(self) -> None: + """Update installed and latest controller versions.""" + update = self._update_data + if update is None: + self._attr_installed_version = self.coordinator.data.current_version + self._attr_latest_version = self._attr_installed_version + self._attr_supported_features = UpdateEntityFeature(0) + return + + active_update = update.update + self._attr_installed_version = ( + active_update.current_version + if update.hardware is not None and active_update is not None + else self.coordinator.data.current_version or update.current_version + ) + self._attr_latest_version = ( + active_update.latest_version + if active_update is not None + else self._attr_installed_version + ) + self._attr_supported_features = UpdateEntityFeature.RELEASE_NOTES + if update.hardware is not None: + self._attr_supported_features |= UpdateEntityFeature.INSTALL + + @override + def release_notes(self) -> str | None: + """Return the release notes for the latest controller update.""" + if (update := self._update_data) is None: + return None + return update.release_notes + + @property + @override + def extra_state_attributes(self) -> dict[str, str] | None: + """Return the controller update download URL.""" + update = self._update_data + if ( + update is None + or update.update is None + or update.update.download_link is None + ): + return None + + return {"download_url": update.update.download_link} + + @override + async def async_install( + self, version: str | None, backup: bool, **kwargs: Any + ) -> None: + """Install a controller firmware update.""" + update = self._update_data + + if update is None or update.hardware is None: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="firmware_update_rejected", + ) + + target_version = version or update.latest_version + if target_version is None: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="firmware_update_rejected", + ) + + try: + await self._omada_client.install_controller_firmware(target_version) + except RequestFailed as ex: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="firmware_update_rejected", + ) from ex + except OmadaClientException as ex: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="firmware_update_failed", + ) from ex + finally: + await self._update_coordinator.async_request_refresh() + + class OmadaDeviceUpdate( OmadaDeviceEntity[OmadaFirmwareUpdateCoordinator], UpdateEntity, diff --git a/tests/components/tplink_omada/conftest.py b/tests/components/tplink_omada/conftest.py index d0209fdee923..7964e2c68aea 100644 --- a/tests/components/tplink_omada/conftest.py +++ b/tests/components/tplink_omada/conftest.py @@ -5,7 +5,11 @@ from functools import partial from unittest.mock import AsyncMock, MagicMock, patch import pytest -from tplink_omada_client import OmadaSite +from tplink_omada_client import ( + OmadaControllerStatus, + OmadaControllerUpdateInfo, + OmadaSite, +) from tplink_omada_client.clients import ( OmadaConnectedClient, OmadaNetworkClient, @@ -192,6 +196,28 @@ def mock_omada_client(mock_omada_site_client: AsyncMock) -> Generator[MagicMock] client.get_site_client.return_value = mock_omada_site_client client.login.return_value = "12345" client.get_controller_name.return_value = "OC200" + client.get_controller_status.return_value = OmadaControllerStatus( + { + "name": "Test Omada Controller", + "macAddress": "00-11-22-33-44-55", + "upTime": 123456, + "controllerVersion": "6.2.10.17", + "model": "OC200", + } + ) + client.check_firmware_updates.return_value = OmadaControllerUpdateInfo( + { + "software": { + "upgrade": True, + "currentVersion": "6.2.10.17", + "latestVersion": "6.3.0.45 Build 20260903171910", + "releaseLog": "Release notes for Omada SDN Controller.", + "releaseUrl": "https://example.com/controller-release-notes", + "downloadLink": "https://example.com/controller-update.tar.gz", + } + } + ) + client.install_controller_firmware = AsyncMock() client.get_sites.return_value = [OmadaSite("Display Name", "SiteId")] yield client @@ -208,6 +234,31 @@ def mock_omada_clients_only_client( client = client_mock.return_value client.get_site_client.return_value = mock_omada_clients_only_site_client + client.login.return_value = "12345" + client.get_controller_name.return_value = "OC200" + client.get_controller_status.return_value = OmadaControllerStatus( + { + "name": "Test Omada Controller", + "macAddress": "00-11-22-33-44-55", + "upTime": 123456, + "controllerVersion": "6.2.10.17", + "model": "OC200", + } + ) + client.check_firmware_updates.return_value = OmadaControllerUpdateInfo( + { + "software": { + "upgrade": True, + "currentVersion": "6.2.10.17", + "latestVersion": "6.3.0.45 Build 20260903171910", + "releaseLog": "Release notes for Omada SDN Controller.", + "releaseUrl": "https://example.com/controller-release-notes", + "downloadLink": "https://example.com/controller-update.tar.gz", + } + } + ) + client.install_controller_firmware = AsyncMock() + client.get_sites.return_value = [OmadaSite("Display Name", "SiteId")] yield client diff --git a/tests/components/tplink_omada/snapshots/test_sensor.ambr b/tests/components/tplink_omada/snapshots/test_sensor.ambr index baab53e1b347..eeac3be5208a 100644 --- a/tests/components/tplink_omada/snapshots/test_sensor.ambr +++ b/tests/components/tplink_omada/snapshots/test_sensor.ambr @@ -1,4 +1,74 @@ # serializer version: 1 +# name: test_entities[sensor.oc200_test_omada_controller_device_status-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : list([ + 'disconnected', + 'connected', + 'pending', + 'heartbeat_missed', + 'isolated', + 'adopt_failed', + 'managed_externally', + ]), + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.oc200_test_omada_controller_device_status', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Device status', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Device status', + 'platform': 'tplink_omada', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'device_status', + 'unique_id': '00-11-22-33-44-55_Default_device_status', + 'unit_of_measurement': None, + }) +# --- +# name: test_entities[sensor.oc200_test_omada_controller_device_status-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'enum', + : 'OC200 - Test Omada Controller Device status', + : list([ + 'disconnected', + 'connected', + 'pending', + 'heartbeat_missed', + 'isolated', + 'adopt_failed', + 'managed_externally', + ]), + }), + 'context': , + 'entity_id': 'sensor.oc200_test_omada_controller_device_status', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'connected', + }) +# --- # name: test_entities[sensor.test_poe_switch_cpu_usage-entry] EntityRegistryEntrySnapshot({ 'aliases': list([ diff --git a/tests/components/tplink_omada/snapshots/test_update.ambr b/tests/components/tplink_omada/snapshots/test_update.ambr index 46cd30814fcd..ad5960727c68 100644 --- a/tests/components/tplink_omada/snapshots/test_update.ambr +++ b/tests/components/tplink_omada/snapshots/test_update.ambr @@ -1,4 +1,68 @@ # serializer version: 1 +# name: test_entities[update.oc200_test_omada_controller_firmware-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'update', + 'entity_category': , + 'entity_id': 'update.oc200_test_omada_controller_firmware', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Firmware', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Firmware', + 'platform': 'tplink_omada', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': , + 'translation_key': 'firmware', + 'unique_id': '00-11-22-33-44-55_Default_firmware', + 'unit_of_measurement': None, + }) +# --- +# name: test_entities[update.oc200_test_omada_controller_firmware-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : False, + : 'firmware', + : 0, + 'download_url': 'https://example.com/controller-update.tar.gz', + : '/api/brands/integration/tplink_omada/icon.png', + : 'OC200 - Test Omada Controller Firmware', + : False, + : '6.2.10.17', + : '6.3.0.45 Build 20260903171910', + : None, + : None, + : None, + : , + : None, + : None, + }), + 'context': , + 'entity_id': 'update.oc200_test_omada_controller_firmware', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'on', + }) +# --- # name: test_entities[update.test_poe_switch_firmware-entry] EntityRegistryEntrySnapshot({ 'aliases': list([ diff --git a/tests/components/tplink_omada/test_config_flow.py b/tests/components/tplink_omada/test_config_flow.py index f2ac7107f186..f6967243004f 100644 --- a/tests/components/tplink_omada/test_config_flow.py +++ b/tests/components/tplink_omada/test_config_flow.py @@ -54,7 +54,7 @@ async def test_form_single_site( ) assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == "OC200 (Display Name)" + assert result["title"] == "Test Omada Controller (Display Name)" assert result["data"] == MOCK_ENTRY_DATA assert result["result"].unique_id == "12345_SiteId" assert len(mock_setup_entry.mock_calls) == 1 @@ -94,7 +94,7 @@ async def test_form_multiple_sites( ) assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == "OC200 (Site 2)" + assert result["title"] == "Test Omada Controller (Site 2)" assert result["data"] == { "host": "https://fake.omada.host", "verify_ssl": True, @@ -148,7 +148,7 @@ async def test_form_errors_and_recovery( ) assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == "OC200 (Display Name)" + assert result["title"] == "Test Omada Controller (Display Name)" assert result["data"] == MOCK_ENTRY_DATA assert len(mock_setup_entry.mock_calls) == 1 diff --git a/tests/components/tplink_omada/test_update.py b/tests/components/tplink_omada/test_update.py index d2d4457fee53..8fc42fc97bdd 100644 --- a/tests/components/tplink_omada/test_update.py +++ b/tests/components/tplink_omada/test_update.py @@ -6,20 +6,36 @@ from unittest.mock import AsyncMock, MagicMock, patch from freezegun.api import FrozenDateTimeFactory import pytest from syrupy.assertion import SnapshotAssertion +from tplink_omada_client import OmadaControllerStatus, OmadaControllerUpdateInfo from tplink_omada_client.devices import OmadaListDevice from tplink_omada_client.exceptions import OmadaClientException, RequestFailed from homeassistant.components.tplink_omada.const import DOMAIN -from homeassistant.components.tplink_omada.coordinator import POLL_DEVICES +from homeassistant.components.tplink_omada.coordinator import ( + POLL_CONTROLLER, + POLL_DEVICES, +) from homeassistant.components.update import ( ATTR_IN_PROGRESS, + ATTR_INSTALLED_VERSION, + ATTR_LATEST_VERSION, + DATA_COMPONENT, DOMAIN as UPDATE_DOMAIN, SERVICE_INSTALL, + UpdateEntityFeature, +) +from homeassistant.const import ( + ATTR_ENTITY_ID, + ATTR_SUPPORTED_FEATURES, + STATE_OFF, + STATE_ON, + STATE_UNAVAILABLE, + Platform, ) -from homeassistant.const import ATTR_ENTITY_ID, STATE_ON, Platform from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError -from homeassistant.helpers import entity_registry as er +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.update_coordinator import REQUEST_REFRESH_DEFAULT_COOLDOWN from tests.common import ( MockConfigEntry, @@ -30,6 +46,8 @@ from tests.common import ( from tests.typing import WebSocketGenerator POLL_INTERVAL = timedelta(seconds=POLL_DEVICES) +CONTROLLER_POLL_INTERVAL = timedelta(seconds=POLL_CONTROLLER) +REFRESH_COOLDOWN = timedelta(seconds=REQUEST_REFRESH_DEFAULT_COOLDOWN) async def _rebuild_device_list_with_update( @@ -131,6 +149,160 @@ async def test_install_firmware_success( assert await_args[0].mac == "54-AF-97-00-00-01" +async def test_install_controller_firmware_success( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_omada_client: MagicMock, + freezer: FrozenDateTimeFactory, +) -> None: + """Test successful controller firmware installation.""" + entity_id = "update.oc200_test_omada_controller_firmware" + mock_omada_client.check_firmware_updates.return_value = OmadaControllerUpdateInfo( + { + "hardware": { + "upgrade": True, + "currentVersion": "1.0.0", + "latestVersion": "1.0.1", + "fwReleaseLog": "Fixed things.", + "releaseUrl": "https://example.com/firmware-release-notes", + "downloadLink": "https://example.com/firmware.bin", + } + } + ) + mock_config_entry.add_to_hass(hass) + + with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]): + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + entity = hass.states.get(entity_id) + assert entity is not None + assert entity.state == STATE_ON + assert entity.attributes[ATTR_INSTALLED_VERSION] == "1.0.0" + assert entity.attributes[ATTR_LATEST_VERSION] == "1.0.1" + assert entity.attributes[ATTR_SUPPORTED_FEATURES] == ( + UpdateEntityFeature.RELEASE_NOTES | UpdateEntityFeature.INSTALL + ) + + mock_omada_client.check_firmware_updates.reset_mock() + + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + + freezer.tick(REFRESH_COOLDOWN) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + mock_omada_client.install_controller_firmware.assert_awaited_once_with("1.0.1") + mock_omada_client.check_firmware_updates.assert_awaited_once() + + +async def test_controller_update_check_failure_does_not_block_setup( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_omada_client: MagicMock, +) -> None: + """Test controller update check failures do not block setup.""" + entity_id = "update.oc200_test_omada_controller_firmware" + mock_omada_client.check_firmware_updates.side_effect = OmadaClientException( + "Connection error" + ) + mock_config_entry.add_to_hass(hass) + + with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]): + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + entity = hass.states.get(entity_id) + assert entity is not None + assert entity.state == STATE_UNAVAILABLE + + +async def test_controller_software_update_installed_version_prefers_status_coordinator( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_omada_client: MagicMock, +) -> None: + """Test controller software update installed version prefers controller status.""" + entity_id = "update.oc200_test_omada_controller_firmware" + mock_omada_client.get_controller_status.return_value = OmadaControllerStatus( + { + "name": "Test Omada Controller", + "macAddress": "00-11-22-33-44-55", + "upTime": 123456, + "controllerVersion": "6.3.0.45", + "model": "OC200", + } + ) + mock_omada_client.check_firmware_updates.return_value = OmadaControllerUpdateInfo( + { + "software": { + "upgrade": True, + "currentVersion": "6.2.10.17", + "latestVersion": "6.3.0.45", + } + } + ) + mock_config_entry.add_to_hass(hass) + + with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]): + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + entity = hass.states.get(entity_id) + assert entity is not None + assert entity.state == STATE_OFF + assert entity.attributes[ATTR_INSTALLED_VERSION] == "6.3.0.45" + assert entity.attributes[ATTR_LATEST_VERSION] == "6.3.0.45" + + +async def test_controller_device_sw_version_updates_with_status_coordinator( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_omada_client: MagicMock, + device_registry: dr.DeviceRegistry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test controller device software version updates with controller status.""" + mock_config_entry.add_to_hass(hass) + + with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]): + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + device_entry = device_registry.async_get_device_by_identifier( + (DOMAIN, "00-11-22-33-44-55"), + mock_config_entry.entry_id, + ) + assert device_entry is not None + assert device_entry.sw_version == "6.2.10.17" + + mock_omada_client.get_controller_status.return_value = OmadaControllerStatus( + { + "name": "Test Omada Controller", + "macAddress": "00-11-22-33-44-55", + "upTime": 123456, + "controllerVersion": "6.3.0.45", + "model": "OC200", + } + ) + + freezer.tick(CONTROLLER_POLL_INTERVAL) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + device_entry = device_registry.async_get_device_by_identifier( + (DOMAIN, "00-11-22-33-44-55"), + mock_config_entry.entry_id, + ) + assert device_entry is not None + assert device_entry.sw_version == "6.3.0.45" + + @pytest.mark.parametrize( ("exception_type", "translation_key"), [ @@ -172,9 +344,86 @@ async def test_install_firmware_exceptions( assert err.value.translation_domain == DOMAIN +@pytest.mark.parametrize( + ("exception_type", "translation_key"), + [ + ( + RequestFailed(500, "Update rejected"), + "firmware_update_rejected", + ), + ( + OmadaClientException("Connection error"), + "firmware_update_failed", + ), + ], +) +async def test_install_controller_firmware_exceptions( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_omada_client: MagicMock, + freezer: FrozenDateTimeFactory, + exception_type: Exception, + translation_key: str, +) -> None: + """Test controller firmware installation exception handling.""" + entity_id = "update.oc200_test_omada_controller_firmware" + mock_omada_client.check_firmware_updates.return_value = OmadaControllerUpdateInfo( + { + "hardware": { + "upgrade": True, + "currentVersion": "1.0.0", + "latestVersion": "1.0.1", + } + } + ) + mock_omada_client.install_controller_firmware = AsyncMock( + side_effect=exception_type + ) + mock_config_entry.add_to_hass(hass) + + with patch("homeassistant.components.tplink_omada.PLATFORMS", [Platform.UPDATE]): + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + mock_omada_client.check_firmware_updates.reset_mock() + + with pytest.raises(HomeAssistantError) as err: + await hass.services.async_call( + UPDATE_DOMAIN, + SERVICE_INSTALL, + {ATTR_ENTITY_ID: entity_id}, + blocking=True, + ) + + assert err.value.translation_key == translation_key + assert err.value.translation_domain == DOMAIN + freezer.tick(REFRESH_COOLDOWN) + async_fire_time_changed(hass) + await hass.async_block_till_done() + mock_omada_client.check_firmware_updates.assert_awaited_once() + + +async def test_install_controller_firmware_rejected_without_hardware( + hass: HomeAssistant, + init_integration: MockConfigEntry, +) -> None: + """Test controller firmware installation rejects software-only updates.""" + entity = hass.data[DATA_COMPONENT].get_entity( + "update.oc200_test_omada_controller_firmware" + ) + assert entity is not None + + with pytest.raises(HomeAssistantError) as err: + await entity.async_install(version=None, backup=False) + + assert err.value.translation_key == "firmware_update_rejected" + assert err.value.translation_domain == DOMAIN + + @pytest.mark.parametrize( ("entity_name", "expected_notes"), [ + ("oc200_test_omada_controller", "Release notes for Omada SDN Controller."), ("test_router", None), ("test_poe_switch", "Bug fixes and performance improvements"), ],