Add web socket API command to get a single ZHA device (#26196)

* get single device web socket command

* test get single device

* add not found error

* fix handling when device doesn't exist

* add test for zha device not found

homeassistant/components/zha/api.py

@@ -148,6 +148,31 @@ async def websocket_get_devices(hass, connection, msg):
     connection.send_result(msg[ID], devices)
 
 
+@websocket_api.require_admin
+@websocket_api.async_response
+@websocket_api.websocket_command(
+    {vol.Required(TYPE): "zha/device", vol.Required(ATTR_IEEE): convert_ieee}
+)
+async def websocket_get_device(hass, connection, msg):
+    """Get ZHA devices."""
+    zha_gateway = hass.data[DATA_ZHA][DATA_ZHA_GATEWAY]
+    ha_device_registry = await async_get_registry(hass)
+    ieee = msg[ATTR_IEEE]
+    device = None
+    if ieee in zha_gateway.devices:
+        device = async_get_device_info(
+            hass, zha_gateway.devices[ieee], ha_device_registry=ha_device_registry
+        )
+    if not device:
+        connection.send_message(
+            websocket_api.error_message(
+                msg[ID], websocket_api.const.ERR_NOT_FOUND, "ZHA Device not found"
+            )
+        )
+        return
+    connection.send_result(msg[ID], device)
+
+
 @callback
 def async_get_device_info(hass, device, ha_device_registry=None):
     """Get ZHA device."""
@@ -587,6 +612,7 @@ def async_load_api(hass):
 
     websocket_api.async_register_command(hass, websocket_permit_devices)
     websocket_api.async_register_command(hass, websocket_get_devices)
+    websocket_api.async_register_command(hass, websocket_get_device)
     websocket_api.async_register_command(hass, websocket_reconfigure_node)
     websocket_api.async_register_command(hass, websocket_device_clusters)
     websocket_api.async_register_command(hass, websocket_device_cluster_attributes)

tests/components/zha/test_api.py

@@ -13,6 +13,7 @@ from homeassistant.components.zha.core.const import (
     ATTR_MANUFACTURER,
     ATTR_ENDPOINT_ID,
 )
+from homeassistant.components.websocket_api import const
 from .common import async_init_zigpy_device
 
 
@@ -126,3 +127,22 @@ async def test_list_devices(hass, config_entry, zha_gateway, zha_client):
         for entity_reference in device["entities"]:
             assert entity_reference[ATTR_NAME] is not None
             assert entity_reference["entity_id"] is not None
+
+        await zha_client.send_json(
+            {ID: 6, TYPE: "zha/device", ATTR_IEEE: device[ATTR_IEEE]}
+        )
+        msg = await zha_client.receive_json()
+        device2 = msg["result"]
+        assert device == device2
+
+
+async def test_device_not_found(hass, config_entry, zha_gateway, zha_client):
+    """Test not found response from get device API."""
+    await zha_client.send_json(
+        {ID: 6, TYPE: "zha/device", ATTR_IEEE: "28:6d:97:00:01:04:11:8c"}
+    )
+    msg = await zha_client.receive_json()
+    assert msg["id"] == 6
+    assert msg["type"] == const.TYPE_RESULT
+    assert not msg["success"]
+    assert msg["error"]["code"] == const.ERR_NOT_FOUND