Add validation for lock component (#85842)

* Add validation for lock integration

* Add LockEntityFeature.OPEN for lock group

* Correct tests google_assistant for extra entity

* Validate feature when registering service

* Update tests

* Add LockFeature.OPEN with group
This commit is contained in:
Jan Bouwhuis 2023-01-23 13:08:00 +01:00 committed by GitHub
parent ea43effcc9
commit b03677db1c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 383 additions and 22 deletions

View File

@ -6,7 +6,12 @@ from typing import Any
import voluptuous as vol import voluptuous as vol
from homeassistant.components.lock import DOMAIN, PLATFORM_SCHEMA, LockEntity from homeassistant.components.lock import (
DOMAIN,
PLATFORM_SCHEMA,
LockEntity,
LockEntityFeature,
)
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_ENTITY_ID,
@ -100,6 +105,7 @@ class LockGroup(GroupEntity, LockEntity):
) -> None: ) -> None:
"""Initialize a lock group.""" """Initialize a lock group."""
self._entity_ids = entity_ids self._entity_ids = entity_ids
self._attr_supported_features = LockEntityFeature.OPEN
self._attr_name = name self._attr_name = name
self._attr_extra_state_attributes = {ATTR_ENTITY_ID: entity_ids} self._attr_extra_state_attributes = {ATTR_ENTITY_ID: entity_ids}

View File

@ -25,9 +25,7 @@ import homeassistant.util.dt as dt_util
DOMAIN = "kitchen_sink" DOMAIN = "kitchen_sink"
COMPONENTS_WITH_DEMO_PLATFORM = [ COMPONENTS_WITH_DEMO_PLATFORM = [Platform.SENSOR, Platform.LOCK]
Platform.SENSOR,
]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:

View File

