mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Add sensors to Omada (#127767)
Co-authored-by: Joostlek <joostlek@outlook.com>
This commit is contained in:
parent
275c86a0a9
commit
57ef175050
@ -24,6 +24,7 @@ from .controller import OmadaSiteController
|
|||||||
PLATFORMS: list[Platform] = [
|
PLATFORMS: list[Platform] = [
|
||||||
Platform.BINARY_SENSOR,
|
Platform.BINARY_SENSOR,
|
||||||
Platform.DEVICE_TRACKER,
|
Platform.DEVICE_TRACKER,
|
||||||
|
Platform.SENSOR,
|
||||||
Platform.SWITCH,
|
Platform.SWITCH,
|
||||||
Platform.UPDATE,
|
Platform.UPDATE,
|
||||||
]
|
]
|
||||||
|
@ -99,7 +99,6 @@ class OmadaGatewayPortBinarySensor(
|
|||||||
"""Binary status of a property on an internet gateway."""
|
"""Binary status of a property on an internet gateway."""
|
||||||
|
|
||||||
entity_description: GatewayPortBinarySensorEntityDescription
|
entity_description: GatewayPortBinarySensorEntityDescription
|
||||||
_attr_has_entity_name = True
|
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
|
@ -1,3 +1,17 @@
|
|||||||
"""Constants for the TP-Link Omada integration."""
|
"""Constants for the TP-Link Omada integration."""
|
||||||
|
|
||||||
|
from enum import StrEnum
|
||||||
|
|
||||||
DOMAIN = "tplink_omada"
|
DOMAIN = "tplink_omada"
|
||||||
|
|
||||||
|
|
||||||
|
class OmadaDeviceStatus(StrEnum):
|
||||||
|
"""Possible composite status values for Omada devices."""
|
||||||
|
|
||||||
|
DISCONNECTED = "disconnected"
|
||||||
|
CONNECTED = "connected"
|
||||||
|
PENDING = "pending"
|
||||||
|
HEARTBEAT_MISSED = "heartbeat_missed"
|
||||||
|
ISOLATED = "isolated"
|
||||||
|
ADOPT_FAILED = "adopt_failed"
|
||||||
|
MANAGED_EXTERNALLY = "managed_externally"
|
||||||
|
@ -17,7 +17,7 @@ _LOGGER = logging.getLogger(__name__)
|
|||||||
POLL_SWITCH_PORT = 300
|
POLL_SWITCH_PORT = 300
|
||||||
POLL_GATEWAY = 300
|
POLL_GATEWAY = 300
|
||||||
POLL_CLIENTS = 300
|
POLL_CLIENTS = 300
|
||||||
POLL_DEVICES = 900
|
POLL_DEVICES = 300
|
||||||
|
|
||||||
|
|
||||||
class OmadaCoordinator[_T](DataUpdateCoordinator[dict[str, _T]]):
|
class OmadaCoordinator[_T](DataUpdateCoordinator[dict[str, _T]]):
|
||||||
|
@ -14,6 +14,8 @@ from .coordinator import OmadaCoordinator
|
|||||||
class OmadaDeviceEntity[_T: OmadaCoordinator[Any]](CoordinatorEntity[_T]):
|
class OmadaDeviceEntity[_T: OmadaCoordinator[Any]](CoordinatorEntity[_T]):
|
||||||
"""Common base class for all entities associated with Omada SDN Devices."""
|
"""Common base class for all entities associated with Omada SDN Devices."""
|
||||||
|
|
||||||
|
_attr_has_entity_name = True
|
||||||
|
|
||||||
def __init__(self, coordinator: _T, device: OmadaDevice) -> None:
|
def __init__(self, coordinator: _T, device: OmadaDevice) -> None:
|
||||||
"""Initialize the device."""
|
"""Initialize the device."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator)
|
||||||
|
@ -18,6 +18,14 @@
|
|||||||
"off": "mdi:cloud-cancel"
|
"off": "mdi:cloud-cancel"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
},
|
||||||
|
"sensor": {
|
||||||
|
"cpu_usage": {
|
||||||
|
"default": "mdi:cpu-32-bit"
|
||||||
|
},
|
||||||
|
"mem_usage": {
|
||||||
|
"default": "mdi:memory"
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
132
homeassistant/components/tplink_omada/sensor.py
Normal file
132
homeassistant/components/tplink_omada/sensor.py
Normal file
@ -0,0 +1,132 @@
|
|||||||
|
"""Support for TPLink Omada binary sensors."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from collections.abc import Callable
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from tplink_omada_client.definitions import DeviceStatus, DeviceStatusCategory
|
||||||
|
from tplink_omada_client.devices import OmadaListDevice
|
||||||
|
|
||||||
|
from homeassistant.components.sensor import (
|
||||||
|
SensorDeviceClass,
|
||||||
|
SensorEntity,
|
||||||
|
SensorEntityDescription,
|
||||||
|
SensorStateClass,
|
||||||
|
)
|
||||||
|
from homeassistant.const import PERCENTAGE, EntityCategory
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
|
from homeassistant.helpers.typing import StateType
|
||||||
|
|
||||||
|
from . import OmadaConfigEntry
|
||||||
|
from .const import OmadaDeviceStatus
|
||||||
|
from .coordinator import OmadaDevicesCoordinator
|
||||||
|
from .entity import OmadaDeviceEntity
|
||||||
|
|
||||||
|
# Useful low level status categories, mapped to a more descriptive status.
|
||||||
|
DEVICE_STATUS_MAP = {
|
||||||
|
DeviceStatus.PROVISIONING: OmadaDeviceStatus.PENDING,
|
||||||
|
DeviceStatus.CONFIGURING: OmadaDeviceStatus.PENDING,
|
||||||
|
DeviceStatus.UPGRADING: OmadaDeviceStatus.PENDING,
|
||||||
|
DeviceStatus.REBOOTING: OmadaDeviceStatus.PENDING,
|
||||||
|
DeviceStatus.ADOPT_FAILED: OmadaDeviceStatus.ADOPT_FAILED,
|
||||||
|
DeviceStatus.ADOPT_FAILED_WIRELESS: OmadaDeviceStatus.ADOPT_FAILED,
|
||||||
|
DeviceStatus.MANAGED_EXTERNALLY: OmadaDeviceStatus.MANAGED_EXTERNALLY,
|
||||||
|
DeviceStatus.MANAGED_EXTERNALLY_WIRELESS: OmadaDeviceStatus.MANAGED_EXTERNALLY,
|
||||||
|
}
|
||||||
|
|
||||||
|
# High level status categories, suitable for most device statuses.
|
||||||
|
DEVICE_STATUS_CATEGORY_MAP = {
|
||||||
|
DeviceStatusCategory.DISCONNECTED: OmadaDeviceStatus.DISCONNECTED,
|
||||||
|
DeviceStatusCategory.CONNECTED: OmadaDeviceStatus.CONNECTED,
|
||||||
|
DeviceStatusCategory.PENDING: OmadaDeviceStatus.PENDING,
|
||||||
|
DeviceStatusCategory.HEARTBEAT_MISSED: OmadaDeviceStatus.HEARTBEAT_MISSED,
|
||||||
|
DeviceStatusCategory.ISOLATED: OmadaDeviceStatus.ISOLATED,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _map_device_status(device: OmadaListDevice) -> str | None:
|
||||||
|
"""Map the API device status to the best available descriptive device status."""
|
||||||
|
display_status = DEVICE_STATUS_MAP.get(
|
||||||
|
device.status
|
||||||
|
) or DEVICE_STATUS_CATEGORY_MAP.get(device.status_category)
|
||||||
|
return display_status.value if display_status else None
|
||||||
|
|
||||||
|
|
||||||
|
async def async_setup_entry(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
config_entry: OmadaConfigEntry,
|
||||||
|
async_add_entities: AddEntitiesCallback,
|
||||||
|
) -> None:
|
||||||
|
"""Set up sensors."""
|
||||||
|
controller = config_entry.runtime_data
|
||||||
|
|
||||||
|
devices_coordinator = controller.devices_coordinator
|
||||||
|
|
||||||
|
async_add_entities(
|
||||||
|
OmadaDeviceSensor(devices_coordinator, device, desc)
|
||||||
|
for device in devices_coordinator.data.values()
|
||||||
|
for desc in OMADA_DEVICE_SENSORS
|
||||||
|
if desc.exists_func(device)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, kw_only=True)
|
||||||
|
class OmadaDeviceSensorEntityDescription(SensorEntityDescription):
|
||||||
|
"""Entity description for a status derived from an Omada device in the device list."""
|
||||||
|
|
||||||
|
exists_func: Callable[[OmadaListDevice], bool] = lambda _: True
|
||||||
|
update_func: Callable[[OmadaListDevice], StateType]
|
||||||
|
|
||||||
|
|
||||||
|
OMADA_DEVICE_SENSORS: list[OmadaDeviceSensorEntityDescription] = [
|
||||||
|
OmadaDeviceSensorEntityDescription(
|
||||||
|
key="device_status",
|
||||||
|
translation_key="device_status",
|
||||||
|
device_class=SensorDeviceClass.ENUM,
|
||||||
|
entity_category=EntityCategory.DIAGNOSTIC,
|
||||||
|
update_func=_map_device_status,
|
||||||
|
options=[v.value for v in OmadaDeviceStatus],
|
||||||
|
),
|
||||||
|
OmadaDeviceSensorEntityDescription(
|
||||||
|
key="cpu_usage",
|
||||||
|
translation_key="cpu_usage",
|
||||||
|
entity_category=EntityCategory.DIAGNOSTIC,
|
||||||
|
state_class=SensorStateClass.MEASUREMENT,
|
||||||
|
native_unit_of_measurement=PERCENTAGE,
|
||||||
|
update_func=lambda device: device.cpu_usage,
|
||||||
|
),
|
||||||
|
OmadaDeviceSensorEntityDescription(
|
||||||
|
key="mem_usage",
|
||||||
|
translation_key="mem_usage",
|
||||||
|
entity_category=EntityCategory.DIAGNOSTIC,
|
||||||
|
state_class=SensorStateClass.MEASUREMENT,
|
||||||
|
native_unit_of_measurement=PERCENTAGE,
|
||||||
|
update_func=lambda device: device.mem_usage,
|
||||||
|
),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
class OmadaDeviceSensor(OmadaDeviceEntity[OmadaDevicesCoordinator], SensorEntity):
|
||||||
|
"""Sensor for property of a generic Omada device."""
|
||||||
|
|
||||||
|
entity_description: OmadaDeviceSensorEntityDescription
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
coordinator: OmadaDevicesCoordinator,
|
||||||
|
device: OmadaListDevice,
|
||||||
|
entity_description: OmadaDeviceSensorEntityDescription,
|
||||||
|
) -> None:
|
||||||
|
"""Initialize the device sensor."""
|
||||||
|
super().__init__(coordinator, device)
|
||||||
|
self.entity_description = entity_description
|
||||||
|
self._attr_unique_id = f"{device.mac}_{entity_description.key}"
|
||||||
|
|
||||||
|
@property
|
||||||
|
def native_value(self) -> StateType:
|
||||||
|
"""Return the state of the sensor."""
|
||||||
|
return self.entity_description.update_func(
|
||||||
|
self.coordinator.data[self.device.mac]
|
||||||
|
)
|
@ -65,6 +65,27 @@
|
|||||||
"poe_delivery": {
|
"poe_delivery": {
|
||||||
"name": "Port {port_name} PoE Delivery"
|
"name": "Port {port_name} PoE Delivery"
|
||||||
}
|
}
|
||||||
|
},
|
||||||
|
"sensor": {
|
||||||
|
"device_status": {
|
||||||
|
"name": "Device status",
|
||||||
|
"state": {
|
||||||
|
"error": "Error",
|
||||||
|
"disconnected": "[%key:common::state::disconnected%]",
|
||||||
|
"connected": "[%key:common::state::connected%]",
|
||||||
|
"pending": "Pending",
|
||||||
|
"heartbeat_missed": "Heartbeat missed",
|
||||||
|
"isolated": "Isolated",
|
||||||
|
"adopt_failed": "Adopt failed",
|
||||||
|
"managed_externally": "Managed externally"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"cpu_usage": {
|
||||||
|
"name": "CPU usage"
|
||||||
|
},
|
||||||
|
"mem_usage": {
|
||||||
|
"name": "Memory usage"
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -229,7 +229,6 @@ class OmadaDevicePortSwitchEntity(
|
|||||||
):
|
):
|
||||||
"""Generic toggle switch entity for a Netork Port of an Omada Device."""
|
"""Generic toggle switch entity for a Netork Port of an Omada Device."""
|
||||||
|
|
||||||
_attr_has_entity_name = True
|
|
||||||
entity_description: OmadaDevicePortSwitchEntityDescription[
|
entity_description: OmadaDevicePortSwitchEntityDescription[
|
||||||
TCoordinator, TDevice, TPort
|
TCoordinator, TDevice, TPort
|
||||||
]
|
]
|
||||||
|
@ -119,7 +119,6 @@ class OmadaDeviceUpdate(
|
|||||||
| UpdateEntityFeature.PROGRESS
|
| UpdateEntityFeature.PROGRESS
|
||||||
| UpdateEntityFeature.RELEASE_NOTES
|
| UpdateEntityFeature.RELEASE_NOTES
|
||||||
)
|
)
|
||||||
_attr_has_entity_name = True
|
|
||||||
_attr_device_class = UpdateDeviceClass.FIRMWARE
|
_attr_device_class = UpdateDeviceClass.FIRMWARE
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
|
@ -163,21 +163,10 @@ def mock_omada_clients_only_client(
|
|||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
async def init_integration(
|
async def init_integration(
|
||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
|
mock_config_entry: MockConfigEntry,
|
||||||
mock_omada_client: MagicMock,
|
mock_omada_client: MagicMock,
|
||||||
) -> MockConfigEntry:
|
) -> MockConfigEntry:
|
||||||
"""Set up the TP-Link Omada integration for testing."""
|
"""Set up the TP-Link Omada integration for testing."""
|
||||||
mock_config_entry = MockConfigEntry(
|
|
||||||
title="Test Omada Controller",
|
|
||||||
domain=DOMAIN,
|
|
||||||
data={
|
|
||||||
CONF_HOST: "127.0.0.1",
|
|
||||||
CONF_PASSWORD: "mocked-password",
|
|
||||||
CONF_USERNAME: "mocked-user",
|
|
||||||
CONF_VERIFY_SSL: False,
|
|
||||||
CONF_SITE: "Default",
|
|
||||||
},
|
|
||||||
unique_id="12345",
|
|
||||||
)
|
|
||||||
mock_config_entry.add_to_hass(hass)
|
mock_config_entry.add_to_hass(hass)
|
||||||
|
|
||||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||||
|
333
tests/components/tplink_omada/snapshots/test_sensor.ambr
Normal file
333
tests/components/tplink_omada/snapshots/test_sensor.ambr
Normal file
@ -0,0 +1,333 @@
|
|||||||
|
# serializer version: 1
|
||||||
|
# name: test_entities[sensor.test_poe_switch_cpu_usage-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_cpu_usage',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': None,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'CPU usage',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'cpu_usage',
|
||||||
|
'unique_id': '54-AF-97-00-00-01_cpu_usage',
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_poe_switch_cpu_usage-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'friendly_name': 'Test PoE Switch CPU usage',
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_cpu_usage',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': '10',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_poe_switch_device_status-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'options': list([
|
||||||
|
'disconnected',
|
||||||
|
'connected',
|
||||||
|
'pending',
|
||||||
|
'heartbeat_missed',
|
||||||
|
'isolated',
|
||||||
|
'adopt_failed',
|
||||||
|
'managed_externally',
|
||||||
|
]),
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_device_status',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'Device status',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'device_status',
|
||||||
|
'unique_id': '54-AF-97-00-00-01_device_status',
|
||||||
|
'unit_of_measurement': None,
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_poe_switch_device_status-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'device_class': 'enum',
|
||||||
|
'friendly_name': 'Test PoE Switch Device status',
|
||||||
|
'options': list([
|
||||||
|
'disconnected',
|
||||||
|
'connected',
|
||||||
|
'pending',
|
||||||
|
'heartbeat_missed',
|
||||||
|
'isolated',
|
||||||
|
'adopt_failed',
|
||||||
|
'managed_externally',
|
||||||
|
]),
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_device_status',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': 'connected',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_poe_switch_memory_usage-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_memory_usage',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': None,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'Memory usage',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'mem_usage',
|
||||||
|
'unique_id': '54-AF-97-00-00-01_mem_usage',
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_poe_switch_memory_usage-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'friendly_name': 'Test PoE Switch Memory usage',
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_poe_switch_memory_usage',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': '20',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_cpu_usage-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_router_cpu_usage',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': None,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'CPU usage',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'cpu_usage',
|
||||||
|
'unique_id': 'AA-BB-CC-DD-EE-FF_cpu_usage',
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_cpu_usage-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'friendly_name': 'Test Router CPU usage',
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_router_cpu_usage',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': '16',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_device_status-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'options': list([
|
||||||
|
'disconnected',
|
||||||
|
'connected',
|
||||||
|
'pending',
|
||||||
|
'heartbeat_missed',
|
||||||
|
'isolated',
|
||||||
|
'adopt_failed',
|
||||||
|
'managed_externally',
|
||||||
|
]),
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_router_device_status',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': <SensorDeviceClass.ENUM: 'enum'>,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'Device status',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'device_status',
|
||||||
|
'unique_id': 'AA-BB-CC-DD-EE-FF_device_status',
|
||||||
|
'unit_of_measurement': None,
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_device_status-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'device_class': 'enum',
|
||||||
|
'friendly_name': 'Test Router Device status',
|
||||||
|
'options': list([
|
||||||
|
'disconnected',
|
||||||
|
'connected',
|
||||||
|
'pending',
|
||||||
|
'heartbeat_missed',
|
||||||
|
'isolated',
|
||||||
|
'adopt_failed',
|
||||||
|
'managed_externally',
|
||||||
|
]),
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_router_device_status',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': 'connected',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_memory_usage-entry]
|
||||||
|
EntityRegistryEntrySnapshot({
|
||||||
|
'aliases': set({
|
||||||
|
}),
|
||||||
|
'area_id': None,
|
||||||
|
'capabilities': dict({
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
}),
|
||||||
|
'config_entry_id': <ANY>,
|
||||||
|
'device_class': None,
|
||||||
|
'device_id': <ANY>,
|
||||||
|
'disabled_by': None,
|
||||||
|
'domain': 'sensor',
|
||||||
|
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
|
||||||
|
'entity_id': 'sensor.test_router_memory_usage',
|
||||||
|
'has_entity_name': True,
|
||||||
|
'hidden_by': None,
|
||||||
|
'icon': None,
|
||||||
|
'id': <ANY>,
|
||||||
|
'labels': set({
|
||||||
|
}),
|
||||||
|
'name': None,
|
||||||
|
'options': dict({
|
||||||
|
}),
|
||||||
|
'original_device_class': None,
|
||||||
|
'original_icon': None,
|
||||||
|
'original_name': 'Memory usage',
|
||||||
|
'platform': 'tplink_omada',
|
||||||
|
'previous_unique_id': None,
|
||||||
|
'supported_features': 0,
|
||||||
|
'translation_key': 'mem_usage',
|
||||||
|
'unique_id': 'AA-BB-CC-DD-EE-FF_mem_usage',
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
})
|
||||||
|
# ---
|
||||||
|
# name: test_entities[sensor.test_router_memory_usage-state]
|
||||||
|
StateSnapshot({
|
||||||
|
'attributes': ReadOnlyDict({
|
||||||
|
'friendly_name': 'Test Router Memory usage',
|
||||||
|
'state_class': <SensorStateClass.MEASUREMENT: 'measurement'>,
|
||||||
|
'unit_of_measurement': '%',
|
||||||
|
}),
|
||||||
|
'context': <ANY>,
|
||||||
|
'entity_id': 'sensor.test_router_memory_usage',
|
||||||
|
'last_changed': <ANY>,
|
||||||
|
'last_reported': <ANY>,
|
||||||
|
'last_updated': <ANY>,
|
||||||
|
'state': '47',
|
||||||
|
})
|
||||||
|
# ---
|
117
tests/components/tplink_omada/test_sensor.py
Normal file
117
tests/components/tplink_omada/test_sensor.py
Normal file
@ -0,0 +1,117 @@
|
|||||||
|
"""Tests for TP-Link Omada sensor entities."""
|
||||||
|
|
||||||
|
from datetime import timedelta
|
||||||
|
import json
|
||||||
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
from freezegun.api import FrozenDateTimeFactory
|
||||||
|
import pytest
|
||||||
|
from syrupy.assertion import SnapshotAssertion
|
||||||
|
from tplink_omada_client.definitions import DeviceStatus, DeviceStatusCategory
|
||||||
|
from tplink_omada_client.devices import OmadaGatewayPortStatus, OmadaListDevice
|
||||||
|
|
||||||
|
from homeassistant.components.tplink_omada.const import DOMAIN
|
||||||
|
from homeassistant.components.tplink_omada.coordinator import POLL_DEVICES
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
from homeassistant.helpers import entity_registry as er
|
||||||
|
|
||||||
|
from tests.common import (
|
||||||
|
MockConfigEntry,
|
||||||
|
async_fire_time_changed,
|
||||||
|
load_fixture,
|
||||||
|
snapshot_platform,
|
||||||
|
)
|
||||||
|
|
||||||
|
POLL_INTERVAL = timedelta(seconds=POLL_DEVICES)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def init_integration(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
mock_config_entry: MockConfigEntry,
|
||||||
|
mock_omada_client: MagicMock,
|
||||||
|
) -> MockConfigEntry:
|
||||||
|
"""Set up the TP-Link Omada integration for testing."""
|
||||||
|
mock_config_entry.add_to_hass(hass)
|
||||||
|
|
||||||
|
with patch("homeassistant.components.tplink_omada.PLATFORMS", ["sensor"]):
|
||||||
|
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
return mock_config_entry
|
||||||
|
|
||||||
|
|
||||||
|
async def test_entities(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
entity_registry: er.EntityRegistry,
|
||||||
|
init_integration: MockConfigEntry,
|
||||||
|
snapshot: SnapshotAssertion,
|
||||||
|
) -> None:
|
||||||
|
"""Test the creation of the TP-Link Omada sensor entities."""
|
||||||
|
await snapshot_platform(hass, entity_registry, snapshot, init_integration.entry_id)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_device_specific_status(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
init_integration: MockConfigEntry,
|
||||||
|
mock_omada_site_client: MagicMock,
|
||||||
|
freezer: FrozenDateTimeFactory,
|
||||||
|
) -> None:
|
||||||
|
"""Test a connection status is reported from known detailed status."""
|
||||||
|
entity_id = "sensor.test_poe_switch_device_status"
|
||||||
|
entity = hass.states.get(entity_id)
|
||||||
|
assert entity is not None
|
||||||
|
assert entity.state == "connected"
|
||||||
|
|
||||||
|
_set_test_device_status(
|
||||||
|
mock_omada_site_client,
|
||||||
|
DeviceStatus.ADOPT_FAILED.value,
|
||||||
|
DeviceStatusCategory.CONNECTED.value,
|
||||||
|
)
|
||||||
|
|
||||||
|
freezer.tick(POLL_INTERVAL)
|
||||||
|
async_fire_time_changed(hass)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity = hass.states.get(entity_id)
|
||||||
|
assert entity.state == "adopt_failed"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_device_category_status(
|
||||||
|
hass: HomeAssistant,
|
||||||
|
init_integration: MockConfigEntry,
|
||||||
|
mock_omada_site_client: MagicMock,
|
||||||
|
freezer: FrozenDateTimeFactory,
|
||||||
|
) -> None:
|
||||||
|
"""Test a connection status is reported, with fallback to status category."""
|
||||||
|
entity_id = "sensor.test_poe_switch_device_status"
|
||||||
|
entity = hass.states.get(entity_id)
|
||||||
|
assert entity is not None
|
||||||
|
assert entity.state == "connected"
|
||||||
|
|
||||||
|
_set_test_device_status(
|
||||||
|
mock_omada_site_client,
|
||||||
|
DeviceStatus.PENDING_WIRELESS,
|
||||||
|
DeviceStatusCategory.PENDING.value,
|
||||||
|
)
|
||||||
|
|
||||||
|
freezer.tick(POLL_INTERVAL)
|
||||||
|
async_fire_time_changed(hass)
|
||||||
|
await hass.async_block_till_done()
|
||||||
|
|
||||||
|
entity = hass.states.get(entity_id)
|
||||||
|
assert entity.state == "pending"
|
||||||
|
|
||||||
|
|
||||||
|
def _set_test_device_status(
|
||||||
|
mock_omada_site_client: MagicMock,
|
||||||
|
status: int,
|
||||||
|
status_category: int,
|
||||||
|
) -> OmadaGatewayPortStatus:
|
||||||
|
devices_data = json.loads(load_fixture("devices.json", DOMAIN))
|
||||||
|
devices_data[1]["status"] = status
|
||||||
|
devices_data[1]["statusCategory"] = status_category
|
||||||
|
devices = [OmadaListDevice(d) for d in devices_data]
|
||||||
|
|
||||||
|
mock_omada_site_client.get_devices.reset_mock()
|
||||||
|
mock_omada_site_client.get_devices.return_value = devices
|
Loading…
x
Reference in New Issue
Block a user