Template Lock (#17288)

* Template Lock component

* Tests

* CI Fix

* Don't track templates if they have result in MATCH_ALL

* async/await

* houndci-bot review fix

homeassistant/components/lock/template.py

@@ -0,0 +1,141 @@
+"""
+Support for locks which integrates with other components.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/lock.template/
+"""
+import logging
+
+import voluptuous as vol
+
+import homeassistant.helpers.config_validation as cv
+
+from homeassistant.core import callback
+from homeassistant.components.lock import (LockDevice, PLATFORM_SCHEMA)
+from homeassistant.const import (
+    CONF_NAME, CONF_OPTIMISTIC, CONF_VALUE_TEMPLATE,
+    EVENT_HOMEASSISTANT_START, STATE_ON, STATE_LOCKED, MATCH_ALL)
+from homeassistant.exceptions import TemplateError
+from homeassistant.helpers.event import async_track_state_change
+from homeassistant.helpers.script import Script
+
+_LOGGER = logging.getLogger(__name__)
+
+CONF_LOCK = 'lock'
+CONF_UNLOCK = 'unlock'
+
+DEFAULT_NAME = 'Template Lock'
+DEFAULT_OPTIMISTIC = False
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
+    vol.Required(CONF_LOCK): cv.SCRIPT_SCHEMA,
+    vol.Required(CONF_UNLOCK): cv.SCRIPT_SCHEMA,
+    vol.Required(CONF_VALUE_TEMPLATE): cv.template,
+    vol.Optional(CONF_OPTIMISTIC, default=DEFAULT_OPTIMISTIC): cv.boolean
+})
+
+
+async def async_setup_platform(hass, config, async_add_devices,
+                               discovery_info=None):
+    """Set up the Template lock."""
+    name = config.get(CONF_NAME)
+    value_template = config.get(CONF_VALUE_TEMPLATE)
+    value_template.hass = hass
+    value_template_entity_ids = value_template.extract_entities()
+
+    if value_template_entity_ids == MATCH_ALL:
+        _LOGGER.warning(
+            'Template lock %s has no entity ids configured to track nor '
+            'were we able to extract the entities to track from the %s '
+            'template. This entity will only be able to be updated '
+            'manually.', name, CONF_VALUE_TEMPLATE)
+
+    async_add_devices([TemplateLock(
+        hass,
+        name,
+        value_template,
+        value_template_entity_ids,
+        config.get(CONF_LOCK),
+        config.get(CONF_UNLOCK),
+        config.get(CONF_OPTIMISTIC)
+    )])
+
+
+class TemplateLock(LockDevice):
+    """Representation of a template lock."""
+
+    def __init__(self, hass, name, value_template, entity_ids,
+                 command_lock, command_unlock, optimistic):
+        """Initialize the lock."""
+        self._state = None
+        self._hass = hass
+        self._name = name
+        self._state_template = value_template
+        self._state_entities = entity_ids
+        self._command_lock = Script(hass, command_lock)
+        self._command_unlock = Script(hass, command_unlock)
+        self._optimistic = optimistic
+
+    async def async_added_to_hass(self):
+        """Register callbacks."""
+        @callback
+        def template_lock_state_listener(entity, old_state, new_state):
+            """Handle target device state changes."""
+            self.async_schedule_update_ha_state(True)
+
+        @callback
+        def template_lock_startup(event):
+            """Update template on startup."""
+            if self._state_entities != MATCH_ALL:
+                # Track state change only for valid templates
+                async_track_state_change(
+                    self._hass, self._state_entities,
+                    template_lock_state_listener)
+            self.async_schedule_update_ha_state(True)
+
+        self._hass.bus.async_listen_once(
+            EVENT_HOMEASSISTANT_START, template_lock_startup)
+
+    @property
+    def assumed_state(self):
+        """Return true if we do optimistic updates."""
+        return self._optimistic
+
+    @property
+    def should_poll(self):
+        """No polling needed."""
+        return False
+
+    @property
+    def name(self):
+        """Return the name of the lock."""
+        return self._name
+
+    @property
+    def is_locked(self):
+        """Return true if lock is locked."""
+        return self._state
+
+    async def async_update(self):
+        """Update the state from the template."""
+        try:
+            self._state = self._state_template.async_render().lower() in (
+                'true', STATE_ON, STATE_LOCKED)
+        except TemplateError as ex:
+            self._state = None
+            _LOGGER.error('Could not render template %s: %s', self._name, ex)
+
+    async def async_lock(self, **kwargs):
+        """Lock the device."""
+        if self._optimistic:
+            self._state = True
+            self.async_schedule_update_ha_state()
+        await self._command_lock.async_run()
+
+    async def async_unlock(self, **kwargs):
+        """Unlock the device."""
+        if self._optimistic:
+            self._state = False
+            self.async_schedule_update_ha_state()
+        await self._command_unlock.async_run()

tests/components/lock/test_template.py

