From aec7147921cbd4621def56934ceedc4cb7998e4b Mon Sep 17 00:00:00 2001 From: Petar Petrov Date: Tue, 29 Sep 2026 16:15:13 +0300 Subject: [PATCH] Add esphome/get_device_capabilities websocket command (#180135) Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../components/esphome/websocket_api.py | 126 ++++++- .../components/esphome/test_websocket_api.py | 328 +++++++++++++++++- 2 files changed, 445 insertions(+), 9 deletions(-) diff --git a/homeassistant/components/esphome/websocket_api.py b/homeassistant/components/esphome/websocket_api.py index 271bb7718e7a..ed600a2efad1 100644 --- a/homeassistant/components/esphome/websocket_api.py +++ b/homeassistant/components/esphome/websocket_api.py @@ -1,26 +1,36 @@ """ESPHome websocket API.""" -import logging -from typing import Any +from typing import Any, cast import probatio from homeassistant.components import websocket_api +from homeassistant.config_entries import ConfigEntryState from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import device_registry as dr -from .const import CONF_NOISE_PSK - -_LOGGER = logging.getLogger(__name__) - +from .const import CONF_NOISE_PSK, DOMAIN +from .entry_data import ESPHomeConfigEntry +from .serial_proxy import build_url TYPE = "type" ENTRY_ID = "entry_id" +DEVICE_ID = "device_id" + +ZWAVE_JS_DOMAIN = "zwave_js" + +_UNAVAILABLE_CAPABILITIES: dict[str, Any] = { + "bluetooth_proxy": {"supported": False}, + "zwave_proxy": {"supported": False, "home_id": 0, "config_entry_id": None}, + "serial_proxies": [], +} @callback def async_setup(hass: HomeAssistant) -> None: """Set up the websocket API.""" websocket_api.async_register_command(hass, get_encryption_key) + websocket_api.async_register_command(hass, get_device_capabilities) @callback @@ -50,3 +60,107 @@ def get_encryption_key( "encryption_key": entry.data.get(CONF_NOISE_PSK), }, ) + + +@callback +@websocket_api.require_admin +@websocket_api.websocket_command( + { + probatio.Required(TYPE): "esphome/get_device_capabilities", + probatio.Required(DEVICE_ID): str, + } +) +def get_device_capabilities( + hass: HomeAssistant, + connection: websocket_api.connection.ActiveConnection, + msg: dict[str, Any], +) -> None: + """Return cached ESPHome DeviceInfo capabilities for the device page.""" + device, candidate = dr.async_get_device_and_config_entry_for_domain( + hass, msg[DEVICE_ID], domain=DOMAIN, include_child_devices=False + ) + if device is None: + connection.send_error( + msg["id"], websocket_api.ERR_NOT_FOUND, "Device not found" + ) + return + + if candidate is None: + connection.send_error( + msg["id"], + websocket_api.ERR_NOT_FOUND, + "Device is not an ESPHome device", + ) + return + entry = cast(ESPHomeConfigEntry, candidate) + + if not _is_main_esphome_device(device): + connection.send_error( + msg["id"], + websocket_api.ERR_NOT_FOUND, + "Device is not the main ESPHome device", + ) + return + + device_info = None + if entry.state is ConfigEntryState.LOADED: + device_info = entry.runtime_data.device_info + + if device_info is None: + connection.send_result(msg["id"], _UNAVAILABLE_CAPABILITIES) + return + + entry_data = entry.runtime_data + home_id = device_info.zwave_home_id or 0 + connection.send_result( + msg["id"], + { + "bluetooth_proxy": { + "supported": bool( + device_info.bluetooth_proxy_feature_flags_compat( + entry_data.api_version + ) + ), + }, + "zwave_proxy": { + "supported": bool(device_info.zwave_proxy_feature_flags), + "home_id": home_id, + "config_entry_id": _zwave_js_config_entry_id(hass, home_id), + }, + "serial_proxies": [ + { + "name": proxy.name, + "port_type": ( + proxy.port_type.name if proxy.port_type is not None else None + ), + "url": str(build_url(entry.entry_id, proxy.name)), + } + for proxy in device_info.serial_proxies + ], + }, + ) + + +def _is_main_esphome_device(device: dr.DeviceEntry) -> bool: + """Return True if device is the MAC-connected ESPHome node.""" + return any( + conn_type == dr.CONNECTION_NETWORK_MAC for conn_type, _ in device.connections + ) + + +def _zwave_js_config_entry_id(hass: HomeAssistant, home_id: int) -> str | None: + """Return the entry ID of the zwave_js entry whose unique ID is this home ID.""" + if not home_id: + return None + home_id_str = str(home_id) + # Ignored and disabled entries are not the configured network. Scan all + # matches because the unique ID index returns only the first one, and + # compare with str() so legacy integer unique IDs still match. + for entry in hass.config_entries.async_entries( + ZWAVE_JS_DOMAIN, + include_ignore=False, + include_disabled=False, + ): + if str(entry.unique_id) == home_id_str: + return entry.entry_id + return None diff --git a/tests/components/esphome/test_websocket_api.py b/tests/components/esphome/test_websocket_api.py index 649935b72ded..749d208d8c7f 100644 --- a/tests/components/esphome/test_websocket_api.py +++ b/tests/components/esphome/test_websocket_api.py @@ -1,14 +1,37 @@ """Tests for ESPHome websocket API.""" -from aioesphomeapi import APIClient +from typing import Any -from homeassistant.components.esphome.const import CONF_NOISE_PSK -from homeassistant.components.esphome.websocket_api import ENTRY_ID, TYPE +from aioesphomeapi import APIClient +from aioesphomeapi.model import SerialProxyInfo, SerialProxyPortType, SubDeviceInfo +import pytest + +from homeassistant.components.esphome.const import CONF_NOISE_PSK, DOMAIN +from homeassistant.components.esphome.serial_proxy import build_url +from homeassistant.components.esphome.websocket_api import DEVICE_ID, ENTRY_ID, TYPE +from homeassistant.config_entries import SOURCE_IGNORE, ConfigEntryDisabler +from homeassistant.core import HomeAssistant +from homeassistant.helpers import device_registry as dr + +from .conftest import MockESPHomeDeviceType from tests.common import MockConfigEntry from tests.typing import WebSocketGenerator +def _device_id_for_mac( + device_registry: dr.DeviceRegistry, + entry: MockConfigEntry, + mac: str = "11:22:33:44:55:aa", +) -> str: + """Return the device registry id for an ESPHome MAC.""" + device = device_registry.async_get_device_by_connection( + (dr.CONNECTION_NETWORK_MAC, mac), entry.entry_id + ) + assert device is not None + return device.id + + async def test_get_encryption_key( mock_client: APIClient, init_integration: MockConfigEntry, @@ -30,3 +53,302 @@ async def test_get_encryption_key( assert response["result"] == { "encryption_key": mock_config_entry.data.get(CONF_NOISE_PSK) } + + +async def test_get_device_capabilities( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + mock_client: APIClient, + mock_esphome_device: MockESPHomeDeviceType, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test capabilities from cached DeviceInfo.""" + mock_client.connected_address = "192.168.1.2" + device = await mock_esphome_device( + mock_client=mock_client, + device_info={ + "bluetooth_proxy_feature_flags": 1, + "zwave_proxy_feature_flags": 1, + "zwave_home_id": 1234567890, + "serial_proxies": [ + SerialProxyInfo(name="uart0", port_type=SerialProxyPortType.TTL), + SerialProxyInfo(name="amp", port_type=SerialProxyPortType.RS232), + SerialProxyInfo(name="bus", port_type=SerialProxyPortType.RS485), + SerialProxyInfo(name="unknown", port_type=None), + ], + }, + ) + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: _device_id_for_mac(device_registry, device.entry), + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is True + assert response["result"] == { + "bluetooth_proxy": {"supported": True}, + "zwave_proxy": { + "supported": True, + "home_id": 1234567890, + "config_entry_id": None, + }, + "serial_proxies": [ + { + "name": "uart0", + "port_type": "TTL", + "url": str(build_url(device.entry.entry_id, "uart0")), + }, + { + "name": "amp", + "port_type": "RS232", + "url": str(build_url(device.entry.entry_id, "amp")), + }, + { + "name": "bus", + "port_type": "RS485", + "url": str(build_url(device.entry.entry_id, "bus")), + }, + { + "name": "unknown", + "port_type": None, + "url": str(build_url(device.entry.entry_id, "unknown")), + }, + ], + } + + +async def test_get_device_capabilities_device_not_found( + init_integration: MockConfigEntry, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test get_device_capabilities when the device registry id is unknown.""" + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: "not-a-device", + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is False + assert response["error"]["code"] == "not_found" + assert response["error"]["message"] == "Device not found" + + +async def test_get_device_capabilities_wrong_domain( + hass: HomeAssistant, + init_integration: MockConfigEntry, + device_registry: dr.DeviceRegistry, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test get_device_capabilities when the device is not ESPHome.""" + other_entry = MockConfigEntry(domain="switch", data={}) + other_entry.add_to_hass(hass) + device = device_registry.async_get_or_create( + config_entry_id=other_entry.entry_id, + connections={(dr.CONNECTION_NETWORK_MAC, "aa:bb:cc:dd:ee:ff")}, + ) + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: device.id, + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is False + assert response["error"]["code"] == "not_found" + assert response["error"]["message"] == "Device is not an ESPHome device" + + +async def test_get_device_capabilities_sub_device( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + mock_client: APIClient, + mock_esphome_device: MockESPHomeDeviceType, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test capabilities are not exposed on ESPHome sub-devices.""" + device = await mock_esphome_device( + mock_client=mock_client, + device_info={ + "devices": [ + SubDeviceInfo(device_id=11111111, name="Motion Sensor", area_id=0), + ], + }, + ) + + sub_device = device_registry.async_get_device_by_identifier( + (DOMAIN, f"{device.device_info.mac_address}_11111111"), device.entry.entry_id + ) + assert sub_device is not None + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: sub_device.id, + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is False + assert response["error"]["code"] == "not_found" + assert response["error"]["message"] == "Device is not the main ESPHome device" + + +async def test_get_device_capabilities_unavailable( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + mock_client: APIClient, + mock_esphome_device: MockESPHomeDeviceType, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test cached capabilities are returned when the device is unavailable.""" + mock_client.connected_address = "192.168.1.2" + device = await mock_esphome_device( + mock_client=mock_client, + device_info={ + "bluetooth_proxy_feature_flags": 1, + "zwave_proxy_feature_flags": 1, + "zwave_home_id": 1234567890, + }, + ) + device.entry.runtime_data.available = False + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: _device_id_for_mac(device_registry, device.entry), + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is True + assert response["result"] == { + "bluetooth_proxy": {"supported": True}, + "zwave_proxy": { + "supported": True, + "home_id": 1234567890, + "config_entry_id": None, + }, + "serial_proxies": [], + } + + +async def test_get_device_capabilities_no_device_info( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + mock_client: APIClient, + mock_esphome_device: MockESPHomeDeviceType, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test a useful empty payload when cached DeviceInfo is missing.""" + device = await mock_esphome_device(mock_client=mock_client) + device.entry.runtime_data.device_info = None + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: _device_id_for_mac(device_registry, device.entry), + } + ) + + response = await websocket_client.receive_json() + assert response["success"] is True + assert response["result"] == { + "bluetooth_proxy": {"supported": False}, + "zwave_proxy": { + "supported": False, + "home_id": 0, + "config_entry_id": None, + }, + "serial_proxies": [], + } + + +@pytest.mark.parametrize( + ("zwave_entries", "home_id", "config_entry_id"), + [ + pytest.param( + [{"entry_id": "zwave", "unique_id": "1234567890"}], + 1234567890, + "zwave", + id="configured", + ), + pytest.param( + [{"entry_id": "zwave", "unique_id": 1234567890}], + 1234567890, + "zwave", + id="legacy_int_unique_id", + ), + pytest.param( + [ + {"unique_id": "1234567890", "source": SOURCE_IGNORE}, + {"entry_id": "zwave", "unique_id": "1234567890"}, + ], + 1234567890, + "zwave", + id="ignored_duplicate", + ), + pytest.param([{"unique_id": "999"}], 1234567890, None, id="other_network"), + pytest.param( + [{"unique_id": "1234567890", "source": SOURCE_IGNORE}], + 1234567890, + None, + id="ignored", + ), + pytest.param( + [{"unique_id": "1234567890", "disabled_by": ConfigEntryDisabler.USER}], + 1234567890, + None, + id="disabled", + ), + pytest.param([{"unique_id": "0"}], 0, None, id="home_id_zero"), + ], +) +async def test_get_device_capabilities_zwave_js_config_entry( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + mock_client: APIClient, + mock_esphome_device: MockESPHomeDeviceType, + hass_ws_client: WebSocketGenerator, + zwave_entries: list[dict[str, Any]], + home_id: int, + config_entry_id: str | None, +) -> None: + """Test matching the advertised home ID to a zwave_js config entry.""" + for entry_kwargs in zwave_entries: + MockConfigEntry(domain="zwave_js", **entry_kwargs).add_to_hass(hass) + mock_client.connected_address = "192.168.1.2" + device = await mock_esphome_device( + mock_client=mock_client, + device_info={ + "zwave_proxy_feature_flags": 1, + "zwave_home_id": home_id, + }, + ) + + websocket_client = await hass_ws_client() + await websocket_client.send_json_auto_id( + { + TYPE: "esphome/get_device_capabilities", + DEVICE_ID: _device_id_for_mac(device_registry, device.entry), + } + ) + response = await websocket_client.receive_json() + assert response["success"] is True + assert response["result"]["zwave_proxy"] == { + "supported": True, + "home_id": home_id, + "config_entry_id": config_entry_id, + }