diff --git a/custom_components/lock_code_manager/domain/credentials.py b/custom_components/lock_code_manager/domain/credentials.py index b9582727..eec24279 100644 --- a/custom_components/lock_code_manager/domain/credentials.py +++ b/custom_components/lock_code_manager/domain/credentials.py @@ -16,7 +16,7 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Iterable, Mapping from dataclasses import dataclass, field from enum import StrEnum from types import MappingProxyType @@ -241,6 +241,13 @@ class CredentialTypeCapability: ``supports_learn`` is True when the lock can enroll the credential at the device (for example a fingerprint learn flow) rather than being told the value. + + Length convention shared by every provider: a non-positive ``max_length`` + means "no advertised maximum / unknown" -- never a literal zero-length + limit, which would be meaningless -- so providers map an absent or + unreadable maximum to ``0`` (Matter's ``max_pin_length or 0`` idiom). A + non-positive ``min_length`` means "no minimum". ``length_bounds`` applies + this normalization; do not emit a literal ``0`` to express a real limit. """ num_slots: int @@ -287,6 +294,58 @@ def supports(self, credential_type: CredentialType) -> bool: """Return True when the lock advertises ``credential_type``.""" return credential_type in self.credential_types + def length_bounds( + self, credential_type: CredentialType + ) -> tuple[int, int | None] | None: + """ + Return the effective ``(min, max)`` value length for a credential type. + + ``None`` when the type is unsupported. A non-positive advertised + bound means "unbounded" rather than a literal limit: Matter reports + ``max_pin_length`` as ``... or 0``, where ``0`` is "unknown", so it + normalizes to no upper bound (``max`` of ``None``). A non-positive + minimum normalizes to ``0`` (no minimum). + """ + cap = self.capability_for(credential_type) + if cap is None: + return None + return ( + max(cap.min_length, 0), + cap.max_length if cap.max_length > 0 else None, + ) + + +def aggregate_length_bounds( + capabilities: Iterable[LockCapabilities | None], + credential_type: CredentialType, +) -> tuple[int | None, int | None]: + """ + Fold many locks' length limits into one tightest-common ``(min, max)``. + + Each bound is ``None`` when nothing constrains it. Locks with no + capabilities (``None``) or that do not support ``credential_type`` + contribute nothing, so an all-unknown set yields ``(None, None)``. + + The result is the tightest range every lock can satisfy: the largest + minimum and the smallest maximum. A returned ``min`` greater than ``max`` + signals an unsatisfiable intersection across locks; the caller decides how + to present it. User-interface defaults deliberately live in the caller, not + here, so non-interface callers can reuse this unchanged. + """ + mins: list[int] = [] + maxes: list[int] = [] + for caps in capabilities: + if caps is None: + continue + bounds = caps.length_bounds(credential_type) + if bounds is None: + continue + lo, hi = bounds + mins.append(lo) + if hi is not None: + maxes.append(hi) + return (max(mins) if mins else None, min(maxes) if maxes else None) + def credential_from_slot(slot: int, state: SlotCredential) -> Credential: """ diff --git a/custom_components/lock_code_manager/domain/slot_coordinator.py b/custom_components/lock_code_manager/domain/slot_coordinator.py index ad59eb79..0636e7c6 100644 --- a/custom_components/lock_code_manager/domain/slot_coordinator.py +++ b/custom_components/lock_code_manager/domain/slot_coordinator.py @@ -33,6 +33,7 @@ HomeAssistant, callback, ) +from homeassistant.exceptions import ServiceValidationError from homeassistant.helpers.event import async_track_state_change_event from homeassistant.helpers.issue_registry import ( IssueSeverity, @@ -42,6 +43,7 @@ from ..const import ATTR_IN_SYNC, DOMAIN, EVENT_PIN_USED from .config import EntryConfig +from .credentials import CredentialType from .queries import get_entry_config if TYPE_CHECKING: @@ -214,10 +216,21 @@ async def async_request_pin_update(self, value: str) -> None: Normalizing whitespace and the empty-PIN side effect (disabling the slot on an active slot whose PIN was cleared) live here so entities do not have to coordinate sibling state themselves. + + A non-empty PIN is validated against every bound lock's advertised + length range before it is written; an empty PIN clears the slot and + is exempt. This is the authoritative *minimum* gate: the text entity + keeps ``native_min`` permissive so Home Assistant's ``text.set_value`` + service neither rejects the empty clear nor pre-empts the per-lock + error built here. The maximum is additionally surfaced as the entity's + ``native_max`` ceiling, which Home Assistant does enforce. """ if not value.strip(): value = "" + if value: + self._validate_credential_length(value, CredentialType.PIN) + updates: dict[str, Any] = {CONF_PIN: value} if not value and self.is_enabled: _LOGGER.debug( @@ -228,6 +241,46 @@ async def async_request_pin_update(self, value: str) -> None: self._write_config_fields(updates) + def _validate_credential_length( + self, value: str, credential_type: CredentialType + ) -> None: + """ + Reject ``value`` if it violates any bound lock's length range. + + Authoritative gate for credential length. Iterates every bound lock so + the error names each offending lock with its required range. The lock + set is the entry-wide ``runtime_data.locks`` -- the same set the text + entity mirrors in ``self.locks`` to size its surfaced bounds, since LCM + binds every lock to every slot; a future per-slot binding must update + both sites together. Locks whose capabilities are not cached + (disconnected or not yet probed) and locks that do not advertise + ``credential_type`` are skipped -- the write proceeds rather than + blocking on unknown limits, and the sync layer surfaces any later + device rejection. + """ + length = len(value) + violations: list[str] = [] + for lock in self._config_entry.runtime_data.locks.values(): + caps = lock.cached_capabilities + if caps is None: + continue + bounds = caps.length_bounds(credential_type) + if bounds is None: + continue + lo, hi = bounds + if length < lo or (hi is not None and length > hi): + required = ( + f"at least {lo} characters" + if hi is None + else f"{lo}-{hi} characters" + ) + violations.append(f"{required} for {lock.display_name}") + if violations: + raise ServiceValidationError( + f"{credential_type.value.upper()} length {length} is not accepted " + f"by all locks: {'; '.join(violations)}" + ) + async def async_request_active_toggle(self, enabled: bool) -> None: """ Apply an enabled/disabled toggle requested by the switch entity. diff --git a/custom_components/lock_code_manager/providers/_base.py b/custom_components/lock_code_manager/providers/_base.py index b6b9572a..346c9a2f 100644 --- a/custom_components/lock_code_manager/providers/_base.py +++ b/custom_components/lock_code_manager/providers/_base.py @@ -1163,6 +1163,19 @@ async def async_get_usercodes(self) -> dict[int, SlotCredential]: """ return await self._project_users_to_slots(CredentialType.PIN) + @final + @property + def cached_capabilities(self) -> LockCapabilities | None: + """ + Return the already-probed capabilities, or ``None``. Never performs I/O. + + Synchronous read of the same cache ``_get_cached_capabilities`` + populates. Lets synchronous callers (e.g. the PIN text entity sizing + its length bounds) consult capabilities without awaiting; an unprobed + or disconnected lock reads ``None`` and contributes no constraint. + """ + return self._capabilities_cache + @final async def _get_cached_capabilities(self) -> LockCapabilities: """ diff --git a/custom_components/lock_code_manager/text.py b/custom_components/lock_code_manager/text.py index dfe0058a..de18c513 100644 --- a/custom_components/lock_code_manager/text.py +++ b/custom_components/lock_code_manager/text.py @@ -2,7 +2,9 @@ from __future__ import annotations +from collections.abc import Mapping import logging +from typing import TYPE_CHECKING from homeassistant.components.text import TextEntity, TextMode from homeassistant.const import CONF_NAME, CONF_PIN @@ -10,11 +12,24 @@ from homeassistant.helpers import entity_registry as er from homeassistant.helpers.entity_platform import AddEntitiesCallback +from .domain.credentials import CredentialType, aggregate_length_bounds +from .domain.exceptions import LockCodeManagerProviderError from .domain.models import LockCodeManagerConfigEntry from .entity import BaseLockCodeManagerEntity +if TYPE_CHECKING: + from .providers import BaseLock + _LOGGER = logging.getLogger(__name__) +# The single credential-type-specific knob. A text entity's ``key`` maps to a +# credential type when its value length is governed by a lock capability; +# keys absent here (for example the slot name) carry no length constraint. +# Adding password support later is a one-line entry once a password entity exists. +CREDENTIAL_TYPE_BY_CONF_KEY: Mapping[str, CredentialType] = { + CONF_PIN: CredentialType.PIN, +} + async def async_setup_entry( hass: HomeAssistant, @@ -46,8 +61,10 @@ def add_standard_text_entities(slot_num: int, ent_reg: er.EntityRegistry) -> Non class LockCodeManagerText(BaseLockCodeManagerEntity, TextEntity): """Text entity for lock code manager.""" - _attr_native_min = 0 - _attr_native_max = 9999 + # Defaults for keys with no length constraint (the slot name) and the + # fallback when bound locks advertise nothing or an unsatisfiable range. + _DEFAULT_MIN = 0 + _DEFAULT_MAX = 9999 def __init__( self, @@ -64,6 +81,103 @@ def __init__( ) self._attr_mode = text_mode + @property + def native_min(self) -> int: + """ + Return the minimum value length -- always the permissive default. + + The advertised per-lock minimum is deliberately NOT surfaced here. + Home Assistant's ``text.set_value`` service rejects + ``len(value) < native_min`` before the value reaches the coordinator, + which would block the empty string that clears a slot and would replace + the coordinator's per-lock error with a generic one. The coordinator + (``SlotEntityCoordinator._validate_credential_length``) is the + authoritative minimum gate; an empty PIN is exempt because it clears + the slot. + """ + return self._DEFAULT_MIN + + @property + def native_max(self) -> int: + """ + Return the maximum value length advertised by the bound locks. + + Unlike the minimum, the maximum is surfaced as a hard ceiling: no bound + lock accepts a longer value, so Home Assistant enforcing it at the + widget is correct rather than a bypass of the coordinator gate. It also + gives the frontend a ``maxlength`` so over-long input is prevented + rather than merely rejected. + """ + return self._max_bound() + + def _max_bound(self) -> int: + """ + Compute the live tightest-common maximum length across the bound locks. + + Reads each lock's synchronously cached capabilities (uncached or + disconnected locks contribute nothing). Non-credential keys and an + all-unknown set fall back to the default ceiling. + """ + credential_type = CREDENTIAL_TYPE_BY_CONF_KEY.get(self.key) + if credential_type is None: + return self._DEFAULT_MAX + # ``self.locks`` mirrors the entry-wide ``runtime_data.locks`` that the + # coordinator's length gate iterates -- LCM binds every lock to every + # slot. If per-slot lock binding is ever added, this surface and the + # coordinator gate must both switch to the slot's subset together. + _lo, hi = aggregate_length_bounds( + (lock.cached_capabilities for lock in self.locks), credential_type + ) + hi = self._DEFAULT_MAX if hi is None else hi + # Home Assistant validates the stored value against this ceiling when it + # renders state and raises if the value is longer, so the ceiling must + # admit a PIN written before a (now tighter) lock advertised its limit. + # New out-of-range input is rejected by the coordinator gate, not here. + value = self.native_value + if value is not None: + hi = max(hi, len(value)) + return hi + + @callback + def _handle_add_locks(self, locks: list[BaseLock]) -> None: + """Refresh advertised bounds when locks are added to the slot.""" + super()._handle_add_locks(locks) + self._write_bounds_update() + # A freshly added lock has not been probed, so its advertised maximum is + # unknown and contributes nothing yet. Probe in the background and + # re-push once known, otherwise native_max stays at the default ceiling + # until some later state write happens to re-read the warmed cache. + if self.hass is not None: + self.config_entry.async_create_background_task( + self.hass, + self._probe_and_refresh_bounds(locks), + f"{self.entity_id} refresh length bounds", + ) + + @callback + def _handle_remove_lock(self, lock_entity_id: str) -> None: + """Refresh advertised bounds when a lock is removed from the slot.""" + super()._handle_remove_lock(lock_entity_id) + self._write_bounds_update() + + async def _probe_and_refresh_bounds(self, locks: list[BaseLock]) -> None: + """Populate newly added locks' capabilities, then re-push the bounds.""" + for lock in locks: + try: + await lock._get_cached_capabilities() + except LockCodeManagerProviderError: + # A lock that cannot be probed (disconnected, operation failed, + # or no capability support) advertises no ceiling; bounds stay + # at the default until a later write re-reads a warmed cache. + continue + self._write_bounds_update() + + @callback + def _write_bounds_update(self) -> None: + """Re-push state so the frontend re-reads the length bounds.""" + if self.hass is not None and self.entity_id: + self.async_write_ha_state() + @property def native_value(self) -> str | None: """Return native value.""" diff --git a/tests/providers/test_base.py b/tests/providers/test_base.py index 45f7d3b6..32bace04 100644 --- a/tests/providers/test_base.py +++ b/tests/providers/test_base.py @@ -20,6 +20,11 @@ DOMAIN, EVENT_LOCK_STATE_CHANGED, ) +from custom_components.lock_code_manager.domain.credentials import ( + CredentialType, + CredentialTypeCapability, + LockCapabilities, +) from custom_components.lock_code_manager.domain.exceptions import ( DuplicateCodeError, LockDisconnected, @@ -56,6 +61,39 @@ def teardown_push_subscription(self) -> None: self.unsubscribe_calls += 1 +class _CapsLock(MockLCMLock): + """Mock lock that advertises PIN capabilities.""" + + async def async_get_capabilities(self) -> LockCapabilities: + """Report a single PIN credential type with a 4-8 length range.""" + return LockCapabilities( + supports_user_management=True, + max_users=30, + credential_types={ + CredentialType.PIN: CredentialTypeCapability( + num_slots=30, min_length=4, max_length=8, supports_learn=False + ) + }, + ) + + +async def test_cached_capabilities_exposes_warmed_cache(hass: HomeAssistant): + """cached_capabilities is None until probed, then returns the cached snapshot.""" + entity_reg = er.async_get(hass) + config_entry = MockConfigEntry(domain=DOMAIN) + config_entry.add_to_hass(hass) + lock_entity = entity_reg.async_get_or_create( + "lock", "test", "caps_lock", config_entry=config_entry + ) + lock = _CapsLock(hass, dr.async_get(hass), entity_reg, config_entry, lock_entity) + + assert lock.cached_capabilities is None + + caps = await lock._get_cached_capabilities() + assert lock.cached_capabilities is caps + assert lock.cached_capabilities.length_bounds(CredentialType.PIN) == (4, 8) + + async def test_base(hass: HomeAssistant): """Test base class.""" entity_reg = er.async_get(hass) diff --git a/tests/test_credentials.py b/tests/test_credentials.py index c7551333..c3bd2f0c 100644 --- a/tests/test_credentials.py +++ b/tests/test_credentials.py @@ -13,6 +13,7 @@ SetUserResult, User, UserType, + aggregate_length_bounds, credential_from_slot, slot_credential_of, user_from_slot, @@ -276,6 +277,73 @@ def test_credential_types_is_snapshotted(self) -> None: caps.credential_types[CredentialType.RFID] = pin_cap # type: ignore[index] +def _caps(min_length: int, max_length: int) -> LockCapabilities: + """Build a single-PIN-type LockCapabilities with the given length bounds.""" + return LockCapabilities( + supports_user_management=True, + max_users=30, + credential_types={ + CredentialType.PIN: CredentialTypeCapability( + num_slots=30, + min_length=min_length, + max_length=max_length, + supports_learn=False, + ) + }, + ) + + +class TestLengthBounds: + """LockCapabilities.length_bounds normalizes per-type length limits.""" + + def test_returns_min_and_max_for_supported_type(self) -> None: + assert _caps(4, 8).length_bounds(CredentialType.PIN) == (4, 8) + + def test_unsupported_type_returns_none(self) -> None: + assert _caps(4, 8).length_bounds(CredentialType.RFID) is None + + def test_non_positive_max_means_unbounded(self) -> None: + # Matter reports max_pin_length as `... or 0` -- 0 is "unknown", not + # "zero characters", so it must normalize to no upper bound. + assert _caps(4, 0).length_bounds(CredentialType.PIN) == (4, None) + + def test_negative_min_clamps_to_zero(self) -> None: + assert _caps(-1, 8).length_bounds(CredentialType.PIN) == (0, 8) + + +class TestAggregateLengthBounds: + """aggregate_length_bounds folds many locks into one tightest-common range.""" + + def test_no_capabilities_is_unconstrained(self) -> None: + assert aggregate_length_bounds([], CredentialType.PIN) == (None, None) + + def test_skips_none_and_unsupported_caps(self) -> None: + unsupported = LockCapabilities( + supports_user_management=False, max_users=0, credential_types={} + ) + assert aggregate_length_bounds([None, unsupported], CredentialType.PIN) == ( + None, + None, + ) + + def test_single_lock_passes_bounds_through(self) -> None: + assert aggregate_length_bounds([_caps(4, 8)], CredentialType.PIN) == (4, 8) + + def test_tightest_common_takes_max_of_mins_and_min_of_maxes(self) -> None: + assert aggregate_length_bounds( + [_caps(4, 8), _caps(6, 10)], CredentialType.PIN + ) == (6, 8) + + def test_unbounded_max_does_not_constrain_aggregate_max(self) -> None: + assert aggregate_length_bounds( + [_caps(4, 0), _caps(6, 8)], CredentialType.PIN + ) == (6, 8) + + def test_empty_intersection_returns_min_greater_than_max(self) -> None: + lo, hi = aggregate_length_bounds([_caps(6, 6), _caps(4, 4)], CredentialType.PIN) + assert lo is not None and hi is not None and lo > hi + + class TestProjectionHelpers: """Pure 1:1:1 projection between a managed slot and the User/Credential model.""" diff --git a/tests/test_slot_coordinator.py b/tests/test_slot_coordinator.py index a3ca6396..a6bdf0a2 100644 --- a/tests/test_slot_coordinator.py +++ b/tests/test_slot_coordinator.py @@ -29,7 +29,7 @@ STATE_ON, ) from homeassistant.core import HomeAssistant, callback -from homeassistant.exceptions import HomeAssistantError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError from homeassistant.helpers.issue_registry import async_get as async_get_issue_registry from custom_components.lock_code_manager.binary_sensor import ( @@ -40,6 +40,11 @@ CONF_SLOTS, DOMAIN, ) +from custom_components.lock_code_manager.domain.credentials import ( + CredentialType, + CredentialTypeCapability, + LockCapabilities, +) from custom_components.lock_code_manager.domain.queries import get_entry_config from custom_components.lock_code_manager.domain.slot_coordinator import ( PinRequiredError, @@ -48,6 +53,7 @@ from .common import ( LOCK_1_ENTITY_ID, + LOCK_2_ENTITY_ID, SLOT_1_ACTIVE_ENTITY, SLOT_1_ENABLED_ENTITY, SLOT_1_PIN_ENTITY, @@ -59,6 +65,168 @@ _LOGGER = logging.getLogger(__name__) +def _pin_caps(min_length: int, max_length: int) -> LockCapabilities: + """Build LockCapabilities advertising a PIN type with the given bounds.""" + return LockCapabilities( + supports_user_management=True, + max_users=30, + credential_types={ + CredentialType.PIN: CredentialTypeCapability( + num_slots=30, + min_length=min_length, + max_length=max_length, + supports_learn=False, + ) + }, + ) + + +async def test_request_pin_update_rejects_out_of_range_pin( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """A PIN shorter than the lock minimum is rejected and never written.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + coordinator = runtime_data.slot_coordinators[1] + + with pytest.raises(ServiceValidationError): + await coordinator.async_request_pin_update("12") + + # The rejected PIN must not reach config. + assert get_entry_config(lock_code_manager_config_entry).slot(1).get(CONF_PIN) == ( + "1234" + ) + + +async def test_request_pin_update_rejects_too_long_pin( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """A PIN longer than the lock maximum is rejected.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + coordinator = runtime_data.slot_coordinators[1] + + with pytest.raises(ServiceValidationError): + await coordinator.async_request_pin_update("123456789") + + +async def test_request_pin_update_allows_in_range_pin( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """A PIN within the advertised range is written normally.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + coordinator = runtime_data.slot_coordinators[1] + + await coordinator.async_request_pin_update("567890") + await hass.async_block_till_done() + + assert get_entry_config(lock_code_manager_config_entry).slot(1).get(CONF_PIN) == ( + "567890" + ) + + +async def test_request_pin_update_empty_pin_exempt_from_length( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """Clearing the PIN is allowed even when the lock requires a minimum length.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + coordinator = runtime_data.slot_coordinators[1] + + await coordinator.async_request_pin_update("") + await hass.async_block_till_done() + + assert get_entry_config(lock_code_manager_config_entry).slot(1).get(CONF_PIN) == "" + + +async def test_request_pin_update_fails_open_without_capabilities( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """Locks with unknown capabilities do not block a write.""" + coordinator = lock_code_manager_config_entry.runtime_data.slot_coordinators[1] + + await coordinator.async_request_pin_update("12") + await hass.async_block_till_done() + + assert ( + get_entry_config(lock_code_manager_config_entry).slot(1).get(CONF_PIN) == "12" + ) + + +async def test_validation_names_each_offending_lock( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """The rejection message names every lock the PIN violates.""" + runtime_data = lock_code_manager_config_entry.runtime_data + runtime_data.locks[LOCK_1_ENTITY_ID]._capabilities_cache = _pin_caps(6, 8) + runtime_data.locks[LOCK_2_ENTITY_ID]._capabilities_cache = _pin_caps(6, 8) + coordinator = runtime_data.slot_coordinators[1] + + with pytest.raises(ServiceValidationError) as exc: + await coordinator.async_request_pin_update("12") + + message = str(exc.value) + assert runtime_data.locks[LOCK_1_ENTITY_ID].display_name in message + assert runtime_data.locks[LOCK_2_ENTITY_ID].display_name in message + + +async def test_request_pin_update_accepts_boundary_lengths( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """A PIN exactly at the min or max is accepted; one past either end is rejected.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + coordinator = runtime_data.slot_coordinators[1] + + for ok in ("1234", "12345678"): # exactly the min (4) and the max (8) + await coordinator.async_request_pin_update(ok) + await hass.async_block_till_done() + assert ( + get_entry_config(lock_code_manager_config_entry).slot(1).get(CONF_PIN) == ok + ) + + for bad in ("123", "123456789"): # one under the min and one over the max + with pytest.raises(ServiceValidationError): + await coordinator.async_request_pin_update(bad) + + +async def test_validation_message_unbounded_max_says_at_least( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """A lock advertising a minimum but no maximum yields an 'at least N' message.""" + runtime_data = lock_code_manager_config_entry.runtime_data + for lock in runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(6, 0) # max 0 == unbounded + coordinator = runtime_data.slot_coordinators[1] + + with pytest.raises(ServiceValidationError) as exc: + await coordinator.async_request_pin_update("12") + + assert "at least 6 characters" in str(exc.value) + + async def test_coordinator_registered_for_each_slot( hass: HomeAssistant, mock_lock_config_entry, diff --git a/tests/test_text.py b/tests/test_text.py index 6418950e..41557573 100644 --- a/tests/test_text.py +++ b/tests/test_text.py @@ -1,20 +1,254 @@ """Test text platform.""" import logging +from types import SimpleNamespace +from unittest.mock import PropertyMock, patch + +from pytest_homeassistant_custom_component.common import MockConfigEntry from homeassistant.components.text import ( + ATTR_MAX, + ATTR_MIN, ATTR_VALUE, DOMAIN as TEXT_DOMAIN, SERVICE_SET_VALUE, + TextMode, ) -from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF +from homeassistant.const import ATTR_ENTITY_ID, CONF_NAME, CONF_PIN, STATE_OFF from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er + +from custom_components.lock_code_manager.const import DOMAIN +from custom_components.lock_code_manager.domain.credentials import ( + CredentialType, + CredentialTypeCapability, + LockCapabilities, +) +from custom_components.lock_code_manager.domain.models import ( + LockCodeManagerConfigEntryRuntimeData, +) +from custom_components.lock_code_manager.text import ( + CREDENTIAL_TYPE_BY_CONF_KEY, + LockCodeManagerText, +) from .common import SLOT_2_ENABLED_ENTITY, SLOT_2_NAME_ENTITY, SLOT_2_PIN_ENTITY _LOGGER = logging.getLogger(__name__) +def _pin_caps(min_length: int, max_length: int) -> LockCapabilities: + """Build LockCapabilities advertising a PIN type with the given bounds.""" + return LockCapabilities( + supports_user_management=True, + max_users=30, + credential_types={ + CredentialType.PIN: CredentialTypeCapability( + num_slots=30, + min_length=min_length, + max_length=max_length, + supports_learn=False, + ) + }, + ) + + +def _fake_lock(entity_id: str, caps: LockCapabilities | None): + """A stand-in lock exposing only what the text entity reads.""" + + async def _get_cached_capabilities() -> LockCapabilities | None: + """Stand in for the async probe the add hook runs in the background.""" + return caps + + return SimpleNamespace( + cached_capabilities=caps, + lock=SimpleNamespace(entity_id=entity_id), + _get_cached_capabilities=_get_cached_capabilities, + ) + + +def _make_text_entity( + hass: HomeAssistant, key: str, locks: list +) -> LockCodeManagerText: + """Construct a text entity with a controlled lock list for bounds tests.""" + config_entry = MockConfigEntry(domain=DOMAIN, title="Test") + config_entry.add_to_hass(hass) + config_entry.runtime_data = LockCodeManagerConfigEntryRuntimeData() + + entity = LockCodeManagerText( + hass, + er.async_get(hass), + config_entry, + 1, + key, + TextMode.PASSWORD if key == CONF_PIN else TextMode.TEXT, + ) + entity.locks = locks + return entity + + +def test_conf_key_credential_type_map() -> None: + """The map exposes PIN and excludes the (non-credential) name key.""" + assert CREDENTIAL_TYPE_BY_CONF_KEY[CONF_PIN] is CredentialType.PIN + assert CONF_NAME not in CREDENTIAL_TYPE_BY_CONF_KEY + + +def test_pin_bounds_default_without_capabilities(hass: HomeAssistant) -> None: + """An uncached lock contributes no constraint; bounds stay 0/9999.""" + entity = _make_text_entity(hass, CONF_PIN, [_fake_lock("lock.a", None)]) + assert (entity.native_min, entity.native_max) == (0, 9999) + + +def test_pin_max_reflects_single_lock(hass: HomeAssistant) -> None: + """A lock advertising 4-8 surfaces a max of 8; the min stays permissive.""" + entity = _make_text_entity(hass, CONF_PIN, [_fake_lock("lock.a", _pin_caps(4, 8))]) + assert (entity.native_min, entity.native_max) == (0, 8) + + +def test_pin_max_takes_tightest_common(hass: HomeAssistant) -> None: + """Two locks collapse to the smallest advertised maximum.""" + entity = _make_text_entity( + hass, + CONF_PIN, + [_fake_lock("lock.a", _pin_caps(4, 8)), _fake_lock("lock.b", _pin_caps(6, 10))], + ) + assert (entity.native_min, entity.native_max) == (0, 8) + + +def test_native_min_stays_zero_despite_advertised_minimum(hass: HomeAssistant) -> None: + """The advertised minimum is never surfaced as a hard floor. + + Surfacing it would make HA's text service reject the empty string that + clears a slot; the coordinator owns the minimum instead. + """ + entity = _make_text_entity(hass, CONF_PIN, [_fake_lock("lock.a", _pin_caps(6, 8))]) + assert entity.native_min == 0 + assert entity.native_max == 8 + + +def test_native_max_widens_to_admit_longer_stored_value(hass: HomeAssistant) -> None: + """A stored PIN longer than the advertised max still renders (ceiling widens). + + HA raises at state-render time if the value exceeds native_max, so a PIN + written before a (now tighter) lock advertised its limit forces the ceiling + up to admit it. + """ + entity = _make_text_entity(hass, CONF_PIN, [_fake_lock("lock.a", _pin_caps(4, 8))]) + with patch.object( + LockCodeManagerText, + "native_value", + new_callable=PropertyMock, + return_value="1234567890", + ): + assert entity.native_min == 0 + assert entity.native_max == 10 # widened up to admit the length-10 value + + +def test_name_entity_ignores_capabilities(hass: HomeAssistant) -> None: + """The name entity is not a credential key; bounds stay 0/9999.""" + entity = _make_text_entity(hass, CONF_NAME, [_fake_lock("lock.a", _pin_caps(4, 8))]) + assert (entity.native_min, entity.native_max) == (0, 9999) + + +async def test_lock_add_remove_rewrites_state(hass: HomeAssistant) -> None: + """Lock set changes re-push state so the frontend re-reads bounds.""" + entity = _make_text_entity(hass, CONF_PIN, []) + entity.hass = hass + entity.entity_id = "text.test" + + with patch.object(entity, "async_write_ha_state") as mock_write: + added = _fake_lock("lock.a", _pin_caps(4, 8)) + entity._handle_add_locks([added]) + await hass.async_block_till_done() + assert added in entity.locks + # The added lock's max is surfaced; the min stays permissive. + assert (entity.native_min, entity.native_max) == (0, 8) + assert mock_write.called # immediate re-push, plus one after the probe + mock_write.reset_mock() + + entity._handle_remove_lock("lock.a") + await hass.async_block_till_done() + assert entity.locks == [] + # Removing the only lock reverts the surfaced ceiling to the default. + assert (entity.native_min, entity.native_max) == (0, 9999) + assert mock_write.called # the removal re-pushed state + + +async def test_pin_entity_surfaces_lock_bounds( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """The PIN entity's state exposes the bound lock's length range.""" + # Without advertised capabilities the entity uses the default range. + # Home Assistant clamps the reported max to its 255-char state ceiling. + state = hass.states.get(SLOT_2_PIN_ENTITY) + assert state + assert state.attributes[ATTR_MIN] == 0 + assert state.attributes[ATTR_MAX] == 255 + + # Warm the lock's capability cache, then a write re-reads the bounds. + for lock in lock_code_manager_config_entry.runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(4, 8) + + await hass.services.async_call( + TEXT_DOMAIN, + SERVICE_SET_VALUE, + service_data={ATTR_VALUE: "1234"}, + target={ATTR_ENTITY_ID: SLOT_2_PIN_ENTITY}, + blocking=True, + ) + + state = hass.states.get(SLOT_2_PIN_ENTITY) + assert state + # The minimum is owned by the coordinator, not surfaced as a hard floor. + assert state.attributes[ATTR_MIN] == 0 + assert state.attributes[ATTR_MAX] == 8 + + +async def test_pin_clear_through_service_with_minimum_advertised( + hass: HomeAssistant, + mock_lock_config_entry, + lock_code_manager_config_entry, +): + """Clearing a PIN via text.set_value works even when locks advertise a minimum. + + Regression: surfacing the advertised minimum as ``native_min`` made HA's + text service reject the empty string (``len 0 < min``) before the + coordinator's empty-PIN exemption ran, so a slot could not be cleared. + """ + for lock in lock_code_manager_config_entry.runtime_data.locks.values(): + lock._capabilities_cache = _pin_caps(6, 8) + + # An in-range PIN goes through the service normally. + await hass.services.async_call( + TEXT_DOMAIN, + SERVICE_SET_VALUE, + service_data={ATTR_VALUE: "654321"}, + target={ATTR_ENTITY_ID: SLOT_2_PIN_ENTITY}, + blocking=True, + ) + state = hass.states.get(SLOT_2_PIN_ENTITY) + assert state + assert state.state == "654321" + + # Clearing must reach the coordinator (empty is exempt) rather than being + # rejected by HA's service-level minimum check. + await hass.services.async_call( + TEXT_DOMAIN, + SERVICE_SET_VALUE, + service_data={ATTR_VALUE: ""}, + target={ATTR_ENTITY_ID: SLOT_2_PIN_ENTITY}, + blocking=True, + ) + state = hass.states.get(SLOT_2_PIN_ENTITY) + assert state + assert state.state == "" + state = hass.states.get(SLOT_2_ENABLED_ENTITY) + assert state + assert state.state == STATE_OFF + + async def test_text_entities( hass: HomeAssistant, mock_lock_config_entry,