evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add SmartStart provisioning support to zwave_js WS API (#59037)

Browse source 
Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>
Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Raman Gupta 2021-12-01 05:43:51 -05:00 • committed by GitHub
parent 694f6d0abe
commit eeafa36abc
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 1213 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

846
tests/components/zwave_js/test_api.py
View file

@ -9,7 +9,10 @@ from zwave_js_server.const import (
CommandClass,
InclusionStrategy,
LogLevel,
Protocols,
QRCodeVersion,
SecurityClass,
ZwaveFeature,
)
from zwave_js_server.event import Event
from zwave_js_server.exceptions import (
@ -19,31 +22,49 @@ from zwave_js_server.exceptions import (
NotFoundError,
SetValueFailed,
)
from zwave_js_server.model.controller import (
ProvisioningEntry,
QRProvisioningInformation,
)
from zwave_js_server.model.node import Node
from zwave_js_server.model.value import _get_value_id_from_dict, get_value_id
from homeassistant.components.websocket_api.const import ERR_NOT_FOUND
from homeassistant.components.zwave_js.api import (
APPLICATION_VERSION,
CLIENT_SIDE_AUTH,
COMMAND_CLASS_ID,
CONFIG,
DSK,
ENABLED,
ENTRY_ID,
ERR_NOT_LOADED,
FEATURE,
FILENAME,
FORCE_CONSOLE,
GENERIC_DEVICE_CLASS,
ID,
INCLUSION_STRATEGY,
INSTALLER_ICON_TYPE,
LEVEL,
LOG_TO_FILE,
MANUFACTURER_ID,
NODE_ID,
OPTED_IN,
PIN,
PLANNED_PROVISIONING_ENTRY,
PRODUCT_ID,
PRODUCT_TYPE,
PROPERTY,
PROPERTY_KEY,
QR_CODE_STRING,
QR_PROVISIONING_INFORMATION,
SECURITY_CLASSES,
SPECIFIC_DEVICE_CLASS,
TYPE,
UNPROVISION,
VALUE,
VERSION,
)
from homeassistant.components.zwave_js.const import (
CONF_DATA_COLLECTION_OPTED_IN,
@ -421,9 +442,10 @@ async def test_add_node(
client.async_send_command.return_value = {"success": True}
# Test inclusion with no provisioning input
await ws_client.send_json(
{
ID: 3,
ID: 1,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
@ -542,6 +564,193 @@ async def test_add_node(
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(
QRCodeVersion.S2,
[SecurityClass.S2_UNAUTHENTICATED],
"test",
1,
1,
1,
1,
1,
1,
"test",
None,
None,
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/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 Smart Start QR provisioning information with S2 inclusion strategy fails
await ws_client.send_json(
{
ID: 5,
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: 5,
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: 6,
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(
"zwave_js_server.model.controller.Controller.async_begin_inclusion",
@ -549,7 +758,7 @@ async def test_add_node(
):
await ws_client.send_json(
{
ID: 4,
ID: 7,
TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id,
}
@ -565,7 +774,7 @@ async def test_add_node(
await hass.async_block_till_done()
await ws_client.send_json(
{ID: 5, TYPE: "zwave_js/add_node", ENTRY_ID: entry.entry_id}
{ID: 8, TYPE: "zwave_js/add_node", ENTRY_ID: entry.entry_id}
)
msg = await ws_client.receive_json()
@ -661,6 +870,465 @@ async def test_validate_dsk_and_enter_pin(hass, integration, client, hass_ws_cli
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_provision_smart_start_node(hass, integration, client, hass_ws_client):
"""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",
},
}
)
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(
QRCodeVersion.SMART_START,
[SecurityClass.S2_UNAUTHENTICATED],
"test",
1,
1,
1,
1,
1,
1,
"test",
None,
None,
None,
).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(
"zwave_js_server.model.controller.Controller.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"] == "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, integration, client, hass_ws_client):
"""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(
"zwave_js_server.model.controller.Controller.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"] == "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, integration, client, hass_ws_client):
"""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": [SecurityClass.S2_UNAUTHENTICATED],
"additional_properties": {"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(
"zwave_js_server.model.controller.Controller.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"] == "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, integration, client, hass_ws_client):
"""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": [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": 1,
"uuid": "test",
"supported_protocols": [Protocols.ZWAVE],
}
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"] == "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_supports_feature(hass, integration, client, hass_ws_client):
"""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, integration, client, hass_ws_client):
"""Test cancelling the inclusion and exclusion process."""
entry = integration
@ -754,12 +1422,17 @@ async def test_remove_node(
client.async_send_command.return_value = {"success": True}
await ws_client.send_json(
{ID: 3, TYPE: "zwave_js/remove_node", ENTRY_ID: entry.entry_id}
{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={
@ -792,6 +1465,28 @@ async def test_remove_node(
)
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,
UNPROVISION: True,
}
)
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",
"unprovision": True,
}
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.controller.Controller.async_begin_exclusion",
@ -847,11 +1542,12 @@ async def test_replace_failed_node(
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: 3,
ID: 1,
TYPE: "zwave_js/replace_failed_node",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
@ -997,6 +1693,140 @@ async def test_replace_failed_node(
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",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
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",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
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(
QRCodeVersion.S2,
[SecurityClass.S2_UNAUTHENTICATED],
"test",
1,
1,
1,
1,
1,
1,
"test",
None,
None,
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",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
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",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
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(
"zwave_js_server.model.controller.Controller.async_replace_failed_node",
@ -1004,7 +1834,7 @@ async def test_replace_failed_node(
):
await ws_client.send_json(
{
ID: 4,
ID: 7,
TYPE: "zwave_js/replace_failed_node",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
@ -1022,7 +1852,7 @@ async def test_replace_failed_node(
await ws_client.send_json(
{
ID: 5,
ID: 8,
TYPE: "zwave_js/replace_failed_node",
ENTRY_ID: entry.entry_id,
NODE_ID: 67,
@ -2324,7 +3154,7 @@ async def test_update_log_config(hass, client, integration, hass_ws_client):
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert "error" in msg and "value must be one of" in msg["error"]["message"]
assert "error" in msg and msg["error"]["code"] == "invalid_format"
# Test error without service data
await ws_client.send_json(