@ -0,0 +1,92 @@
"""Demo platform that has a couple of fake locks."""
from __future__ import annotations
from typing import Any
from homeassistant.components.lock import LockEntity, LockEntityFeature
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_LOCKED, STATE_UNLOCKED, STATE_UNLOCKING
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Demo sensors."""
async_add_entities(
[
DemoLock(
"kitchen_sink_lock_001",
"Openable kitchen sink lock",
STATE_LOCKED,
LockEntityFeature.OPEN,
),
DemoLock(
"kitchen_sink_lock_002",
"Another kitchen sink openable lock",
STATE_UNLOCKED,
LockEntityFeature.OPEN,
),
DemoLock(
"kitchen_sink_lock_003",
"Basic kitchen sink lock",
STATE_LOCKED,
),
DemoLock(
"kitchen_sink_lock_004",
"Another kitchen sink lock",
STATE_UNLOCKED,
),
]
)
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Everything but the Kitchen Sink config entry."""
await async_setup_platform(hass, {}, async_add_entities)
class DemoLock(LockEntity):
"""Representation of a Demo lock."""
def __init__(
self,
unique_id: str,
name: str,
state: str,
features: LockEntityFeature = LockEntityFeature(0),
) -> None:
"""Initialize the sensor."""
self._attr_name = name
self._attr_unique_id = unique_id
self._attr_supported_features = features
self._state = state
@property
def is_locked(self) -> bool:
"""Return true if lock is locked."""
return self._state == STATE_LOCKED
async def async_lock(self, **kwargs: Any) -> None:
"""Lock the device."""
self._state = STATE_LOCKED
self.async_write_ha_state()
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the device."""
self._state = STATE_UNLOCKING
self.async_write_ha_state()
async def async_open(self, **kwargs: Any) -> None:
"""Open the door latch."""
self._state = STATE_UNLOCKED
self.async_write_ha_state()

View File

@ -6,6 +6,7 @@ from datetime import timedelta
from enum import IntFlag from enum import IntFlag
import functools as ft import functools as ft
import logging import logging
import re
from typing import Any, final from typing import Any, final
import voluptuous as vol import voluptuous as vol
@ -23,7 +24,7 @@ from homeassistant.const import (
STATE_UNLOCKED, STATE_UNLOCKED,
STATE_UNLOCKING, STATE_UNLOCKING,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant, ServiceCall
import homeassistant.helpers.config_validation as cv import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.config_validation import ( # noqa: F401 from homeassistant.helpers.config_validation import ( # noqa: F401
PLATFORM_SCHEMA, PLATFORM_SCHEMA,
@ -72,18 +73,48 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
await component.async_setup(config) await component.async_setup(config)
component.async_register_entity_service( component.async_register_entity_service(
SERVICE_UNLOCK, LOCK_SERVICE_SCHEMA, "async_unlock" SERVICE_UNLOCK, LOCK_SERVICE_SCHEMA, _async_unlock
) )
component.async_register_entity_service( component.async_register_entity_service(
SERVICE_LOCK, LOCK_SERVICE_SCHEMA, "async_lock" SERVICE_LOCK, LOCK_SERVICE_SCHEMA, _async_lock
) )
component.async_register_entity_service( component.async_register_entity_service(
SERVICE_OPEN, LOCK_SERVICE_SCHEMA, "async_open" SERVICE_OPEN, LOCK_SERVICE_SCHEMA, _async_open, [LockEntityFeature.OPEN]
) )
return True return True
async def _async_lock(entity: LockEntity, service_call: ServiceCall) -> None:
"""Lock the lock."""
code: str = service_call.data.get(ATTR_CODE, "")
if entity.code_format_cmp and not entity.code_format_cmp.match(code):
raise ValueError(
f"Code '{code}' for locking {entity.name} doesn't match pattern {entity.code_format}"
)
await entity.async_lock(**service_call.data)
async def _async_unlock(entity: LockEntity, service_call: ServiceCall) -> None:
"""Unlock the lock."""
code: str = service_call.data.get(ATTR_CODE, "")
if entity.code_format_cmp and not entity.code_format_cmp.match(code):
raise ValueError(
f"Code '{code}' for unlocking {entity.name} doesn't match pattern {entity.code_format}"
)
await entity.async_unlock(**service_call.data)
async def _async_open(entity: LockEntity, service_call: ServiceCall) -> None:
"""Open the door latch."""
code: str = service_call.data.get(ATTR_CODE, "")
if entity.code_format_cmp and not entity.code_format_cmp.match(code):
raise ValueError(
f"Code '{code}' for opening {entity.name} doesn't match pattern {entity.code_format}"
)
await entity.async_open(**service_call.data)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry.""" """Set up a config entry."""
component: EntityComponent[LockEntity] = hass.data[DOMAIN] component: EntityComponent[LockEntity] = hass.data[DOMAIN]
@ -113,6 +144,7 @@ class LockEntity(Entity):
_attr_is_jammed: bool | None = None _attr_is_jammed: bool | None = None
_attr_state: None = None _attr_state: None = None
_attr_supported_features: LockEntityFeature = LockEntityFeature(0) _attr_supported_features: LockEntityFeature = LockEntityFeature(0)
__code_format_cmp: re.Pattern[str] | None = None
@property @property
def changed_by(self) -> str | None: def changed_by(self) -> str | None:
@ -124,6 +156,20 @@ class LockEntity(Entity):
"""Regex for code format or None if no code is required.""" """Regex for code format or None if no code is required."""
return self._attr_code_format return self._attr_code_format
@property
@final
def code_format_cmp(self) -> re.Pattern[str] | None:
"""Return a compiled code_format."""
if self.code_format is None:
self.__code_format_cmp = None
return None
if (
not self.__code_format_cmp
or self.code_format != self.__code_format_cmp.pattern
):
self.__code_format_cmp = re.compile(self.code_format)
return self.__code_format_cmp
@property @property
def is_locked(self) -> bool | None: def is_locked(self) -> bool | None:
"""Return true if the lock is locked.""" """Return true if the lock is locked."""

View File

