starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-10-06 14:29:21 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add esphome/get_device_capabilities websocket command (#180135)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Petar Petrov
2026-09-29 15:15:13 +02:00
committed by GitHub
co-authored by Copilot Autofix powered by AI
parent 0523a26e57
commit aec7147921
2 changed files with 445 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/esphome/websocket_api.py
+120 -6
View File
@@ -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
tests/components/esphome/test_websocket_api.py
+325 -3
View File
@@ -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,
}