Add reproduce state support to text platform (#82772)

Co-authored-by: Franck Nijhof <frenck@frenck.nl>

homeassistant/components/text/reproduce_state.py

@@ -0,0 +1,57 @@
+"""Reproduce a Text entity state."""
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Iterable
+import logging
+from typing import Any
+
+from homeassistant.const import ATTR_ENTITY_ID
+from homeassistant.core import Context, HomeAssistant, State
+
+from .const import ATTR_VALUE, DOMAIN, SERVICE_SET_VALUE
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def _async_reproduce_state(
+    hass: HomeAssistant,
+    state: State,
+    *,
+    context: Context | None = None,
+    reproduce_options: dict[str, Any] | None = None,
+) -> None:
+    """Reproduce a single state."""
+    if (cur_state := hass.states.get(state.entity_id)) is None:
+        _LOGGER.warning("Unable to find entity %s", state.entity_id)
+        return
+
+    # Return if we are already at the right state.
+    if cur_state.state == state.state:
+        return
+
+    service = SERVICE_SET_VALUE
+    service_data = {ATTR_ENTITY_ID: state.entity_id, ATTR_VALUE: state.state}
+
+    await hass.services.async_call(
+        DOMAIN, service, service_data, context=context, blocking=True
+    )
+
+
+async def async_reproduce_states(
+    hass: HomeAssistant,
+    states: Iterable[State],
+    *,
+    context: Context | None = None,
+    reproduce_options: dict[str, Any] | None = None,
+) -> None:
+    """Reproduce multiple Text states."""
+    # Reproduce states in parallel.
+    await asyncio.gather(
+        *(
+            _async_reproduce_state(
+                hass, state, context=context, reproduce_options=reproduce_options
+            )
+            for state in states
+        )
+    )

tests/components/text/test_reproduce_state.py

@@ -0,0 +1,53 @@
+"""Test reproduce state for Text entities."""
+from homeassistant.components.text.const import (
+    ATTR_MAX,
+    ATTR_MIN,
+    ATTR_MODE,
+    ATTR_PATTERN,
+    DOMAIN,
+    SERVICE_SET_VALUE,
+)
+from homeassistant.core import State
+from homeassistant.helpers.state import async_reproduce_state
+
+from tests.common import async_mock_service
+
+VALID_TEXT1 = "Hello"
+VALID_TEXT2 = "World"
+
+
+async def test_reproducing_states(hass, caplog):
+    """Test reproducing Text states."""
+
+    hass.states.async_set(
+        "text.test_text",
+        VALID_TEXT1,
+        {ATTR_MIN: 1, ATTR_MAX: 5, ATTR_MODE: "text", ATTR_PATTERN: None},
+    )
+
+    # These calls should do nothing as entities already in desired state
+    await async_reproduce_state(
+        hass,
+        [
+            State("text.test_text", VALID_TEXT1),
+            # Should not raise
+            State("text.non_existing", "234"),
+        ],
+    )
+
+    assert hass.states.get("text.test_text").state == VALID_TEXT1
+
+    # Test reproducing with different state
+    calls = async_mock_service(hass, DOMAIN, SERVICE_SET_VALUE)
+    await async_reproduce_state(
+        hass,
+        [
+            State("text.test_text", VALID_TEXT2),
+            # Should not raise
+            State("text.non_existing", "234"),
+        ],
+    )
+
+    assert len(calls) == 1
+    assert calls[0].domain == DOMAIN
+    assert calls[0].data == {"entity_id": "text.test_text", "value": VALID_TEXT2}