Add zwave_js node_capabilities and invoke_cc_api websocket commands (#125327)

* Add zwave_js node_capabilities and invoke_cc_api websocket commands * Map isSecure to is_secure * Add tests * Add error handling * fix * Use to_dict function * Make response compatible with current expectations --------- Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
2025-04-23 16:57:53 +00:00 · 2024-11-01 00:32:01 +01:00 · 2024-11-01 00:32:01 +01:00 · 5900413c08
commit 5900413c08
parent c2ceab741f
2 changed files with 246 additions and 1 deletions
--- a/homeassistant/components/zwave_js/api.py
+++ b/homeassistant/components/zwave_js/api.py
@ -43,6 +43,7 @@ from zwave_js_server.model.controller.firmware import (
    ControllerFirmwareUpdateResult,
 )
 from zwave_js_server.model.driver import Driver
+from zwave_js_server.model.endpoint import Endpoint
 from zwave_js_server.model.log_config import LogConfig
 from zwave_js_server.model.log_message import LogMessage
 from zwave_js_server.model.node import Node, NodeStatistics
@ -75,6 +76,11 @@ from homeassistant.helpers.dispatcher import async_dispatcher_connect

 from .config_validation import BITMASK_SCHEMA
 from .const import (
+    ATTR_COMMAND_CLASS,
+    ATTR_ENDPOINT,
+    ATTR_METHOD_NAME,
+    ATTR_PARAMETERS,
+    ATTR_WAIT_FOR_RESULT,
    CONF_DATA_COLLECTION_OPTED_IN,
    DATA_CLIENT,
    EVENT_DEVICE_ADDED_TO_REGISTRY,
@ -437,6 +443,8 @@ def async_register_api(hass: HomeAssistant) -> None:
    )
    websocket_api.async_register_command(hass, websocket_subscribe_node_statistics)
    websocket_api.async_register_command(hass, websocket_hard_reset_controller)
+    websocket_api.async_register_command(hass, websocket_node_capabilities)
+    websocket_api.async_register_command(hass, websocket_invoke_cc_api)
    hass.http.register_view(FirmwareUploadView(dr.async_get(hass)))


@ -2525,3 +2533,81 @@ async def websocket_hard_reset_controller(
        )
    ]
    await driver.async_hard_reset()
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zwave_js/node_capabilities",
+        vol.Required(DEVICE_ID): str,
+    }
+)
+@websocket_api.async_response
+@async_handle_failed_command
+@async_get_node
+async def websocket_node_capabilities(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict[str, Any],
+    node: Node,
+) -> None:
+    """Get node endpoints with their support command classes."""
+    # consumers expect snake_case at the moment
+    # remove that addition when consumers are updated
+    connection.send_result(
+        msg[ID],
+        {
+            idx: [
+                command_class.to_dict() | {"is_secure": command_class.is_secure}
+                for command_class in endpoint.command_classes
+            ]
+            for idx, endpoint in node.endpoints.items()
+        },
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zwave_js/invoke_cc_api",
+        vol.Required(DEVICE_ID): str,
+        vol.Required(ATTR_COMMAND_CLASS): vol.All(
+            vol.Coerce(int), vol.Coerce(CommandClass)
+        ),
+        vol.Optional(ATTR_ENDPOINT): vol.Coerce(int),
+        vol.Required(ATTR_METHOD_NAME): cv.string,
+        vol.Required(ATTR_PARAMETERS): list,
+        vol.Optional(ATTR_WAIT_FOR_RESULT): cv.boolean,
+    }
+)
+@websocket_api.async_response
+@async_handle_failed_command
+@async_get_node
+async def websocket_invoke_cc_api(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict[str, Any],
+    node: Node,
+) -> None:
+    """Call invokeCCAPI on the node or provided endpoint."""
+    command_class: CommandClass = msg[ATTR_COMMAND_CLASS]
+    method_name: str = msg[ATTR_METHOD_NAME]
+    parameters: list[Any] = msg[ATTR_PARAMETERS]
+
+    node_or_endpoint: Node | Endpoint = node
+    if (endpoint := msg.get(ATTR_ENDPOINT)) is not None:
+        node_or_endpoint = node.endpoints[endpoint]
+
+    try:
+        result = await node_or_endpoint.async_invoke_cc_api(
+            command_class,
+            method_name,
+            *parameters,
+            wait_for_result=msg.get(ATTR_WAIT_FOR_RESULT, False),
+        )
+    except BaseZwaveJSServerError as err:
+        connection.send_error(msg[ID], err.__class__.__name__, str(err))
+    else:
+        connection.send_result(
+            msg[ID],
+            result,
+        )
--- a/tests/components/zwave_js/test_api.py
+++ b/tests/components/zwave_js/test_api.py
@ -81,6 +81,11 @@ from homeassistant.components.zwave_js.api import (
    VERSION,
 )
 from homeassistant.components.zwave_js.const import (
+    ATTR_COMMAND_CLASS,
+    ATTR_ENDPOINT,
+    ATTR_METHOD_NAME,
+    ATTR_PARAMETERS,
+    ATTR_WAIT_FOR_RESULT,
    CONF_DATA_COLLECTION_OPTED_IN,
    DOMAIN,
 )
