diff --git a/homeassistant/components/zwave_js/api.py b/homeassistant/components/zwave_js/api.py index a3d1416962e..ec164e2b505 100644 --- a/homeassistant/components/zwave_js/api.py +++ b/homeassistant/components/zwave_js/api.py @@ -405,6 +405,7 @@ def async_register_api(hass: HomeAssistant) -> None: websocket_api.async_register_command( hass, websocket_try_parse_dsk_from_qr_code_string ) + websocket_api.async_register_command(hass, websocket_lookup_device) websocket_api.async_register_command(hass, websocket_supports_feature) websocket_api.async_register_command(hass, websocket_stop_inclusion) websocket_api.async_register_command(hass, websocket_stop_exclusion) @@ -1138,6 +1139,41 @@ async def websocket_try_parse_dsk_from_qr_code_string( ) +@websocket_api.require_admin +@websocket_api.websocket_command( + { + vol.Required(TYPE): "zwave_js/lookup_device", + vol.Required(ENTRY_ID): str, + vol.Required(MANUFACTURER_ID): int, + vol.Required(PRODUCT_TYPE): int, + vol.Required(PRODUCT_ID): int, + vol.Optional(APPLICATION_VERSION): str, + } +) +@websocket_api.async_response +@async_handle_failed_command +@async_get_entry +async def websocket_lookup_device( + hass: HomeAssistant, + connection: ActiveConnection, + msg: dict[str, Any], + entry: ConfigEntry, + client: Client, + driver: Driver, +) -> None: + """Look up the definition of a given device in the configuration DB.""" + device = await driver.config_manager.lookup_device( + msg[MANUFACTURER_ID], + msg[PRODUCT_TYPE], + msg[PRODUCT_ID], + msg.get(APPLICATION_VERSION), + ) + if device is None: + connection.send_error(msg[ID], ERR_NOT_FOUND, "Device not found") + else: + connection.send_result(msg[ID], device.to_dict()) + + @websocket_api.require_admin @websocket_api.websocket_command( { diff --git a/tests/components/zwave_js/test_api.py b/tests/components/zwave_js/test_api.py index 07c874197b6..b2741a53a92 100644 --- a/tests/components/zwave_js/test_api.py +++ b/tests/components/zwave_js/test_api.py @@ -5,7 +5,7 @@ from http import HTTPStatus from io import BytesIO import json from typing import Any -from unittest.mock import MagicMock, PropertyMock, patch +from unittest.mock import AsyncMock, MagicMock, PropertyMock, patch import pytest from zwave_js_server.const import ( @@ -5577,3 +5577,127 @@ async def test_subscribe_s2_inclusion( msg = await ws_client.receive_json() assert not msg["success"] assert msg["error"]["code"] == ERR_NOT_FOUND + + +async def test_lookup_device( + hass: HomeAssistant, + integration: MockConfigEntry, + client: MagicMock, + hass_ws_client: WebSocketGenerator, +) -> None: + """Test lookup_device websocket command.""" + entry = integration + ws_client = await hass_ws_client(hass) + + # Create mock device response + mock_device = MagicMock() + mock_device.to_dict.return_value = { + "manufacturer": "Test Manufacturer", + "label": "Test Device", + "description": "Test Device Description", + "devices": [{"productType": 1, "productId": 2}], + "firmwareVersion": {"min": "1.0", "max": "2.0"}, + } + + # Test successful lookup + client.driver.config_manager.lookup_device = AsyncMock(return_value=mock_device) + + await ws_client.send_json_auto_id( + { + TYPE: "zwave_js/lookup_device", + ENTRY_ID: entry.entry_id, + MANUFACTURER_ID: 1, + PRODUCT_TYPE: 2, + PRODUCT_ID: 3, + APPLICATION_VERSION: "1.5", + } + ) + msg = await ws_client.receive_json() + + assert msg["success"] + assert msg["result"] == mock_device.to_dict.return_value + + client.driver.config_manager.lookup_device.assert_called_once_with(1, 2, 3, "1.5") + + # Reset mock + client.driver.config_manager.lookup_device.reset_mock() + + # Test lookup without optional application_version + await ws_client.send_json_auto_id( + { + TYPE: "zwave_js/lookup_device", + ENTRY_ID: entry.entry_id, + MANUFACTURER_ID: 4, + PRODUCT_TYPE: 5, + PRODUCT_ID: 6, + } + ) + msg = await ws_client.receive_json() + + assert msg["success"] + assert msg["result"] == mock_device.to_dict.return_value + + client.driver.config_manager.lookup_device.assert_called_once_with(4, 5, 6, None) + + # Test device not found + with patch.object( + client.driver.config_manager, + "lookup_device", + return_value=None, + ): + await ws_client.send_json_auto_id( + { + TYPE: "zwave_js/lookup_device", + ENTRY_ID: entry.entry_id, + MANUFACTURER_ID: 99, + PRODUCT_TYPE: 99, + PRODUCT_ID: 99, + APPLICATION_VERSION: "9.9", + } + ) + msg = await ws_client.receive_json() + + assert not msg["success"] + assert msg["error"]["code"] == ERR_NOT_FOUND + assert msg["error"]["message"] == "Device not found" + + # Test sending command with improper entry ID fails + await ws_client.send_json_auto_id( + { + TYPE: "zwave_js/lookup_device", + ENTRY_ID: "invalid_entry_id", + MANUFACTURER_ID: 1, + PRODUCT_TYPE: 1, + PRODUCT_ID: 1, + APPLICATION_VERSION: "1.0", + } + ) + msg = await ws_client.receive_json() + assert not msg["success"] + assert msg["error"]["code"] == ERR_NOT_FOUND + assert msg["error"]["message"] == "Config entry invalid_entry_id not found" + + # Test FailedCommand exception + error_message = "Failed to execute lookup_device command" + with patch.object( + client.driver.config_manager, + "lookup_device", + side_effect=FailedCommand("lookup_device", error_message), + ): + # Send the subscription request + await ws_client.send_json_auto_id( + { + TYPE: "zwave_js/lookup_device", + ENTRY_ID: entry.entry_id, + MANUFACTURER_ID: 1, + PRODUCT_TYPE: 2, + PRODUCT_ID: 3, + APPLICATION_VERSION: "1.0", + } + ) + + # Verify error response + msg = await ws_client.receive_json() + assert not msg["success"] + assert msg["error"]["code"] == error_message + assert msg["error"]["message"] == f"Command failed: {error_message}"