Add input_button (#62008)

* Add input_button

* Update homeassistant/components/input_button/__init__.py

Co-authored-by: Erik Montnemery <erik@montnemery.com>

* Improve test coverage

* Add reload test: not affecting state

Co-authored-by: Erik Montnemery <erik@montnemery.com>

.core_files.yaml

@@ -65,6 +65,7 @@ components: &components
   - homeassistant/components/homeassistant/**
   - homeassistant/components/image/*
   - homeassistant/components/input_boolean/*
+  - homeassistant/components/input_button/*
   - homeassistant/components/input_datetime/*
   - homeassistant/components/input_number/*
   - homeassistant/components/input_select/*

.strict-typing

@@ -65,6 +65,7 @@ homeassistant.components.http.*
 homeassistant.components.huawei_lte.*
 homeassistant.components.hyperion.*
 homeassistant.components.image_processing.*
+homeassistant.components.input_button.*
 homeassistant.components.input_select.*
 homeassistant.components.integration.*
 homeassistant.components.iqvia.*

CODEOWNERS

@@ -425,6 +425,8 @@ homeassistant/components/influxdb/* @fabaff @mdegat01
 tests/components/influxdb/* @fabaff @mdegat01
 homeassistant/components/input_boolean/* @home-assistant/core
 tests/components/input_boolean/* @home-assistant/core
+homeassistant/components/input_button/* @home-assistant/core
+tests/components/input_button/* @home-assistant/core
 homeassistant/components/input_datetime/* @home-assistant/core
 tests/components/input_datetime/* @home-assistant/core
 homeassistant/components/input_number/* @home-assistant/core

homeassistant/components/default_config/manifest.json

@@ -11,6 +11,7 @@
     "frontend",
     "history",
     "input_boolean",
+    "input_button",
     "input_datetime",
     "input_number",
     "input_select",

homeassistant/components/demo/__init__.py

@@ -119,6 +119,22 @@ async def async_setup(hass, config):
         )
     )
 
+    # Set up input button
+    tasks.append(
+        bootstrap.async_setup_component(
+            hass,
+            "input_button",
+            {
+                "input_button": {
+                    "bell": {
+                        "icon": "mdi:bell-ring-outline",
+                        "name": "Ring bell",
+                    }
+                }
+            },
+        )
+    )
+
     # Set up input number
     tasks.append(
         bootstrap.async_setup_component(

homeassistant/components/input_button/__init__.py

@@ -0,0 +1,171 @@
+"""Support to keep track of user controlled buttons which can be used in automations."""
+from __future__ import annotations
+
+import logging
+from typing import cast
+
+import voluptuous as vol
+
+from homeassistant.components.button import SERVICE_PRESS, ButtonEntity
+from homeassistant.const import (
+    ATTR_EDITABLE,
+    CONF_ICON,
+    CONF_ID,
+    CONF_NAME,
+    SERVICE_RELOAD,
+)
+from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.helpers import collection
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.entity_component import EntityComponent
+from homeassistant.helpers.restore_state import RestoreEntity
+import homeassistant.helpers.service
+from homeassistant.helpers.storage import Store
+from homeassistant.helpers.typing import ConfigType
+
+DOMAIN = "input_button"
+
+_LOGGER = logging.getLogger(__name__)
+
+CREATE_FIELDS = {
+    vol.Required(CONF_NAME): vol.All(str, vol.Length(min=1)),
+    vol.Optional(CONF_ICON): cv.icon,
+}
+
+UPDATE_FIELDS = {
+    vol.Optional(CONF_NAME): cv.string,
+    vol.Optional(CONF_ICON): cv.icon,
+}
+
+CONFIG_SCHEMA = vol.Schema(
+    {DOMAIN: cv.schema_with_slug_keys(vol.Any(UPDATE_FIELDS, None))},
+    extra=vol.ALLOW_EXTRA,
+)
+
+RELOAD_SERVICE_SCHEMA = vol.Schema({})
+STORAGE_KEY = DOMAIN
+STORAGE_VERSION = 1
+
+
+class InputButtonStorageCollection(collection.StorageCollection):
+    """Input button collection stored in storage."""
+
+    CREATE_SCHEMA = vol.Schema(CREATE_FIELDS)
+    UPDATE_SCHEMA = vol.Schema(UPDATE_FIELDS)
+
+    async def _process_create_data(self, data: dict) -> vol.Schema:
+        """Validate the config is valid."""
+        return self.CREATE_SCHEMA(data)
+
+    @callback
+    def _get_suggested_id(self, info: dict) -> str:
+        """Suggest an ID based on the config."""
+        return cast(str, info[CONF_NAME])
+
+    async def _update_data(self, data: dict, update_data: dict) -> dict:
+        """Return a new updated data object."""
+        update_data = self.UPDATE_SCHEMA(update_data)
+        return {**data, **update_data}
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up an input button."""
+    component = EntityComponent(_LOGGER, DOMAIN, hass)
+    id_manager = collection.IDManager()
+
+    yaml_collection = collection.YamlCollection(
+        logging.getLogger(f"{__name__}.yaml_collection"), id_manager
+    )
+    collection.sync_entity_lifecycle(
+        hass, DOMAIN, DOMAIN, component, yaml_collection, InputButton.from_yaml
+    )
+
+    storage_collection = InputButtonStorageCollection(
+        Store(hass, STORAGE_VERSION, STORAGE_KEY),
+        logging.getLogger(f"{__name__}.storage_collection"),
+        id_manager,
+    )
+    collection.sync_entity_lifecycle(
+        hass, DOMAIN, DOMAIN, component, storage_collection, InputButton
+    )
+
+    await yaml_collection.async_load(
+        [{CONF_ID: id_, **(conf or {})} for id_, conf in config.get(DOMAIN, {}).items()]
+    )
+    await storage_collection.async_load()
+
+    collection.StorageCollectionWebsocket(
+        storage_collection, DOMAIN, DOMAIN, CREATE_FIELDS, UPDATE_FIELDS
+    ).async_setup(hass)
+
+    async def reload_service_handler(service_call: ServiceCall) -> None:
+        """Remove all input buttons and load new ones from config."""
+        conf = await component.async_prepare_reload(skip_reset=True)
+        if conf is None:
+            return
+        await yaml_collection.async_load(
+            [
+                {CONF_ID: id_, **(conf or {})}
+                for id_, conf in conf.get(DOMAIN, {}).items()
+            ]
+        )
+
+    homeassistant.helpers.service.async_register_admin_service(
+        hass,
+        DOMAIN,
+        SERVICE_RELOAD,
+        reload_service_handler,
+        schema=RELOAD_SERVICE_SCHEMA,
+    )
+
+    component.async_register_entity_service(SERVICE_PRESS, {}, "_async_press_action")
+
+    return True
+
+
+class InputButton(ButtonEntity, RestoreEntity):
+    """Representation of a button."""
+
+    _attr_should_poll = False
+
+    def __init__(self, config: ConfigType) -> None:
+        """Initialize a button."""
+        self._config = config
+        self.editable = True
+        self._attr_unique_id = config[CONF_ID]
+
+    @classmethod
+    def from_yaml(cls, config: ConfigType) -> ButtonEntity:
+        """Return entity instance initialized from yaml storage."""
+        button = cls(config)
+        button.entity_id = f"{DOMAIN}.{config[CONF_ID]}"
+        button.editable = False
+        return button
+
+    @property
+    def name(self) -> str | None:
+        """Return name of the button."""
+        return self._config.get(CONF_NAME)
+
+    @property
+    def icon(self) -> str | None:
+        """Return the icon to be used for this entity."""
+        return self._config.get(CONF_ICON)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, bool]:
+        """Return the state attributes of the entity."""
+        return {ATTR_EDITABLE: self.editable}
+
+    async def async_press(self) -> None:
+        """Press the button.
+
+        Left emtpty intentionally.
+        The input button itself doesn't trigger anything.
+        """
+        return None
+
+    async def async_update_config(self, config: ConfigType) -> None:
+        """Handle when the config is updated."""
+        self._config = config
+        self.async_write_ha_state()

homeassistant/components/input_button/manifest.json

@@ -0,0 +1,7 @@
+{
+  "domain": "input_button",
+  "name": "Input Button",
+  "documentation": "https://www.home-assistant.io/integrations/input_button",
+  "codeowners": ["@home-assistant/core"],
+  "quality_scale": "internal"
+}

homeassistant/components/input_button/services.yaml

@@ -0,0 +1,6 @@
+press:
+  name: Press
+  description: Press the input button entity.
+  target:
+    entity:
+      domain: input_button

mypy.ini

@@ -726,6 +726,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.input_button.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.input_select.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

script/hassfest/dependencies.py

@@ -102,6 +102,7 @@ ALLOWED_USED_COMPONENTS = {
     "hassio",
     "homeassistant",
     "input_boolean",
+    "input_button",
     "input_datetime",
     "input_number",
     "input_select",

script/hassfest/manifest.py

@@ -64,6 +64,7 @@ NO_IOT_CLASS = [
     "image_processing",
     "image",
     "input_boolean",
+    "input_button",
     "input_datetime",
     "input_number",
     "input_select",

tests/components/input_button/__init__.py

@@ -0,0 +1 @@
+"""Tests for the input_test component."""

tests/components/input_button/test_init.py

@@ -0,0 +1,357 @@
+"""The tests for the input_test component."""
+import logging
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.input_button import DOMAIN, SERVICE_PRESS
+from homeassistant.const import (
+    ATTR_EDITABLE,
+    ATTR_ENTITY_ID,
+    ATTR_FRIENDLY_NAME,
+    ATTR_ICON,
+    ATTR_NAME,
+    SERVICE_RELOAD,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import Context, CoreState, State
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.event import async_track_state_change
+from homeassistant.setup import async_setup_component
+
+from tests.common import mock_component, mock_restore_cache
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@pytest.fixture
+def storage_setup(hass, hass_storage):
+    """Storage setup."""
+
+    async def _storage(items=None, config=None):
+        if items is None:
+            hass_storage[DOMAIN] = {
+                "key": DOMAIN,
+                "version": 1,
+                "data": {"items": [{"id": "from_storage", "name": "from storage"}]},
+            }
+        else:
+            hass_storage[DOMAIN] = items
+        if config is None:
+            config = {DOMAIN: {}}
+        return await async_setup_component(hass, DOMAIN, config)
+
+    return _storage
+
+
+async def test_config(hass):
+    """Test config."""
+    invalid_configs = [None, 1, {}, {"name with space": None}]
+
+    for cfg in invalid_configs:
+        assert not await async_setup_component(hass, DOMAIN, {DOMAIN: cfg})
+
+
+async def test_config_options(hass):
+    """Test configuration options."""
+    count_start = len(hass.states.async_entity_ids())
+
+    _LOGGER.debug("ENTITIES @ start: %s", hass.states.async_entity_ids())
+
+    assert await async_setup_component(
+        hass,
+        DOMAIN,
+        {
+            DOMAIN: {
+                "test_1": None,
+                "test_2": {"name": "Hello World", "icon": "mdi:work"},
+            }
+        },
+    )
+
+    _LOGGER.debug("ENTITIES: %s", hass.states.async_entity_ids())
+
+    assert count_start + 2 == len(hass.states.async_entity_ids())
+
+    state_1 = hass.states.get("input_button.test_1")
+    state_2 = hass.states.get("input_button.test_2")
+
+    assert state_1 is not None
+    assert state_2 is not None
+
+    assert state_1.state == STATE_UNKNOWN
+    assert ATTR_ICON not in state_1.attributes
+    assert ATTR_FRIENDLY_NAME not in state_1.attributes
+
+    assert state_2.state == STATE_UNKNOWN
+    assert state_2.attributes.get(ATTR_FRIENDLY_NAME) == "Hello World"
+    assert state_2.attributes.get(ATTR_ICON) == "mdi:work"
+
+
+async def test_restore_state(hass):
+    """Ensure states are restored on startup."""
+    mock_restore_cache(
+        hass,
+        (State("input_button.b1", "2021-01-01T23:59:59+00:00"),),
+    )
+
+    hass.state = CoreState.starting
+    mock_component(hass, "recorder")
+
+    await async_setup_component(hass, DOMAIN, {DOMAIN: {"b1": None, "b2": None}})
+
+    state = hass.states.get("input_button.b1")
+    assert state
+    assert state.state == "2021-01-01T23:59:59+00:00"
+
+    state = hass.states.get("input_button.b2")
+    assert state
+    assert state.state == STATE_UNKNOWN
+
+
+async def test_input_button_context(hass, hass_admin_user):
+    """Test that input_button context works."""
+    assert await async_setup_component(hass, DOMAIN, {DOMAIN: {"update": {}}})
+
+    state = hass.states.get("input_button.update")
+    assert state is not None
+
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state.entity_id},
+        True,
+        Context(user_id=hass_admin_user.id),
+    )
+
+    state2 = hass.states.get("input_button.update")
+    assert state2 is not None
+    assert state.state != state2.state
+    assert state2.context.user_id == hass_admin_user.id
+
+
+async def test_reload(hass, hass_admin_user):
+    """Test reload service."""
+    count_start = len(hass.states.async_entity_ids())
+    ent_reg = er.async_get(hass)
+
+    _LOGGER.debug("ENTITIES @ start: %s", hass.states.async_entity_ids())
+
+    assert await async_setup_component(
+        hass,
+        DOMAIN,
+        {
+            DOMAIN: {
+                "test_1": None,
+                "test_2": {"name": "Hello World", "icon": "mdi:work"},
+            }
+        },
+    )
+
+    _LOGGER.debug("ENTITIES: %s", hass.states.async_entity_ids())
+
+    assert count_start + 2 == len(hass.states.async_entity_ids())
+
+    state_1 = hass.states.get("input_button.test_1")
+    state_2 = hass.states.get("input_button.test_2")
+    state_3 = hass.states.get("input_button.test_3")
+
+    assert state_1 is not None
+    assert state_2 is not None
+    assert state_3 is None
+
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_1") is not None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_2") is not None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_3") is None
+
+    with patch(
+        "homeassistant.config.load_yaml_config_file",
+        autospec=True,
+        return_value={
+            DOMAIN: {
+                "test_2": {
+                    "name": "Hello World reloaded",
+                    "icon": "mdi:work_reloaded",
+                },
+                "test_3": None,
+            }
+        },
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_RELOAD,
+            blocking=True,
+            context=Context(user_id=hass_admin_user.id),
+        )
+
+    assert count_start + 2 == len(hass.states.async_entity_ids())
+
+    state_1 = hass.states.get("input_button.test_1")
+    state_2 = hass.states.get("input_button.test_2")
+    state_3 = hass.states.get("input_button.test_3")
+
+    assert state_1 is None
+    assert state_2 is not None
+    assert state_3 is not None
+
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_1") is None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_2") is not None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "test_3") is not None
+
+
+async def test_reload_not_changing_state(hass, storage_setup):
+    """Test reload not changing state."""
+    assert await storage_setup()
+    state_changes = []
+
+    def state_changed_listener(entity_id, from_s, to_s):
+        state_changes.append(to_s)
+
+    state = hass.states.get(f"{DOMAIN}.from_storage")
+    assert state is not None
+
+    async_track_state_change(hass, [f"{DOMAIN}.from_storage"], state_changed_listener)
+
+    # Pressing button changes state
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state.entity_id},
+        True,
+    )
+    await hass.async_block_till_done()
+    assert len(state_changes) == 1
+
+    # Reloading does not
+    with patch(
+        "homeassistant.config.load_yaml_config_file", autospec=True, return_value={}
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_RELOAD,
+            blocking=True,
+        )
+
+    await hass.async_block_till_done()
+
+    state = hass.states.get(f"{DOMAIN}.from_storage")
+    assert state is not None
+    assert len(state_changes) == 1
+
+
+async def test_load_from_storage(hass, storage_setup):
+    """Test set up from storage."""
+    assert await storage_setup()
+    state = hass.states.get(f"{DOMAIN}.from_storage")
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_FRIENDLY_NAME) == "from storage"
+    assert state.attributes.get(ATTR_EDITABLE)
+
+
+async def test_editable_state_attribute(hass, storage_setup):
+    """Test editable attribute."""
+    assert await storage_setup(config={DOMAIN: {"from_yaml": None}})
+
+    state = hass.states.get(f"{DOMAIN}.from_storage")
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_FRIENDLY_NAME) == "from storage"
+    assert state.attributes.get(ATTR_EDITABLE)
+
+    state = hass.states.get(f"{DOMAIN}.from_yaml")
+    assert state.state == STATE_UNKNOWN
+    assert not state.attributes.get(ATTR_EDITABLE)
+
+
+async def test_ws_list(hass, hass_ws_client, storage_setup):
+    """Test listing via WS."""
+    assert await storage_setup(config={DOMAIN: {"from_yaml": None}})
+
+    client = await hass_ws_client(hass)
+
+    await client.send_json({"id": 6, "type": f"{DOMAIN}/list"})
+    resp = await client.receive_json()
+    assert resp["success"]
+
+    storage_ent = "from_storage"
+    yaml_ent = "from_yaml"
+    result = {item["id"]: item for item in resp["result"]}
+
+    assert len(result) == 1
+    assert storage_ent in result
+    assert yaml_ent not in result
+    assert result[storage_ent][ATTR_NAME] == "from storage"
+
+
+async def test_ws_create_update(hass, hass_ws_client, storage_setup):
+    """Test creating and updating via WS."""
+    assert await storage_setup(config={DOMAIN: {}})
+
+    client = await hass_ws_client(hass)
+
+    await client.send_json({"id": 7, "type": f"{DOMAIN}/create", "name": "new"})
+    resp = await client.receive_json()
+    assert resp["success"]
+
+    state = hass.states.get(f"{DOMAIN}.new")
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_FRIENDLY_NAME) == "new"
+
+    ent_reg = er.async_get(hass)
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "new") is not None
+
+    await client.send_json(
+        {"id": 8, "type": f"{DOMAIN}/update", f"{DOMAIN}_id": "new", "name": "newer"}
+    )
+    resp = await client.receive_json()
+    assert resp["success"]
+
+    state = hass.states.get(f"{DOMAIN}.new")
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_FRIENDLY_NAME) == "newer"
+
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, "new") is not None
+
+
+async def test_ws_delete(hass, hass_ws_client, storage_setup):
+    """Test WS delete cleans up entity registry."""
+    assert await storage_setup()
+
+    input_id = "from_storage"
+    input_entity_id = f"{DOMAIN}.{input_id}"
+    ent_reg = er.async_get(hass)
+
+    state = hass.states.get(input_entity_id)
+    assert state is not None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, input_id) is not None
+
+    client = await hass_ws_client(hass)
+
+    await client.send_json(
+        {"id": 6, "type": f"{DOMAIN}/delete", f"{DOMAIN}_id": f"{input_id}"}
+    )
+    resp = await client.receive_json()
+    assert resp["success"]
+
+    state = hass.states.get(input_entity_id)
+    assert state is None
+    assert ent_reg.async_get_entity_id(DOMAIN, DOMAIN, input_id) is None
+
+
+async def test_setup_no_config(hass, hass_admin_user):
+    """Test component setup with no config."""
+    count_start = len(hass.states.async_entity_ids())
+    assert await async_setup_component(hass, DOMAIN, {})
+
+    with patch(
+        "homeassistant.config.load_yaml_config_file", autospec=True, return_value={}
+    ):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_RELOAD,
+            blocking=True,
+            context=Context(user_id=hass_admin_user.id),
+        )
+
+    assert count_start == len(hass.states.async_entity_ids())