@ -88,7 +93,7 @@ from homeassistant.components.zwave_js.helpers import get_device_id
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import device_registry as dr

-from tests.common import MockUser
+from tests.common import MockConfigEntry, MockUser
 from tests.typing import ClientSessionGenerator, WebSocketGenerator

 CONTROLLER_PATCH_PREFIX = "zwave_js_server.model.controller.Controller"
@ -4828,3 +4833,157 @@ async def test_hard_reset_controller(

    assert not msg["success"]
    assert msg["error"]["code"] == ERR_NOT_FOUND
+
+
+async def test_node_capabilities(
+    hass: HomeAssistant,
+    multisensor_6: Node,
+    integration: MockConfigEntry,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test the node_capabilities websocket command."""
+    entry = integration
+    ws_client = await hass_ws_client(hass)
+
+    node = multisensor_6
+    device = get_device(hass, node)
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/node_capabilities",
+            DEVICE_ID: device.id,
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert msg["result"] == {
+        "0": [
+            {
+                "id": 113,
+                "name": "Notification",
+                "version": 8,
+                "isSecure": False,
+                "is_secure": False,
+            }
+        ]
+    }
+
+    # Test getting non-existent node fails
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/node_status",
+            DEVICE_ID: "fake_device",
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_FOUND
+
+    # Test sending command with not loaded entry fails
+    await hass.config_entries.async_unload(entry.entry_id)
+    await hass.async_block_till_done()
+
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/node_status",
+            DEVICE_ID: device.id,
+        }
+    )
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_LOADED
+
+
+async def test_invoke_cc_api(
+    hass: HomeAssistant,
+    client,
+    climate_radio_thermostat_ct100_plus_different_endpoints: Node,
+    integration: MockConfigEntry,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test the invoke_cc_api websocket command."""
+    ws_client = await hass_ws_client(hass)
+
+    device_radio_thermostat = get_device(
+        hass, climate_radio_thermostat_ct100_plus_different_endpoints
+    )
+    assert device_radio_thermostat
+
+    # Test successful invoke_cc_api call with a static endpoint
+    client.async_send_command.return_value = {"response": True}
+    client.async_send_command_no_wait.return_value = {"response": True}
+
+    # Test with wait_for_result=False (default)
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/invoke_cc_api",
+            DEVICE_ID: device_radio_thermostat.id,
+            ATTR_COMMAND_CLASS: 67,
+            ATTR_METHOD_NAME: "someMethod",
+            ATTR_PARAMETERS: [1, 2],
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert msg["success"]
+    assert msg["result"] is None  # We did not specify wait_for_result=True
+
+    await hass.async_block_till_done()
+
+    assert len(client.async_send_command_no_wait.call_args_list) == 1
+    args = client.async_send_command_no_wait.call_args[0][0]
+    assert args == {
+        "command": "endpoint.invoke_cc_api",
+        "nodeId": 26,
+        "endpoint": 0,
+        "commandClass": 67,
+        "methodName": "someMethod",
+        "args": [1, 2],
+    }
+
+    client.async_send_command_no_wait.reset_mock()
+
+    # Test with wait_for_result=True
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/invoke_cc_api",
+            DEVICE_ID: device_radio_thermostat.id,
+            ATTR_COMMAND_CLASS: 67,
+            ATTR_ENDPOINT: 0,
+            ATTR_METHOD_NAME: "someMethod",
+            ATTR_PARAMETERS: [1, 2],
+            ATTR_WAIT_FOR_RESULT: True,
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert msg["success"]
+    assert msg["result"] is True
+
+    await hass.async_block_till_done()
+
+    assert len(client.async_send_command.call_args_list) == 1
+    args = client.async_send_command.call_args[0][0]
+    assert args == {
+        "command": "endpoint.invoke_cc_api",
+        "nodeId": 26,
+        "endpoint": 0,
+        "commandClass": 67,
+        "methodName": "someMethod",
+        "args": [1, 2],
+    }
+
+    client.async_send_command.side_effect = NotFoundError
+
+    # Ensure an error is returned
+    await ws_client.send_json_auto_id(
+        {
+            TYPE: "zwave_js/invoke_cc_api",
+            DEVICE_ID: device_radio_thermostat.id,
+            ATTR_COMMAND_CLASS: 67,
+            ATTR_ENDPOINT: 0,
+            ATTR_METHOD_NAME: "someMethod",
+            ATTR_PARAMETERS: [1, 2],
+            ATTR_WAIT_FOR_RESULT: True,
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert not msg["success"]
+    assert msg["error"] == {"code": "NotFoundError", "message": ""}