@ -1,6 +1,9 @@
"""The tests for the Group Lock platform.""" """The tests for the Group Lock platform."""
from unittest.mock import patch from unittest.mock import patch
import pytest
from homeassistant import config as hass_config from homeassistant import config as hass_config
from homeassistant.components.demo import lock as demo_lock from homeassistant.components.demo import lock as demo_lock
from homeassistant.components.group import DOMAIN, SERVICE_RELOAD from homeassistant.components.group import DOMAIN, SERVICE_RELOAD
@ -20,6 +23,8 @@ from homeassistant.const import (
STATE_UNLOCKED, STATE_UNLOCKED,
STATE_UNLOCKING, STATE_UNLOCKING,
) )
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@ -167,20 +172,19 @@ async def test_state_reporting(hass):
assert hass.states.get("lock.lock_group").state == STATE_UNAVAILABLE assert hass.states.get("lock.lock_group").state == STATE_UNAVAILABLE
@patch.object(demo_lock, "LOCK_UNLOCK_DELAY", 0) async def test_service_calls_openable(hass: HomeAssistant) -> None:
async def test_service_calls(hass, enable_custom_integrations): """Test service calls with open support."""
"""Test service calls."""
await async_setup_component( await async_setup_component(
hass, hass,
LOCK_DOMAIN, LOCK_DOMAIN,
{ {
LOCK_DOMAIN: [ LOCK_DOMAIN: [
{"platform": "demo"}, {"platform": "kitchen_sink"},
{ {
"platform": DOMAIN, "platform": DOMAIN,
"entities": [ "entities": [
"lock.front_door", "lock.openable_kitchen_sink_lock",
"lock.kitchen_door", "lock.another_kitchen_sink_openable_lock",
], ],
}, },
] ]
@ -190,8 +194,11 @@ async def test_service_calls(hass, enable_custom_integrations):
group_state = hass.states.get("lock.lock_group") group_state = hass.states.get("lock.lock_group")
assert group_state.state == STATE_UNLOCKED assert group_state.state == STATE_UNLOCKED
assert hass.states.get("lock.front_door").state == STATE_LOCKED assert hass.states.get("lock.openable_kitchen_sink_lock").state == STATE_LOCKED
assert hass.states.get("lock.kitchen_door").state == STATE_UNLOCKED assert (
hass.states.get("lock.another_kitchen_sink_openable_lock").state
== STATE_UNLOCKED
)
await hass.services.async_call( await hass.services.async_call(
LOCK_DOMAIN, LOCK_DOMAIN,
@ -199,8 +206,11 @@ async def test_service_calls(hass, enable_custom_integrations):
{ATTR_ENTITY_ID: "lock.lock_group"}, {ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True, blocking=True,
) )
assert hass.states.get("lock.front_door").state == STATE_UNLOCKED assert hass.states.get("lock.openable_kitchen_sink_lock").state == STATE_UNLOCKED
assert hass.states.get("lock.kitchen_door").state == STATE_UNLOCKED assert (
hass.states.get("lock.another_kitchen_sink_openable_lock").state
== STATE_UNLOCKED
)
await hass.services.async_call( await hass.services.async_call(
LOCK_DOMAIN, LOCK_DOMAIN,
@ -208,8 +218,10 @@ async def test_service_calls(hass, enable_custom_integrations):
{ATTR_ENTITY_ID: "lock.lock_group"}, {ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True, blocking=True,
) )
assert hass.states.get("lock.front_door").state == STATE_LOCKED assert hass.states.get("lock.openable_kitchen_sink_lock").state == STATE_LOCKED
assert hass.states.get("lock.kitchen_door").state == STATE_LOCKED assert (
hass.states.get("lock.another_kitchen_sink_openable_lock").state == STATE_LOCKED
)
await hass.services.async_call( await hass.services.async_call(
LOCK_DOMAIN, LOCK_DOMAIN,
@ -217,8 +229,63 @@ async def test_service_calls(hass, enable_custom_integrations):
{ATTR_ENTITY_ID: "lock.lock_group"}, {ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True, blocking=True,
) )
assert hass.states.get("lock.front_door").state == STATE_UNLOCKED assert hass.states.get("lock.openable_kitchen_sink_lock").state == STATE_UNLOCKED
assert hass.states.get("lock.kitchen_door").state == STATE_UNLOCKED assert (
hass.states.get("lock.another_kitchen_sink_openable_lock").state
== STATE_UNLOCKED
)
async def test_service_calls_basic(hass: HomeAssistant) -> None:
"""Test service calls without open support."""
await async_setup_component(
hass,
LOCK_DOMAIN,
{
LOCK_DOMAIN: [
{"platform": "kitchen_sink"},
{
"platform": DOMAIN,
"entities": [
"lock.basic_kitchen_sink_lock",
"lock.another_kitchen_sink_lock",
],
},
]
},
)
await hass.async_block_till_done()
group_state = hass.states.get("lock.lock_group")
assert group_state.state == STATE_UNLOCKED
assert hass.states.get("lock.basic_kitchen_sink_lock").state == STATE_LOCKED
assert hass.states.get("lock.another_kitchen_sink_lock").state == STATE_UNLOCKED
await hass.services.async_call(
LOCK_DOMAIN,
SERVICE_LOCK,
{ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True,
)
assert hass.states.get("lock.basic_kitchen_sink_lock").state == STATE_LOCKED
assert hass.states.get("lock.another_kitchen_sink_lock").state == STATE_LOCKED
await hass.services.async_call(
LOCK_DOMAIN,
SERVICE_UNLOCK,
{ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True,
)
assert hass.states.get("lock.basic_kitchen_sink_lock").state == STATE_UNLOCKED
assert hass.states.get("lock.another_kitchen_sink_lock").state == STATE_UNLOCKED
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
LOCK_DOMAIN,
SERVICE_OPEN,
{ATTR_ENTITY_ID: "lock.lock_group"},
blocking=True,
)
async def test_reload(hass): async def test_reload(hass):

View File

@ -0,0 +1,152 @@
"""The tests for the lock component."""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
import pytest
from homeassistant.components.lock import (
ATTR_CODE,
DOMAIN,
SERVICE_LOCK,
SERVICE_OPEN,
SERVICE_UNLOCK,
STATE_JAMMED,
STATE_LOCKED,
STATE_LOCKING,
STATE_UNLOCKED,
STATE_UNLOCKING,
LockEntity,
LockEntityFeature,
_async_lock,
_async_open,
_async_unlock,
)
from homeassistant.core import HomeAssistant, ServiceCall
class MockLockEntity(LockEntity):
"""Mock lock to use in tests."""
def __init__(
self,
code_format: str | None = None,
supported_features: LockEntityFeature = LockEntityFeature(0),
) -> None:
"""Initialize mock lock entity."""
self._attr_supported_features = supported_features
self.calls_open = MagicMock()
if code_format is not None:
self._attr_code_format = code_format
async def async_lock(self, **kwargs: Any) -> None:
"""Lock the lock."""
self._attr_is_locking = False
self._attr_is_locked = True
async def async_unlock(self, **kwargs: Any) -> None:
"""Unlock the lock."""
self._attr_is_unlocking = False
self._attr_is_locked = False
async def async_open(self, **kwargs: Any) -> None:
"""Open the door latch."""
self.calls_open(kwargs)
async def test_lock_default(hass: HomeAssistant) -> None:
"""Test lock entity with defaults."""
lock = MockLockEntity()
lock.hass = hass
assert lock.code_format is None
assert lock.state is None
async def test_lock_states(hass: HomeAssistant) -> None:
"""Test lock entity states."""
# pylint: disable=protected-access
lock = MockLockEntity()
lock.hass = hass
assert lock.state is None
lock._attr_is_locking = True
assert lock.is_locking
assert lock.state == STATE_LOCKING
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_LOCK, {}))
assert lock.is_locked
assert lock.state == STATE_LOCKED
lock._attr_is_unlocking = True
assert lock.is_unlocking
assert lock.state == STATE_UNLOCKING
await _async_unlock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {}))
assert not lock.is_locked
assert lock.state == STATE_UNLOCKED
lock._attr_is_jammed = True
assert lock.is_jammed
assert lock.state == STATE_JAMMED
assert not lock.is_locked
async def test_lock_open_with_code(hass: HomeAssistant) -> None:
"""Test lock entity with open service."""
lock = MockLockEntity(
code_format=r"^\d{4}$", supported_features=LockEntityFeature.OPEN
)
lock.hass = hass
assert lock.state_attributes == {"code_format": r"^\d{4}$"}
with pytest.raises(ValueError):
await _async_open(lock, ServiceCall(DOMAIN, SERVICE_OPEN, {}))
with pytest.raises(ValueError):
await _async_open(lock, ServiceCall(DOMAIN, SERVICE_OPEN, {ATTR_CODE: ""}))
with pytest.raises(ValueError):
await _async_open(lock, ServiceCall(DOMAIN, SERVICE_OPEN, {ATTR_CODE: "HELLO"}))
await _async_open(lock, ServiceCall(DOMAIN, SERVICE_OPEN, {ATTR_CODE: "1234"}))
assert lock.calls_open.call_count == 1
async def test_lock_lock_with_code(hass: HomeAssistant) -> None:
"""Test lock entity with open service."""
lock = MockLockEntity(code_format=r"^\d{4}$")
lock.hass = hass
await _async_unlock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {ATTR_CODE: "1234"}))
assert not lock.is_locked
with pytest.raises(ValueError):
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_LOCK, {}))
with pytest.raises(ValueError):
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_LOCK, {ATTR_CODE: ""}))
with pytest.raises(ValueError):
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_LOCK, {ATTR_CODE: "HELLO"}))
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_LOCK, {ATTR_CODE: "1234"}))
assert lock.is_locked
async def test_lock_unlock_with_code(hass: HomeAssistant) -> None:
"""Test unlock entity with open service."""
lock = MockLockEntity(code_format=r"^\d{4}$")
lock.hass = hass
await _async_lock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {ATTR_CODE: "1234"}))
assert lock.is_locked
with pytest.raises(ValueError):
await _async_unlock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {}))
with pytest.raises(ValueError):
await _async_unlock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {ATTR_CODE: ""}))
with pytest.raises(ValueError):
await _async_unlock(
lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {ATTR_CODE: "HELLO"})
)
await _async_unlock(lock, ServiceCall(DOMAIN, SERVICE_UNLOCK, {ATTR_CODE: "1234"}))
assert not lock.is_locked