Fix zwave_js add node schemas (#73343)

* Fix zwave_js add node schemas

* Code cleanup

* Add test
This commit is contained in:
Raman Gupta 2022-06-11 02:13:50 -04:00 committed by GitHub
parent b1f2e5f897
commit 63b51f566d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 90 additions and 12 deletions

View file

@ -14,6 +14,7 @@ from zwave_js_server.const import (
InclusionStrategy, InclusionStrategy,
LogLevel, LogLevel,
Protocols, Protocols,
ProvisioningEntryStatus,
QRCodeVersion, QRCodeVersion,
SecurityClass, SecurityClass,
ZwaveFeature, ZwaveFeature,
@ -148,6 +149,8 @@ MAX_INCLUSION_REQUEST_INTERVAL = "max_inclusion_request_interval"
UUID = "uuid" UUID = "uuid"
SUPPORTED_PROTOCOLS = "supported_protocols" SUPPORTED_PROTOCOLS = "supported_protocols"
ADDITIONAL_PROPERTIES = "additional_properties" ADDITIONAL_PROPERTIES = "additional_properties"
STATUS = "status"
REQUESTED_SECURITY_CLASSES = "requested_security_classes"
FEATURE = "feature" FEATURE = "feature"
UNPROVISION = "unprovision" UNPROVISION = "unprovision"
@ -160,19 +163,22 @@ def convert_planned_provisioning_entry(info: dict) -> ProvisioningEntry:
"""Handle provisioning entry dict to ProvisioningEntry.""" """Handle provisioning entry dict to ProvisioningEntry."""
return ProvisioningEntry( return ProvisioningEntry(
dsk=info[DSK], dsk=info[DSK],
security_classes=[SecurityClass(sec_cls) for sec_cls in info[SECURITY_CLASSES]], security_classes=info[SECURITY_CLASSES],
status=info[STATUS],
requested_security_classes=info.get(REQUESTED_SECURITY_CLASSES),
additional_properties={ additional_properties={
k: v for k, v in info.items() if k not in (DSK, SECURITY_CLASSES) k: v
for k, v in info.items()
if k not in (DSK, SECURITY_CLASSES, STATUS, REQUESTED_SECURITY_CLASSES)
}, },
) )
def convert_qr_provisioning_information(info: dict) -> QRProvisioningInformation: def convert_qr_provisioning_information(info: dict) -> QRProvisioningInformation:
"""Convert QR provisioning information dict to QRProvisioningInformation.""" """Convert QR provisioning information dict to QRProvisioningInformation."""
protocols = [Protocols(proto) for proto in info.get(SUPPORTED_PROTOCOLS, [])]
return QRProvisioningInformation( return QRProvisioningInformation(
version=QRCodeVersion(info[VERSION]), version=info[VERSION],
security_classes=[SecurityClass(sec_cls) for sec_cls in info[SECURITY_CLASSES]], security_classes=info[SECURITY_CLASSES],
dsk=info[DSK], dsk=info[DSK],
generic_device_class=info[GENERIC_DEVICE_CLASS], generic_device_class=info[GENERIC_DEVICE_CLASS],
specific_device_class=info[SPECIFIC_DEVICE_CLASS], specific_device_class=info[SPECIFIC_DEVICE_CLASS],
@ -183,7 +189,9 @@ def convert_qr_provisioning_information(info: dict) -> QRProvisioningInformation
application_version=info[APPLICATION_VERSION], application_version=info[APPLICATION_VERSION],
max_inclusion_request_interval=info.get(MAX_INCLUSION_REQUEST_INTERVAL), max_inclusion_request_interval=info.get(MAX_INCLUSION_REQUEST_INTERVAL),
uuid=info.get(UUID), uuid=info.get(UUID),
supported_protocols=protocols if protocols else None, supported_protocols=info.get(SUPPORTED_PROTOCOLS),
status=info[STATUS],
requested_security_classes=info.get(REQUESTED_SECURITY_CLASSES),
additional_properties=info.get(ADDITIONAL_PROPERTIES, {}), additional_properties=info.get(ADDITIONAL_PROPERTIES, {}),
) )
@ -197,6 +205,12 @@ PLANNED_PROVISIONING_ENTRY_SCHEMA = vol.All(
cv.ensure_list, cv.ensure_list,
[vol.Coerce(SecurityClass)], [vol.Coerce(SecurityClass)],
), ),
vol.Optional(STATUS, default=ProvisioningEntryStatus.ACTIVE): vol.Coerce(
ProvisioningEntryStatus
),
vol.Optional(REQUESTED_SECURITY_CLASSES): vol.All(
cv.ensure_list, [vol.Coerce(SecurityClass)]
),
}, },
# Provisioning entries can have extra keys for SmartStart # Provisioning entries can have extra keys for SmartStart
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
@ -226,6 +240,12 @@ QR_PROVISIONING_INFORMATION_SCHEMA = vol.All(
cv.ensure_list, cv.ensure_list,
[vol.Coerce(Protocols)], [vol.Coerce(Protocols)],
), ),
vol.Optional(STATUS, default=ProvisioningEntryStatus.ACTIVE): vol.Coerce(
ProvisioningEntryStatus
),
vol.Optional(REQUESTED_SECURITY_CLASSES): vol.All(
cv.ensure_list, [vol.Coerce(SecurityClass)]
),
vol.Optional(ADDITIONAL_PROPERTIES): dict, vol.Optional(ADDITIONAL_PROPERTIES): dict,
} }
), ),

