diff --git a/README.md b/README.md index 417753df8..a5202bc35 100644 --- a/README.md +++ b/README.md @@ -32,3 +32,15 @@ try: except zwave_js_server.client.FailedCommand as err: print("Command failed with", err.error_code) ``` + +## Access control + +Schema 48 adds typed access-control helpers exposed via the `AccessControlAPI` +wrapper: + +- `node.access_control`: shortcut for the root endpoint API. +- `endpoint.access_control`: API for a specific endpoint. +- Call `await endpoint.access_control.is_supported()` before using other + methods. +- Credential payloads accept `str | bytes`; binary credentials are converted to + the websocket Buffer transport shape internally. diff --git a/test/conftest.py b/test/conftest.py index 85574a79a..c7e786a16 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -262,7 +262,7 @@ def version_data_fixture() -> dict[str, Any]: "serverVersion": "test_server_version", "homeId": "test_home_id", "minSchemaVersion": 0, - "maxSchemaVersion": 47, + "maxSchemaVersion": 48, } diff --git a/test/model/test_access_control.py b/test/model/test_access_control.py new file mode 100644 index 000000000..a29793e54 --- /dev/null +++ b/test/model/test_access_control.py @@ -0,0 +1,1268 @@ +"""Test access-control models and API wrappers.""" + +from __future__ import annotations + +from typing import Any + +from zwave_js_server.const import SupervisionStatus +from zwave_js_server.const.command_class.access_control import ( + AssignCredentialResult, + SetCredentialResult, + SetUserResult, + UserCredentialLearnStatus, + UserCredentialRule, + UserCredentialType, + UserCredentialUserType, +) +from zwave_js_server.event import Event +from zwave_js_server.model.access_control import ( + CredentialCapabilities, + CredentialChangedArgs, + CredentialData, + CredentialDeletedArgs, + CredentialLearnCompletedArgs, + CredentialLearnProgressArgs, + SetUserOptions, + UserCapabilities, + UserCredentialCapability, + UserData, + UserDeletedArgs, + deserialize_credential_data, + serialize_credential_data, +) +from zwave_js_server.model.node import Node + +from ..common import MockCommandProtocol + + +def test_credential_capabilities_from_dict() -> None: + """Test credential capability map normalization.""" + + capabilities = CredentialCapabilities.from_dict( + { + "supportedCredentialTypes": { + str(UserCredentialType.PIN_CODE.value): { + "numberOfCredentialSlots": 2, + "minCredentialLength": 4, + "maxCredentialLength": 8, + "maxCredentialHashLength": 0, + "supportsCredentialLearn": True, + "credentialLearnRecommendedTimeout": 30, + "credentialLearnNumberOfSteps": 3, + } + }, + "supportsAdminCode": True, + "supportsAdminCodeDeactivation": False, + "supportsCredentialAssignment": True, + } + ) + + assert capabilities.supports_admin_code + assert not capabilities.supports_admin_code_deactivation + assert capabilities.supports_credential_assignment + assert UserCredentialType.PIN_CODE in capabilities.supported_credential_types + pin_code = capabilities.supported_credential_types[UserCredentialType.PIN_CODE] + assert pin_code.number_of_credential_slots == 2 + assert pin_code.supports_credential_learn + assert pin_code.credential_learn_number_of_steps == 3 + + +async def test_access_control_support_and_capabilities( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test support checks and cached capability commands.""" + + node = lock_schlage_be469 + endpoint = node.endpoints[0] + ack_commands = mock_command( + { + "command": "endpoint.access_control.is_supported", + "nodeId": node.node_id, + "endpoint": endpoint.index, + }, + {"supported": True}, + ) + mock_command( + { + "command": "endpoint.access_control.get_user_capabilities_cached", + "nodeId": node.node_id, + "endpoint": endpoint.index, + }, + { + "capabilities": { + "maxUsers": 25, + "supportedUserTypes": [ + UserCredentialUserType.GENERAL, + UserCredentialUserType.EXPIRING, + ], + "maxUserNameLength": 10, + "supportedCredentialRules": [ + UserCredentialRule.SINGLE, + UserCredentialRule.DUAL, + ], + } + }, + ) + mock_command( + { + "command": "endpoint.access_control.get_credential_capabilities_cached", + "nodeId": node.node_id, + "endpoint": endpoint.index, + }, + { + "capabilities": { + "supportedCredentialTypes": { + str(UserCredentialType.PIN_CODE.value): { + "numberOfCredentialSlots": 5, + "minCredentialLength": 4, + "maxCredentialLength": 8, + "maxCredentialHashLength": 0, + "supportsCredentialLearn": False, + } + }, + "supportsAdminCode": True, + "supportsAdminCodeDeactivation": True, + "supportsCredentialAssignment": True, + } + }, + ) + + assert await endpoint.access_control.is_supported() + user_capabilities = await node.access_control.get_user_capabilities_cached() + credential_capabilities = ( + await node.access_control.get_credential_capabilities_cached() + ) + + assert len(ack_commands) == 3 + assert ack_commands[0] == { + "command": "endpoint.access_control.is_supported", + "nodeId": node.node_id, + "endpoint": endpoint.index, + "messageId": uuid4, + } + assert ack_commands[1]["command"] == ( + "endpoint.access_control.get_user_capabilities_cached" + ) + assert ack_commands[2]["command"] == ( + "endpoint.access_control.get_credential_capabilities_cached" + ) + + assert user_capabilities == UserCapabilities( + max_users=25, + supported_user_types=[ + UserCredentialUserType.GENERAL, + UserCredentialUserType.EXPIRING, + ], + max_user_name_length=10, + supported_credential_rules=[ + UserCredentialRule.SINGLE, + UserCredentialRule.DUAL, + ], + ) + assert credential_capabilities.supports_admin_code + assert credential_capabilities.supports_admin_code_deactivation + assert credential_capabilities.supports_credential_assignment + assert ( + credential_capabilities.supported_credential_types[ + UserCredentialType.PIN_CODE + ].number_of_credential_slots + == 5 + ) + + +async def test_access_control_get_user_and_credential( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test fetching access-control users and credentials through node proxies.""" + + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.get_user", + "nodeId": node.node_id, + "endpoint": 0, + }, + { + "user": { + "userId": 3, + "active": True, + "userType": UserCredentialUserType.EXPIRING, + "userName": "Guest", + "credentialRule": UserCredentialRule.DUAL, + "expiringTimeoutMinutes": 60, + } + }, + ) + mock_command( + { + "command": "endpoint.access_control.get_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + { + "credential": { + "userId": 3, + "type": UserCredentialType.PIN_CODE, + "slot": 0, + "data": {"type": "Buffer", "data": [49, 50, 51, 52]}, + } + }, + ) + + user = await node.access_control.get_user(user_id=3) + credential = await node.access_control.get_credential( + credential_type=UserCredentialType.PIN_CODE, credential_slot=0 + ) + + assert len(ack_commands) == 2 + assert ack_commands[0] == { + "command": "endpoint.access_control.get_user", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "messageId": uuid4, + } + assert ack_commands[1] == { + "command": "endpoint.access_control.get_credential", + "nodeId": node.node_id, + "endpoint": 0, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "messageId": uuid4, + } + assert user == UserData( + user_id=3, + active=True, + user_type=UserCredentialUserType.EXPIRING, + user_name="Guest", + credential_rule=UserCredentialRule.DUAL, + expiring_timeout_minutes=60, + ) + assert credential == CredentialData( + user_id=3, + type=UserCredentialType.PIN_CODE, + slot=0, + data=b"1234", + ) + + +async def test_access_control_set_user( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test set_user returns SetUserResult.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.set_user", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetUserResult.OK}, + ) + + result = await node.access_control.set_user( + user_id=3, + options=SetUserOptions( + active=True, + user_type=UserCredentialUserType.GENERAL, + user_name="Guest", + credential_rule=UserCredentialRule.SINGLE, + ), + ) + + assert result is SetUserResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.set_user", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "options": { + "active": True, + "userType": UserCredentialUserType.GENERAL, + "userName": "Guest", + "credentialRule": UserCredentialRule.SINGLE, + }, + "messageId": uuid4, + } + ] + + +async def test_access_control_delete_user( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test delete_user returns SetUserResult.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.delete_user", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetUserResult.OK}, + ) + + result = await node.access_control.delete_user(user_id=3) + + assert result is SetUserResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.delete_user", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "messageId": uuid4, + } + ] + + +async def test_access_control_delete_all_users( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test delete_all_users returns SetUserResult.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.delete_all_users", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetUserResult.OK}, + ) + + result = await node.access_control.delete_all_users() + + assert result is SetUserResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.delete_all_users", + "nodeId": node.node_id, + "endpoint": 0, + "messageId": uuid4, + } + ] + + +async def test_access_control_set_credential( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test set_credential returns SetCredentialResult and wraps bytes payload.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.set_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetCredentialResult.OK}, + ) + + result = await node.access_control.set_credential( + user_id=3, + credential_type=UserCredentialType.PIN_CODE, + credential_slot=0, + data=b"1234", + ) + + assert result is SetCredentialResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.set_credential", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": {"type": "Buffer", "data": [49, 50, 51, 52]}, + "messageId": uuid4, + } + ] + + +async def test_access_control_delete_credential( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test delete_credential returns SetCredentialResult.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.delete_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetCredentialResult.OK}, + ) + + result = await node.access_control.delete_credential( + user_id=3, + credential_type=UserCredentialType.PIN_CODE, + credential_slot=0, + ) + + assert result is SetCredentialResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.delete_credential", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "messageId": uuid4, + } + ] + + +async def test_access_control_assign_credential( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test assign_credential returns AssignCredentialResult.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.assign_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": AssignCredentialResult.OK}, + ) + + result = await node.access_control.assign_credential( + credential_type=UserCredentialType.PIN_CODE, + credential_slot=1, + destination_user_id=5, + ) + + assert result is AssignCredentialResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.assign_credential", + "nodeId": node.node_id, + "endpoint": 0, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 1, + "destinationUserId": 5, + "messageId": uuid4, + } + ] + + +async def test_access_control_start_credential_learn( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test start_credential_learn returns supervision result.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.start_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": {"status": SupervisionStatus.SUCCESS}}, + ) + + result = await node.access_control.start_credential_learn( + user_id=3, + credential_type=UserCredentialType.PIN_CODE, + credential_slot=0, + timeout=30, + ) + + assert result is not None + assert result.status is SupervisionStatus.SUCCESS + assert ack_commands == [ + { + "command": "endpoint.access_control.start_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "timeout": 30, + "messageId": uuid4, + } + ] + + +async def test_access_control_cancel_credential_learn( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test cancel_credential_learn returns supervision result.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.cancel_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": {"status": SupervisionStatus.SUCCESS}}, + ) + + result = await node.access_control.cancel_credential_learn() + + assert result is not None + assert result.status is SupervisionStatus.SUCCESS + assert ack_commands == [ + { + "command": "endpoint.access_control.cancel_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + "messageId": uuid4, + } + ] + + +async def test_access_control_set_admin_code( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test set_admin_code returns supervision result.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.set_admin_code", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": {"status": SupervisionStatus.SUCCESS}}, + ) + + result = await node.access_control.set_admin_code(code="2468") + + assert result is not None + assert result.status is SupervisionStatus.SUCCESS + assert ack_commands == [ + { + "command": "endpoint.access_control.set_admin_code", + "nodeId": node.node_id, + "endpoint": 0, + "code": "2468", + "messageId": uuid4, + } + ] + + +async def test_access_control_get_admin_code( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test fetching access-control admin code.""" + + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.get_admin_code", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"code": "2468"}, + ) + + assert await node.access_control.get_admin_code() == "2468" + assert ack_commands == [ + { + "command": "endpoint.access_control.get_admin_code", + "nodeId": node.node_id, + "endpoint": 0, + "messageId": uuid4, + } + ] + + +def _dispatch_access_control_event( + node: Node, event_type: str, args: dict[str, Any] +) -> dict[str, Any]: + """Dispatch a schema-48 access-control event and return the observed payload.""" + observed: list[dict[str, Any]] = [] + node.on(event_type, observed.append) + + node.receive_event( + Event( + event_type, + { + "source": "node", + "event": event_type, + "nodeId": node.node_id, + "endpointIndex": 0, + "args": args, + }, + ) + ) + + assert len(observed) == 1 + event_data = observed[0] + assert event_data["node"] is node + assert event_data["endpoint"] is node.endpoints[0] + assert event_data["endpointIndex"] == 0 + return event_data + + +def test_access_control_user_added_event(lock_schlage_be469: Node) -> None: + """Test `user added` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "user added", + { + "userId": 3, + "active": True, + "userType": UserCredentialUserType.GENERAL, + "userName": "Guest", + "credentialRule": UserCredentialRule.SINGLE, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, UserData) + assert parsed.user_id == 3 + assert parsed.user_type is UserCredentialUserType.GENERAL + assert parsed.user_name == "Guest" + assert parsed.credential_rule is UserCredentialRule.SINGLE + + +def test_access_control_user_modified_event(lock_schlage_be469: Node) -> None: + """Test `user modified` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "user modified", + { + "userId": 3, + "active": False, + "userType": UserCredentialUserType.EXPIRING, + "userName": "Temp", + "credentialRule": UserCredentialRule.DUAL, + "expiringTimeoutMinutes": 30, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, UserData) + assert parsed.user_id == 3 + assert parsed.user_type is UserCredentialUserType.EXPIRING + assert parsed.expiring_timeout_minutes == 30 + + +def test_access_control_user_deleted_event(lock_schlage_be469: Node) -> None: + """Test `user deleted` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, "user deleted", {"userId": 3} + ) + parsed = event_data["args"] + assert isinstance(parsed, UserDeletedArgs) + assert parsed.user_id == 3 + + +def test_access_control_credential_added_event(lock_schlage_be469: Node) -> None: + """Test `credential added` event normalization with bytes payload.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "credential added", + { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": {"type": "Buffer", "data": [49, 50, 51, 52]}, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, CredentialChangedArgs) + assert parsed.user_id == 3 + assert parsed.credential_type is UserCredentialType.PIN_CODE + assert parsed.data == b"1234" + + +def test_access_control_credential_modified_event(lock_schlage_be469: Node) -> None: + """Test `credential modified` event normalization with str payload.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "credential modified", + { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": "2468", + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, CredentialChangedArgs) + assert parsed.user_id == 3 + assert parsed.credential_type is UserCredentialType.PIN_CODE + assert parsed.data == "2468" + + +def test_access_control_credential_deleted_event(lock_schlage_be469: Node) -> None: + """Test `credential deleted` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "credential deleted", + { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, CredentialDeletedArgs) + assert parsed.credential_slot == 0 + assert parsed.credential_type is UserCredentialType.PIN_CODE + + +def test_access_control_credential_learn_progress_event( + lock_schlage_be469: Node, +) -> None: + """Test `credential learn progress` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "credential learn progress", + { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "stepsRemaining": 2, + "status": UserCredentialLearnStatus.STEP_RETRY, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, CredentialLearnProgressArgs) + assert parsed.steps_remaining == 2 + assert parsed.status is UserCredentialLearnStatus.STEP_RETRY + + +def test_access_control_credential_learn_completed_event( + lock_schlage_be469: Node, +) -> None: + """Test `credential learn completed` event normalization.""" + event_data = _dispatch_access_control_event( + lock_schlage_be469, + "credential learn completed", + { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "status": UserCredentialLearnStatus.SUCCESS, + "success": True, + }, + ) + parsed = event_data["args"] + assert isinstance(parsed, CredentialLearnCompletedArgs) + assert parsed.success + assert parsed.status is UserCredentialLearnStatus.SUCCESS + + +async def test_access_control_list_commands( + lock_schlage_be469: Node, mock_command: MockCommandProtocol +) -> None: + """Test fetching lists of users and credentials (fresh and cached).""" + node = lock_schlage_be469 + user_payload = { + "userId": 1, + "active": True, + "userType": UserCredentialUserType.GENERAL, + "userName": "Owner", + "credentialRule": UserCredentialRule.SINGLE, + } + credential_payload = { + "userId": 1, + "type": UserCredentialType.PIN_CODE, + "slot": 0, + "data": "1234", + } + mock_command( + { + "command": "endpoint.access_control.get_users", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"users": [user_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_users_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"users": [user_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_user_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"user": user_payload}, + ) + mock_command( + { + "command": "endpoint.access_control.get_credentials", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_credentials_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_credential_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credential": credential_payload}, + ) + mock_command( + { + "command": "endpoint.access_control.get_credentials_by_type", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_credentials_by_type_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_all_credentials", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + mock_command( + { + "command": "endpoint.access_control.get_all_credentials_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"credentials": [credential_payload]}, + ) + + assert await node.access_control.get_users() == [UserData.from_dict(user_payload)] + assert await node.access_control.get_users_cached() == [ + UserData.from_dict(user_payload) + ] + assert await node.access_control.get_user_cached(user_id=1) == UserData.from_dict( + user_payload + ) + assert await node.access_control.get_credentials(user_id=1) == [ + CredentialData.from_dict(credential_payload) + ] + assert await node.access_control.get_credentials_cached(user_id=1) == [ + CredentialData.from_dict(credential_payload) + ] + assert await node.access_control.get_credential_cached( + credential_type=UserCredentialType.PIN_CODE, credential_slot=0 + ) == CredentialData.from_dict(credential_payload) + assert await node.access_control.get_credentials_by_type( + credential_type=UserCredentialType.PIN_CODE + ) == [CredentialData.from_dict(credential_payload)] + assert await node.access_control.get_credentials_by_type_cached( + credential_type=UserCredentialType.PIN_CODE + ) == [CredentialData.from_dict(credential_payload)] + assert await node.access_control.get_all_credentials() == [ + CredentialData.from_dict(credential_payload) + ] + assert await node.access_control.get_all_credentials_cached() == [ + CredentialData.from_dict(credential_payload) + ] + + +async def test_access_control_get_user_none_result( + lock_schlage_be469: Node, mock_command: MockCommandProtocol +) -> None: + """Test that a missing user response returns None.""" + node = lock_schlage_be469 + mock_command( + { + "command": "endpoint.access_control.get_user", + "nodeId": node.node_id, + "endpoint": 0, + }, + {}, + ) + + assert await node.access_control.get_user(user_id=1) is None + + +async def test_access_control_get_user_cached_none_result( + lock_schlage_be469: Node, mock_command: MockCommandProtocol +) -> None: + """Test that a missing cached user response returns None.""" + node = lock_schlage_be469 + mock_command( + { + "command": "endpoint.access_control.get_user_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {}, + ) + + assert await node.access_control.get_user_cached(user_id=1) is None + + +async def test_access_control_get_credential_none_result( + lock_schlage_be469: Node, mock_command: MockCommandProtocol +) -> None: + """Test that a missing credential response returns None.""" + node = lock_schlage_be469 + mock_command( + { + "command": "endpoint.access_control.get_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + {}, + ) + + assert ( + await node.access_control.get_credential( + credential_type=UserCredentialType.PIN_CODE, credential_slot=0 + ) + is None + ) + + +async def test_access_control_get_credential_cached_none_result( + lock_schlage_be469: Node, mock_command: MockCommandProtocol +) -> None: + """Test that a missing cached credential response returns None.""" + node = lock_schlage_be469 + mock_command( + { + "command": "endpoint.access_control.get_credential_cached", + "nodeId": node.node_id, + "endpoint": 0, + }, + {}, + ) + + assert ( + await node.access_control.get_credential_cached( + credential_type=UserCredentialType.PIN_CODE, credential_slot=0 + ) + is None + ) + + +async def test_access_control_set_credential_str_data( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test that string credential payloads are sent without Buffer wrapping.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.set_credential", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": SetCredentialResult.OK}, + ) + + result = await node.access_control.set_credential( + user_id=3, + credential_type=UserCredentialType.PIN_CODE, + credential_slot=0, + data="1234", + ) + + assert result is SetCredentialResult.OK + assert ack_commands == [ + { + "command": "endpoint.access_control.set_credential", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": "1234", + "messageId": uuid4, + } + ] + + +async def test_access_control_start_credential_learn_no_timeout( + lock_schlage_be469: Node, mock_command: MockCommandProtocol, uuid4: str +) -> None: + """Test that start_credential_learn omits timeout when not provided.""" + node = lock_schlage_be469 + ack_commands = mock_command( + { + "command": "endpoint.access_control.start_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + }, + {"result": {"status": SupervisionStatus.SUCCESS}}, + ) + + result = await node.access_control.start_credential_learn( + user_id=3, + credential_type=UserCredentialType.PIN_CODE, + credential_slot=0, + ) + + assert result is not None + assert "timeout" not in ack_commands[0] + assert ack_commands == [ + { + "command": "endpoint.access_control.start_credential_learn", + "nodeId": node.node_id, + "endpoint": 0, + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "messageId": uuid4, + } + ] + + +def test_credential_data_serialize_helpers() -> None: + """Test (de)serialization helpers for credential payloads.""" + + assert deserialize_credential_data(None) is None + assert deserialize_credential_data("abcd") == "abcd" + assert deserialize_credential_data({"type": "Buffer", "data": [1, 2]}) == bytes( + [1, 2] + ) + assert serialize_credential_data("abcd") == "abcd" + assert serialize_credential_data(b"\x01\x02") == { + "type": "Buffer", + "data": [1, 2], + } + + +def test_user_credential_capability_round_trip() -> None: + """Test UserCredentialCapability round-trips via from_dict/to_dict.""" + + full_payload = { + "numberOfCredentialSlots": 5, + "minCredentialLength": 4, + "maxCredentialLength": 8, + "maxCredentialHashLength": 0, + "supportsCredentialLearn": True, + "credentialLearnRecommendedTimeout": 30, + "credentialLearnNumberOfSteps": 3, + } + minimal_payload = { + "numberOfCredentialSlots": 5, + "minCredentialLength": 4, + "maxCredentialLength": 8, + "maxCredentialHashLength": 0, + "supportsCredentialLearn": False, + } + + assert UserCredentialCapability.from_dict(full_payload).to_dict() == full_payload + assert ( + UserCredentialCapability.from_dict(minimal_payload).to_dict() == minimal_payload + ) + + +def test_credential_capabilities_to_dict() -> None: + """Test CredentialCapabilities serialization including nested capability.""" + + payload = { + "supportedCredentialTypes": { + str(UserCredentialType.PIN_CODE.value): { + "numberOfCredentialSlots": 2, + "minCredentialLength": 4, + "maxCredentialLength": 8, + "maxCredentialHashLength": 0, + "supportsCredentialLearn": True, + "credentialLearnRecommendedTimeout": 30, + "credentialLearnNumberOfSteps": 3, + } + }, + "supportsAdminCode": True, + "supportsAdminCodeDeactivation": False, + "supportsCredentialAssignment": True, + } + + assert CredentialCapabilities.from_dict(payload).to_dict() == payload + + +def test_user_capabilities_to_dict() -> None: + """Test UserCapabilities serialization with and without optional fields.""" + + full_payload = { + "maxUsers": 25, + "supportedUserTypes": [ + UserCredentialUserType.GENERAL, + UserCredentialUserType.EXPIRING, + ], + "maxUserNameLength": 10, + "supportedCredentialRules": [ + UserCredentialRule.SINGLE, + UserCredentialRule.DUAL, + ], + } + minimal_payload = { + "maxUsers": 25, + "supportedUserTypes": [], + "supportedCredentialRules": [], + } + + assert UserCapabilities.from_dict(full_payload).to_dict() == full_payload + assert UserCapabilities.from_dict(minimal_payload).to_dict() == minimal_payload + + +def test_user_data_to_dict() -> None: + """Test UserData serialization with all and minimal fields.""" + + full_payload = { + "userId": 3, + "active": True, + "userType": UserCredentialUserType.EXPIRING, + "userName": "Temp", + "credentialRule": UserCredentialRule.DUAL, + "expiringTimeoutMinutes": 30, + } + minimal_payload = { + "userId": 7, + "active": False, + "userType": UserCredentialUserType.GENERAL, + } + + assert UserData.from_dict(full_payload).to_dict() == full_payload + assert UserData.from_dict(minimal_payload).to_dict() == minimal_payload + + +def test_set_user_options_to_dict_with_expiring_timeout() -> None: + """Test SetUserOptions serialization includes expiringTimeoutMinutes.""" + + options = SetUserOptions( + active=True, + user_type=UserCredentialUserType.EXPIRING, + user_name="Temp", + credential_rule=UserCredentialRule.DUAL, + expiring_timeout_minutes=15, + ) + + assert options.to_dict() == { + "active": True, + "userType": UserCredentialUserType.EXPIRING, + "userName": "Temp", + "credentialRule": UserCredentialRule.DUAL, + "expiringTimeoutMinutes": 15, + } + + +def test_set_user_options_to_dict_empty() -> None: + """Test SetUserOptions serialization omits all unset fields.""" + + assert SetUserOptions().to_dict() == {} + + +def test_set_user_options_to_dict_partial() -> None: + """Test SetUserOptions serialization omits fields left as None.""" + + options = SetUserOptions( + active=False, + user_name="Guest", + ) + + assert options.to_dict() == { + "active": False, + "userName": "Guest", + } + + +def test_credential_data_to_dict() -> None: + """Test CredentialData serialization for bytes, str, and missing data.""" + + bytes_payload = { + "userId": 3, + "type": UserCredentialType.PIN_CODE, + "slot": 0, + "data": {"type": "Buffer", "data": [49, 50, 51, 52]}, + } + str_payload = { + "userId": 3, + "type": UserCredentialType.PIN_CODE, + "slot": 0, + "data": "1234", + } + no_data_payload = { + "userId": 3, + "type": UserCredentialType.PIN_CODE, + "slot": 0, + } + + assert CredentialData.from_dict(bytes_payload).to_dict() == bytes_payload + assert CredentialData.from_dict(str_payload).to_dict() == str_payload + assert CredentialData.from_dict(no_data_payload).to_dict() == no_data_payload + + +def test_user_deleted_args_to_dict() -> None: + """Test UserDeletedArgs round-trips via from_dict/to_dict.""" + + payload = {"userId": 3} + assert UserDeletedArgs.from_dict(payload).to_dict() == payload + + +def test_credential_changed_args_to_dict() -> None: + """Test CredentialChangedArgs serialization for bytes, str, and missing data.""" + + bytes_payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": {"type": "Buffer", "data": [49, 50, 51, 52]}, + } + str_payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "data": "2468", + } + no_data_payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + } + + assert CredentialChangedArgs.from_dict(bytes_payload).to_dict() == bytes_payload + assert CredentialChangedArgs.from_dict(str_payload).to_dict() == str_payload + assert CredentialChangedArgs.from_dict(no_data_payload).to_dict() == no_data_payload + + +def test_credential_deleted_args_to_dict() -> None: + """Test CredentialDeletedArgs round-trips via from_dict/to_dict.""" + + payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + } + assert CredentialDeletedArgs.from_dict(payload).to_dict() == payload + + +def test_credential_learn_progress_args_to_dict() -> None: + """Test CredentialLearnProgressArgs round-trips via from_dict/to_dict.""" + + payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "stepsRemaining": 2, + "status": UserCredentialLearnStatus.STEP_RETRY, + } + assert CredentialLearnProgressArgs.from_dict(payload).to_dict() == payload + + +def test_credential_learn_completed_args_to_dict() -> None: + """Test CredentialLearnCompletedArgs round-trips via from_dict/to_dict.""" + + payload = { + "userId": 3, + "credentialType": UserCredentialType.PIN_CODE, + "credentialSlot": 0, + "status": UserCredentialLearnStatus.SUCCESS, + "success": True, + } + assert CredentialLearnCompletedArgs.from_dict(payload).to_dict() == payload diff --git a/test/test_client.py b/test/test_client.py index 79e4765cb..6277236a4 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -463,7 +463,7 @@ async def test_additional_user_agent_components(client_session, url): { "command": "initialize", "messageId": "initialize", - "schemaVersion": 47, + "schemaVersion": 48, "additionalUserAgentComponents": { "zwave-js-server-python": __version__, "foo": "bar", diff --git a/test/test_dump.py b/test/test_dump.py index 6c6be46cf..3eeefd202 100644 --- a/test/test_dump.py +++ b/test/test_dump.py @@ -106,7 +106,7 @@ async def test_dump_additional_user_agent_components( { "command": "initialize", "messageId": "initialize", - "schemaVersion": 47, + "schemaVersion": 48, "additionalUserAgentComponents": { "zwave-js-server-python": __version__, "foo": "bar", diff --git a/test/test_main.py b/test/test_main.py index c23800015..6244138ea 100644 --- a/test/test_main.py +++ b/test/test_main.py @@ -58,7 +58,7 @@ def test_dump_state( assert captured.out == ( "{'type': 'version', 'driverVersion': 'test_driver_version', " "'serverVersion': 'test_server_version', 'homeId': 'test_home_id', " - "'minSchemaVersion': 0, 'maxSchemaVersion': 47}\n" + "'minSchemaVersion': 0, 'maxSchemaVersion': 48}\n" "{'type': 'result', 'success': True, 'result': {}, 'messageId': 'initialize'}\n" "test_result\n" ) diff --git a/zwave_js_server/const/__init__.py b/zwave_js_server/const/__init__.py index 218b6efaf..f2a57b9f8 100644 --- a/zwave_js_server/const/__init__.py +++ b/zwave_js_server/const/__init__.py @@ -13,7 +13,7 @@ # minimal server schema version we can handle MIN_SERVER_SCHEMA_VERSION = 47 # max server schema version we can handle (and our code is compatible with) -MAX_SERVER_SCHEMA_VERSION = 47 +MAX_SERVER_SCHEMA_VERSION = 48 VALUE_UNKNOWN = "unknown" diff --git a/zwave_js_server/const/command_class/access_control.py b/zwave_js_server/const/command_class/access_control.py new file mode 100644 index 000000000..0d4cf58a2 --- /dev/null +++ b/zwave_js_server/const/command_class/access_control.py @@ -0,0 +1,87 @@ +"""Constants for access control related command classes.""" + +from __future__ import annotations + +from enum import IntEnum + + +class UserCredentialType(IntEnum): + """Credential types supported by User Credential CC.""" + + NONE = 0 + PIN_CODE = 1 + PASSWORD = 2 + RFID_CODE = 3 + BLE = 4 + NFC = 5 + UWB = 6 + EYE_BIOMETRIC = 7 + FACE_BIOMETRIC = 8 + FINGER_BIOMETRIC = 9 + HAND_BIOMETRIC = 10 + UNSPECIFIED_BIOMETRIC = 11 + DESFIRE = 12 + + +class UserCredentialUserType(IntEnum): + """User types supported by User Credential CC.""" + + GENERAL = 0 + PROGRAMMING = 3 + NON_ACCESS = 4 + DURESS = 5 + DISPOSABLE = 6 + EXPIRING = 7 + REMOTE_ONLY = 9 + + +class UserCredentialRule(IntEnum): + """Credential rules supported by User Credential CC.""" + + SINGLE = 1 + DUAL = 2 + TRIPLE = 3 + + +class UserCredentialLearnStatus(IntEnum): + """Credential learn statuses supported by User Credential CC.""" + + STARTED = 0 + SUCCESS = 1 + ALREADY_IN_PROGRESS = 2 + ENDED_NOT_DUE_TO_TIMEOUT = 3 + TIMEOUT = 4 + STEP_RETRY = 5 + INVALID_ADD_OPERATION_TYPE = 254 + INVALID_MODIFY_OPERATION_TYPE = 255 + + +class SetUserResult(IntEnum): + """Result for set_user / delete_user / delete_all_users commands.""" + + OK = 0 + ERROR_ADD_REJECTED_LOCATION_OCCUPIED = 1 + ERROR_MODIFY_REJECTED_LOCATION_EMPTY = 2 + ERROR_UNKNOWN = 255 + + +class SetCredentialResult(IntEnum): + """Result for set_credential / delete_credential commands.""" + + OK = 0 + ERROR_ADD_REJECTED_LOCATION_OCCUPIED = 1 + ERROR_MODIFY_REJECTED_LOCATION_EMPTY = 2 + ERROR_DUPLICATE_CREDENTIAL = 3 + ERROR_MANUFACTURER_SECURITY_RULES = 4 + ERROR_DUPLICATE_ADMIN_PIN_CODE = 5 + ERROR_WRONG_USER_UNIQUE_IDENTIFIER = 6 + ERROR_UNKNOWN = 255 + + +class AssignCredentialResult(IntEnum): + """Result for assign_credential commands.""" + + OK = 0 + ERROR_INVALID_CREDENTIAL = 1 + ERROR_INVALID_USER = 2 + ERROR_UNKNOWN = 255 diff --git a/zwave_js_server/model/access_control.py b/zwave_js_server/model/access_control.py new file mode 100644 index 000000000..05537b039 --- /dev/null +++ b/zwave_js_server/model/access_control.py @@ -0,0 +1,885 @@ +"""Shared models and API wrapper for endpoint access-control support.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Self, TypedDict, cast + +from ..const.command_class.access_control import ( + AssignCredentialResult, + SetCredentialResult, + SetUserResult, + UserCredentialLearnStatus, + UserCredentialRule, + UserCredentialType, + UserCredentialUserType, +) +from ..util.helpers import ( + BufferObjectDataType, + buffer_object_to_bytes, + bytes_to_buffer_object, +) +from .value import SupervisionResult, parse_supervision_result + +if TYPE_CHECKING: + from .endpoint import Endpoint + + +def serialize_credential_data(data: str | bytes) -> str | BufferObjectDataType: + """Serialize a credential payload for websocket transport.""" + if isinstance(data, bytes): + return bytes_to_buffer_object(data) + return data + + +def deserialize_credential_data( + data: str | BufferObjectDataType | None, +) -> str | bytes | None: + """Deserialize a credential payload from websocket transport.""" + if data is None: + return None + if isinstance(data, str): + return data + return buffer_object_to_bytes(data) + + +class UserCredentialCapabilityDataType(TypedDict, total=False): + """Represent a credential capability payload.""" + + numberOfCredentialSlots: int + minCredentialLength: int + maxCredentialLength: int + maxCredentialHashLength: int + supportsCredentialLearn: bool + credentialLearnRecommendedTimeout: int + credentialLearnNumberOfSteps: int + + +@dataclass +class UserCredentialCapability: + """Credential capability for a single credential type.""" + + number_of_credential_slots: int + min_credential_length: int + max_credential_length: int + max_credential_hash_length: int + supports_credential_learn: bool + credential_learn_recommended_timeout: int | None = None + credential_learn_number_of_steps: int | None = None + + @classmethod + def from_dict(cls, data: UserCredentialCapabilityDataType) -> Self: + """Return capability from serialized data.""" + return cls( + number_of_credential_slots=data["numberOfCredentialSlots"], + min_credential_length=data["minCredentialLength"], + max_credential_length=data["maxCredentialLength"], + max_credential_hash_length=data["maxCredentialHashLength"], + supports_credential_learn=data["supportsCredentialLearn"], + credential_learn_recommended_timeout=data.get( + "credentialLearnRecommendedTimeout" + ), + credential_learn_number_of_steps=data.get("credentialLearnNumberOfSteps"), + ) + + def to_dict(self) -> UserCredentialCapabilityDataType: + """Return serialized capability data.""" + data: UserCredentialCapabilityDataType = { + "numberOfCredentialSlots": self.number_of_credential_slots, + "minCredentialLength": self.min_credential_length, + "maxCredentialLength": self.max_credential_length, + "maxCredentialHashLength": self.max_credential_hash_length, + "supportsCredentialLearn": self.supports_credential_learn, + } + if self.credential_learn_recommended_timeout is not None: + data["credentialLearnRecommendedTimeout"] = ( + self.credential_learn_recommended_timeout + ) + if self.credential_learn_number_of_steps is not None: + data["credentialLearnNumberOfSteps"] = self.credential_learn_number_of_steps + return data + + +class UserCapabilitiesDataType(TypedDict, total=False): + """Represent user capabilities payload.""" + + maxUsers: int + supportedUserTypes: list[UserCredentialUserType] + maxUserNameLength: int + supportedCredentialRules: list[UserCredentialRule] + + +@dataclass +class UserCapabilities: + """User-related capabilities for an access-control endpoint.""" + + max_users: int + supported_user_types: list[UserCredentialUserType] = field(default_factory=list) + max_user_name_length: int | None = None + supported_credential_rules: list[UserCredentialRule] = field(default_factory=list) + + @classmethod + def from_dict(cls, data: UserCapabilitiesDataType) -> Self: + """Return user capabilities from serialized data.""" + return cls( + max_users=data["maxUsers"], + supported_user_types=[ + UserCredentialUserType(user_type) + for user_type in data.get("supportedUserTypes", []) + ], + max_user_name_length=data.get("maxUserNameLength"), + supported_credential_rules=[ + UserCredentialRule(rule) + for rule in data.get("supportedCredentialRules", []) + ], + ) + + def to_dict(self) -> UserCapabilitiesDataType: + """Return serialized user capabilities.""" + data: UserCapabilitiesDataType = { + "maxUsers": self.max_users, + "supportedUserTypes": list(self.supported_user_types), + "supportedCredentialRules": list(self.supported_credential_rules), + } + if self.max_user_name_length is not None: + data["maxUserNameLength"] = self.max_user_name_length + return data + + +class CredentialCapabilitiesDataType(TypedDict, total=False): + """Represent credential capabilities payload.""" + + supportedCredentialTypes: dict[str, UserCredentialCapabilityDataType] + supportsAdminCode: bool + supportsAdminCodeDeactivation: bool + supportsCredentialAssignment: bool + + +@dataclass +class CredentialCapabilities: + """Credential-related capabilities for an access-control endpoint.""" + + supported_credential_types: dict[UserCredentialType, UserCredentialCapability] = ( + field(default_factory=dict) + ) + supports_admin_code: bool = False + supports_admin_code_deactivation: bool = False + supports_credential_assignment: bool = False + + @classmethod + def from_dict(cls, data: CredentialCapabilitiesDataType) -> Self: + """Return credential capabilities from serialized data.""" + supported_credential_types = { + UserCredentialType( + int(credential_type) + ): UserCredentialCapability.from_dict(capability) + for credential_type, capability in data.get( + "supportedCredentialTypes", {} + ).items() + } + return cls( + supported_credential_types=supported_credential_types, + supports_admin_code=data.get("supportsAdminCode", False), + supports_admin_code_deactivation=data.get( + "supportsAdminCodeDeactivation", False + ), + supports_credential_assignment=data.get( + "supportsCredentialAssignment", False + ), + ) + + def to_dict(self) -> CredentialCapabilitiesDataType: + """Return serialized credential capabilities.""" + return { + "supportedCredentialTypes": { + str(int(credential_type)): capability.to_dict() + for credential_type, capability in self.supported_credential_types.items() + }, + "supportsAdminCode": self.supports_admin_code, + "supportsAdminCodeDeactivation": self.supports_admin_code_deactivation, + "supportsCredentialAssignment": self.supports_credential_assignment, + } + + +class UserDataDataType(TypedDict, total=False): + """Represent access-control user payload.""" + + userId: int + active: bool + userType: UserCredentialUserType + userName: str + credentialRule: UserCredentialRule + expiringTimeoutMinutes: int + + +@dataclass +class UserData: + """User data returned by endpoint access-control commands.""" + + user_id: int + active: bool + user_type: UserCredentialUserType + user_name: str | None = None + credential_rule: UserCredentialRule | None = None + expiring_timeout_minutes: int | None = None + + @classmethod + def from_dict(cls, data: UserDataDataType) -> Self: + """Return user data from serialized payload.""" + return cls( + user_id=data["userId"], + active=data.get("active", False), + user_type=UserCredentialUserType(data.get("userType", 0)), + user_name=data.get("userName"), + credential_rule=( + UserCredentialRule(credential_rule) + if (credential_rule := data.get("credentialRule")) is not None + else None + ), + expiring_timeout_minutes=data.get("expiringTimeoutMinutes"), + ) + + def to_dict(self) -> UserDataDataType: + """Return serialized user payload.""" + data: UserDataDataType = { + "userId": self.user_id, + "active": self.active, + "userType": self.user_type, + } + if self.user_name is not None: + data["userName"] = self.user_name + if self.credential_rule is not None: + data["credentialRule"] = self.credential_rule + if self.expiring_timeout_minutes is not None: + data["expiringTimeoutMinutes"] = self.expiring_timeout_minutes + return data + + +class SetUserOptionsDataType(TypedDict, total=False): + """Represent serialized set-user options.""" + + active: bool + userType: UserCredentialUserType + userName: str + credentialRule: UserCredentialRule + expiringTimeoutMinutes: int + + +@dataclass +class SetUserOptions: + """Options for creating or updating a user.""" + + active: bool | None = None + user_type: UserCredentialUserType | None = None + user_name: str | None = None + credential_rule: UserCredentialRule | None = None + expiring_timeout_minutes: int | None = None + + def to_dict(self) -> SetUserOptionsDataType: + """Return serialized set-user options.""" + data: SetUserOptionsDataType = {} + if self.active is not None: + data["active"] = self.active + if self.user_type is not None: + data["userType"] = self.user_type + if self.user_name is not None: + data["userName"] = self.user_name + if self.credential_rule is not None: + data["credentialRule"] = self.credential_rule + if self.expiring_timeout_minutes is not None: + data["expiringTimeoutMinutes"] = self.expiring_timeout_minutes + return data + + +class CredentialDataDataType(TypedDict, total=False): + """Represent access-control credential payload.""" + + userId: int + type: UserCredentialType + slot: int + data: str | BufferObjectDataType + + +@dataclass +class CredentialData: + """Credential data returned by endpoint access-control commands.""" + + user_id: int + type: UserCredentialType + slot: int + data: str | bytes | None = None + + @classmethod + def from_dict(cls, data: CredentialDataDataType) -> Self: + """Return credential data from serialized payload.""" + return cls( + user_id=data["userId"], + type=UserCredentialType(data["type"]), + slot=data["slot"], + data=deserialize_credential_data(data.get("data")), + ) + + def to_dict(self) -> CredentialDataDataType: + """Return serialized credential payload.""" + data: CredentialDataDataType = { + "userId": self.user_id, + "type": self.type, + "slot": self.slot, + } + if self.data is not None: + data["data"] = serialize_credential_data(self.data) + return data + + +class UserDeletedArgsDataType(TypedDict): + """Represent access-control user deleted event args.""" + + userId: int + + +@dataclass +class UserDeletedArgs: + """Args for access-control user deleted events.""" + + user_id: int + + @classmethod + def from_dict(cls, data: UserDeletedArgsDataType) -> Self: + """Return user deleted args from serialized payload.""" + return cls(user_id=data["userId"]) + + def to_dict(self) -> UserDeletedArgsDataType: + """Return serialized user deleted args.""" + return {"userId": self.user_id} + + +class CredentialChangedArgsDataType(TypedDict, total=False): + """Represent access-control credential changed event args.""" + + userId: int + credentialType: UserCredentialType + credentialSlot: int + data: str | BufferObjectDataType + + +@dataclass +class CredentialChangedArgs: + """Args for access-control credential add/modify events.""" + + user_id: int + credential_type: UserCredentialType + credential_slot: int + data: str | bytes | None = None + + @classmethod + def from_dict(cls, data: CredentialChangedArgsDataType) -> Self: + """Return credential changed args from serialized payload.""" + return cls( + user_id=data["userId"], + credential_type=UserCredentialType(data["credentialType"]), + credential_slot=data["credentialSlot"], + data=deserialize_credential_data(data.get("data")), + ) + + def to_dict(self) -> CredentialChangedArgsDataType: + """Return serialized credential changed args.""" + data: CredentialChangedArgsDataType = { + "userId": self.user_id, + "credentialType": self.credential_type, + "credentialSlot": self.credential_slot, + } + if self.data is not None: + data["data"] = serialize_credential_data(self.data) + return data + + +class CredentialDeletedArgsDataType(TypedDict): + """Represent access-control credential deleted event args.""" + + userId: int + credentialType: UserCredentialType + credentialSlot: int + + +@dataclass +class CredentialDeletedArgs: + """Args for access-control credential deleted events.""" + + user_id: int + credential_type: UserCredentialType + credential_slot: int + + @classmethod + def from_dict(cls, data: CredentialDeletedArgsDataType) -> Self: + """Return credential deleted args from serialized payload.""" + return cls( + user_id=data["userId"], + credential_type=UserCredentialType(data["credentialType"]), + credential_slot=data["credentialSlot"], + ) + + def to_dict(self) -> CredentialDeletedArgsDataType: + """Return serialized credential deleted args.""" + return { + "userId": self.user_id, + "credentialType": self.credential_type, + "credentialSlot": self.credential_slot, + } + + +class CredentialLearnProgressArgsDataType(TypedDict): + """Represent access-control credential learn progress event args.""" + + userId: int + credentialType: UserCredentialType + credentialSlot: int + stepsRemaining: int + status: UserCredentialLearnStatus + + +@dataclass +class CredentialLearnProgressArgs: + """Args for access-control credential learn progress events.""" + + user_id: int + credential_type: UserCredentialType + credential_slot: int + steps_remaining: int + status: UserCredentialLearnStatus + + @classmethod + def from_dict(cls, data: CredentialLearnProgressArgsDataType) -> Self: + """Return credential learn progress args from serialized payload.""" + return cls( + user_id=data["userId"], + credential_type=UserCredentialType(data["credentialType"]), + credential_slot=data["credentialSlot"], + steps_remaining=data["stepsRemaining"], + status=UserCredentialLearnStatus(data["status"]), + ) + + def to_dict(self) -> CredentialLearnProgressArgsDataType: + """Return serialized credential learn progress args.""" + return { + "userId": self.user_id, + "credentialType": self.credential_type, + "credentialSlot": self.credential_slot, + "stepsRemaining": self.steps_remaining, + "status": self.status, + } + + +class CredentialLearnCompletedArgsDataType(TypedDict): + """Represent access-control credential learn completed event args.""" + + userId: int + credentialType: UserCredentialType + credentialSlot: int + status: UserCredentialLearnStatus + success: bool + + +@dataclass +class CredentialLearnCompletedArgs: + """Args for access-control credential learn completed events.""" + + user_id: int + credential_type: UserCredentialType + credential_slot: int + status: UserCredentialLearnStatus + success: bool + + @classmethod + def from_dict(cls, data: CredentialLearnCompletedArgsDataType) -> Self: + """Return credential learn completed args from serialized payload.""" + return cls( + user_id=data["userId"], + credential_type=UserCredentialType(data["credentialType"]), + credential_slot=data["credentialSlot"], + status=UserCredentialLearnStatus(data["status"]), + success=data["success"], + ) + + def to_dict(self) -> CredentialLearnCompletedArgsDataType: + """Return serialized credential learn completed args.""" + return { + "userId": self.user_id, + "credentialType": self.credential_type, + "credentialSlot": self.credential_slot, + "status": self.status, + "success": self.success, + } + + +class AccessControlAPI: + """Access-control command API wrapper for a single endpoint. + + Schema 48+. Obtain via :attr:`Endpoint.access_control` or + :attr:`Node.access_control` (root endpoint shortcut). + """ + + def __init__(self, endpoint: Endpoint) -> None: + """Initialize the API wrapper for the given endpoint.""" + self._endpoint = endpoint + + async def is_supported(self) -> bool: + """Return whether the endpoint supports access-control methods.""" + result = await self._endpoint.async_send_command( + "access_control.is_supported", + require_schema=48, + wait_for_result=True, + ) + assert result + return cast(bool, result["supported"]) + + async def get_user_capabilities_cached(self) -> UserCapabilities: + """Return cached user capabilities for access control.""" + result = await self._endpoint.async_send_command( + "access_control.get_user_capabilities_cached", + require_schema=48, + wait_for_result=True, + ) + assert result + capabilities = result["capabilities"] + assert capabilities is not None + return UserCapabilities.from_dict(cast(UserCapabilitiesDataType, capabilities)) + + async def get_credential_capabilities_cached( + self, + ) -> CredentialCapabilities: + """Return cached credential capabilities for access control.""" + result = await self._endpoint.async_send_command( + "access_control.get_credential_capabilities_cached", + require_schema=48, + wait_for_result=True, + ) + assert result + capabilities = result["capabilities"] + assert capabilities is not None + return CredentialCapabilities.from_dict( + cast(CredentialCapabilitiesDataType, capabilities) + ) + + async def get_user(self, user_id: int) -> UserData | None: + """Return fresh data for a single access-control user.""" + result = await self._endpoint.async_send_command( + "access_control.get_user", + userId=user_id, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + if (user := result.get("user")) is None: + return None + return UserData.from_dict(cast(UserDataDataType, user)) + + async def get_user_cached(self, user_id: int) -> UserData | None: + """Return cached data for a single access-control user.""" + result = await self._endpoint.async_send_command( + "access_control.get_user_cached", + userId=user_id, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + if (user := result.get("user")) is None: + return None + return UserData.from_dict(cast(UserDataDataType, user)) + + async def get_users(self) -> list[UserData]: + """Return fresh data for all configured access-control users.""" + result = await self._endpoint.async_send_command( + "access_control.get_users", + require_schema=48, + wait_for_result=True, + ) + assert result + users = result["users"] + assert users is not None + return [UserData.from_dict(cast(UserDataDataType, user)) for user in users] + + async def get_users_cached(self) -> list[UserData]: + """Return cached data for all configured access-control users.""" + result = await self._endpoint.async_send_command( + "access_control.get_users_cached", + require_schema=48, + wait_for_result=True, + ) + assert result + users = result["users"] + assert users is not None + return [UserData.from_dict(cast(UserDataDataType, user)) for user in users] + + async def set_user(self, user_id: int, options: SetUserOptions) -> SetUserResult: + """Create or update an access-control user.""" + result = await self._endpoint.async_send_command( + "access_control.set_user", + userId=user_id, + options=options.to_dict(), + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return SetUserResult(result["result"]) + + async def delete_user(self, user_id: int) -> SetUserResult: + """Delete an access-control user.""" + result = await self._endpoint.async_send_command( + "access_control.delete_user", + userId=user_id, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return SetUserResult(result["result"]) + + async def delete_all_users(self) -> SetUserResult: + """Delete all configured access-control users.""" + result = await self._endpoint.async_send_command( + "access_control.delete_all_users", + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return SetUserResult(result["result"]) + + async def get_credential( + self, + credential_type: UserCredentialType, + credential_slot: int, + ) -> CredentialData | None: + """Return fresh data for a single access-control credential.""" + result = await self._endpoint.async_send_command( + "access_control.get_credential", + credentialType=credential_type, + credentialSlot=credential_slot, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + if (credential := result.get("credential")) is None: + return None + return CredentialData.from_dict(cast(CredentialDataDataType, credential)) + + async def get_credential_cached( + self, + credential_type: UserCredentialType, + credential_slot: int, + ) -> CredentialData | None: + """Return cached data for a single access-control credential.""" + result = await self._endpoint.async_send_command( + "access_control.get_credential_cached", + credentialType=credential_type, + credentialSlot=credential_slot, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + if (credential := result.get("credential")) is None: + return None + return CredentialData.from_dict(cast(CredentialDataDataType, credential)) + + async def get_credentials(self, user_id: int) -> list[CredentialData]: + """Return fresh data for all credentials assigned to a user.""" + result = await self._endpoint.async_send_command( + "access_control.get_credentials", + userId=user_id, + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def get_credentials_cached(self, user_id: int) -> list[CredentialData]: + """Return cached data for all credentials assigned to a user.""" + result = await self._endpoint.async_send_command( + "access_control.get_credentials_cached", + userId=user_id, + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def get_credentials_by_type( + self, credential_type: UserCredentialType + ) -> list[CredentialData]: + """Return fresh data for all credentials of the given type.""" + result = await self._endpoint.async_send_command( + "access_control.get_credentials_by_type", + credentialType=credential_type, + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def get_credentials_by_type_cached( + self, credential_type: UserCredentialType + ) -> list[CredentialData]: + """Return cached data for all credentials of the given type.""" + result = await self._endpoint.async_send_command( + "access_control.get_credentials_by_type_cached", + credentialType=credential_type, + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def get_all_credentials(self) -> list[CredentialData]: + """Return fresh data for all credentials regardless of type or user.""" + result = await self._endpoint.async_send_command( + "access_control.get_all_credentials", + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def get_all_credentials_cached(self) -> list[CredentialData]: + """Return cached data for all credentials regardless of type or user.""" + result = await self._endpoint.async_send_command( + "access_control.get_all_credentials_cached", + require_schema=48, + wait_for_result=True, + ) + assert result + credentials = result["credentials"] + assert credentials is not None + return [ + CredentialData.from_dict(cast(CredentialDataDataType, credential)) + for credential in credentials + ] + + async def assign_credential( + self, + credential_type: UserCredentialType, + credential_slot: int, + destination_user_id: int, + ) -> AssignCredentialResult: + """Reassign an existing credential to a different user.""" + result = await self._endpoint.async_send_command( + "access_control.assign_credential", + credentialType=credential_type, + credentialSlot=credential_slot, + destinationUserId=destination_user_id, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return AssignCredentialResult(result["result"]) + + async def set_credential( + self, + user_id: int, + credential_type: UserCredentialType, + credential_slot: int, + data: str | bytes, + ) -> SetCredentialResult: + """Create or update an access-control credential.""" + result = await self._endpoint.async_send_command( + "access_control.set_credential", + userId=user_id, + credentialType=credential_type, + credentialSlot=credential_slot, + data=serialize_credential_data(data), + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return SetCredentialResult(result["result"]) + + async def delete_credential( + self, + user_id: int, + credential_type: UserCredentialType, + credential_slot: int, + ) -> SetCredentialResult: + """Delete an access-control credential.""" + result = await self._endpoint.async_send_command( + "access_control.delete_credential", + userId=user_id, + credentialType=credential_type, + credentialSlot=credential_slot, + require_schema=48, + wait_for_result=True, + ) + assert result is not None + return SetCredentialResult(result["result"]) + + async def start_credential_learn( + self, + user_id: int, + credential_type: UserCredentialType, + credential_slot: int, + timeout: int | None = None, + ) -> SupervisionResult | None: + """Start learning mode for an access-control credential.""" + cmd_kwargs: dict[str, Any] = {} + if timeout is not None: + cmd_kwargs["timeout"] = timeout + result = await self._endpoint.async_send_command( + "access_control.start_credential_learn", + userId=user_id, + credentialType=credential_type, + credentialSlot=credential_slot, + require_schema=48, + wait_for_result=None, + **cmd_kwargs, + ) + return parse_supervision_result(result) + + async def cancel_credential_learn(self) -> SupervisionResult | None: + """Cancel any active credential learn operation.""" + result = await self._endpoint.async_send_command( + "access_control.cancel_credential_learn", + require_schema=48, + wait_for_result=None, + ) + return parse_supervision_result(result) + + async def get_admin_code(self) -> str | None: + """Return the configured admin code for access control.""" + result = await self._endpoint.async_send_command( + "access_control.get_admin_code", + require_schema=48, + wait_for_result=True, + ) + assert result + return cast(str | None, result.get("code")) + + async def set_admin_code(self, code: str) -> SupervisionResult | None: + """Set the admin code for access control.""" + result = await self._endpoint.async_send_command( + "access_control.set_admin_code", + code=code, + require_schema=48, + wait_for_result=None, + ) + return parse_supervision_result(result) diff --git a/zwave_js_server/model/endpoint.py b/zwave_js_server/model/endpoint.py index 14dc0ba7a..650109432 100644 --- a/zwave_js_server/model/endpoint.py +++ b/zwave_js_server/model/endpoint.py @@ -13,6 +13,7 @@ from ..const import NodeStatus from ..event import EventBase from ..exceptions import FailedCommand, NotFoundError +from .access_control import AccessControlAPI from .command_class import CommandClass, CommandClassInfo, CommandClassInfoDataType from .device_class import DeviceClass, DeviceClassDataType from .value import ( @@ -293,6 +294,11 @@ async def async_get_node_unsafe(self) -> NodeDataType: assert result return cast("NodeDataType", result["node"]) + @cached_property + def access_control(self) -> AccessControlAPI: + """Return the access-control API wrapper for this endpoint.""" + return AccessControlAPI(self) + async def async_set_raw_config_parameter_value( self, new_value: int, diff --git a/zwave_js_server/model/node/__init__.py b/zwave_js_server/model/node/__init__.py index adcd02f12..356d0d2c9 100644 --- a/zwave_js_server/model/node/__init__.py +++ b/zwave_js_server/model/node/__init__.py @@ -20,6 +20,15 @@ ) from ...event import Event, EventBase from ...exceptions import NotFoundError, UnparseableValue, UnwriteableValue +from ..access_control import ( + AccessControlAPI, + CredentialChangedArgs, + CredentialDeletedArgs, + CredentialLearnCompletedArgs, + CredentialLearnProgressArgs, + UserData, + UserDeletedArgs, +) from ..command_class import CommandClassInfo from ..device_class import DeviceClass from ..device_config import DeviceConfig @@ -738,6 +747,11 @@ async def async_get_node_unsafe(self) -> NodeDataType: """Call endpoint.get_node_unsafe command.""" return await self.endpoints[0].async_get_node_unsafe() + @property + def access_control(self) -> AccessControlAPI: + """Return the access-control API wrapper for the root endpoint.""" + return self.endpoints[0].access_control + async def async_has_security_class( self, security_class: SecurityClass ) -> bool | None: @@ -1023,6 +1037,25 @@ async def async_get_raw_config_parameter_value( property_, property_key, allow_unexpected_response ) + def _handle_access_control_event( + self, + event: Event, + args_factory: type[ + UserData + | UserDeletedArgs + | CredentialChangedArgs + | CredentialDeletedArgs + | CredentialLearnProgressArgs + | CredentialLearnCompletedArgs + ], + ) -> None: + """Resolve endpoint and normalize access-control event args.""" + if (endpoint_index := event.data.get("endpointIndex")) is not None and ( + endpoint := self.endpoints.get(endpoint_index) + ): + event.data["endpoint"] = endpoint + event.data["args"] = args_factory.from_dict(event.data["args"]) + def handle_test_powerlevel_progress(self, event: Event) -> None: """Process a test power level progress event.""" event.data["test_power_level_progress"] = TestPowerLevelProgress( @@ -1174,6 +1207,38 @@ def handle_node_info_received(self, event: Event) -> None: """Process a node info received event.""" # Nothing to do for now + def handle_user_added(self, event: Event) -> None: + """Process a node user added event.""" + self._handle_access_control_event(event, UserData) + + def handle_user_modified(self, event: Event) -> None: + """Process a node user modified event.""" + self._handle_access_control_event(event, UserData) + + def handle_user_deleted(self, event: Event) -> None: + """Process a node user deleted event.""" + self._handle_access_control_event(event, UserDeletedArgs) + + def handle_credential_added(self, event: Event) -> None: + """Process a node credential added event.""" + self._handle_access_control_event(event, CredentialChangedArgs) + + def handle_credential_modified(self, event: Event) -> None: + """Process a node credential modified event.""" + self._handle_access_control_event(event, CredentialChangedArgs) + + def handle_credential_deleted(self, event: Event) -> None: + """Process a node credential deleted event.""" + self._handle_access_control_event(event, CredentialDeletedArgs) + + def handle_credential_learn_progress(self, event: Event) -> None: + """Process a node credential learn progress event.""" + self._handle_access_control_event(event, CredentialLearnProgressArgs) + + def handle_credential_learn_completed(self, event: Event) -> None: + """Process a node credential learn completed event.""" + self._handle_access_control_event(event, CredentialLearnCompletedArgs) + def handle_firmware_update_progress(self, event: Event) -> None: """Process a node firmware update progress event.""" self._firmware_update_progress = event.data["firmware_update_progress"] = ( diff --git a/zwave_js_server/model/node/event_model.py b/zwave_js_server/model/node/event_model.py index b0e6e6a43..40fafe69b 100644 --- a/zwave_js_server/model/node/event_model.py +++ b/zwave_js_server/model/node/event_model.py @@ -2,12 +2,20 @@ from __future__ import annotations -from typing import Literal +from typing import Any, Literal, Self from pydantic import BaseModel from ...const import CommandClass from ...event import BaseEventModel +from ..access_control import ( + CredentialChangedArgsDataType, + CredentialDeletedArgsDataType, + CredentialLearnCompletedArgsDataType, + CredentialLearnProgressArgsDataType, + UserDataDataType, + UserDeletedArgsDataType, +) from ..notification import ( EntryControlNotificationArgsDataType, MultilevelSwitchNotificationArgsDataType, @@ -205,6 +213,80 @@ def from_dict(cls, data: dict) -> NotificationEventModel: ) +class BaseNodeEndpointArgsEventModel(BaseNodeEventModel): + """Base model for node endpoint events with args payload.""" + + endpointIndex: int + args: Any + + @classmethod + def from_dict(cls, data: dict) -> Self: + """Initialize from dict.""" + return cls( + source=data["source"], + event=data["event"], + nodeId=data["nodeId"], + endpointIndex=data["endpointIndex"], + args=data["args"], + ) + + +class UserAddedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `user added` event data.""" + + event: Literal["user added"] + args: UserDataDataType + + +class UserModifiedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `user modified` event data.""" + + event: Literal["user modified"] + args: UserDataDataType + + +class UserDeletedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `user deleted` event data.""" + + event: Literal["user deleted"] + args: UserDeletedArgsDataType + + +class CredentialAddedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `credential added` event data.""" + + event: Literal["credential added"] + args: CredentialChangedArgsDataType + + +class CredentialModifiedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `credential modified` event data.""" + + event: Literal["credential modified"] + args: CredentialChangedArgsDataType + + +class CredentialDeletedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `credential deleted` event data.""" + + event: Literal["credential deleted"] + args: CredentialDeletedArgsDataType + + +class CredentialLearnProgressEventModel(BaseNodeEndpointArgsEventModel): + """Model for `credential learn progress` event data.""" + + event: Literal["credential learn progress"] + args: CredentialLearnProgressArgsDataType + + +class CredentialLearnCompletedEventModel(BaseNodeEndpointArgsEventModel): + """Model for `credential learn completed` event data.""" + + event: Literal["credential learn completed"] + args: CredentialLearnCompletedArgsDataType + + class ReadyEventModel(BaseNodeEventModel): """Model for `ready` event data.""" @@ -360,6 +442,11 @@ def from_dict(cls, data: dict) -> FirmwareUpdateProgressEventModel: "check lifeline health progress": CheckLifelineHealthProgressEventModel, "check link reliability progress": CheckLinkReliabilityProgressEventModel, "check route health progress": CheckRouteHealthProgressEventModel, + "credential added": CredentialAddedEventModel, + "credential deleted": CredentialDeletedEventModel, + "credential learn completed": CredentialLearnCompletedEventModel, + "credential learn progress": CredentialLearnProgressEventModel, + "credential modified": CredentialModifiedEventModel, "dead": DeadEventModel, "firmware update finished": FirmwareUpdateFinishedEventModel, "firmware update progress": FirmwareUpdateProgressEventModel, @@ -374,6 +461,9 @@ def from_dict(cls, data: dict) -> FirmwareUpdateProgressEventModel: "sleep": SleepEventModel, "statistics updated": StatisticsUpdatedEventModel, "test powerlevel progress": TestPowerLevelProgressEventModel, + "user added": UserAddedEventModel, + "user deleted": UserDeletedEventModel, + "user modified": UserModifiedEventModel, "value added": ValueAddedEventModel, "value notification": ValueNotificationEventModel, "value removed": ValueRemovedEventModel, diff --git a/zwave_js_server/model/value.py b/zwave_js_server/model/value.py index 73b3607c8..560ae2d79 100644 --- a/zwave_js_server/model/value.py +++ b/zwave_js_server/model/value.py @@ -460,6 +460,15 @@ def __post_init__(self) -> None: ) +def parse_supervision_result( + data: dict[str, Any] | None, +) -> SupervisionResult | None: + """Return a parsed supervision result from a command response.""" + if not data or (result := data.get("result")) is None: + return None + return SupervisionResult(result) + + class ConfigurationValue(Value): """Model for a Configuration Value.""" diff --git a/zwave_js_server/util/helpers.py b/zwave_js_server/util/helpers.py index d2bff41b5..92ea42146 100644 --- a/zwave_js_server/util/helpers.py +++ b/zwave_js_server/util/helpers.py @@ -4,11 +4,38 @@ import base64 import json -from typing import Any +from typing import Any, Literal, TypeGuard, TypedDict from ..exceptions import UnparseableValue +class BufferObjectDataType(TypedDict): + """Buffer object representation used by zwave-js-server JSON transport.""" + + type: Literal["Buffer"] + data: list[int] + + +def is_buffer_object(value: Any) -> TypeGuard[BufferObjectDataType]: + """Return whether value matches zwave-js-server buffer transport shape.""" + return ( + isinstance(value, dict) + and value.get("type") == "Buffer" + and isinstance(value.get("data"), list) + and all(isinstance(item, int) for item in value["data"]) + ) + + +def bytes_to_buffer_object(data: bytes) -> BufferObjectDataType: + """Wrap bytes in the websocket Buffer transport shape.""" + return {"type": "Buffer", "data": list(data)} + + +def buffer_object_to_bytes(value: BufferObjectDataType) -> bytes: + """Unwrap a websocket Buffer transport shape into bytes.""" + return bytes(value["data"]) + + def is_json_string(value: Any) -> bool: """Check if the provided string looks like json.""" # NOTE: we do not use json.loads here as it is not strict enough @@ -38,7 +65,7 @@ def parse_buffer(value: dict[str, Any] | str) -> str: def parse_buffer_from_dict(value: dict[str, Any]) -> str: """Parse value dictionary from a buffer data type.""" - if value.get("type") != "Buffer" or "data" not in value: + if not is_buffer_object(value): raise UnparseableValue(f"Unparseable value: {value}") from ValueError( "JSON does not match expected schema" )