@@ -0,0 +1,309 @@
+"""The tests for the Template lock platform."""
+import logging
+
+from homeassistant.core import callback
+from homeassistant import setup
+from homeassistant.components import lock
+from homeassistant.const import STATE_ON, STATE_OFF
+
+from tests.common import (get_test_home_assistant,
+                          assert_setup_component)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class TestTemplateLock:
+    """Test the Template lock."""
+
+    hass = None
+    calls = None
+    # pylint: disable=invalid-name
+
+    def setup_method(self, method):
+        """Set up things to be run when tests are started."""
+        self.hass = get_test_home_assistant()
+        self.calls = []
+
+        @callback
+        def record_call(service):
+            """Track function calls."""
+            self.calls.append(service)
+
+        self.hass.services.register('test', 'automation', record_call)
+
+    def teardown_method(self, method):
+        """Stop everything that was started."""
+        self.hass.stop()
+
+    def test_template_state(self):
+        """Test template."""
+        with assert_setup_component(1, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'name': 'Test template lock',
+                    'value_template':
+                        "{{ states.switch.test_state.state }}",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        self.hass.states.set('switch.test_state', STATE_ON)
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.test_template_lock')
+        assert state.state == lock.STATE_LOCKED
+
+        self.hass.states.set('switch.test_state', STATE_OFF)
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.test_template_lock')
+        assert state.state == lock.STATE_UNLOCKED
+
+    def test_template_state_boolean_on(self):
+        """Test the setting of the state with boolean on."""
+        with assert_setup_component(1, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'value_template':
+                        "{{ 1 == 1 }}",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_LOCKED
+
+    def test_template_state_boolean_off(self):
+        """Test the setting of the state with off."""
+        with assert_setup_component(1, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'value_template':
+                        "{{ 1 == 2 }}",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_UNLOCKED
+
+    def test_template_syntax_error(self):
+        """Test templating syntax error."""
+        with assert_setup_component(0, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'value_template':
+                        "{% if rubbish %}",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        assert self.hass.states.all() == []
+
+    def test_invalid_name_does_not_create(self):
+        """Test invalid name."""
+        with assert_setup_component(0, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'switch': {
+                    'platform': 'lock',
+                    'name': '{{%}',
+                    'value_template':
+                        "{{ rubbish }",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        assert self.hass.states.all() == []
+
+    def test_invalid_lock_does_not_create(self):
+        """Test invalid lock."""
+        with assert_setup_component(0, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'value_template': "Invalid"
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        assert self.hass.states.all() == []
+
+    def test_missing_template_does_not_create(self):
+        """Test missing template."""
+        with assert_setup_component(0, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'not_value_template':
+                        "{{ states.switch.test_state.state }}",
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        assert self.hass.states.all() == []
+
+    def test_no_template_match_all(self, caplog):
+        """Test that we do not allow locks that match on all."""
+        with assert_setup_component(1, 'lock'):
+            assert setup.setup_component(self.hass, 'lock', {
+                'lock': {
+                    'platform': 'template',
+                    'value_template': '{{ 1 + 1 }}',
+                    'lock': {
+                        'service': 'switch.turn_on',
+                        'entity_id': 'switch.test_state'
+                    },
+                    'unlock': {
+                        'service': 'switch.turn_off',
+                        'entity_id': 'switch.test_state'
+                    }
+                }
+            })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_UNLOCKED
+
+        assert ('Template lock Template Lock has no entity ids configured '
+                'to track nor were we able to extract the entities to track '
+                'from the value_template template. This entity will only '
+                'be able to be updated manually.') in caplog.text
+
+        self.hass.states.set('lock.template_lock', lock.STATE_LOCKED)
+        self.hass.block_till_done()
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_LOCKED
+
+    def test_lock_action(self):
+        """Test lock action."""
+        assert setup.setup_component(self.hass, 'lock', {
+            'lock': {
+                'platform': 'template',
+                'value_template':
+                    "{{ states.switch.test_state.state }}",
+                'lock': {
+                    'service': 'test.automation'
+                },
+                'unlock': {
+                    'service': 'switch.turn_off',
+                    'entity_id': 'switch.test_state'
+                }
+            }
+        })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        self.hass.states.set('switch.test_state', STATE_OFF)
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_UNLOCKED
+
+        self.hass.services.call(lock.DOMAIN, lock.SERVICE_LOCK, {
+            lock.ATTR_ENTITY_ID: 'lock.template_lock'
+        })
+        self.hass.block_till_done()
+
+        assert len(self.calls) == 1
+
+    def test_unlock_action(self):
+        """Test unlock action."""
+        assert setup.setup_component(self.hass, 'lock', {
+            'lock': {
+                'platform': 'template',
+                'value_template':
+                    "{{ states.switch.test_state.state }}",
+                'lock': {
+                    'service': 'switch.turn_on',
+                    'entity_id': 'switch.test_state'
+                },
+                'unlock': {
+                    'service': 'test.automation'
+                }
+            }
+        })
+
+        self.hass.start()
+        self.hass.block_till_done()
+
+        self.hass.states.set('switch.test_state', STATE_ON)
+        self.hass.block_till_done()
+
+        state = self.hass.states.get('lock.template_lock')
+        assert state.state == lock.STATE_LOCKED
+
+        self.hass.services.call(lock.DOMAIN, lock.SERVICE_UNLOCK, {
+            lock.ATTR_ENTITY_ID: 'lock.template_lock'
+        })
+        self.hass.block_till_done()
+
+        assert len(self.calls) == 1