View file

@ -10,6 +10,7 @@ from zwave_js_server.const import (
InclusionStrategy, InclusionStrategy,
LogLevel, LogLevel,
Protocols, Protocols,
ProvisioningEntryStatus,
QRCodeVersion, QRCodeVersion,
SecurityClass, SecurityClass,
ZwaveFeature, ZwaveFeature,
@ -63,8 +64,10 @@ from homeassistant.components.zwave_js.api import (
PROPERTY_KEY, PROPERTY_KEY,
QR_CODE_STRING, QR_CODE_STRING,
QR_PROVISIONING_INFORMATION, QR_PROVISIONING_INFORMATION,
REQUESTED_SECURITY_CLASSES,
SECURITY_CLASSES, SECURITY_CLASSES,
SPECIFIC_DEVICE_CLASS, SPECIFIC_DEVICE_CLASS,
STATUS,
TYPE, TYPE,
UNPROVISION, UNPROVISION,
VALUE, VALUE,
@ -619,13 +622,68 @@ async def test_add_node(
client.async_send_command.reset_mock() client.async_send_command.reset_mock()
client.async_send_command.return_value = {"success": True} client.async_send_command.return_value = {"success": True}
# Test S2 QR code string # Test S2 QR provisioning information
await ws_client.send_json( await ws_client.send_json(
{ {
ID: 4, ID: 4,
TYPE: "zwave_js/add_node", TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id, ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value, 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", QR_CODE_STRING: "90testtesttesttesttesttesttesttesttesttesttesttesttest",
} }
) )
@ -648,7 +706,7 @@ async def test_add_node(
# Test Smart Start QR provisioning information with S2 inclusion strategy fails # Test Smart Start QR provisioning information with S2 inclusion strategy fails
await ws_client.send_json( await ws_client.send_json(
{ {
ID: 5, ID: 6,
TYPE: "zwave_js/add_node", TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id, ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value, INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S2.value,
@ -678,7 +736,7 @@ async def test_add_node(
# Test QR provisioning information with S0 inclusion strategy fails # Test QR provisioning information with S0 inclusion strategy fails
await ws_client.send_json( await ws_client.send_json(
{ {
ID: 5, ID: 7,
TYPE: "zwave_js/add_node", TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id, ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S0, INCLUSION_STRATEGY: InclusionStrategy.SECURITY_S0,
@ -708,7 +766,7 @@ async def test_add_node(
# Test ValueError is caught as failure # Test ValueError is caught as failure
await ws_client.send_json( await ws_client.send_json(
{ {
ID: 6, ID: 8,
TYPE: "zwave_js/add_node", TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id, ENTRY_ID: entry.entry_id,
INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value, INCLUSION_STRATEGY: InclusionStrategy.DEFAULT.value,
@ -728,7 +786,7 @@ async def test_add_node(
): ):
await ws_client.send_json( await ws_client.send_json(
{ {
ID: 7, ID: 9,
TYPE: "zwave_js/add_node", TYPE: "zwave_js/add_node",
ENTRY_ID: entry.entry_id, ENTRY_ID: entry.entry_id,
} }
@ -744,7 +802,7 @@ async def test_add_node(
await hass.async_block_till_done() await hass.async_block_till_done()
await ws_client.send_json( await ws_client.send_json(
{ID: 8, TYPE: "zwave_js/add_node", ENTRY_ID: entry.entry_id} {ID: 10, TYPE: "zwave_js/add_node", ENTRY_ID: entry.entry_id}
) )
msg = await ws_client.receive_json() msg = await ws_client.receive_json()