From ee05f14530786b76b7d4203d1f287839ef6d9336 Mon Sep 17 00:00:00 2001 From: Alex Brown Date: Sat, 28 Feb 2026 04:43:09 -0500 Subject: [PATCH] Add Matter lock user and credential management services (#161936) Co-authored-by: Claude Opus 4.5 --- homeassistant/components/matter/const.py | 99 + homeassistant/components/matter/icons.json | 21 + homeassistant/components/matter/lock.py | 180 +- .../components/matter/lock_helpers.py | 881 +++++++ homeassistant/components/matter/services.py | 125 +- homeassistant/components/matter/services.yaml | 174 ++ homeassistant/components/matter/strings.json | 109 + tests/components/matter/test_lock.py | 2151 ++++++++++++++++- 8 files changed, 3702 insertions(+), 38 deletions(-) create mode 100644 homeassistant/components/matter/lock_helpers.py diff --git a/homeassistant/components/matter/const.py b/homeassistant/components/matter/const.py index 8018d5e09edf..cb42401725a5 100644 --- a/homeassistant/components/matter/const.py +++ b/homeassistant/components/matter/const.py @@ -2,6 +2,8 @@ import logging +from chip.clusters import Objects as clusters + ADDON_SLUG = "core_matter_server" CONF_INTEGRATION_CREATED_ADDON = "integration_created_addon" @@ -15,3 +17,100 @@ ID_TYPE_DEVICE_ID = "deviceid" ID_TYPE_SERIAL = "serial" FEATUREMAP_ATTRIBUTE_ID = 65532 + +# --- Lock domain constants --- + +# Shared field keys +ATTR_CREDENTIAL_RULE = "credential_rule" +ATTR_MAX_CREDENTIALS_PER_USER = "max_credentials_per_user" +ATTR_MAX_PIN_USERS = "max_pin_users" +ATTR_MAX_RFID_USERS = "max_rfid_users" +ATTR_MAX_USERS = "max_users" +ATTR_SUPPORTS_USER_MGMT = "supports_user_management" +ATTR_USER_INDEX = "user_index" +ATTR_USER_NAME = "user_name" +ATTR_USER_STATUS = "user_status" +ATTR_USER_TYPE = "user_type" + +# Magic values +CLEAR_ALL_INDEX = 0xFFFE # Matter spec: pass to ClearUser/ClearCredential to clear all + +# Timed request timeout for lock commands that modify state. +# 10 seconds accounts for Thread network latency and retransmissions. +LOCK_TIMED_REQUEST_TIMEOUT_MS = 10000 + +# Credential field keys +ATTR_CREDENTIAL_DATA = "credential_data" +ATTR_CREDENTIAL_INDEX = "credential_index" +ATTR_CREDENTIAL_TYPE = "credential_type" + +# Credential type strings +CRED_TYPE_FACE = "face" +CRED_TYPE_FINGERPRINT = "fingerprint" +CRED_TYPE_FINGER_VEIN = "finger_vein" +CRED_TYPE_PIN = "pin" +CRED_TYPE_RFID = "rfid" + +# User status mapping (Matter DoorLock UserStatusEnum) +_UserStatus = clusters.DoorLock.Enums.UserStatusEnum +USER_STATUS_MAP: dict[int, str] = { + _UserStatus.kAvailable: "available", + _UserStatus.kOccupiedEnabled: "occupied_enabled", + _UserStatus.kOccupiedDisabled: "occupied_disabled", +} +USER_STATUS_REVERSE_MAP: dict[str, int] = {v: k for k, v in USER_STATUS_MAP.items()} + +# User type mapping (Matter DoorLock UserTypeEnum) +_UserType = clusters.DoorLock.Enums.UserTypeEnum +USER_TYPE_MAP: dict[int, str] = { + _UserType.kUnrestrictedUser: "unrestricted_user", + _UserType.kYearDayScheduleUser: "year_day_schedule_user", + _UserType.kWeekDayScheduleUser: "week_day_schedule_user", + _UserType.kProgrammingUser: "programming_user", + _UserType.kNonAccessUser: "non_access_user", + _UserType.kForcedUser: "forced_user", + _UserType.kDisposableUser: "disposable_user", + _UserType.kExpiringUser: "expiring_user", + _UserType.kScheduleRestrictedUser: "schedule_restricted_user", + _UserType.kRemoteOnlyUser: "remote_only_user", +} +USER_TYPE_REVERSE_MAP: dict[str, int] = {v: k for k, v in USER_TYPE_MAP.items()} + +# Credential type mapping (Matter DoorLock CredentialTypeEnum) +_CredentialType = clusters.DoorLock.Enums.CredentialTypeEnum +CREDENTIAL_TYPE_MAP: dict[int, str] = { + _CredentialType.kProgrammingPIN: "programming_pin", + _CredentialType.kPin: CRED_TYPE_PIN, + _CredentialType.kRfid: CRED_TYPE_RFID, + _CredentialType.kFingerprint: CRED_TYPE_FINGERPRINT, + _CredentialType.kFingerVein: CRED_TYPE_FINGER_VEIN, + _CredentialType.kFace: CRED_TYPE_FACE, + _CredentialType.kAliroCredentialIssuerKey: "aliro_credential_issuer_key", + _CredentialType.kAliroEvictableEndpointKey: "aliro_evictable_endpoint_key", + _CredentialType.kAliroNonEvictableEndpointKey: "aliro_non_evictable_endpoint_key", +} + +# Credential rule mapping (Matter DoorLock CredentialRuleEnum) +_CredentialRule = clusters.DoorLock.Enums.CredentialRuleEnum +CREDENTIAL_RULE_MAP: dict[int, str] = { + _CredentialRule.kSingle: "single", + _CredentialRule.kDual: "dual", + _CredentialRule.kTri: "tri", +} +CREDENTIAL_RULE_REVERSE_MAP: dict[str, int] = { + v: k for k, v in CREDENTIAL_RULE_MAP.items() +} + +# Reverse mapping for credential types (str -> int) +CREDENTIAL_TYPE_REVERSE_MAP: dict[str, int] = { + v: k for k, v in CREDENTIAL_TYPE_MAP.items() +} + +# Credential types allowed in set/clear services (excludes programming_pin, aliro_*) +SERVICE_CREDENTIAL_TYPES = [ + CRED_TYPE_PIN, + CRED_TYPE_RFID, + CRED_TYPE_FINGERPRINT, + CRED_TYPE_FINGER_VEIN, + CRED_TYPE_FACE, +] diff --git a/homeassistant/components/matter/icons.json b/homeassistant/components/matter/icons.json index ec96875c06b4..be65b4621080 100644 --- a/homeassistant/components/matter/icons.json +++ b/homeassistant/components/matter/icons.json @@ -174,6 +174,27 @@ } }, "services": { + "clear_lock_credential": { + "service": "mdi:key-remove" + }, + "clear_lock_user": { + "service": "mdi:account-remove" + }, + "get_lock_credential_status": { + "service": "mdi:key-chain" + }, + "get_lock_info": { + "service": "mdi:lock-question" + }, + "get_lock_users": { + "service": "mdi:account-multiple" + }, + "set_lock_credential": { + "service": "mdi:key-plus" + }, + "set_lock_user": { + "service": "mdi:account-lock" + }, "water_heater_boost": { "service": "mdi:water-boiler" } diff --git a/homeassistant/components/matter/lock.py b/homeassistant/components/matter/lock.py index 330735f338b0..80316ea80148 100644 --- a/homeassistant/components/matter/lock.py +++ b/homeassistant/components/matter/lock.py @@ -7,6 +7,7 @@ from dataclasses import dataclass from typing import Any from chip.clusters import Objects as clusters +from matter_server.common.errors import MatterError from matter_server.common.models import EventType, MatterNodeEvent from homeassistant.components.lock import ( @@ -17,32 +18,56 @@ from homeassistant.components.lock import ( from homeassistant.config_entries import ConfigEntry from homeassistant.const import ATTR_CODE, Platform from homeassistant.core import HomeAssistant, callback +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback -from .const import LOGGER +from .const import ( + ATTR_CREDENTIAL_DATA, + ATTR_CREDENTIAL_INDEX, + ATTR_CREDENTIAL_RULE, + ATTR_CREDENTIAL_TYPE, + ATTR_USER_INDEX, + ATTR_USER_NAME, + ATTR_USER_STATUS, + ATTR_USER_TYPE, + LOCK_TIMED_REQUEST_TIMEOUT_MS, + LOGGER, +) from .entity import MatterEntity, MatterEntityDescription from .helpers import get_matter +from .lock_helpers import ( + DoorLockFeature, + GetLockCredentialStatusResult, + GetLockInfoResult, + GetLockUsersResult, + SetLockCredentialResult, + clear_lock_credential, + clear_lock_user, + get_lock_credential_status, + get_lock_info, + get_lock_users, + set_lock_credential, + set_lock_user, +) from .models import MatterDiscoverySchema -DOOR_LOCK_OPERATION_SOURCE = { - # mapping from operation source id's to textual representation - 0: "Unspecified", - 1: "Manual", # [Optional] - 2: "Proprietary Remote", # [Optional] - 3: "Keypad", # [Optional] - 4: "Auto", # [Optional] - 5: "Button", # [Optional] - 6: "Schedule", # [HDSCH] - 7: "Remote", # [M] - 8: "RFID", # [RID] - 9: "Biometric", # [USR] - 10: "Aliro", # [Aliro] +# Door lock operation source mapping (Matter DoorLock OperationSourceEnum) +_OperationSource = clusters.DoorLock.Enums.OperationSourceEnum +DOOR_LOCK_OPERATION_SOURCE: dict[int, str] = { + _OperationSource.kUnspecified: "Unspecified", + _OperationSource.kManual: "Manual", + _OperationSource.kProprietaryRemote: "Proprietary Remote", + _OperationSource.kKeypad: "Keypad", + _OperationSource.kAuto: "Auto", + _OperationSource.kButton: "Button", + _OperationSource.kSchedule: "Schedule", + _OperationSource.kRemote: "Remote", + _OperationSource.kRfid: "RFID", + _OperationSource.kBiometric: "Biometric", + _OperationSource.kAliro: "Aliro", } -DoorLockFeature = clusters.DoorLock.Bitmaps.Feature - - async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, @@ -98,17 +123,15 @@ class MatterLock(MatterEntity, LockEntity): node_event.data, ) - # handle the DoorLock events + # Handle the DoorLock events node_event_data: dict[str, int] = node_event.data or {} match node_event.event_id: - case ( - clusters.DoorLock.Events.LockOperation.event_id - ): # Lock cluster event 2 - # update the changed_by attribute to indicate lock operation source + case clusters.DoorLock.Events.LockOperation.event_id: operation_source: int = node_event_data.get("operationSource", -1) - self._attr_changed_by = DOOR_LOCK_OPERATION_SOURCE.get( + source_name = DOOR_LOCK_OPERATION_SOURCE.get( operation_source, "Unknown" ) + self._attr_changed_by = source_name self.async_write_ha_state() @property @@ -146,7 +169,7 @@ class MatterLock(MatterEntity, LockEntity): code_bytes = code.encode() if code else None await self.send_device_command( command=clusters.DoorLock.Commands.LockDoor(code_bytes), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, ) async def async_unlock(self, **kwargs: Any) -> None: @@ -168,12 +191,12 @@ class MatterLock(MatterEntity, LockEntity): # and unlatch on the HA 'open' command. await self.send_device_command( command=clusters.DoorLock.Commands.UnboltDoor(code_bytes), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, ) else: await self.send_device_command( command=clusters.DoorLock.Commands.UnlockDoor(code_bytes), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, ) async def async_open(self, **kwargs: Any) -> None: @@ -190,7 +213,7 @@ class MatterLock(MatterEntity, LockEntity): code_bytes = code.encode() if code else None await self.send_device_command( command=clusters.DoorLock.Commands.UnlockDoor(code_bytes), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, ) @callback @@ -256,6 +279,109 @@ class MatterLock(MatterEntity, LockEntity): supported_features |= LockEntityFeature.OPEN self._attr_supported_features = supported_features + # --- Entity service methods --- + + async def async_set_lock_user(self, **kwargs: Any) -> None: + """Set a lock user (full CRUD).""" + try: + await set_lock_user( + self.matter_client, + self._endpoint.node, + user_index=kwargs.get(ATTR_USER_INDEX), + user_name=kwargs.get(ATTR_USER_NAME), + user_type=kwargs.get(ATTR_USER_TYPE), + credential_rule=kwargs.get(ATTR_CREDENTIAL_RULE), + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to set lock user on {self.entity_id}: {err}" + ) from err + + async def async_clear_lock_user(self, **kwargs: Any) -> None: + """Clear a lock user.""" + try: + await clear_lock_user( + self.matter_client, + self._endpoint.node, + kwargs[ATTR_USER_INDEX], + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to clear lock user on {self.entity_id}: {err}" + ) from err + + async def async_get_lock_info(self) -> GetLockInfoResult: + """Get lock capabilities and configuration info.""" + try: + return await get_lock_info( + self.matter_client, + self._endpoint.node, + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to get lock info for {self.entity_id}: {err}" + ) from err + + async def async_get_lock_users(self) -> GetLockUsersResult: + """Get all users from the lock.""" + try: + return await get_lock_users( + self.matter_client, + self._endpoint.node, + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to get lock users for {self.entity_id}: {err}" + ) from err + + async def async_set_lock_credential(self, **kwargs: Any) -> SetLockCredentialResult: + """Set a credential on the lock.""" + try: + return await set_lock_credential( + self.matter_client, + self._endpoint.node, + credential_type=kwargs[ATTR_CREDENTIAL_TYPE], + credential_data=kwargs[ATTR_CREDENTIAL_DATA], + credential_index=kwargs.get(ATTR_CREDENTIAL_INDEX), + user_index=kwargs.get(ATTR_USER_INDEX), + user_status=kwargs.get(ATTR_USER_STATUS), + user_type=kwargs.get(ATTR_USER_TYPE), + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to set lock credential on {self.entity_id}: {err}" + ) from err + + async def async_clear_lock_credential(self, **kwargs: Any) -> None: + """Clear a credential from the lock.""" + try: + await clear_lock_credential( + self.matter_client, + self._endpoint.node, + credential_type=kwargs[ATTR_CREDENTIAL_TYPE], + credential_index=kwargs[ATTR_CREDENTIAL_INDEX], + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to clear lock credential on {self.entity_id}: {err}" + ) from err + + async def async_get_lock_credential_status( + self, **kwargs: Any + ) -> GetLockCredentialStatusResult: + """Get the status of a credential slot on the lock.""" + try: + return await get_lock_credential_status( + self.matter_client, + self._endpoint.node, + credential_type=kwargs[ATTR_CREDENTIAL_TYPE], + credential_index=kwargs[ATTR_CREDENTIAL_INDEX], + ) + except MatterError as err: + raise HomeAssistantError( + f"Failed to get credential status for {self.entity_id}: {err}" + ) from err + DISCOVERY_SCHEMAS = [ MatterDiscoverySchema( diff --git a/homeassistant/components/matter/lock_helpers.py b/homeassistant/components/matter/lock_helpers.py new file mode 100644 index 000000000000..45cb014dffe5 --- /dev/null +++ b/homeassistant/components/matter/lock_helpers.py @@ -0,0 +1,881 @@ +"""Lock-specific helpers for the Matter integration. + +Provides DoorLock cluster endpoint resolution, feature detection, and +business logic for lock user/credential management. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, TypedDict + +from chip.clusters import Objects as clusters + +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError + +from .const import ( + CLEAR_ALL_INDEX, + CRED_TYPE_FACE, + CRED_TYPE_FINGER_VEIN, + CRED_TYPE_FINGERPRINT, + CRED_TYPE_PIN, + CRED_TYPE_RFID, + CREDENTIAL_RULE_MAP, + CREDENTIAL_RULE_REVERSE_MAP, + CREDENTIAL_TYPE_MAP, + CREDENTIAL_TYPE_REVERSE_MAP, + LOCK_TIMED_REQUEST_TIMEOUT_MS, + USER_STATUS_MAP, + USER_STATUS_REVERSE_MAP, + USER_TYPE_MAP, + USER_TYPE_REVERSE_MAP, +) + +# Error translation keys (used in ServiceValidationError/HomeAssistantError) +ERR_CREDENTIAL_TYPE_NOT_SUPPORTED = "credential_type_not_supported" +ERR_INVALID_CREDENTIAL_DATA = "invalid_credential_data" + +# SetCredential response status mapping (Matter DlStatus) +_DlStatus = clusters.DoorLock.Enums.DlStatus +SET_CREDENTIAL_STATUS_MAP: dict[int, str] = { + _DlStatus.kSuccess: "success", + _DlStatus.kFailure: "failure", + _DlStatus.kDuplicate: "duplicate", + _DlStatus.kOccupied: "occupied", +} + +if TYPE_CHECKING: + from matter_server.client import MatterClient + from matter_server.client.models.node import MatterEndpoint, MatterNode + +# DoorLock Feature bitmap from Matter SDK +DoorLockFeature = clusters.DoorLock.Bitmaps.Feature + + +# --- TypedDicts for service action responses --- + + +class LockUserCredentialData(TypedDict): + """Credential data within a user response.""" + + type: str + index: int | None + + +class LockUserData(TypedDict): + """User data returned from lock queries.""" + + user_index: int | None + user_name: str | None + user_unique_id: int | None + user_status: str + user_type: str + credential_rule: str + credentials: list[LockUserCredentialData] + next_user_index: int | None + + +class SetLockUserResult(TypedDict): + """Result of set_lock_user service action.""" + + user_index: int + + +class GetLockUsersResult(TypedDict): + """Result of get_lock_users service action.""" + + max_users: int + users: list[LockUserData] + + +class GetLockInfoResult(TypedDict): + """Result of get_lock_info service action.""" + + supports_user_management: bool + supported_credential_types: list[str] + max_users: int | None + max_pin_users: int | None + max_rfid_users: int | None + max_credentials_per_user: int | None + min_pin_length: int | None + max_pin_length: int | None + min_rfid_length: int | None + max_rfid_length: int | None + + +class SetLockCredentialResult(TypedDict): + """Result of set_lock_credential service action.""" + + credential_index: int + user_index: int | None + next_credential_index: int | None + + +class GetLockCredentialStatusResult(TypedDict): + """Result of get_lock_credential_status service action.""" + + credential_exists: bool + user_index: int | None + next_credential_index: int | None + + +def _get_lock_endpoint_from_node(node: MatterNode) -> MatterEndpoint | None: + """Get the DoorLock endpoint from a node. + + Returns the first endpoint that has the DoorLock cluster, or None if not found. + """ + for endpoint in node.endpoints.values(): + if endpoint.has_cluster(clusters.DoorLock): + return endpoint + return None + + +def _get_feature_map(endpoint: MatterEndpoint) -> int | None: + """Read the DoorLock FeatureMap attribute from an endpoint.""" + value: int | None = endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.FeatureMap + ) + return value + + +def _lock_supports_usr_feature(endpoint: MatterEndpoint) -> bool: + """Check if lock endpoint supports USR (User) feature. + + The USR feature indicates the lock supports user and credential management + commands like SetUser, GetUser, SetCredential, etc. + """ + feature_map = _get_feature_map(endpoint) + if feature_map is None: + return False + return bool(feature_map & DoorLockFeature.kUser) + + +# --- Pure utility functions --- + + +def _get_attr(obj: Any, attr: str) -> Any: + """Get attribute from object or dict. + + Matter SDK responses can be either dataclass objects or dicts depending on + the SDK version and serialization context. + """ + if isinstance(obj, dict): + return obj.get(attr) + return getattr(obj, attr, None) + + +def _get_supported_credential_types(feature_map: int) -> list[str]: + """Get list of supported credential types from feature map.""" + types = [] + if feature_map & DoorLockFeature.kPinCredential: + types.append(CRED_TYPE_PIN) + if feature_map & DoorLockFeature.kRfidCredential: + types.append(CRED_TYPE_RFID) + if feature_map & DoorLockFeature.kFingerCredentials: + types.append(CRED_TYPE_FINGERPRINT) + if feature_map & DoorLockFeature.kFaceCredentials: + types.append(CRED_TYPE_FACE) + return types + + +def _format_user_response(user_data: Any) -> LockUserData | None: + """Format GetUser response to API response format. + + Returns None if the user slot is empty (no userStatus). + """ + if user_data is None: + return None + + user_status = _get_attr(user_data, "userStatus") + if user_status is None: + return None + + creds = _get_attr(user_data, "credentials") + credentials: list[LockUserCredentialData] = [ + LockUserCredentialData( + type=CREDENTIAL_TYPE_MAP.get(_get_attr(cred, "credentialType"), "unknown"), + index=_get_attr(cred, "credentialIndex"), + ) + for cred in (creds or []) + ] + + return LockUserData( + user_index=_get_attr(user_data, "userIndex"), + user_name=_get_attr(user_data, "userName"), + user_unique_id=_get_attr(user_data, "userUniqueID"), + user_status=USER_STATUS_MAP.get(user_status, "unknown"), + user_type=USER_TYPE_MAP.get(_get_attr(user_data, "userType"), "unknown"), + credential_rule=CREDENTIAL_RULE_MAP.get( + _get_attr(user_data, "credentialRule"), "unknown" + ), + credentials=credentials, + next_user_index=_get_attr(user_data, "nextUserIndex"), + ) + + +# --- Credential management helpers --- + + +async def _clear_user_credentials( + matter_client: MatterClient, + node_id: int, + endpoint_id: int, + user_index: int, +) -> None: + """Clear all credentials for a specific user. + + Fetches the user to get credential list, then clears each credential. + """ + get_user_response = await matter_client.send_device_command( + node_id=node_id, + endpoint_id=endpoint_id, + command=clusters.DoorLock.Commands.GetUser(userIndex=user_index), + ) + + creds = _get_attr(get_user_response, "credentials") + if not creds: + return + + for cred in creds: + cred_type = _get_attr(cred, "credentialType") + cred_index = _get_attr(cred, "credentialIndex") + await matter_client.send_device_command( + node_id=node_id, + endpoint_id=endpoint_id, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type, + credentialIndex=cred_index, + ), + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + + +class LockEndpointNotFoundError(HomeAssistantError): + """Lock endpoint not found on node.""" + + +class UsrFeatureNotSupportedError(ServiceValidationError): + """Lock does not support USR (user management) feature.""" + + +class UserSlotEmptyError(ServiceValidationError): + """User slot is empty.""" + + +class NoAvailableUserSlotsError(ServiceValidationError): + """No available user slots on the lock.""" + + +class CredentialTypeNotSupportedError(ServiceValidationError): + """Lock does not support the requested credential type.""" + + +class CredentialDataInvalidError(ServiceValidationError): + """Credential data fails validation.""" + + +class SetCredentialFailedError(HomeAssistantError): + """SetCredential command returned a non-success status.""" + + +def _get_lock_endpoint_or_raise(node: MatterNode) -> MatterEndpoint: + """Get the DoorLock endpoint from a node or raise an error.""" + lock_endpoint = _get_lock_endpoint_from_node(node) + if lock_endpoint is None: + raise LockEndpointNotFoundError("No lock endpoint found on this device") + return lock_endpoint + + +def _ensure_usr_support(lock_endpoint: MatterEndpoint) -> None: + """Ensure the lock endpoint supports USR (user management) feature. + + Raises UsrFeatureNotSupportedError if the lock doesn't support user management. + """ + if not _lock_supports_usr_feature(lock_endpoint): + raise UsrFeatureNotSupportedError( + "Lock does not support user/credential management" + ) + + +# --- High-level business logic functions --- + + +async def get_lock_info( + matter_client: MatterClient, + node: MatterNode, +) -> GetLockInfoResult: + """Get lock capabilities and configuration info. + + Returns a typed dict with lock capability information. + Raises HomeAssistantError if lock endpoint not found. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + supports_usr = _lock_supports_usr_feature(lock_endpoint) + + # Get feature map for credential type detection + feature_map = ( + lock_endpoint.get_attribute_value(None, clusters.DoorLock.Attributes.FeatureMap) + or 0 + ) + + result = GetLockInfoResult( + supports_user_management=supports_usr, + supported_credential_types=_get_supported_credential_types(feature_map), + max_users=None, + max_pin_users=None, + max_rfid_users=None, + max_credentials_per_user=None, + min_pin_length=None, + max_pin_length=None, + min_rfid_length=None, + max_rfid_length=None, + ) + + # Populate capacity info if USR feature is supported + if supports_usr: + result["max_users"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfTotalUsersSupported + ) + result["max_pin_users"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfPINUsersSupported + ) + result["max_rfid_users"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfRFIDUsersSupported + ) + result["max_credentials_per_user"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfCredentialsSupportedPerUser + ) + result["min_pin_length"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MinPINCodeLength + ) + result["max_pin_length"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MaxPINCodeLength + ) + result["min_rfid_length"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MinRFIDCodeLength + ) + result["max_rfid_length"] = lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MaxRFIDCodeLength + ) + + return result + + +async def set_lock_user( + matter_client: MatterClient, + node: MatterNode, + *, + user_index: int | None = None, + user_name: str | None = None, + user_unique_id: int | None = None, + user_status: str | None = None, + user_type: str | None = None, + credential_rule: str | None = None, +) -> SetLockUserResult: + """Add or update a user on the lock. + + When user_status, user_type, or credential_rule is None, defaults are used + for new users and existing values are preserved for modifications. + + Returns typed dict with user_index on success. + Raises HomeAssistantError on failure. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + + if user_index is None: + # Adding new user - find first available slot + max_users = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfTotalUsersSupported + ) + or 0 + ) + + for idx in range(1, max_users + 1): + get_user_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetUser(userIndex=idx), + ) + if _get_attr(get_user_response, "userStatus") is None: + user_index = idx + break + + if user_index is None: + raise NoAvailableUserSlotsError("No available user slots on the lock") + + user_status_enum = ( + USER_STATUS_REVERSE_MAP.get( + user_status, + clusters.DoorLock.Enums.UserStatusEnum.kOccupiedEnabled, + ) + if user_status is not None + else clusters.DoorLock.Enums.UserStatusEnum.kOccupiedEnabled + ) + + await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd, + userIndex=user_index, + userName=user_name, + userUniqueID=user_unique_id, + userStatus=user_status_enum, + userType=USER_TYPE_REVERSE_MAP.get( + user_type, + clusters.DoorLock.Enums.UserTypeEnum.kUnrestrictedUser, + ) + if user_type is not None + else clusters.DoorLock.Enums.UserTypeEnum.kUnrestrictedUser, + credentialRule=CREDENTIAL_RULE_REVERSE_MAP.get( + credential_rule, + clusters.DoorLock.Enums.CredentialRuleEnum.kSingle, + ) + if credential_rule is not None + else clusters.DoorLock.Enums.CredentialRuleEnum.kSingle, + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + else: + # Updating existing user - preserve existing values when not specified + get_user_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetUser(userIndex=user_index), + ) + + if _get_attr(get_user_response, "userStatus") is None: + raise UserSlotEmptyError(f"User slot {user_index} is empty") + + resolved_user_name = ( + user_name + if user_name is not None + else _get_attr(get_user_response, "userName") + ) + resolved_unique_id = ( + user_unique_id + if user_unique_id is not None + else _get_attr(get_user_response, "userUniqueID") + ) + + resolved_status = ( + USER_STATUS_REVERSE_MAP[user_status] + if user_status is not None + else _get_attr(get_user_response, "userStatus") + ) + + resolved_type = ( + USER_TYPE_REVERSE_MAP[user_type] + if user_type is not None + else _get_attr(get_user_response, "userType") + ) + + resolved_rule = ( + CREDENTIAL_RULE_REVERSE_MAP[credential_rule] + if credential_rule is not None + else _get_attr(get_user_response, "credentialRule") + ) + + await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kModify, + userIndex=user_index, + userName=resolved_user_name, + userUniqueID=resolved_unique_id, + userStatus=resolved_status, + userType=resolved_type, + credentialRule=resolved_rule, + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + + return SetLockUserResult(user_index=user_index) + + +async def get_lock_users( + matter_client: MatterClient, + node: MatterNode, +) -> GetLockUsersResult: + """Get all users from the lock. + + Returns typed dict with users list and max_users capacity. + Raises HomeAssistantError on failure. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + + max_users = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.NumberOfTotalUsersSupported + ) + or 0 + ) + + users: list[LockUserData] = [] + current_index = 1 + + # Iterate through users using next_user_index for efficiency + while current_index is not None and current_index <= max_users: + get_user_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetUser( + userIndex=current_index, + ), + ) + + user_data = _format_user_response(get_user_response) + if user_data is not None: + users.append(user_data) + + # Move to next user index + next_index = _get_attr(get_user_response, "nextUserIndex") + if next_index is None or next_index <= current_index: + break + current_index = next_index + + return GetLockUsersResult( + max_users=max_users, + users=users, + ) + + +async def clear_lock_user( + matter_client: MatterClient, + node: MatterNode, + user_index: int, +) -> None: + """Clear a user from the lock, cleaning up credentials first. + + Use index 0xFFFE (CLEAR_ALL_INDEX) to clear all users. + Raises HomeAssistantError on failure. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + + if user_index == CLEAR_ALL_INDEX: + # Clear all: clear all credentials first, then all users + await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.ClearCredential( + credential=None, + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + else: + # Clear credentials for this specific user before deleting them + await _clear_user_credentials( + matter_client, + node.node_id, + lock_endpoint.endpoint_id, + user_index, + ) + + await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.ClearUser( + userIndex=user_index, + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + + +# --- Credential validation helpers --- + +# Map credential type strings to the feature bit that must be set +_CREDENTIAL_TYPE_FEATURE_MAP: dict[str, int] = { + CRED_TYPE_PIN: DoorLockFeature.kPinCredential, + CRED_TYPE_RFID: DoorLockFeature.kRfidCredential, + CRED_TYPE_FINGERPRINT: DoorLockFeature.kFingerCredentials, + CRED_TYPE_FINGER_VEIN: DoorLockFeature.kFingerCredentials, + CRED_TYPE_FACE: DoorLockFeature.kFaceCredentials, +} + + +def _validate_credential_type_support( + lock_endpoint: MatterEndpoint, credential_type: str +) -> None: + """Validate the lock supports the requested credential type. + + Raises CredentialTypeNotSupportedError if not supported. + """ + required_bit = _CREDENTIAL_TYPE_FEATURE_MAP.get(credential_type) + if required_bit is None: + raise CredentialTypeNotSupportedError( + translation_domain="matter", + translation_key=ERR_CREDENTIAL_TYPE_NOT_SUPPORTED, + translation_placeholders={"credential_type": credential_type}, + ) + + feature_map = _get_feature_map(lock_endpoint) or 0 + if not (feature_map & required_bit): + raise CredentialTypeNotSupportedError( + translation_domain="matter", + translation_key=ERR_CREDENTIAL_TYPE_NOT_SUPPORTED, + translation_placeholders={"credential_type": credential_type}, + ) + + +def _validate_credential_data( + lock_endpoint: MatterEndpoint, credential_type: str, credential_data: str +) -> None: + """Validate credential data against lock constraints. + + For PIN: checks digits-only and length against Min/MaxPINCodeLength. + For RFID: checks valid hex and byte length against Min/MaxRFIDCodeLength. + Raises CredentialDataInvalidError on failure. + """ + if credential_type == CRED_TYPE_PIN: + if not credential_data.isdigit(): + raise CredentialDataInvalidError( + translation_domain="matter", + translation_key=ERR_INVALID_CREDENTIAL_DATA, + translation_placeholders={"reason": "PIN must contain only digits"}, + ) + min_len = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MinPINCodeLength + ) + or 0 + ) + max_len = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MaxPINCodeLength + ) + or 255 + ) + if not min_len <= len(credential_data) <= max_len: + raise CredentialDataInvalidError( + translation_domain="matter", + translation_key=ERR_INVALID_CREDENTIAL_DATA, + translation_placeholders={ + "reason": (f"PIN length must be between {min_len} and {max_len}") + }, + ) + + elif credential_type == CRED_TYPE_RFID: + try: + rfid_bytes = bytes.fromhex(credential_data) + except ValueError as err: + raise CredentialDataInvalidError( + translation_domain="matter", + translation_key=ERR_INVALID_CREDENTIAL_DATA, + translation_placeholders={ + "reason": "RFID data must be valid hexadecimal" + }, + ) from err + min_len = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MinRFIDCodeLength + ) + or 0 + ) + max_len = ( + lock_endpoint.get_attribute_value( + None, clusters.DoorLock.Attributes.MaxRFIDCodeLength + ) + or 255 + ) + if not min_len <= len(rfid_bytes) <= max_len: + raise CredentialDataInvalidError( + translation_domain="matter", + translation_key=ERR_INVALID_CREDENTIAL_DATA, + translation_placeholders={ + "reason": ( + f"RFID data length must be between" + f" {min_len} and {max_len} bytes" + ) + }, + ) + + +def _credential_data_to_bytes(credential_type: str, credential_data: str) -> bytes: + """Convert credential data string to bytes for the Matter command.""" + if credential_type == CRED_TYPE_RFID: + return bytes.fromhex(credential_data) + # PIN and other types: encode as UTF-8 + return credential_data.encode() + + +# --- Credential business logic functions --- + + +async def set_lock_credential( + matter_client: MatterClient, + node: MatterNode, + *, + credential_type: str, + credential_data: str, + credential_index: int | None = None, + user_index: int | None = None, + user_status: str | None = None, + user_type: str | None = None, +) -> SetLockCredentialResult: + """Add or modify a credential on the lock. + + Returns typed dict with credential_index, user_index, and next_credential_index. + Raises ServiceValidationError for validation failures. + Raises HomeAssistantError for device communication failures. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + _validate_credential_type_support(lock_endpoint, credential_type) + _validate_credential_data(lock_endpoint, credential_type, credential_data) + + cred_type_int = CREDENTIAL_TYPE_REVERSE_MAP[credential_type] + cred_data_bytes = _credential_data_to_bytes(credential_type, credential_data) + + # Determine operation type and credential index + operation_type = clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd + + if credential_index is None: + # Auto-find first available credential slot + max_creds = ( + lock_endpoint.get_attribute_value( + None, + clusters.DoorLock.Attributes.NumberOfCredentialsSupportedPerUser, + ) + or 5 + ) + for idx in range(1, max_creds + 1): + status_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type_int, + credentialIndex=idx, + ), + ), + ) + if not _get_attr(status_response, "credentialExists"): + credential_index = idx + break + + if credential_index is None: + raise NoAvailableUserSlotsError("No available credential slots on the lock") + else: + # Check if slot is occupied to determine Add vs Modify + status_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type_int, + credentialIndex=credential_index, + ), + ), + ) + if _get_attr(status_response, "credentialExists"): + operation_type = clusters.DoorLock.Enums.DataOperationTypeEnum.kModify + + # Resolve optional user_status and user_type enums + resolved_user_status = ( + USER_STATUS_REVERSE_MAP.get(user_status) if user_status is not None else None + ) + resolved_user_type = ( + USER_TYPE_REVERSE_MAP.get(user_type) if user_type is not None else None + ) + + set_cred_response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.SetCredential( + operationType=operation_type, + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type_int, + credentialIndex=credential_index, + ), + credentialData=cred_data_bytes, + userIndex=user_index, + userStatus=resolved_user_status, + userType=resolved_user_type, + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + + status_code = _get_attr(set_cred_response, "status") + status_str = SET_CREDENTIAL_STATUS_MAP.get(status_code, f"unknown({status_code})") + if status_str != "success": + raise SetCredentialFailedError( + translation_domain="matter", + translation_key="set_credential_failed", + translation_placeholders={"status": status_str}, + ) + + return SetLockCredentialResult( + credential_index=credential_index, + user_index=_get_attr(set_cred_response, "userIndex"), + next_credential_index=_get_attr(set_cred_response, "nextCredentialIndex"), + ) + + +async def clear_lock_credential( + matter_client: MatterClient, + node: MatterNode, + *, + credential_type: str, + credential_index: int, +) -> None: + """Clear a credential from the lock. + + Raises HomeAssistantError on failure. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + + cred_type_int = CREDENTIAL_TYPE_REVERSE_MAP[credential_type] + + await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type_int, + credentialIndex=credential_index, + ), + ), + timed_request_timeout_ms=LOCK_TIMED_REQUEST_TIMEOUT_MS, + ) + + +async def get_lock_credential_status( + matter_client: MatterClient, + node: MatterNode, + *, + credential_type: str, + credential_index: int, +) -> GetLockCredentialStatusResult: + """Get the status of a credential slot on the lock. + + Returns typed dict with credential_exists, user_index, next_credential_index. + Raises HomeAssistantError on failure. + """ + lock_endpoint = _get_lock_endpoint_or_raise(node) + _ensure_usr_support(lock_endpoint) + + cred_type_int = CREDENTIAL_TYPE_REVERSE_MAP[credential_type] + + response = await matter_client.send_device_command( + node_id=node.node_id, + endpoint_id=lock_endpoint.endpoint_id, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=cred_type_int, + credentialIndex=credential_index, + ), + ), + ) + + return GetLockCredentialStatusResult( + credential_exists=bool(_get_attr(response, "credentialExists")), + user_index=_get_attr(response, "userIndex"), + next_credential_index=_get_attr(response, "nextCredentialIndex"), + ) diff --git a/homeassistant/components/matter/services.py b/homeassistant/components/matter/services.py index 62a2da51a967..e8076d76cfc1 100644 --- a/homeassistant/components/matter/services.py +++ b/homeassistant/components/matter/services.py @@ -4,11 +4,27 @@ from __future__ import annotations import voluptuous as vol +from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN from homeassistant.components.water_heater import DOMAIN as WATER_HEATER_DOMAIN -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import HomeAssistant, SupportsResponse, callback from homeassistant.helpers import config_validation as cv, service -from .const import DOMAIN +from .const import ( + ATTR_CREDENTIAL_DATA, + ATTR_CREDENTIAL_INDEX, + ATTR_CREDENTIAL_RULE, + ATTR_CREDENTIAL_TYPE, + ATTR_USER_INDEX, + ATTR_USER_NAME, + ATTR_USER_STATUS, + ATTR_USER_TYPE, + CLEAR_ALL_INDEX, + CREDENTIAL_RULE_REVERSE_MAP, + CREDENTIAL_TYPE_REVERSE_MAP, + DOMAIN, + SERVICE_CREDENTIAL_TYPES, + USER_TYPE_REVERSE_MAP, +) ATTR_DURATION = "duration" ATTR_EMERGENCY_BOOST = "emergency_boost" @@ -36,3 +52,108 @@ def async_setup_services(hass: HomeAssistant) -> None: }, func="async_set_boost", ) + + # Lock services - Full user CRUD + service.async_register_platform_entity_service( + hass, + DOMAIN, + "set_lock_user", + entity_domain=LOCK_DOMAIN, + schema={ + vol.Optional(ATTR_USER_INDEX): vol.All(vol.Coerce(int), vol.Range(min=1)), + vol.Optional(ATTR_USER_NAME): vol.Any(str, None), + vol.Optional(ATTR_USER_TYPE): vol.In(USER_TYPE_REVERSE_MAP.keys()), + vol.Optional(ATTR_CREDENTIAL_RULE): vol.In( + CREDENTIAL_RULE_REVERSE_MAP.keys() + ), + }, + func="async_set_lock_user", + ) + + service.async_register_platform_entity_service( + hass, + DOMAIN, + "clear_lock_user", + entity_domain=LOCK_DOMAIN, + schema={ + vol.Required(ATTR_USER_INDEX): vol.All( + vol.Coerce(int), + vol.Any(vol.Range(min=1), CLEAR_ALL_INDEX), + ), + }, + func="async_clear_lock_user", + ) + + # Lock services - Query operations + service.async_register_platform_entity_service( + hass, + DOMAIN, + "get_lock_info", + entity_domain=LOCK_DOMAIN, + schema={}, + func="async_get_lock_info", + supports_response=SupportsResponse.ONLY, + ) + + service.async_register_platform_entity_service( + hass, + DOMAIN, + "get_lock_users", + entity_domain=LOCK_DOMAIN, + schema={}, + func="async_get_lock_users", + supports_response=SupportsResponse.ONLY, + ) + + # Lock services - Credential management + service.async_register_platform_entity_service( + hass, + DOMAIN, + "set_lock_credential", + entity_domain=LOCK_DOMAIN, + schema={ + vol.Required(ATTR_CREDENTIAL_TYPE): vol.In(SERVICE_CREDENTIAL_TYPES), + vol.Required(ATTR_CREDENTIAL_DATA): str, + vol.Optional(ATTR_CREDENTIAL_INDEX): vol.All( + vol.Coerce(int), vol.Range(min=0) + ), + vol.Optional(ATTR_USER_INDEX): vol.All(vol.Coerce(int), vol.Range(min=1)), + vol.Optional(ATTR_USER_STATUS): vol.In( + ["occupied_enabled", "occupied_disabled"] + ), + vol.Optional(ATTR_USER_TYPE): vol.In(USER_TYPE_REVERSE_MAP.keys()), + }, + func="async_set_lock_credential", + supports_response=SupportsResponse.ONLY, + ) + + service.async_register_platform_entity_service( + hass, + DOMAIN, + "clear_lock_credential", + entity_domain=LOCK_DOMAIN, + schema={ + vol.Required(ATTR_CREDENTIAL_TYPE): vol.In(SERVICE_CREDENTIAL_TYPES), + vol.Required(ATTR_CREDENTIAL_INDEX): vol.All( + vol.Coerce(int), vol.Range(min=0) + ), + }, + func="async_clear_lock_credential", + ) + + service.async_register_platform_entity_service( + hass, + DOMAIN, + "get_lock_credential_status", + entity_domain=LOCK_DOMAIN, + schema={ + vol.Required(ATTR_CREDENTIAL_TYPE): vol.In( + CREDENTIAL_TYPE_REVERSE_MAP.keys() + ), + vol.Required(ATTR_CREDENTIAL_INDEX): vol.All( + vol.Coerce(int), vol.Range(min=0) + ), + }, + func="async_get_lock_credential_status", + supports_response=SupportsResponse.ONLY, + ) diff --git a/homeassistant/components/matter/services.yaml b/homeassistant/components/matter/services.yaml index a0f8b9d7862a..5127c7b602f3 100644 --- a/homeassistant/components/matter/services.yaml +++ b/homeassistant/components/matter/services.yaml @@ -1,3 +1,177 @@ +clear_lock_credential: + target: + entity: + domain: lock + integration: matter + fields: + credential_type: + selector: + select: + options: + - pin + - rfid + - fingerprint + - finger_vein + - face + required: true + credential_index: + selector: + number: + min: 0 + max: 65534 + step: 1 + mode: box + required: true + +clear_lock_user: + target: + entity: + domain: lock + integration: matter + fields: + user_index: + selector: + number: + min: 1 + max: 65534 + step: 1 + mode: box + required: true + +get_lock_credential_status: + target: + entity: + domain: lock + integration: matter + fields: + credential_type: + selector: + select: + options: + - programming_pin + - pin + - rfid + - fingerprint + - finger_vein + - face + - aliro_credential_issuer_key + - aliro_evictable_endpoint_key + - aliro_non_evictable_endpoint_key + required: true + credential_index: + selector: + number: + min: 0 + max: 65534 + step: 1 + mode: box + required: true + +get_lock_info: + target: + entity: + domain: lock + integration: matter + +get_lock_users: + target: + entity: + domain: lock + integration: matter + +set_lock_credential: + target: + entity: + domain: lock + integration: matter + fields: + credential_type: + selector: + select: + options: + - pin + - rfid + - fingerprint + - finger_vein + - face + required: true + credential_data: + selector: + text: + required: true + credential_index: + selector: + number: + min: 0 + max: 65534 + step: 1 + mode: box + user_index: + selector: + number: + min: 1 + max: 65534 + step: 1 + mode: box + user_status: + selector: + select: + options: + - occupied_enabled + - occupied_disabled + user_type: + selector: + select: + options: + - unrestricted_user + - year_day_schedule_user + - week_day_schedule_user + - programming_user + - non_access_user + - forced_user + - disposable_user + - expiring_user + - schedule_restricted_user + - remote_only_user + +set_lock_user: + target: + entity: + domain: lock + integration: matter + fields: + user_index: + selector: + number: + min: 1 + max: 255 + step: 1 + mode: box + user_name: + selector: + text: + user_type: + selector: + select: + options: + - unrestricted_user + - year_day_schedule_user + - week_day_schedule_user + - programming_user + - non_access_user + - forced_user + - disposable_user + - expiring_user + - schedule_restricted_user + - remote_only_user + credential_rule: + selector: + select: + options: + - single + - dual + - tri + water_heater_boost: target: entity: diff --git a/homeassistant/components/matter/strings.json b/homeassistant/components/matter/strings.json index 42d8d1cc0f05..436e1dd6b1a9 100644 --- a/homeassistant/components/matter/strings.json +++ b/homeassistant/components/matter/strings.json @@ -619,6 +619,17 @@ } } }, + "exceptions": { + "credential_type_not_supported": { + "message": "The lock does not support credential type `{credential_type}`." + }, + "invalid_credential_data": { + "message": "Invalid credential data: {reason}." + }, + "set_credential_failed": { + "message": "Failed to set credential: lock returned status `{status}`." + } + }, "issues": { "server_version_version_too_new": { "description": "The version of the Matter Server you are currently running is too new for this version of Home Assistant. Please update Home Assistant or downgrade the Matter Server to an older version to fix this issue.", @@ -630,6 +641,52 @@ } }, "services": { + "clear_lock_credential": { + "description": "Removes a credential from the lock.", + "fields": { + "credential_index": { + "description": "The credential slot index to clear.", + "name": "Credential index" + }, + "credential_type": { + "description": "The type of credential to clear.", + "name": "Credential type" + } + }, + "name": "Clear lock credential" + }, + "clear_lock_user": { + "description": "Deletes a lock user and all associated credentials. Use index 65534 to clear all users.", + "fields": { + "user_index": { + "description": "The user slot index (1-based) to clear, or 65534 to clear all.", + "name": "User index" + } + }, + "name": "Clear lock user" + }, + "get_lock_credential_status": { + "description": "Returns the status of a credential slot on the lock.", + "fields": { + "credential_index": { + "description": "The credential slot index to query.", + "name": "Credential index" + }, + "credential_type": { + "description": "The type of credential to query.", + "name": "Credential type" + } + }, + "name": "Get lock credential status" + }, + "get_lock_info": { + "description": "Returns lock capabilities including supported credential types, user capacity, and PIN length constraints.", + "name": "Get lock info" + }, + "get_lock_users": { + "description": "Returns all users configured on the lock with their credentials.", + "name": "Get lock users" + }, "open_commissioning_window": { "description": "Allows adding one of your devices to another Matter network by opening the commissioning window for this Matter device for 60 seconds.", "fields": { @@ -640,6 +697,58 @@ }, "name": "Open commissioning window" }, + "set_lock_credential": { + "description": "Adds or updates a credential on the lock.", + "fields": { + "credential_data": { + "description": "The credential data. For PIN: digits only. For RFID: hexadecimal string.", + "name": "Credential data" + }, + "credential_index": { + "description": "The credential slot index. Leave empty to auto-find an available slot.", + "name": "Credential index" + }, + "credential_type": { + "description": "The type of credential (e.g., pin, rfid, fingerprint).", + "name": "Credential type" + }, + "user_index": { + "description": "The user index to associate the credential with. Leave empty for automatic assignment.", + "name": "User index" + }, + "user_status": { + "description": "The user status to set when creating a new user for this credential.", + "name": "User status" + }, + "user_type": { + "description": "The user type to set when creating a new user for this credential.", + "name": "User type" + } + }, + "name": "Set lock credential" + }, + "set_lock_user": { + "description": "Creates or updates a lock user.", + "fields": { + "credential_rule": { + "description": "The credential rule for the user.", + "name": "Credential rule" + }, + "user_index": { + "description": "The user slot index (1-based). Leave empty to auto-find an available slot.", + "name": "User index" + }, + "user_name": { + "description": "The name for the user.", + "name": "User name" + }, + "user_type": { + "description": "The type of user to create.", + "name": "User type" + } + }, + "name": "Set lock user" + }, "water_heater_boost": { "description": "Enables water heater boost for a specific duration.", "fields": { diff --git a/tests/components/matter/test_lock.py b/tests/components/matter/test_lock.py index 1151d250da68..cac0f2bb59a7 100644 --- a/tests/components/matter/test_lock.py +++ b/tests/components/matter/test_lock.py @@ -1,17 +1,31 @@ """Test Matter locks.""" -from unittest.mock import MagicMock, call +from typing import Any +from unittest.mock import AsyncMock, MagicMock, call from chip.clusters import Objects as clusters from matter_server.client.models.node import MatterNode +from matter_server.common.errors import MatterError from matter_server.common.models import EventType, MatterNodeEvent import pytest from syrupy.assertion import SnapshotAssertion from homeassistant.components.lock import ATTR_CHANGED_BY, LockEntityFeature, LockState -from homeassistant.const import ATTR_CODE, STATE_UNKNOWN, Platform +from homeassistant.components.matter.const import ( + ATTR_CREDENTIAL_DATA, + ATTR_CREDENTIAL_INDEX, + ATTR_CREDENTIAL_RULE, + ATTR_CREDENTIAL_TYPE, + ATTR_USER_INDEX, + ATTR_USER_NAME, + ATTR_USER_STATUS, + ATTR_USER_TYPE, + CLEAR_ALL_INDEX, + DOMAIN, +) +from homeassistant.const import ATTR_CODE, ATTR_ENTITY_ID, STATE_UNKNOWN, Platform from homeassistant.core import HomeAssistant -from homeassistant.exceptions import ServiceValidationError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers import entity_registry as er from .common import ( @@ -20,6 +34,14 @@ from .common import ( trigger_subscription_callback, ) +# Feature map bits +_FEATURE_PIN = 1 # kPinCredential (bit 0) +_FEATURE_RFID = 2 # kRfidCredential (bit 1) +_FEATURE_USR = 256 # kUser (bit 8) +_FEATURE_USR_PIN = _FEATURE_USR | _FEATURE_PIN # 257 +_FEATURE_USR_RFID = _FEATURE_USR | _FEATURE_RFID # 258 +_FEATURE_USR_PIN_RFID = _FEATURE_USR | _FEATURE_PIN | _FEATURE_RFID # 259 + @pytest.mark.usefixtures("matter_devices") async def test_locks( @@ -52,7 +74,7 @@ async def test_lock( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.UnlockDoor(), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) matter_client.send_device_command.reset_mock() @@ -70,7 +92,7 @@ async def test_lock( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.LockDoor(), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) matter_client.send_device_command.reset_mock() @@ -173,7 +195,7 @@ async def test_lock_requires_pin( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.LockDoor(code.encode()), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) # Lock door using default code @@ -193,7 +215,7 @@ async def test_lock_requires_pin( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.LockDoor(default_code.encode()), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) @@ -223,7 +245,7 @@ async def test_lock_with_unbolt( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.UnboltDoor(), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) matter_client.send_device_command.reset_mock() # test open / unlatch @@ -240,7 +262,7 @@ async def test_lock_with_unbolt( node_id=matter_node.node_id, endpoint_id=1, command=clusters.DoorLock.Commands.UnlockDoor(), - timed_request_timeout_ms=1000, + timed_request_timeout_ms=10000, ) await hass.async_block_till_done() @@ -261,3 +283,2114 @@ async def test_lock_with_unbolt( state = hass.states.get("lock.mock_door_lock_with_unbolt") assert state assert state.state == LockState.OPEN + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_lock_operation_updates_changed_by( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test lock operation event updates changed_by with source.""" + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=257, + event_id=2, + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={"operationSource": 7, "lockOperationType": 1}, + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state + assert state.attributes[ATTR_CHANGED_BY] == "Remote" + + +# --- Entity service tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_service( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user entity service creates user.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + {"userStatus": None}, # GetUser(1): empty slot + None, # SetUser: success + ] + ) + + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_NAME: "TestUser", + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + # Verify GetUser was called to find empty slot + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + # Verify SetUser was called with kAdd operation + set_user_cmd = matter_client.send_device_command.call_args_list[1] + assert set_user_cmd == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd, + userIndex=1, + userName="TestUser", + userUniqueID=None, + userStatus=clusters.DoorLock.Enums.UserStatusEnum.kOccupiedEnabled, + userType=clusters.DoorLock.Enums.UserTypeEnum.kUnrestrictedUser, + credentialRule=clusters.DoorLock.Enums.CredentialRuleEnum.kSingle, + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_update_existing( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user service updates existing user.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + { # GetUser: existing user + "userStatus": 1, + "userName": "Old Name", + "userUniqueID": 123, + "userType": 0, + "credentialRule": 0, + "credentials": None, + }, + None, # SetUser: modify + ] + ) + + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 1, + ATTR_USER_NAME: "New Name", + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + # Verify GetUser was called to check existing user + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + # Verify SetUser was called with kModify, preserving existing values + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kModify, + userIndex=1, + userName="New Name", + userUniqueID=123, + userStatus=1, # Preserved from existing user + userType=0, # Preserved from existing user + credentialRule=0, # Preserved from existing user + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_no_available_slots( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user when no user slots are available.""" + # All user slots are occupied + matter_client.send_device_command = AsyncMock( + return_value={"userStatus": 1} # All slots occupied + ) + + with pytest.raises(ServiceValidationError, match="No available user slots"): + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_NAME: "Test User", + }, + blocking=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_empty_slot_error( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user errors when updating non-existent user.""" + matter_client.send_device_command = AsyncMock( + return_value={"userStatus": None} # User doesn't exist + ) + + with pytest.raises(ServiceValidationError, match="is empty"): + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 5, + ATTR_USER_NAME: "Test User", + }, + blocking=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_clear_lock_user_service( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_user entity service.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # clear_user_credentials: GetUser returns user with no creds + {"userStatus": 1, "credentials": None}, + None, # ClearUser + ] + ) + + await hass.services.async_call( + DOMAIN, + "clear_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 1, + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + # Verify GetUser was called to check credentials + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + # Verify ClearUser was called + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearUser(userIndex=1), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_clear_lock_user_clears_credentials_first( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_user clears credentials before clearing user.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # clear_user_credentials: GetUser returns user with credentials + { + "userStatus": 1, + "credentials": [ + {"credentialType": 1, "credentialIndex": 1}, + {"credentialType": 1, "credentialIndex": 2}, + ], + }, + None, # ClearCredential for first + None, # ClearCredential for second + None, # ClearUser + ] + ) + + await hass.services.async_call( + DOMAIN, + "clear_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 1, + }, + blocking=True, + ) + + # GetUser + 2 ClearCredential + ClearUser + assert matter_client.send_device_command.call_count == 4 + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=1, + credentialIndex=1, + ), + ), + timed_request_timeout_ms=10000, + ) + assert matter_client.send_device_command.call_args_list[2] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=1, + credentialIndex=2, + ), + ), + timed_request_timeout_ms=10000, + ) + assert matter_client.send_device_command.call_args_list[3] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearUser(userIndex=1), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_info_service( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_info entity service returns capabilities.""" + result = await hass.services.async_call( + DOMAIN, + "get_lock_info", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "supports_user_management": True, + "supported_credential_types": ["pin"], + "max_users": 10, + "max_pin_users": 10, + "max_rfid_users": 10, + "max_credentials_per_user": 5, + "min_pin_length": 6, + "max_pin_length": 8, + "min_rfid_length": 10, + "max_rfid_length": 20, + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_users_service( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_users entity service returns users.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + { + "userIndex": 1, + "userName": "Alice", + "userUniqueID": None, + "userStatus": 1, + "userType": 0, + "credentialRule": 0, + "credentials": None, + "nextUserIndex": None, + }, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_users", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + # Verify GetUser command was sent + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + + assert result["lock.mock_door_lock"] == { + "max_users": 10, + "users": [ + { + "user_index": 1, + "user_name": "Alice", + "user_unique_id": None, + "user_status": "occupied_enabled", + "user_type": "unrestricted_user", + "credential_rule": "single", + "credentials": [], + "next_user_index": None, + } + ], + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_service_on_lock_without_user_management( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test entity services on lock without USR feature raise error.""" + # Default door_lock fixture has featuremap=0, no USR support + with pytest.raises(ServiceValidationError, match="does not support"): + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_NAME: "Test", + }, + blocking=True, + ) + + with pytest.raises(ServiceValidationError, match="does not support"): + await hass.services.async_call( + DOMAIN, + "clear_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 1, + }, + blocking=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_on_matter_node_event_filters_non_matching_events( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test that node events for different endpoints/clusters are filtered.""" + state = hass.states.get("lock.mock_door_lock") + assert state is not None + original_changed_by = state.attributes.get(ATTR_CHANGED_BY) + + # Fire event for different endpoint - should be ignored + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=99, # Different endpoint + cluster_id=257, + event_id=2, + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={"operationSource": 7}, # Remote source + ), + ) + + # changed_by should not have changed + state = hass.states.get("lock.mock_door_lock") + assert state.attributes.get(ATTR_CHANGED_BY) == original_changed_by + + # Fire event for different cluster - should also be ignored + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=999, # Different cluster + event_id=2, + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={"operationSource": 7}, + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state.attributes.get(ATTR_CHANGED_BY) == original_changed_by + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_users_iterates_with_next_index( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_users uses nextUserIndex for efficient iteration.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + { # First user at index 1 + "userIndex": 1, + "userStatus": 1, + "userName": "User 1", + "userUniqueID": None, + "userType": 0, + "credentialRule": 0, + "credentials": None, + "nextUserIndex": 5, # Next user at index 5 + }, + { # Second user at index 5 + "userIndex": 5, + "userStatus": 1, + "userName": "User 5", + "userUniqueID": None, + "userType": 0, + "credentialRule": 0, + "credentials": None, + "nextUserIndex": None, # No more users + }, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_users", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + assert matter_client.send_device_command.call_count == 2 + # Verify it jumped from index 1 to index 5 via nextUserIndex + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=5), + ) + + entity_result = result["lock.mock_door_lock"] + assert entity_result == { + "max_users": 10, + "users": [ + { + "user_index": 1, + "user_name": "User 1", + "user_unique_id": None, + "user_status": "occupied_enabled", + "user_type": "unrestricted_user", + "credential_rule": "single", + "credentials": [], + "next_user_index": 5, + }, + { + "user_index": 5, + "user_name": "User 5", + "user_unique_id": None, + "user_status": "occupied_enabled", + "user_type": "unrestricted_user", + "credential_rule": "single", + "credentials": [], + "next_user_index": None, + }, + ], + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/51": True, # RequirePINforRemoteOperation (attribute 51) + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_code_format_property_with_pin_required( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test code_format property returns regex when PIN is required.""" + state = hass.states.get("lock.mock_door_lock") + assert state is not None + # code_format should be set when RequirePINforRemoteOperation is True + # The format should be a regex like ^\d{4,8}$ + code_format = state.attributes.get("code_format") + assert code_format is not None + assert "\\d" in code_format + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_users_next_user_index_loop_prevention( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_users handles nextUserIndex <= current to prevent loops.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + { # User at index 1 + "userIndex": 1, + "userStatus": 1, + "userName": "User 1", + "userUniqueID": None, + "userType": 0, + "credentialRule": 0, + "credentials": None, + "nextUserIndex": 1, # Same as current - should break loop + }, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_users", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + + assert result is not None + # Result is keyed by entity_id + lock_users = result["lock.mock_door_lock"] + assert len(lock_users["users"]) == 1 + # Should have stopped after first user due to nextUserIndex <= current + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_users_with_credentials( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_users returns credential info for users.""" + pin_cred_type = clusters.DoorLock.Enums.CredentialTypeEnum.kPin + matter_client.send_device_command = AsyncMock( + side_effect=[ + { # User with credentials + "userIndex": 1, + "userStatus": 1, + "userName": "User With PIN", + "userUniqueID": 123, + "userType": 0, + "credentialRule": 0, + "credentials": [ + {"credentialType": pin_cred_type, "credentialIndex": 1}, + {"credentialType": pin_cred_type, "credentialIndex": 2}, + ], + "nextUserIndex": None, + }, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_users", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetUser(userIndex=1), + ) + + assert result["lock.mock_door_lock"] == { + "max_users": 10, + "users": [ + { + "user_index": 1, + "user_name": "User With PIN", + "user_unique_id": 123, + "user_status": "occupied_enabled", + "user_type": "unrestricted_user", + "credential_rule": "single", + "credentials": [ + {"type": "pin", "index": 1}, + {"type": "pin", "index": 2}, + ], + "next_user_index": None, + } + ], + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +@pytest.mark.parametrize( + ("service_name", "service_data", "return_response"), + [ + ("set_lock_user", {ATTR_USER_NAME: "Test"}, False), + ("clear_lock_user", {ATTR_USER_INDEX: 1}, False), + ("get_lock_users", {}, True), + ( + "set_lock_credential", + { + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "123456", + ATTR_CREDENTIAL_INDEX: 1, + }, + True, + ), + ( + "clear_lock_credential", + {ATTR_CREDENTIAL_TYPE: "pin", ATTR_CREDENTIAL_INDEX: 1}, + False, + ), + ( + "get_lock_credential_status", + {ATTR_CREDENTIAL_TYPE: "pin", ATTR_CREDENTIAL_INDEX: 1}, + True, + ), + ], +) +async def test_matter_error_converted_to_home_assistant_error( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, + service_name: str, + service_data: dict[str, Any], + return_response: bool, +) -> None: + """Test that MatterError from helpers is converted to HomeAssistantError.""" + # Simulate a MatterError from the device command + matter_client.send_device_command = AsyncMock( + side_effect=MatterError("Device communication failed") + ) + + with pytest.raises(HomeAssistantError, match="Device communication failed"): + await hass.services.async_call( + DOMAIN, + service_name, + {ATTR_ENTITY_ID: "lock.mock_door_lock", **service_data}, + blocking=True, + return_response=return_response, + ) + + # Verify a command was attempted before the error + assert matter_client.send_device_command.call_count >= 1 + + +# --- Credential service tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_pin( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential with PIN type.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus: slot occupied -> kModify + {"credentialExists": True, "userIndex": 1, "nextCredentialIndex": 2}, + # SetCredential response + {"status": 0, "userIndex": 1, "nextCredentialIndex": 2}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "credential_index": 1, + "user_index": 1, + "next_credential_index": 2, + } + + assert matter_client.send_device_command.call_count == 2 + # Verify GetCredentialStatus was called first + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kPin, + credentialIndex=1, + ), + ), + ) + # Verify SetCredential was called with kModify (occupied slot) + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetCredential( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kModify, + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kPin, + credentialIndex=1, + ), + credentialData=b"1234", + userIndex=None, + userStatus=None, + userType=None, + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_auto_find_slot( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential auto-finds first available slot.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus(1): occupied + {"credentialExists": True, "userIndex": 1, "nextCredentialIndex": 2}, + # GetCredentialStatus(2): empty + { + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": 3, + }, + # SetCredential response + {"status": 0, "userIndex": 1, "nextCredentialIndex": 3}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "5678", + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "credential_index": 2, + "user_index": 1, + "next_credential_index": 3, + } + + assert matter_client.send_device_command.call_count == 3 + # Verify SetCredential was called with kAdd for the empty slot at index 2 + set_cred_cmd = matter_client.send_device_command.call_args_list[2] + assert ( + set_cred_cmd.kwargs["command"].operationType + == clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd + ) + assert set_cred_cmd.kwargs["command"].credential.credentialIndex == 2 + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_with_user_index( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential passes user_index to command.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus: empty slot + { + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": 2, + }, + # SetCredential response + {"status": 0, "userIndex": 3, "nextCredentialIndex": 2}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + ATTR_USER_INDEX: 3, + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "credential_index": 1, + "user_index": 3, + "next_credential_index": 2, + } + + # Verify user_index was passed in SetCredential command + set_cred_call = matter_client.send_device_command.call_args_list[1] + assert set_cred_call.kwargs["command"].userIndex == 3 + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_invalid_pin_too_short( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects PIN that is too short.""" + with pytest.raises(ServiceValidationError, match="PIN length must be between"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "12", # Too short (min 4) + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_invalid_pin_non_digit( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects non-digit PIN.""" + with pytest.raises(ServiceValidationError, match="PIN must contain only digits"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "abcd", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR}]) +async def test_set_lock_credential_unsupported_type( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects unsupported credential type.""" + # USR feature set but no PIN credential feature + with pytest.raises(ServiceValidationError): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_status_failure( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential raises error on non-success status.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus: empty + { + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": 2, + }, + # SetCredential response with duplicate status + {"status": 2, "userIndex": None, "nextCredentialIndex": None}, + ] + ) + + with pytest.raises(HomeAssistantError, match="duplicate"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_no_available_slot( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential raises error when all slots are full.""" + # All GetCredentialStatus calls return occupied + matter_client.send_device_command = AsyncMock( + return_value={ + "credentialExists": True, + "userIndex": 1, + "nextCredentialIndex": None, + } + ) + + with pytest.raises(ServiceValidationError, match="No available credential slots"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_clear_lock_credential( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_credential sends ClearCredential command.""" + matter_client.send_device_command = AsyncMock(return_value=None) + + await hass.services.async_call( + DOMAIN, + "clear_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kPin, + credentialIndex=1, + ), + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_credential_status( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_credential_status returns credential info.""" + matter_client.send_device_command = AsyncMock( + return_value={ + "credentialExists": True, + "userIndex": 2, + "nextCredentialIndex": 3, + } + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_credential_status", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kPin, + credentialIndex=1, + ), + ), + ) + assert result["lock.mock_door_lock"] == { + "credential_exists": True, + "user_index": 2, + "next_credential_index": 3, + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_get_lock_credential_status_empty_slot( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_credential_status for empty slot.""" + matter_client.send_device_command = AsyncMock( + return_value={ + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": None, + } + ) + + result = await hass.services.async_call( + DOMAIN, + "get_lock_credential_status", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_INDEX: 5, + }, + blocking=True, + return_response=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.GetCredentialStatus( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kPin, + credentialIndex=5, + ), + ), + ) + + assert result["lock.mock_door_lock"] == { + "credential_exists": False, + "user_index": None, + "next_credential_index": None, + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_credential_services_without_usr_feature( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test credential services raise error without USR feature.""" + # Default door_lock fixture has featuremap=0, no USR support + with pytest.raises(ServiceValidationError, match="does not support"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + with pytest.raises(ServiceValidationError, match="does not support"): + await hass.services.async_call( + DOMAIN, + "clear_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + ) + + with pytest.raises(ServiceValidationError, match="does not support"): + await hass.services.async_call( + DOMAIN, + "get_lock_credential_status", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +# --- RFID credential tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_RFID, + "1/257/26": 4, # MinRFIDCodeLength + "1/257/25": 20, # MaxRFIDCodeLength + } + ], +) +async def test_set_lock_credential_rfid( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential with RFID type using hex data.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus: empty slot + { + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": 2, + }, + # SetCredential response + {"status": 0, "userIndex": 1, "nextCredentialIndex": 2}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "rfid", + ATTR_CREDENTIAL_DATA: "AABBCCDD", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "credential_index": 1, + "user_index": 1, + "next_credential_index": 2, + } + + assert matter_client.send_device_command.call_count == 2 + # Verify SetCredential was called with RFID type and hex-decoded bytes + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetCredential( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd, + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kRfid, + credentialIndex=1, + ), + credentialData=bytes.fromhex("AABBCCDD"), + userIndex=None, + userStatus=None, + userType=None, + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_RFID, + "1/257/26": 4, # MinRFIDCodeLength + "1/257/25": 20, # MaxRFIDCodeLength + } + ], +) +async def test_set_lock_credential_rfid_invalid_hex( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects invalid hex RFID data.""" + with pytest.raises( + ServiceValidationError, match="RFID data must be valid hexadecimal" + ): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "rfid", + ATTR_CREDENTIAL_DATA: "ZZZZ", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_RFID, + "1/257/26": 4, # MinRFIDCodeLength (bytes) + "1/257/25": 20, # MaxRFIDCodeLength (bytes) + } + ], +) +async def test_set_lock_credential_rfid_too_short( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects RFID data below min byte length.""" + # "AABB" = 2 bytes, min is 4 + with pytest.raises( + ServiceValidationError, match="RFID data length must be between" + ): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "rfid", + ATTR_CREDENTIAL_DATA: "AABB", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_RFID, + "1/257/26": 4, # MinRFIDCodeLength (bytes) + "1/257/25": 6, # MaxRFIDCodeLength (bytes) + } + ], +) +async def test_set_lock_credential_rfid_too_long( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects RFID data above max byte length.""" + # "AABBCCDDEEFF0011" = 8 bytes, max is 6 + with pytest.raises( + ServiceValidationError, match="RFID data length must be between" + ): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "rfid", + ATTR_CREDENTIAL_DATA: "AABBCCDDEEFF0011", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_RFID}]) +async def test_clear_lock_credential_rfid( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_credential with RFID type.""" + matter_client.send_device_command = AsyncMock(return_value=None) + + await hass.services.async_call( + DOMAIN, + "clear_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "rfid", + ATTR_CREDENTIAL_INDEX: 3, + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 1 + assert matter_client.send_device_command.call_args == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=clusters.DoorLock.Enums.CredentialTypeEnum.kRfid, + credentialIndex=3, + ), + ), + timed_request_timeout_ms=10000, + ) + + +# --- CLEAR_ALL_INDEX tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_clear_lock_user_clear_all( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_user with CLEAR_ALL_INDEX clears all credentials then users.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + None, # ClearCredential(None) - clear all credentials + None, # ClearUser(0xFFFE) - clear all users + ] + ) + + await hass.services.async_call( + DOMAIN, + "clear_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: CLEAR_ALL_INDEX, + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + # First: ClearCredential with None (clear all) + assert matter_client.send_device_command.call_args_list[0] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential(credential=None), + timed_request_timeout_ms=10000, + ) + # Second: ClearUser with CLEAR_ALL_INDEX + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearUser(userIndex=CLEAR_ALL_INDEX), + timed_request_timeout_ms=10000, + ) + + +# --- SetCredential status code tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +@pytest.mark.parametrize( + ("status_code", "expected_match"), + [ + (1, "failure"), # kFailure + (3, "occupied"), # kOccupied + (99, "unknown\\(99\\)"), # Unknown status code + ], +) +async def test_set_lock_credential_status_codes( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, + status_code: int, + expected_match: str, +) -> None: + """Test set_lock_credential raises error for non-success status codes.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetCredentialStatus: empty + { + "credentialExists": False, + "userIndex": None, + "nextCredentialIndex": 2, + }, + # SetCredential response with non-success status + {"status": status_code, "userIndex": None, "nextCredentialIndex": None}, + ] + ) + + with pytest.raises(HomeAssistantError, match=expected_match): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +# --- Node event edge case tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_lock_operation_event_missing_operation_source( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test LockOperation event with missing operationSource uses Unknown.""" + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=257, + event_id=2, # LockOperation + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={}, # No operationSource key + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state.attributes[ATTR_CHANGED_BY] == "Unknown" + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_lock_operation_event_null_data( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test LockOperation event with None data uses Unknown.""" + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=257, + event_id=2, # LockOperation + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data=None, + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state.attributes[ATTR_CHANGED_BY] == "Unknown" + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_lock_operation_event_unknown_source( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test LockOperation event with unknown operationSource value.""" + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=257, + event_id=2, # LockOperation + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={"operationSource": 999}, # Unknown source + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state.attributes[ATTR_CHANGED_BY] == "Unknown" + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_non_lock_operation_event_ignored( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test non-LockOperation events on the DoorLock cluster are ignored.""" + state = hass.states.get("lock.mock_door_lock") + original_changed_by = state.attributes.get(ATTR_CHANGED_BY) + + await trigger_subscription_callback( + hass, + matter_client, + EventType.NODE_EVENT, + MatterNodeEvent( + node_id=matter_node.node_id, + endpoint_id=1, + cluster_id=257, + event_id=99, # Not LockOperation (event_id=2) + event_number=0, + priority=1, + timestamp=0, + timestamp_type=0, + data={"operationSource": 7}, + ), + ) + + state = hass.states.get("lock.mock_door_lock") + assert state.attributes.get(ATTR_CHANGED_BY) == original_changed_by + + +# --- get_lock_info edge case tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +async def test_get_lock_info_without_usr_feature( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_info on lock without USR returns None for capacity fields.""" + # Default mock_door_lock has featuremap=0 (no USR) + result = await hass.services.async_call( + DOMAIN, + "get_lock_info", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"] == { + "supports_user_management": False, + "supported_credential_types": [], + "max_users": None, + "max_pin_users": None, + "max_rfid_users": None, + "max_credentials_per_user": None, + "min_pin_length": None, + "max_pin_length": None, + "min_rfid_length": None, + "max_rfid_length": None, + } + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN_RFID}]) +async def test_get_lock_info_with_multiple_credential_types( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test get_lock_info reports multiple supported credential types.""" + result = await hass.services.async_call( + DOMAIN, + "get_lock_info", + {ATTR_ENTITY_ID: "lock.mock_door_lock"}, + blocking=True, + return_response=True, + ) + + info = result["lock.mock_door_lock"] + assert info["supports_user_management"] is True + assert "pin" in info["supported_credential_types"] + assert "rfid" in info["supported_credential_types"] + + +# --- PIN boundary validation tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_pin_too_long( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential rejects PIN exceeding max length.""" + with pytest.raises(ServiceValidationError, match="PIN length must be between"): + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "123456789", # 9 digits, max is 8 + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_pin_exact_min_length( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential accepts PIN at exact minimum length.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + {"credentialExists": False, "userIndex": None, "nextCredentialIndex": 2}, + {"status": 0, "userIndex": 1, "nextCredentialIndex": 2}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", # Exactly 4 digits (min) + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"]["credential_index"] == 1 + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_pin_exact_max_length( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential accepts PIN at exact maximum length.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + {"credentialExists": False, "userIndex": None, "nextCredentialIndex": 2}, + {"status": 0, "userIndex": 1, "nextCredentialIndex": 2}, + ] + ) + + result = await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "12345678", # Exactly 8 digits (max) + ATTR_CREDENTIAL_INDEX: 1, + }, + blocking=True, + return_response=True, + ) + + assert result["lock.mock_door_lock"]["credential_index"] == 1 + + +# --- set_lock_credential with user_status and user_type params --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize( + "attributes", + [ + { + "1/257/65532": _FEATURE_USR_PIN, + "1/257/24": 4, # MinPINCodeLength + "1/257/23": 8, # MaxPINCodeLength + } + ], +) +async def test_set_lock_credential_with_user_status_and_type( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_credential passes user_status and user_type to command.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + {"credentialExists": False, "userIndex": None, "nextCredentialIndex": 2}, + {"status": 0, "userIndex": 1, "nextCredentialIndex": 2}, + ] + ) + + await hass.services.async_call( + DOMAIN, + "set_lock_credential", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_CREDENTIAL_TYPE: "pin", + ATTR_CREDENTIAL_DATA: "1234", + ATTR_CREDENTIAL_INDEX: 1, + ATTR_USER_STATUS: "occupied_disabled", + ATTR_USER_TYPE: "non_access_user", + }, + blocking=True, + return_response=True, + ) + + # Verify SetCredential was called with resolved user_status and user_type + set_cred_call = matter_client.send_device_command.call_args_list[1] + assert ( + set_cred_call.kwargs["command"].userStatus + == clusters.DoorLock.Enums.UserStatusEnum.kOccupiedDisabled + ) + assert ( + set_cred_call.kwargs["command"].userType + == clusters.DoorLock.Enums.UserTypeEnum.kNonAccessUser + ) + + +# --- set_lock_user with explicit params tests --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_new_with_explicit_params( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user creates new user with explicit type and credential rule.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + {"userStatus": None}, # GetUser(1): empty slot + None, # SetUser: success + ] + ) + + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_NAME: "Restricted", + ATTR_USER_TYPE: "week_day_schedule_user", + ATTR_CREDENTIAL_RULE: "dual", + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + set_user_cmd = matter_client.send_device_command.call_args_list[1] + assert set_user_cmd == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kAdd, + userIndex=1, + userName="Restricted", + userUniqueID=None, + userStatus=clusters.DoorLock.Enums.UserStatusEnum.kOccupiedEnabled, + userType=clusters.DoorLock.Enums.UserTypeEnum.kWeekDayScheduleUser, + credentialRule=clusters.DoorLock.Enums.CredentialRuleEnum.kDual, + ), + timed_request_timeout_ms=10000, + ) + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN}]) +async def test_set_lock_user_update_with_explicit_type_and_rule( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test set_lock_user updates existing user with explicit type and rule.""" + matter_client.send_device_command = AsyncMock( + side_effect=[ + { # GetUser: existing user + "userStatus": 1, + "userName": "Old Name", + "userUniqueID": 42, + "userType": 0, # kUnrestrictedUser + "credentialRule": 0, # kSingle + "credentials": None, + }, + None, # SetUser: modify + ] + ) + + await hass.services.async_call( + DOMAIN, + "set_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 3, + ATTR_USER_TYPE: "programming_user", + ATTR_CREDENTIAL_RULE: "tri", + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 2 + set_user_cmd = matter_client.send_device_command.call_args_list[1] + assert set_user_cmd == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.SetUser( + operationType=clusters.DoorLock.Enums.DataOperationTypeEnum.kModify, + userIndex=3, + userName="Old Name", # Preserved + userUniqueID=42, # Preserved + userStatus=1, # Preserved + userType=clusters.DoorLock.Enums.UserTypeEnum.kProgrammingUser, + credentialRule=clusters.DoorLock.Enums.CredentialRuleEnum.kTri, + ), + timed_request_timeout_ms=10000, + ) + + +# --- clear_lock_user with mixed credential types --- + + +@pytest.mark.parametrize("node_fixture", ["mock_door_lock"]) +@pytest.mark.parametrize("attributes", [{"1/257/65532": _FEATURE_USR_PIN_RFID}]) +async def test_clear_lock_user_mixed_credential_types( + hass: HomeAssistant, + matter_client: MagicMock, + matter_node: MatterNode, +) -> None: + """Test clear_lock_user clears mixed PIN and RFID credentials.""" + pin_type = clusters.DoorLock.Enums.CredentialTypeEnum.kPin + rfid_type = clusters.DoorLock.Enums.CredentialTypeEnum.kRfid + matter_client.send_device_command = AsyncMock( + side_effect=[ + # GetUser returns user with PIN and RFID credentials + { + "userStatus": 1, + "credentials": [ + {"credentialType": pin_type, "credentialIndex": 1}, + {"credentialType": rfid_type, "credentialIndex": 2}, + ], + }, + None, # ClearCredential for PIN + None, # ClearCredential for RFID + None, # ClearUser + ] + ) + + await hass.services.async_call( + DOMAIN, + "clear_lock_user", + { + ATTR_ENTITY_ID: "lock.mock_door_lock", + ATTR_USER_INDEX: 1, + }, + blocking=True, + ) + + assert matter_client.send_device_command.call_count == 4 + # Verify PIN credential was cleared + assert matter_client.send_device_command.call_args_list[1] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=pin_type, + credentialIndex=1, + ), + ), + timed_request_timeout_ms=10000, + ) + # Verify RFID credential was cleared + assert matter_client.send_device_command.call_args_list[2] == call( + node_id=matter_node.node_id, + endpoint_id=1, + command=clusters.DoorLock.Commands.ClearCredential( + credential=clusters.DoorLock.Structs.CredentialStruct( + credentialType=rfid_type, + credentialIndex=2, + ), + ), + timed_request_timeout_ms=10000, + )