Add device uplink mac sensor

homeassistant/components/unifi/sensor.py

@@ -174,6 +174,12 @@ def async_device_outlet_supported_fn(hub: UnifiHub, obj_id: str) -> bool:
     return hub.api.devices[obj_id].outlet_ac_power_budget is not None
 
 
+@callback
+def async_device_uplink_mac_supported_fn(hub: UnifiHub, obj_id: str) -> bool:
+    """Determine if a device supports reading uplink MAC address."""
+    return "uplink_mac" in hub.api.devices[obj_id].raw.get("uplink", {})
+
+
 def device_system_stats_supported_fn(
     stat_index: int, hub: UnifiHub, obj_id: str
 ) -> bool:
@@ -571,6 +577,19 @@ class UnifiSensorEntityDescription(
         unique_id_fn=lambda hub, obj_id: f"device_temperature-{obj_id}",
         value_fn=lambda hub, device: device.general_temperature,
     ),
+    UnifiSensorEntityDescription[Devices, Device](
+        key="Device Uplink MAC",
+        entity_category=EntityCategory.DIAGNOSTIC,
+        api_handler_fn=lambda api: api.devices,
+        available_fn=async_device_available_fn,
+        device_info_fn=async_device_device_info_fn,
+        name_fn=lambda device: "Uplink MAC",
+        object_fn=lambda api, obj_id: api.devices[obj_id],
+        unique_id_fn=lambda hub, obj_id: f"device_uplink_mac-{obj_id}",
+        supported_fn=async_device_uplink_mac_supported_fn,
+        value_fn=lambda hub, device: device.raw.get("uplink", {}).get("uplink_mac"),
+        is_connected_fn=lambda hub, obj_id: hub.api.devices[obj_id].state == 1,
+    ),
     UnifiSensorEntityDescription[Devices, Device](
         key="Device State",
         device_class=SensorDeviceClass.ENUM,

tests/components/unifi/test_sensor.py

@@ -27,7 +27,12 @@
     DEVICE_STATES,
 )
 from homeassistant.config_entries import RELOAD_AFTER_UPDATE_DELAY
-from homeassistant.const import ATTR_DEVICE_CLASS, STATE_UNAVAILABLE, Platform
+from homeassistant.const import (
+    ATTR_DEVICE_CLASS,
+    STATE_UNAVAILABLE,
+    EntityCategory,
+    Platform,
+)
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers.entity_registry import RegistryEntryDisabler
@@ -1681,3 +1686,67 @@ async def test_device_with_no_matching_temperatures(
     )
 
     assert temperature_entity is None
+
+
+@pytest.mark.parametrize(
+    "device_payload",
+    [
+        [
+            {
+                "board_rev": 3,
+                "device_id": "device-with-uplink",
+                "ip": "10.0.1.1",
+                "last_seen": 1562600145,
+                "mac": "00:00:00:00:01:01",
+                "model": "US16P150",
+                "name": "Device",
+                "next_interval": 20,
+                "state": 1,
+                "type": "usw",
+                "upgradable": True,
+                "uptime": 60,
+                "version": "4.0.42.10433",
+                "uplink": {
+                    "uplink_mac": "00:00:00:00:00:02",
+                    "port_idx": 1,
+                },
+            },
+            {
+                "board_rev": 3,
+                "device_id": "device-without-uplink",
+                "ip": "10.0.1.2",
+                "last_seen": 1562600145,
+                "mac": "00:00:00:00:01:02",
+                "model": "US16P150",
+                "name": "Other Device",
+                "next_interval": 20,
+                "state": 1,
+                "type": "usw",
+                "upgradable": True,
+                "uptime": 60,
+                "version": "4.0.42.10433",
+                "uplink": {},
+            },
+        ],
+    ],
+)
+@pytest.mark.usefixtures("config_entry_setup")
+async def test_device_uplink(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    mock_websocket_message,
+    device_payload: list[dict[str, Any]],
+) -> None:
+    """Verify that uplink sensors are working as expected."""
+    assert len(hass.states.async_entity_ids(SENSOR_DOMAIN)) == 5
+    assert hass.states.get("sensor.device_uplink_mac").state == "00:00:00:00:00:02"
+    assert (
+        entity_registry.async_get("sensor.device_uplink_mac").entity_category
+        is EntityCategory.DIAGNOSTIC
+    )
+
+    # Verify new event change temperature
+    device = device_payload[0]
+    device["uplink"]["uplink_mac"] = "00:00:00:00:00:03"
+    mock_websocket_message(message=MessageKey.DEVICE, data=device)
+    assert hass.states.get("sensor.device_uplink_mac").state == "00:00:00:00:00:03"