home-assistant
/
core
mirror of https://github.com/home-assistant/core
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
core/tests/components/zwave_js/test_api.py

5198 lines
152 KiB
Python
Raw Permalink Blame History

"""Test the Z-Wave JS Websocket API."""
from copy import deepcopy
from http import HTTPStatus
from io import BytesIO
import json
from typing import Any
from unittest.mock import MagicMock, PropertyMock, patch
import pytest
from zwave_js_server.const import (
ExclusionStrategy,
InclusionState,
InclusionStrategy,
LogLevel,
Protocols,
ProvisioningEntryStatus,
QRCodeVersion,
SecurityClass,
ZwaveFeature,
)
from zwave_js_server.event import Event
from zwave_js_server.exceptions import (
FailedCommand,
FailedZWaveCommand,
InvalidNewValue,
NotFoundError,
SetValueFailed,
)
from zwave_js_server.model.controller import (
ProvisioningEntry,
QRProvisioningInformation,
)
from zwave_js_server.model.controller.firmware import ControllerFirmwareUpdateData
from zwave_js_server.model.node import Node
from zwave_js_server.model.node.firmware import NodeFirmwareUpdateData
from zwave_js_server.model.value import ConfigurationValue, get_value_id_str
from homeassistant.components.websocket_api import ERR_INVALID_FORMAT, ERR_NOT_FOUND
from homeassistant.components.zwave_js.api import (
APPLICATION_VERSION,
CLIENT_SIDE_AUTH,
COMMAND_CLASS_ID,
CONFIG,
DEVICE_ID,
DSK,
ENABLED,
ENDPOINT,
ENTRY_ID,
ERR_NOT_LOADED,
FEATURE,
FILENAME,
FORCE_CONSOLE,
GENERIC_DEVICE_CLASS,
ID,
INCLUSION_STRATEGY,
INSTALLER_ICON_TYPE,
LEVEL,
LOG_TO_FILE,
MANUFACTURER_ID,
MAX_INCLUSION_REQUEST_INTERVAL,
NODE_ID,
OPTED_IN,
PIN,
PLANNED_PROVISIONING_ENTRY,
PRODUCT_ID,
PRODUCT_TYPE,
PROPERTY,
PROPERTY_KEY,
QR_CODE_STRING,
QR_PROVISIONING_INFORMATION,
REQUESTED_SECURITY_CLASSES,
SECURITY_CLASSES,
SPECIFIC_DEVICE_CLASS,
STATUS,
STRATEGY,
SUPPORTED_PROTOCOLS,
TYPE,
UUID,
VALUE,
VALUE_FORMAT,
VALUE_SIZE,
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,
CONF_INSTALLER_MODE,
DOMAIN,
)
from homeassistant.components.zwave_js.helpers import get_device_id
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, MockUser
from tests.typing import ClientSessionGenerator, WebSocketGenerator
CONTROLLER_PATCH_PREFIX = "zwave_js_server.model.controller.Controller"
def get_device(hass: HomeAssistant, node):
"""Get device ID for a node."""
dev_reg = dr.async_get(hass)
device_id = get_device_id(node.client.driver, node)
return dev_reg.async_get_device(identifiers={device_id})
async def test_no_driver(
hass: HomeAssistant,
client,
multisensor_6,
controller_state,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test driver missing results in error."""
entry = integration
ws_client = await hass_ws_client(hass)
client.driver = None
# Try API call with entry ID
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/network_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
async def test_network_status(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
multisensor_6,
controller_state,
client,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the network status websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.server_logging_enabled = False
# Try API call with entry ID
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_get_state",
return_value=controller_state["controller"],
):
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/network_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["client"]["ws_server_url"] == "ws://test:3000/zjs"
assert result["client"]["server_version"] == "1.0.0"
assert not result["client"]["server_logging_enabled"]
assert result["controller"]["inclusion_state"] == InclusionState.IDLE
# Try API call with device ID
device = device_registry.async_get_device(
identifiers={(DOMAIN, "3245146787-52")},
)
assert device
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_get_state",
return_value=controller_state["controller"],
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/network_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["client"]["ws_server_url"] == "ws://test:3000/zjs"
assert result["client"]["server_version"] == "1.0.0"
assert result["controller"]["inclusion_state"] == InclusionState.IDLE
# Test sending command with invalid config entry ID fails
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/network_status",
ENTRY_ID: "fake_id",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test sending command with invalid device ID fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/network_status",
DEVICE_ID: "fake_id",
}
)
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 with config entry ID
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/network_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with not loaded entry fails with device ID
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/network_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with no device ID or entry ID fails
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/network_status",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_INVALID_FORMAT
async def test_subscribe_node_status(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
multisensor_6_state,
client,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the subscribe node status websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
node_data = deepcopy(multisensor_6_state) # Copy to allow modification in tests.
node = Node(client, node_data)
node.data["ready"] = False
driver = client.driver
driver.controller.nodes[node.node_id] = node
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id, identifiers={get_device_id(driver, node)}
)
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/subscribe_node_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
new_node_data = deepcopy(multisensor_6_state)
new_node_data["ready"] = True
event = Event(
"ready",
{
"source": "node",
"event": "ready",
"nodeId": node.node_id,
"nodeState": new_node_data,
},
)
node.receive_event(event)
await hass.async_block_till_done()
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "ready"
assert msg["event"]["status"] == 1
assert msg["event"]["ready"]
event = Event(
"wake up",
{
"source": "node",
"event": "wake up",
"nodeId": node.node_id,
},
)
node.receive_event(event)
await hass.async_block_till_done()
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "wake up"
assert msg["event"]["status"] == 2
assert msg["event"]["ready"]
async def test_node_status(
hass: HomeAssistant, multisensor_6, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the node status websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
node = multisensor_6
device = get_device(hass, node)
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/node_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result[NODE_ID] == 52
assert result["ready"]
assert result["is_routing"]
assert not result["is_secure"]
assert result["status"] == 1
assert result["zwave_plus_version"] == 1
assert result["highest_security_class"] == SecurityClass.S0_LEGACY
assert not result["is_controller_node"]
assert not result["has_firmware_update_cc"]
# Test getting non-existent node fails
await ws_client.send_json(
{
ID: 4,
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(
{
ID: 5,
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_node_metadata(
hass: HomeAssistant,
wallmote_central_scene,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the node metadata websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
node = wallmote_central_scene
device = get_device(hass, node)
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/node_metadata",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result[NODE_ID] == 35
assert result["inclusion"] == (
"To add the ZP3111 to the Z-Wave network (inclusion), place the Z-Wave "
"primary controller into inclusion mode. Press the Program Switch of ZP3111 "
"for sending the NIF. After sending NIF, Z-Wave will send the auto inclusion, "
"otherwise, ZP3111 will go to sleep after 20 seconds."
)
assert result["exclusion"] == (
"To remove the ZP3111 from the Z-Wave network (exclusion), place the Z-Wave "
"primary controller into \u201cexclusion\u201d mode, and following its "
"instruction to delete the ZP3111 to the controller. Press the Program Switch "
"of ZP3111 once to be excluded."
)
assert result["reset"] == (
"Remove cover to triggered tamper switch, LED flash once & send out Alarm "
"Report. Press Program Switch 10 times within 10 seconds, ZP3111 will send "
"the \u201cDevice Reset Locally Notification\u201d command and reset to the "
"factory default. (Remark: This is to be used only in the case of primary "
"controller being inoperable or otherwise unavailable.)"
)
assert result["manual"] == (
"https://products.z-wavealliance.org/ProductManual/File?folder=&filename="
"MarketCertificationFiles/2479/ZP3111-5_R2_20170316.pdf"
)
assert not result["wakeup"]
assert (
result["device_database_url"]
== "https://devices.zwave-js.io/?jumpTo=0x0086:0x0002:0x0082:0.0"
)
# Test getting non-existent node fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/node_metadata",
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(
{
ID: 5,
TYPE: "zwave_js/node_metadata",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_node_alerts(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
wallmote_central_scene,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the node comments websocket command."""
ws_client = await hass_ws_client(hass)
device = device_registry.async_get_device(identifiers={(DOMAIN, "3245146787-35")})
assert device
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/node_alerts",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result["comments"] == [{"level": "info", "text": "test"}]
assert result["is_embedded"]
async def test_add_node(
hass: HomeAssistant,
nortek_thermostat,
nortek_thermostat_added_event,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the add_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
# Test inclusion with no provisioning input
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {"strategy": InclusionStrategy.DEFAULT},
}
event = Event(
type="inclusion started",
data={
"source": "controller",
"event": "inclusion started",
"strategy": 2,
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "inclusion started"
event = Event(
type="node found",
data={
"source": "controller",
"event": "node found",
"node": {
"nodeId": 67,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node found"
node_details = {
"node_id": 67,
}
assert msg["event"]["node"] == node_details
event = Event(
type="grant security classes",
data={
"source": "controller",
"event": "grant security classes",
"requested": {"securityClasses": [0, 1, 2, 7], "clientSideAuth": False},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "grant security classes"
assert msg["event"]["requested_grant"] == {
"securityClasses": [0, 1, 2, 7],
"clientSideAuth": False,
}
event = Event(
type="validate dsk and enter pin",
data={
"source": "controller",
"event": "validate dsk and enter pin",
"dsk": "test",
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "validate dsk and enter pin"
assert msg["event"]["dsk"] == "test"
client.driver.receive_event(nortek_thermostat_added_event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node added"
node_details = {
"node_id": 67,
"status": 0,
"ready": False,
"low_security": False,
"low_security_reason": None,
}
assert msg["event"]["node"] == node_details
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "device registered"
# Check the keys of the device item
assert list(msg["event"]["device"]) == ["name", "id", "manufacturer", "model"]
# Test receiving interview events
event = Event(
type="interview started",
data={"source": "node", "event": "interview started", "nodeId": 67},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview started"
event = Event(
type="interview stage completed",
data={
"source": "node",
"event": "interview stage completed",
"stageName": "NodeInfo",
"nodeId": 67,
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview stage completed"
assert msg["event"]["stage"] == "NodeInfo"
event = Event(
type="interview completed",
data={"source": "node", "event": "interview completed", "nodeId": 67},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview completed"
event = Event(
type="interview failed",
data={
"source": "node",
"event": "interview failed",
"nodeId": 67,
"args": {
"errorMessage": "error",
"isFinal": True,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview failed"
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 planned provisioning entry
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
PLANNED_PROVISIONING_ENTRY: {
DSK: "test",
SECURITY_CLASSES: [0],
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": ProvisioningEntry(
"test", [SecurityClass.S2_UNAUTHENTICATED]
).to_dict(),
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 QR provisioning information
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_PROVISIONING_INFORMATION: {
VERSION: 0,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": QRProvisioningInformation(
version=QRCodeVersion.S2,
security_classes=[SecurityClass.S2_UNAUTHENTICATED],
dsk="test",
generic_device_class=1,
specific_device_class=1,
installer_icon_type=1,
manufacturer_id=1,
product_type=1,
product_id=1,
application_version="test",
max_inclusion_request_interval=None,
uuid=None,
supported_protocols=None,
).to_dict(),
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 QR provisioning information
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_PROVISIONING_INFORMATION: {
VERSION: 0,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
STATUS: 1,
REQUESTED_SECURITY_CLASSES: [0],
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": QRProvisioningInformation(
version=QRCodeVersion.S2,
security_classes=[SecurityClass.S2_UNAUTHENTICATED],
dsk="test",
generic_device_class=1,
specific_device_class=1,
installer_icon_type=1,
manufacturer_id=1,
product_type=1,
product_id=1,
application_version="test",
max_inclusion_request_interval=None,
uuid=None,
supported_protocols=None,
status=ProvisioningEntryStatus.INACTIVE,
requested_security_classes=[SecurityClass.S2_UNAUTHENTICATED],
).to_dict(),
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 QR code string
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": "90testtesttesttesttesttesttesttesttesttesttesttesttest",
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 DSK string string
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
DSK: "test_dsk",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_inclusion",
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"dsk": "test_dsk",
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test Smart Start QR provisioning information with S2 inclusion strategy fails
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_PROVISIONING_INFORMATION: {
VERSION: 1,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert len(client.async_send_command.call_args_list) == 0
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test QR provisioning information with S0 inclusion strategy fails
await ws_client.send_json(
{
ID: 8,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S0,
QR_PROVISIONING_INFORMATION: {
VERSION: 1,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert len(client.async_send_command.call_args_list) == 0
client.async_send_command.reset_mock()
client.async_send_command.return_value = {" success": True}
# Test ValueError is caught as failure
await ws_client.send_json(
{
ID: 9,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert len(client.async_send_command.call_args_list) == 0
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_begin_inclusion",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 10,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# Test inclusion already in progress
client.async_send_command.reset_mock()
type(client.driver.controller).inclusion_state = PropertyMock(
return_value=InclusionState.INCLUDING
)
# Create a node that's not ready
node_data = deepcopy(nortek_thermostat.data) # Copy to allow modification in tests.
node_data["ready"] = False
node_data["values"] = {}
node_data["endpoints"] = {}
node = Node(client, node_data)
client.driver.controller.nodes[node.node_id] = node
await ws_client.send_json(
{
ID: 11,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
# Verify no command was sent since inclusion is already in progress
assert len(client.async_send_command.call_args_list) == 0
# Verify we got a node added event
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node added"
assert msg["event"]["node"]["node_id"] == node.node_id
# 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(
{ID: 12, TYPE: "zwave_js/add_node", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_grant_security_classes(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the grant_security_classes websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/grant_security_classes",
ENTRY_ID: entry.entry_id,
SECURITY_CLASSES: [SecurityClass.S2_UNAUTHENTICATED],
CLIENT_SIDE_AUTH: False,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.grant_security_classes",
"inclusionGrant": {"securityClasses": [0], "clientSideAuth": False},
}
# 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(
{
ID: 4,
TYPE: "zwave_js/grant_security_classes",
ENTRY_ID: entry.entry_id,
SECURITY_CLASSES: [SecurityClass.S2_UNAUTHENTICATED],
CLIENT_SIDE_AUTH: False,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_validate_dsk_and_enter_pin(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the validate_dsk_and_enter_pin websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/validate_dsk_and_enter_pin",
ENTRY_ID: entry.entry_id,
PIN: "test",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.validate_dsk_and_enter_pin",
"pin": "test",
}
# 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(
{
ID: 4,
TYPE: "zwave_js/validate_dsk_and_enter_pin",
ENTRY_ID: entry.entry_id,
PIN: "test",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_provision_smart_start_node(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test provision_smart_start_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
# Test provisioning entry
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
PLANNED_PROVISIONING_ENTRY: {
DSK: "test",
SECURITY_CLASSES: [0],
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.provision_smart_start_node",
"entry": ProvisioningEntry(
"test", [SecurityClass.S2_UNAUTHENTICATED]
).to_dict(),
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test QR provisioning information
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
QR_PROVISIONING_INFORMATION: {
VERSION: 1,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
"name": "test",
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.provision_smart_start_node",
"entry": QRProvisioningInformation(
version=QRCodeVersion.SMART_START,
security_classes=[SecurityClass.S2_UNAUTHENTICATED],
dsk="test",
generic_device_class=1,
specific_device_class=1,
installer_icon_type=1,
manufacturer_id=1,
product_type=1,
product_id=1,
application_version="test",
max_inclusion_request_interval=None,
uuid=None,
supported_protocols=None,
additional_properties={"name": "test"},
).to_dict(),
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test QR code string
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.provision_smart_start_node",
"entry": "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test QR provisioning information with S2 version throws error
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
QR_PROVISIONING_INFORMATION: {
VERSION: 0,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
assert len(client.async_send_command.call_args_list) == 0
# Test no provisioning parameter provided causes failure
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_provision_smart_start_node",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: (
"90testtesttesttesttesttesttesttesttesttesttesttesttest"
),
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 8,
TYPE: "zwave_js/provision_smart_start_node",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_unprovision_smart_start_node(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test unprovision_smart_start_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {}
# Test node ID as input
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/unprovision_smart_start_node",
ENTRY_ID: entry.entry_id,
NODE_ID: 1,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.unprovision_smart_start_node",
"dskOrNodeId": 1,
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {}
# Test DSK as input
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/unprovision_smart_start_node",
ENTRY_ID: entry.entry_id,
DSK: "test",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.unprovision_smart_start_node",
"dskOrNodeId": "test",
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {}
# Test not including DSK or node ID as input fails
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/unprovision_smart_start_node",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert len(client.async_send_command.call_args_list) == 0
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_unprovision_smart_start_node",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/unprovision_smart_start_node",
ENTRY_ID: entry.entry_id,
DSK: "test",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 7,
TYPE: "zwave_js/unprovision_smart_start_node",
ENTRY_ID: entry.entry_id,
DSK: "test",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_get_provisioning_entries(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test get_provisioning_entries websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {
"entries": [{"dsk": "test", "securityClasses": [0], "fake": "test"}]
}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/get_provisioning_entries",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == [
{DSK: "test", SECURITY_CLASSES: [0], STATUS: 0, "fake": "test"}
]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.get_provisioning_entries",
}
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_get_provisioning_entries",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/get_provisioning_entries",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{ID: 7, TYPE: "zwave_js/get_provisioning_entries", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_parse_qr_code_string(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test parse_qr_code_string websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {
"qrProvisioningInformation": {
"version": 0,
"securityClasses": [0],
"dsk": "test",
"genericDeviceClass": 1,
"specificDeviceClass": 1,
"installerIconType": 1,
"manufacturerId": 1,
"productType": 1,
"productId": 1,
"applicationVersion": "test",
"maxInclusionRequestInterval": 1,
"uuid": "test",
"supportedProtocols": [0],
}
}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/parse_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == {
VERSION: 0,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
MAX_INCLUSION_REQUEST_INTERVAL: 1,
UUID: "test",
SUPPORTED_PROTOCOLS: [Protocols.ZWAVE],
STATUS: 0,
}
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "utils.parse_qr_code_string",
"qr": "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
# Test FailedZWaveCommand is caught
with patch(
"homeassistant.components.zwave_js.api.async_parse_qr_code_string",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/parse_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: (
"90testtesttesttesttesttesttesttesttesttesttesttesttest"
),
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 7,
TYPE: "zwave_js/parse_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_try_parse_dsk_from_qr_code_string(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test try_parse_dsk_from_qr_code_string websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"dsk": "a"}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/try_parse_dsk_from_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == "a"
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "utils.try_parse_dsk_from_qr_code_string",
"qr": "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
# Test FailedZWaveCommand is caught
with patch(
"homeassistant.components.zwave_js.api.async_try_parse_dsk_from_qr_code_string",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/try_parse_dsk_from_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: (
"90testtesttesttesttesttesttesttesttesttesttesttesttest"
),
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 7,
TYPE: "zwave_js/try_parse_dsk_from_qr_code_string",
ENTRY_ID: entry.entry_id,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_supports_feature(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test supports_feature websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"supported": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/supports_feature",
ENTRY_ID: entry.entry_id,
FEATURE: ZwaveFeature.SMART_START,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == {"supported": True}
async def test_cancel_inclusion_exclusion(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test cancelling the inclusion and exclusion process."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{ID: 4, TYPE: "zwave_js/stop_inclusion", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert msg["success"]
await ws_client.send_json(
{ID: 5, TYPE: "zwave_js/stop_exclusion", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert msg["success"]
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_stop_inclusion",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/stop_inclusion",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_stop_exclusion",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/stop_exclusion",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{ID: 8, TYPE: "zwave_js/stop_inclusion", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
await ws_client.send_json(
{ID: 9, TYPE: "zwave_js/stop_exclusion", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_remove_node(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
integration,
client,
hass_ws_client: WebSocketGenerator,
nortek_thermostat,
nortek_thermostat_removed_event,
) -> None:
"""Test the remove_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{ID: 1, TYPE: "zwave_js/remove_node", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_exclusion",
}
event = Event(
type="exclusion started",
data={
"source": "controller",
"event": "exclusion started",
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "exclusion started"
# Create device registry entry for mock node
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, "3245146787-67")},
name="Node 67",
)
# Fire node removed event
client.driver.receive_event(nortek_thermostat_removed_event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node removed"
# Verify device was removed from device registry
device = device_registry.async_get_device(
identifiers={(DOMAIN, "3245146787-67")},
)
assert device is None
# Test unprovision parameter
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/remove_node",
ENTRY_ID: entry.entry_id,
STRATEGY: ExclusionStrategy.EXCLUDE_ONLY,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.begin_exclusion",
"options": {"strategy": 0},
}
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_begin_exclusion",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/remove_node",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{ID: 5, TYPE: "zwave_js/remove_node", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_replace_failed_node(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
nortek_thermostat,
integration,
client,
hass_ws_client: WebSocketGenerator,
nortek_thermostat_added_event,
nortek_thermostat_removed_event,
) -> None:
"""Test the replace_failed_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
# Create device registry entry for mock node
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, "3245146787-67")},
name="Node 67",
)
client.async_send_command.return_value = {"success": True}
# Test replace failed node with no provisioning information
# Order of events we receive for a successful replacement is `inclusion started`,
# `inclusion stopped`, `node removed`, `node added`, then interview stages.
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.replace_failed_node",
"nodeId": nortek_thermostat.node_id,
"options": {"strategy": InclusionStrategy.DEFAULT},
}
client.async_send_command.reset_mock()
event = Event(
type="inclusion started",
data={
"source": "controller",
"event": "inclusion started",
"strategy": 2,
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "inclusion started"
event = Event(
type="node found",
data={
"source": "controller",
"event": "node found",
"node": {
"nodeId": 67,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node found"
node_details = {
"node_id": 67,
}
assert msg["event"]["node"] == node_details
event = Event(
type="grant security classes",
data={
"source": "controller",
"event": "grant security classes",
"requested": {"securityClasses": [0, 1, 2, 7], "clientSideAuth": False},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "grant security classes"
assert msg["event"]["requested_grant"] == {
"securityClasses": [0, 1, 2, 7],
"clientSideAuth": False,
}
event = Event(
type="validate dsk and enter pin",
data={
"source": "controller",
"event": "validate dsk and enter pin",
"dsk": "test",
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "validate dsk and enter pin"
assert msg["event"]["dsk"] == "test"
event = Event(
type="inclusion stopped",
data={
"source": "controller",
"event": "inclusion stopped",
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "inclusion stopped"
# Fire node removed event
client.driver.receive_event(nortek_thermostat_removed_event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node removed"
# Verify device was removed from device registry
assert (
device_registry.async_get_device(
identifiers={(DOMAIN, "3245146787-67")},
)
is None
)
client.driver.receive_event(nortek_thermostat_added_event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node added"
node_details = {
"node_id": 67,
"status": 0,
"ready": False,
}
assert msg["event"]["node"] == node_details
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "device registered"
# Check the keys of the device item
assert list(msg["event"]["device"]) == ["name", "id", "manufacturer", "model"]
# Test receiving interview events
event = Event(
type="interview started",
data={"source": "node", "event": "interview started", "nodeId": 67},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview started"
event = Event(
type="interview stage completed",
data={
"source": "node",
"event": "interview stage completed",
"stageName": "NodeInfo",
"nodeId": 67,
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview stage completed"
assert msg["event"]["stage"] == "NodeInfo"
event = Event(
type="interview completed",
data={"source": "node", "event": "interview completed", "nodeId": 67},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview completed"
event = Event(
type="interview failed",
data={
"source": "node",
"event": "interview failed",
"nodeId": 67,
"args": {
"errorMessage": "error",
"isFinal": True,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview failed"
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 planned provisioning entry
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
PLANNED_PROVISIONING_ENTRY: {
DSK: "test",
SECURITY_CLASSES: [0],
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.replace_failed_node",
"nodeId": 67,
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": ProvisioningEntry(
"test", [SecurityClass.S2_UNAUTHENTICATED]
).to_dict(),
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 QR provisioning information
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_PROVISIONING_INFORMATION: {
VERSION: 0,
SECURITY_CLASSES: [0],
DSK: "test",
GENERIC_DEVICE_CLASS: 1,
SPECIFIC_DEVICE_CLASS: 1,
INSTALLER_ICON_TYPE: 1,
MANUFACTURER_ID: 1,
PRODUCT_TYPE: 1,
PRODUCT_ID: 1,
APPLICATION_VERSION: "test",
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.replace_failed_node",
"nodeId": 67,
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": QRProvisioningInformation(
version=QRCodeVersion.S2,
security_classes=[SecurityClass.S2_UNAUTHENTICATED],
dsk="test",
generic_device_class=1,
specific_device_class=1,
installer_icon_type=1,
manufacturer_id=1,
product_type=1,
product_id=1,
application_version="test",
max_inclusion_request_interval=None,
uuid=None,
supported_protocols=None,
).to_dict(),
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test S2 QR code string
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "controller.replace_failed_node",
"nodeId": 67,
"options": {
"strategy": InclusionStrategy.SECURITY_S2,
"provisioning": "90testtesttesttesttesttesttesttesttesttesttesttesttest",
},
}
client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True}
# Test ValueError is caught as failure
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert len(client.async_send_command.call_args_list) == 0
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_replace_failed_node",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 8,
TYPE: "zwave_js/replace_failed_node",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_remove_failed_node(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
nortek_thermostat,
integration,
client,
hass_ws_client: WebSocketGenerator,
nortek_thermostat_removed_event,
nortek_thermostat_added_event,
) -> None:
"""Test the remove_failed_node websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, nortek_thermostat)
client.async_send_command.return_value = {"success": True}
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_remove_failed_node",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/remove_failed_node",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/remove_failed_node",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
# Create device registry entry for mock node
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={(DOMAIN, "3245146787-67")},
name="Node 67",
)
# Fire node removed event
client.driver.receive_event(nortek_thermostat_removed_event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "node removed"
# Verify device was removed from device registry
assert (
device_registry.async_get_device(
identifiers={(DOMAIN, "3245146787-67")},
)
is None
)
# Re-add node so we can test config entry not loaded
client.driver.receive_event(nortek_thermostat_added_event)
# 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(
{
ID: 3,
TYPE: "zwave_js/remove_failed_node",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_begin_rebuilding_routes(
hass: HomeAssistant,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the begin_rebuilding_routes websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/begin_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_begin_rebuilding_routes",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/begin_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 5,
TYPE: "zwave_js/begin_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_subscribe_rebuild_routes_progress(
hass: HomeAssistant,
integration,
client,
nortek_thermostat,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the subscribe_rebuild_routes_progress command."""
entry = integration
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/subscribe_rebuild_routes_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
# Fire rebuild routes progress
event = Event(
"rebuild routes progress",
{
"source": "controller",
"event": "rebuild routes progress",
"progress": {67: "pending"},
},
)
client.driver.controller.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "rebuild routes progress"
assert msg["event"]["rebuild_routes_status"] == {"67": "pending"}
# 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(
{
ID: 4,
TYPE: "zwave_js/subscribe_rebuild_routes_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_subscribe_rebuild_routes_progress_initial_value(
hass: HomeAssistant,
integration,
client,
nortek_thermostat,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test subscribe_rebuild_routes_progress command when rebuild routes in progress."""
entry = integration
ws_client = await hass_ws_client(hass)
assert not client.driver.controller.rebuild_routes_progress
# Fire rebuild routes progress before sending rebuild routes progress command
event = Event(
"rebuild routes progress",
{
"source": "controller",
"event": "rebuild routes progress",
"progress": {67: "pending"},
},
)
client.driver.controller.receive_event(event)
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/subscribe_rebuild_routes_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == {"67": "pending"}
async def test_stop_rebuilding_routes(
hass: HomeAssistant,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the stop_rebuilding_routes websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/stop_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_stop_rebuilding_routes",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/stop_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 5,
TYPE: "zwave_js/stop_rebuilding_routes",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_rebuild_node_routes(
hass: HomeAssistant,
multisensor_6,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the rebuild_node_routes websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/rebuild_node_routes",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_rebuild_node_routes",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/rebuild_node_routes",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 5,
TYPE: "zwave_js/rebuild_node_routes",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_refresh_node_info(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the refresh_node_info WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command_no_wait.return_value = None
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/refresh_node_info",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
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"] == "node.refresh_info"
assert args["nodeId"] == 52
event = Event(
type="interview started",
data={"source": "node", "event": "interview started", "nodeId": 52},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview started"
event = Event(
type="interview stage completed",
data={
"source": "node",
"event": "interview stage completed",
"stageName": "NodeInfo",
"nodeId": 52,
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview stage completed"
assert msg["event"]["stage"] == "NodeInfo"
event = Event(
type="interview completed",
data={"source": "node", "event": "interview completed", "nodeId": 52},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview completed"
event = Event(
type="interview failed",
data={
"source": "node",
"event": "interview failed",
"nodeId": 52,
"args": {
"errorMessage": "error",
"isFinal": True,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"]["event"] == "interview failed"
client.async_send_command_no_wait.reset_mock()
# Test getting non-existent node fails
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/refresh_node_info",
DEVICE_ID: "fake_device",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_refresh_info",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/refresh_node_info",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 4,
TYPE: "zwave_js/refresh_node_info",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_refresh_node_values(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the refresh_node_values WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command_no_wait.return_value = None
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/refresh_node_values",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
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"] == "node.refresh_values"
assert args["nodeId"] == 52
client.async_send_command_no_wait.reset_mock()
# Test getting non-existent device fails
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/refresh_node_values",
DEVICE_ID: "fake_device",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_refresh_values",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/refresh_node_values",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 5,
TYPE: "zwave_js/refresh_node_values",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_refresh_node_cc_values(
hass: HomeAssistant,
multisensor_6,
client,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the refresh_node_cc_values WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command_no_wait.return_value = None
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/refresh_node_cc_values",
DEVICE_ID: device.id,
COMMAND_CLASS_ID: 112,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
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"] == "node.refresh_cc_values"
assert args["nodeId"] == 52
assert args["commandClass"] == 112
client.async_send_command_no_wait.reset_mock()
# Test using invalid CC ID fails
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/refresh_node_cc_values",
DEVICE_ID: device.id,
COMMAND_CLASS_ID: 9999,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test getting non-existent device fails
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/refresh_node_cc_values",
DEVICE_ID: "fake_device",
COMMAND_CLASS_ID: 112,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_refresh_cc_values",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/refresh_node_cc_values",
DEVICE_ID: device.id,
COMMAND_CLASS_ID: 112,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 5,
TYPE: "zwave_js/refresh_node_cc_values",
DEVICE_ID: device.id,
COMMAND_CLASS_ID: 112,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_set_config_parameter(
hass: HomeAssistant,
multisensor_6,
client,
hass_ws_client: WebSocketGenerator,
integration,
) -> None:
"""Test the set_config_parameter service."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
new_value_data = multisensor_6.values[
get_value_id_str(multisensor_6, 112, 102, 0, 1)
].data.copy()
new_value_data["endpoint"] = 1
new_value = ConfigurationValue(multisensor_6, new_value_data)
multisensor_6.values[get_value_id_str(multisensor_6, 112, 102, 1, 1)] = new_value
client.async_send_command_no_wait.return_value = None
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]["status"] == "queued"
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"] == "node.set_value"
assert args["nodeId"] == 52
assert args["valueId"] == {
"commandClass": 112,
"endpoint": 0,
"property": 102,
"propertyKey": 1,
}
assert args["value"] == 1
client.async_send_command_no_wait.reset_mock()
client.async_send_command_no_wait.return_value = None
# Test using a different endpoint
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
ENDPOINT: 1,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]["status"] == "queued"
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"] == "node.set_value"
assert args["nodeId"] == 52
assert args["valueId"] == {
"commandClass": 112,
"endpoint": 1,
"property": 102,
"propertyKey": 1,
}
assert args["value"] == 1
client.async_send_command_no_wait.reset_mock()
# Test that hex strings are accepted and converted as expected
client.async_send_command_no_wait.return_value = None
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: "0x1",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]["status"] == "queued"
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"] == "node.set_value"
assert args["nodeId"] == 52
assert args["valueId"] == {
"commandClass": 112,
"endpoint": 0,
"property": 102,
"propertyKey": 1,
}
assert args["value"] == 1
client.async_send_command_no_wait.reset_mock()
with patch(
"homeassistant.components.zwave_js.api.async_set_config_parameter",
) as set_param_mock:
set_param_mock.side_effect = InvalidNewValue("test")
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert len(client.async_send_command_no_wait.call_args_list) == 0
assert not msg["success"]
assert msg["error"]["code"] == "not_supported"
assert msg["error"]["message"] == "test"
set_param_mock.side_effect = NotFoundError("test")
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert len(client.async_send_command_no_wait.call_args_list) == 0
assert not msg["success"]
assert msg["error"]["code"] == "not_found"
assert msg["error"]["message"] == "test"
set_param_mock.side_effect = SetValueFailed("test")
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert len(client.async_send_command_no_wait.call_args_list) == 0
assert not msg["success"]
assert msg["error"]["code"] == "unknown_error"
assert msg["error"]["message"] == "test"
# Test getting non-existent node fails
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: "fake_device",
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test FailedZWaveCommand is caught
with patch(
"homeassistant.components.zwave_js.api.async_set_config_parameter",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 8,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 9,
TYPE: "zwave_js/set_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
PROPERTY_KEY: 1,
VALUE: 1,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_get_config_parameters(
hass: HomeAssistant, multisensor_6, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the get config parameters websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
node = multisensor_6
device = get_device(hass, node)
# Test getting configuration parameter values
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/get_config_parameters",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert len(result) == 61
key = "52-112-0-2"
assert result[key]["property"] == 2
assert result[key]["property_key"] is None
assert result[key]["endpoint"] == 0
assert result[key]["configuration_value_type"] == "enumerated"
assert result[key]["metadata"]["states"]
assert (
result[key]["metadata"]["description"]
== "Stay awake for 10 minutes at power on"
)
assert result[key]["metadata"]["label"] == "Stay Awake in Battery Mode"
assert result[key]["metadata"]["type"] == "number"
assert result[key]["metadata"]["min"] == 0
assert result[key]["metadata"]["max"] == 1
assert result[key]["metadata"]["unit"] is None
assert result[key]["metadata"]["writeable"] is True
assert result[key]["metadata"]["readable"] is True
assert result[key]["metadata"]["default"] == 0
assert result[key]["value"] == 0
key = "52-112-0-201-255"
assert result[key]["property_key"] == 255
# Test getting non-existent node config params fails
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/get_config_parameters",
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(
{
ID: 6,
TYPE: "zwave_js/get_config_parameters",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_set_raw_config_parameter(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the set_raw_config_parameter WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
# Change from async_send_command to async_send_command_no_wait
client.async_send_command_no_wait.return_value = None
# Test setting a raw config parameter value
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/set_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
VALUE: 1,
VALUE_SIZE: 2,
VALUE_FORMAT: 1,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]["status"] == "queued"
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.set_raw_config_parameter_value"
assert args["nodeId"] == multisensor_6.node_id
assert args["options"]["parameter"] == 102
assert args["options"]["value"] == 1
assert args["options"]["valueSize"] == 2
assert args["options"]["valueFormat"] == 1
# Reset the mock for async_send_command_no_wait instead
client.async_send_command_no_wait.reset_mock()
# Test getting non-existent node fails
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/set_raw_config_parameter",
DEVICE_ID: "fake_device",
PROPERTY: 102,
VALUE: 1,
VALUE_SIZE: 2,
VALUE_FORMAT: 1,
}
)
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/set_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
VALUE: 1,
VALUE_SIZE: 2,
VALUE_FORMAT: 1,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_get_raw_config_parameter(
hass: HomeAssistant,
multisensor_6,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the get_raw_config_parameter websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command.return_value = {"value": 1}
# Test getting a raw config parameter value
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/get_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]["value"] == 1
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "endpoint.get_raw_config_parameter_value"
assert args["nodeId"] == multisensor_6.node_id
assert args["options"]["parameter"] == 102
client.async_send_command.reset_mock()
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_get_raw_config_parameter_value",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/get_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# Test getting non-existent node fails
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/get_raw_config_parameter",
DEVICE_ID: "fake_device",
PROPERTY: 102,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
# Test FailedCommand exception
client.async_send_command.side_effect = FailedCommand("test", "test")
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/get_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "test"
assert msg["error"]["message"] == "Command failed: test"
# 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/get_raw_config_parameter",
DEVICE_ID: device.id,
PROPERTY: 102,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
@pytest.mark.parametrize(
("firmware_data", "expected_data"),
[({"target": "1"}, {"firmware_target": 1}), ({}, {})],
)
async def test_firmware_upload_view(
hass: HomeAssistant,
multisensor_6,
integration,
hass_client: ClientSessionGenerator,
firmware_file,
firmware_data: dict[str, Any],
expected_data: dict[str, Any],
) -> None:
"""Test the HTTP firmware upload view."""
client = await hass_client()
device = get_device(hass, multisensor_6)
with (
patch(
"homeassistant.components.zwave_js.api.update_firmware",
) as mock_node_cmd,
patch(
"homeassistant.components.zwave_js.api.controller_firmware_update_otw",
) as mock_controller_cmd,
patch.dict(
"homeassistant.components.zwave_js.api.USER_AGENT",
{"HomeAssistant": "0.0.0"},
),
):
data = {"file": firmware_file}
data.update(firmware_data)
resp = await client.post(
f"/api/zwave_js/firmware/upload/{device.id}", data=data
)
update_data = NodeFirmwareUpdateData(
"file", b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
)
for attr, value in expected_data.items():
setattr(update_data, attr, value)
mock_controller_cmd.assert_not_called()
assert mock_node_cmd.call_args[0][1:3] == (multisensor_6, [update_data])
assert mock_node_cmd.call_args[1] == {
"additional_user_agent_components": {"HomeAssistant": "0.0.0"},
}
assert json.loads(await resp.text()) is None
async def test_firmware_upload_view_controller(
hass: HomeAssistant,
client,
integration,
hass_client: ClientSessionGenerator,
firmware_file,
) -> None:
"""Test the HTTP firmware upload view for a controller."""
hass_client = await hass_client()
device = get_device(hass, client.driver.controller.nodes[1])
with (
patch(
"homeassistant.components.zwave_js.api.update_firmware",
) as mock_node_cmd,
patch(
"homeassistant.components.zwave_js.api.controller_firmware_update_otw",
) as mock_controller_cmd,
patch.dict(
"homeassistant.components.zwave_js.api.USER_AGENT",
{"HomeAssistant": "0.0.0"},
),
):
resp = await hass_client.post(
f"/api/zwave_js/firmware/upload/{device.id}",
data={"file": firmware_file},
)
mock_node_cmd.assert_not_called()
assert mock_controller_cmd.call_args[0][1:2] == (
ControllerFirmwareUpdateData(
"file", b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
),
)
assert mock_controller_cmd.call_args[1] == {
"additional_user_agent_components": {"HomeAssistant": "0.0.0"},
}
assert json.loads(await resp.text()) is None
async def test_firmware_upload_view_failed_command(
hass: HomeAssistant,
multisensor_6,
integration,
hass_client: ClientSessionGenerator,
firmware_file,
) -> None:
"""Test failed command for the HTTP firmware upload view."""
client = await hass_client()
device = get_device(hass, multisensor_6)
with patch(
"homeassistant.components.zwave_js.api.update_firmware",
side_effect=FailedCommand("test", "test"),
):
resp = await client.post(
f"/api/zwave_js/firmware/upload/{device.id}",
data={"file": firmware_file},
)
assert resp.status == HTTPStatus.BAD_REQUEST
async def test_firmware_upload_view_invalid_payload(
hass: HomeAssistant, multisensor_6, integration, hass_client: ClientSessionGenerator
) -> None:
"""Test an invalid payload for the HTTP firmware upload view."""
device = get_device(hass, multisensor_6)
client = await hass_client()
resp = await client.post(
f"/api/zwave_js/firmware/upload/{device.id}",
data={"wrong_key": BytesIO(bytes(10))},
)
assert resp.status == HTTPStatus.BAD_REQUEST
async def test_firmware_upload_view_no_driver(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_client: ClientSessionGenerator,
) -> None:
"""Test the HTTP firmware upload view when the driver doesn't exist."""
device = get_device(hass, multisensor_6)
client.driver = None
aiohttp_client = await hass_client()
resp = await aiohttp_client.post(
f"/api/zwave_js/firmware/upload/{device.id}",
data={"wrong_key": BytesIO(bytes(10))},
)
assert resp.status == HTTPStatus.NOT_FOUND
@pytest.mark.parametrize(
("method", "url"),
[("post", "/api/zwave_js/firmware/upload/{}")],
)
async def test_node_view_non_admin_user(
hass: HomeAssistant,
multisensor_6,
integration,
hass_client: ClientSessionGenerator,
hass_admin_user: MockUser,
method,
url,
) -> None:
"""Test node level views for non-admin users."""
client = await hass_client()
device = get_device(hass, multisensor_6)
# Verify we require admin user
hass_admin_user.groups = []
resp = await client.request(method, url.format(device.id))
assert resp.status == HTTPStatus.UNAUTHORIZED
@pytest.mark.parametrize(
("method", "url"),
[
("post", "/api/zwave_js/firmware/upload/{}"),
],
)
async def test_view_unloaded_config_entry(
hass: HomeAssistant,
multisensor_6,
integration,
hass_client: ClientSessionGenerator,
method,
url,
) -> None:
"""Test an unloaded config entry raises Bad Request."""
client = await hass_client()
device = get_device(hass, multisensor_6)
await hass.config_entries.async_unload(integration.entry_id)
resp = await client.request(method, url.format(device.id))
assert resp.status == HTTPStatus.BAD_REQUEST
@pytest.mark.parametrize(
("method", "url"),
[("post", "/api/zwave_js/firmware/upload/INVALID")],
)
async def test_view_invalid_device_id(
integration, hass_client: ClientSessionGenerator, method, url
) -> None:
"""Test an invalid device id parameter."""
client = await hass_client()
resp = await client.request(method, url.format(integration.entry_id))
assert resp.status == HTTPStatus.NOT_FOUND
async def test_subscribe_log_updates(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the subscribe_log_updates websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {}
await ws_client.send_json(
{ID: 1, TYPE: "zwave_js/subscribe_log_updates", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert msg["success"]
event = Event(
type="logging",
data={
"source": "driver",
"event": "logging",
"message": "test",
"formattedMessage": "test",
"direction": ">",
"level": "debug",
"primaryTags": "tag",
"secondaryTags": "tag2",
"secondaryTagPadding": 0,
"multiline": False,
"timestamp": "time",
"label": "label",
"context": {"source": "config"},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"type": "log_message",
"log_message": {
"message": ["test"],
"level": "debug",
"primary_tags": "tag",
"timestamp": "time",
},
}
event = Event(
type="log config updated",
data={
"source": "driver",
"event": "log config updated",
"config": {
"enabled": False,
"level": "error",
"logToFile": True,
"filename": "test",
"forceConsole": True,
},
},
)
client.driver.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"type": "log_config",
"log_config": {
"enabled": False,
"level": "error",
"log_to_file": True,
"filename": "test",
"force_console": True,
},
}
# Test FailedZWaveCommand is caught
client.async_start_listening_logs.side_effect = FailedZWaveCommand(
"failed_command", 1, "error message"
)
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/subscribe_log_updates",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{ID: 3, TYPE: "zwave_js/subscribe_log_updates", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_update_log_config(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that update_log_config WS API call and schema validation works."""
entry = integration
ws_client = await hass_ws_client(hass)
# Test we can set log level
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LEVEL: "Error"},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "driver.update_log_config"
assert args["config"] == {"level": "error"}
client.async_send_command.reset_mock()
# Test we can set logToFile to True
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LOG_TO_FILE: True, FILENAME: "/test"},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "driver.update_log_config"
assert args["config"] == {"logToFile": True, "filename": "/test"}
client.async_send_command.reset_mock()
# Test all parameters
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {
LEVEL: "Error",
LOG_TO_FILE: True,
FILENAME: "/test",
FORCE_CONSOLE: True,
ENABLED: True,
},
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "driver.update_log_config"
assert args["config"] == {
"level": "error",
"logToFile": True,
"filename": "/test",
"forceConsole": True,
"enabled": True,
}
client.async_send_command.reset_mock()
# Test error when setting unrecognized log level
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LEVEL: "bad_log_level"},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert "error" in msg and msg["error"]["code"] == "invalid_format"
# Test error without service data
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert "error" in msg and "must contain at least one of" in msg["error"]["message"]
# Test error if we set logToFile to True without providing filename
await ws_client.send_json(
{
ID: 6,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LOG_TO_FILE: True},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert (
"error" in msg
and "must be provided if logging to file" in msg["error"]["message"]
)
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_update_log_config",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LEVEL: "Error"},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 8,
TYPE: "zwave_js/update_log_config",
ENTRY_ID: entry.entry_id,
CONFIG: {LEVEL: "Error"},
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_get_log_config(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that the get_log_config WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
# Test we can get log configuration
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/get_log_config",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["result"]
assert msg["success"]
log_config = msg["result"]
assert log_config["enabled"]
assert log_config["level"] == LogLevel.INFO
assert log_config["log_to_file"] is False
assert log_config["filename"] == ""
assert log_config["force_console"] is False
# 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(
{
ID: 2,
TYPE: "zwave_js/get_log_config",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_data_collection(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that the data collection WS API commands work."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"statisticsEnabled": False}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/data_collection_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result == {"opted_in": None, "enabled": False}
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "driver.is_statistics_enabled"
}
assert CONF_DATA_COLLECTION_OPTED_IN not in entry.data
client.async_send_command.reset_mock()
client.async_send_command.return_value = {}
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/update_data_collection_preference",
ENTRY_ID: entry.entry_id,
OPTED_IN: True,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result is None
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args_list[0][0][0]
assert args["command"] == "driver.enable_statistics"
assert args["applicationName"] == "Home Assistant"
client.async_send_command.reset_mock()
client.async_send_command.return_value = {}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/update_data_collection_preference",
ENTRY_ID: entry.entry_id,
OPTED_IN: False,
}
)
msg = await ws_client.receive_json()
result = msg["result"]
assert result is None
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {
"command": "driver.disable_statistics"
}
assert not entry.data[CONF_DATA_COLLECTION_OPTED_IN]
client.async_send_command.reset_mock()
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_is_statistics_enabled",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/data_collection_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_enable_statistics",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 5,
TYPE: "zwave_js/update_data_collection_preference",
ENTRY_ID: entry.entry_id,
OPTED_IN: True,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 6,
TYPE: "zwave_js/data_collection_status",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
await ws_client.send_json(
{
ID: 7,
TYPE: "zwave_js/update_data_collection_preference",
ENTRY_ID: entry.entry_id,
OPTED_IN: True,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_abort_firmware_update(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the abort_firmware_update WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/abort_firmware_update",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
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"] == "node.abort_firmware_update"
assert args["nodeId"] == multisensor_6.node_id
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_abort_firmware_update",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/abort_firmware_update",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/abort_firmware_update",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with improper device ID fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/abort_firmware_update",
DEVICE_ID: "fake_device",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_is_node_firmware_update_in_progress(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the is_firmware_update_in_progress WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command.return_value = {"progress": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/is_node_firmware_update_in_progress",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "node.is_firmware_update_in_progress"
assert args["nodeId"] == multisensor_6.node_id
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_is_firmware_update_in_progress",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/is_node_firmware_update_in_progress",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/is_node_firmware_update_in_progress",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_subscribe_firmware_update_status(
hass: HomeAssistant,
multisensor_6,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the subscribe_firmware_update_status websocket command."""
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command_no_wait.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_firmware_update_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
event = Event(
type="firmware update progress",
data={
"source": "node",
"event": "firmware update progress",
"nodeId": multisensor_6.node_id,
"progress": {
"currentFile": 1,
"totalFiles": 1,
"sentFragments": 1,
"totalFragments": 10,
"progress": 10.0,
},
},
)
multisensor_6.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update progress",
"current_file": 1,
"total_files": 1,
"sent_fragments": 1,
"total_fragments": 10,
"progress": 10.0,
}
event = Event(
type="firmware update finished",
data={
"source": "node",
"event": "firmware update finished",
"nodeId": multisensor_6.node_id,
"result": {
"status": 255,
"success": True,
"waitTime": 10,
"reInterview": False,
},
},
)
multisensor_6.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update finished",
"status": 255,
"success": True,
"wait_time": 10,
"reinterview": False,
}
async def test_subscribe_firmware_update_status_initial_value(
hass: HomeAssistant,
multisensor_6,
client,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test subscribe_firmware_update_status WS command with in progress update."""
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
assert multisensor_6.firmware_update_progress is None
# Send a firmware update progress event before the WS command
event = Event(
type="firmware update progress",
data={
"source": "node",
"event": "firmware update progress",
"nodeId": multisensor_6.node_id,
"progress": {
"currentFile": 1,
"totalFiles": 1,
"sentFragments": 1,
"totalFragments": 10,
"progress": 10.0,
},
},
)
multisensor_6.receive_event(event)
client.async_send_command_no_wait.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_firmware_update_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update progress",
"current_file": 1,
"total_files": 1,
"sent_fragments": 1,
"total_fragments": 10,
"progress": 10.0,
}
async def test_subscribe_controller_firmware_update_status(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the subscribe_firmware_update_status websocket command for a node."""
ws_client = await hass_ws_client(hass)
device = get_device(hass, client.driver.controller.nodes[1])
client.async_send_command_no_wait.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_firmware_update_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
event = Event(
type="firmware update progress",
data={
"source": "controller",
"event": "firmware update progress",
"progress": {
"sentFragments": 1,
"totalFragments": 10,
"progress": 10.0,
},
},
)
client.driver.controller.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update progress",
"current_file": 1,
"total_files": 1,
"sent_fragments": 1,
"total_fragments": 10,
"progress": 10.0,
}
event = Event(
type="firmware update finished",
data={
"source": "controller",
"event": "firmware update finished",
"result": {
"status": 255,
"success": True,
},
},
)
client.driver.controller.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update finished",
"status": 255,
"success": True,
}
async def test_subscribe_controller_firmware_update_status_initial_value(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test subscribe_firmware_update_status cmd with in progress update for node."""
ws_client = await hass_ws_client(hass)
device = get_device(hass, client.driver.controller.nodes[1])
assert client.driver.controller.firmware_update_progress is None
# Send a firmware update progress event before the WS command
event = Event(
type="firmware update progress",
data={
"source": "controller",
"event": "firmware update progress",
"progress": {
"sentFragments": 1,
"totalFragments": 10,
"progress": 10.0,
},
},
)
client.driver.controller.receive_event(event)
client.async_send_command_no_wait.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_firmware_update_status",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "firmware update progress",
"current_file": 1,
"total_files": 1,
"sent_fragments": 1,
"total_fragments": 10,
"progress": 10.0,
}
async def test_subscribe_firmware_update_status_failures(
hass: HomeAssistant,
multisensor_6,
client,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test failures for the subscribe_firmware_update_status websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
# Test sending command with improper entry ID fails
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_firmware_update_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(
{
ID: 3,
TYPE: "zwave_js/subscribe_firmware_update_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_get_node_firmware_update_capabilities(
hass: HomeAssistant,
client,
multisensor_6,
integration,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the get_node_firmware_update_capabilities WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = get_device(hass, multisensor_6)
client.async_send_command.return_value = {
"capabilities": {
"firmwareUpgradable": True,
"firmwareTargets": [0],
"continuesToFunction": True,
"supportsActivation": True,
}
}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/get_node_firmware_update_capabilities",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == {
"firmware_upgradable": True,
"firmware_targets": [0],
"continues_to_function": True,
"supports_activation": True,
}
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "node.get_firmware_update_capabilities"
assert args["nodeId"] == multisensor_6.node_id
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.node.Node.async_get_firmware_update_capabilities",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/get_node_firmware_update_capabilities",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/get_node_firmware_update_capabilities",
DEVICE_ID: device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with improper device ID fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/get_node_firmware_update_capabilities",
DEVICE_ID: "fake_device",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_is_any_ota_firmware_update_in_progress(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that the is_any_ota_firmware_update_in_progress WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"progress": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/is_any_ota_firmware_update_in_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "controller.is_any_ota_firmware_update_in_progress"
# Test FailedZWaveCommand is caught
with patch(
f"{CONTROLLER_PATCH_PREFIX}.async_is_any_ota_firmware_update_in_progress",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/is_any_ota_firmware_update_in_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/is_any_ota_firmware_update_in_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with improper device ID fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/is_any_ota_firmware_update_in_progress",
ENTRY_ID: "invalid_entry",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_check_for_config_updates(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that the check_for_config_updates WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
# Test we can get log configuration
client.async_send_command.return_value = {
"updateAvailable": True,
"newVersion": "test",
"installedVersion": "test",
}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/check_for_config_updates",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["result"]
assert msg["success"]
config_update = msg["result"]
assert config_update["update_available"]
assert config_update["new_version"] == "test"
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_check_for_config_updates",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/check_for_config_updates",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/check_for_config_updates",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/check_for_config_updates",
ENTRY_ID: "INVALID",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_install_config_update(
hass: HomeAssistant, client, integration, hass_ws_client: WebSocketGenerator
) -> None:
"""Test that the install_config_update WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
# Test we can get log configuration
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/install_config_update",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["result"]
assert msg["success"]
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_install_config_update",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/install_config_update",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/install_config_update",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/install_config_update",
ENTRY_ID: "INVALID",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_subscribe_controller_statistics(
hass: HomeAssistant, integration, client, hass_ws_client: WebSocketGenerator
) -> None:
"""Test the subscribe_controller_statistics command."""
entry = integration
ws_client = await hass_ws_client(hass)
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_controller_statistics",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "statistics updated",
"source": "controller",
"messages_tx": 0,
"messages_rx": 0,
"messages_dropped_tx": 0,
"messages_dropped_rx": 0,
"nak": 0,
"can": 0,
"timeout_ack": 0,
"timout_response": 0,
"timeout_callback": 0,
}
# Fire statistics updated
event = Event(
"statistics updated",
{
"source": "controller",
"event": "statistics updated",
"statistics": {
"messagesTX": 1,
"messagesRX": 1,
"messagesDroppedTX": 1,
"messagesDroppedRX": 1,
"NAK": 1,
"CAN": 1,
"timeoutACK": 1,
"timeoutResponse": 1,
"timeoutCallback": 1,
},
},
)
client.driver.controller.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "statistics updated",
"source": "controller",
"messages_tx": 1,
"messages_rx": 1,
"messages_dropped_tx": 1,
"messages_dropped_rx": 1,
"nak": 1,
"can": 1,
"timeout_ack": 1,
"timout_response": 1,
"timeout_callback": 1,
}
# Test sending command with improper entry ID fails
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/subscribe_controller_statistics",
ENTRY_ID: "fake_entry_id",
}
)
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(
{
ID: 3,
TYPE: "zwave_js/subscribe_controller_statistics",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_subscribe_node_statistics(
hass: HomeAssistant,
multisensor_6,
wallmote_central_scene,
zen_31,
integration,
client,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the subscribe_node_statistics command."""
entry = integration
ws_client = await hass_ws_client(hass)
multisensor_6_device = get_device(hass, multisensor_6)
zen_31_device = get_device(hass, zen_31)
wallmote_central_scene_device = get_device(hass, wallmote_central_scene)
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/subscribe_node_statistics",
DEVICE_ID: multisensor_6_device.id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] is None
msg = await ws_client.receive_json()
assert msg["event"] == {
"source": "node",
"event": "statistics updated",
"nodeId": multisensor_6.node_id,
"commands_tx": 0,
"commands_rx": 0,
"commands_dropped_tx": 0,
"commands_dropped_rx": 0,
"timeout_response": 0,
"rtt": None,
"rssi": None,
"lwr": None,
"nlwr": None,
}
# Fire statistics updated
event = Event(
"statistics updated",
{
"source": "node",
"event": "statistics updated",
"nodeId": multisensor_6.node_id,
"statistics": {
"commandsTX": 1,
"commandsRX": 2,
"commandsDroppedTX": 3,
"commandsDroppedRX": 4,
"timeoutResponse": 5,
"rtt": 6,
"rssi": 7,
"lwr": {
"protocolDataRate": 1,
"rssi": 1,
"repeaters": [wallmote_central_scene.node_id],
"repeaterRSSI": [1],
"routeFailedBetween": [
zen_31.node_id,
multisensor_6.node_id,
],
},
"nlwr": {
"protocolDataRate": 2,
"rssi": 2,
"repeaters": [],
"repeaterRSSI": [127],
"routeFailedBetween": [
multisensor_6.node_id,
zen_31.node_id,
],
},
},
},
)
client.driver.controller.receive_event(event)
msg = await ws_client.receive_json()
assert msg["event"] == {
"event": "statistics updated",
"source": "node",
"node_id": multisensor_6.node_id,
"commands_tx": 1,
"commands_rx": 2,
"commands_dropped_tx": 3,
"commands_dropped_rx": 4,
"timeout_response": 5,
"rtt": 6,
"rssi": 7,
"lwr": {
"protocol_data_rate": 1,
"rssi": 1,
"repeaters": [wallmote_central_scene_device.id],
"repeater_rssi": [1],
"route_failed_between": [
zen_31_device.id,
multisensor_6_device.id,
],
},
"nlwr": {
"protocol_data_rate": 2,
"rssi": 2,
"repeaters": [],
"repeater_rssi": [127],
"route_failed_between": [
multisensor_6_device.id,
zen_31_device.id,
],
},
}
# Test sending command with improper entry ID fails
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/subscribe_node_statistics",
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(
{
ID: 4,
TYPE: "zwave_js/subscribe_node_statistics",
DEVICE_ID: multisensor_6_device.id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_hard_reset_controller(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
client,
integration,
listen_block,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the hard_reset_controller WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
device = device_registry.async_get_device(
identifiers={get_device_id(client.driver, client.driver.controller.nodes[1])}
)
client.async_send_command.return_value = {}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/hard_reset_controller",
ENTRY_ID: entry.entry_id,
}
)
listen_block.set()
listen_block.clear()
await hass.async_block_till_done()
msg = await ws_client.receive_json()
assert msg["result"] == device.id
assert msg["success"]
assert len(client.async_send_command.call_args_list) == 1
assert client.async_send_command.call_args[0][0] == {"command": "driver.hard_reset"}
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.driver.Driver.async_hard_reset",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/hard_reset_controller",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
# 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(
{
ID: 3,
TYPE: "zwave_js/hard_reset_controller",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/hard_reset_controller",
ENTRY_ID: "INVALID",
}
)
msg = await ws_client.receive_json()
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": ""}
@pytest.mark.parametrize(
("config", "installer_mode"), [({}, False), ({CONF_INSTALLER_MODE: True}, True)]
)
async def test_get_integration_settings(
config: dict[str, Any],
installer_mode: bool,
hass: HomeAssistant,
client: MagicMock,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test that the get_integration_settings WS API call works."""
ws_client = await hass_ws_client(hass)
entry = MockConfigEntry(domain="zwave_js", data={"url": "ws://test.org"})
entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: config})
await hass.async_block_till_done()
await ws_client.send_json_auto_id(
{
TYPE: "zwave_js/get_integration_settings",
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"] == {
CONF_INSTALLER_MODE: installer_mode,
}
Reference in New Issue View Git Blame Copy Permalink