Add improved scene support for input_select (#27697)
* Add improved scene support for input_select * Add tests for reproducing input_select states. * Add some comments. * Add support for set_options Allows defining the options for an input_select in a scene. * Add tests for set_options in test_reproduce_state * Execute for real instead of mock execution.
This commit is contained in:
parent
93f9afcd21
commit
8720ca38b5
2 changed files with 152 additions and 0 deletions
80
homeassistant/components/input_select/reproduce_state.py
Normal file
80
homeassistant/components/input_select/reproduce_state.py
Normal file
|
@ -0,0 +1,80 @@
|
|||
"""Reproduce an Input select state."""
|
||||
import asyncio
|
||||
import logging
|
||||
from types import MappingProxyType
|
||||
from typing import Iterable, Optional
|
||||
|
||||
from homeassistant.const import ATTR_ENTITY_ID
|
||||
from homeassistant.core import Context, State
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from . import (
|
||||
DOMAIN,
|
||||
SERVICE_SELECT_OPTION,
|
||||
SERVICE_SET_OPTIONS,
|
||||
ATTR_OPTION,
|
||||
ATTR_OPTIONS,
|
||||
)
|
||||
|
||||
ATTR_GROUP = [ATTR_OPTION, ATTR_OPTIONS]
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def _async_reproduce_state(
|
||||
hass: HomeAssistantType, state: State, context: Optional[Context] = None
|
||||
) -> None:
|
||||
"""Reproduce a single state."""
|
||||
cur_state = hass.states.get(state.entity_id)
|
||||
|
||||
# Return if we can't find entity
|
||||
if cur_state 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 and all(
|
||||
check_attr_equal(cur_state.attributes, state.attributes, attr)
|
||||
for attr in ATTR_GROUP
|
||||
):
|
||||
return
|
||||
|
||||
# Set service data
|
||||
service_data = {ATTR_ENTITY_ID: state.entity_id}
|
||||
|
||||
# If options are specified, call SERVICE_SET_OPTIONS
|
||||
if ATTR_OPTIONS in state.attributes:
|
||||
service = SERVICE_SET_OPTIONS
|
||||
service_data[ATTR_OPTIONS] = state.attributes[ATTR_OPTIONS]
|
||||
|
||||
await hass.services.async_call(
|
||||
DOMAIN, service, service_data, context=context, blocking=True
|
||||
)
|
||||
|
||||
# Remove ATTR_OPTIONS from service_data so we can reuse service_data in next call
|
||||
del service_data[ATTR_OPTIONS]
|
||||
|
||||
# Call SERVICE_SELECT_OPTION
|
||||
service = SERVICE_SELECT_OPTION
|
||||
service_data[ATTR_OPTION] = state.state
|
||||
|
||||
await hass.services.async_call(
|
||||
DOMAIN, service, service_data, context=context, blocking=True
|
||||
)
|
||||
|
||||
|
||||
async def async_reproduce_states(
|
||||
hass: HomeAssistantType, states: Iterable[State], context: Optional[Context] = None
|
||||
) -> None:
|
||||
"""Reproduce Input select states."""
|
||||
# Reproduce states in parallel.
|
||||
await asyncio.gather(
|
||||
*(_async_reproduce_state(hass, state, context) for state in states)
|
||||
)
|
||||
|
||||
|
||||
def check_attr_equal(
|
||||
attr1: MappingProxyType, attr2: MappingProxyType, attr_str: str
|
||||
) -> bool:
|
||||
"""Return true if the given attributes are equal."""
|
||||
return attr1.get(attr_str) == attr2.get(attr_str)
|
72
tests/components/input_select/test_reproduce_state.py
Normal file
72
tests/components/input_select/test_reproduce_state.py
Normal file
|
@ -0,0 +1,72 @@
|
|||
"""Test reproduce state for Input select."""
|
||||
from homeassistant.core import State
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
VALID_OPTION1 = "Option A"
|
||||
VALID_OPTION2 = "Option B"
|
||||
VALID_OPTION3 = "Option C"
|
||||
VALID_OPTION4 = "Option D"
|
||||
VALID_OPTION5 = "Option E"
|
||||
VALID_OPTION6 = "Option F"
|
||||
INVALID_OPTION = "Option X"
|
||||
VALID_OPTION_SET1 = [VALID_OPTION1, VALID_OPTION2, VALID_OPTION3]
|
||||
VALID_OPTION_SET2 = [VALID_OPTION4, VALID_OPTION5, VALID_OPTION6]
|
||||
ENTITY = "input_select.test_select"
|
||||
|
||||
|
||||
async def test_reproducing_states(hass, caplog):
|
||||
"""Test reproducing Input select states."""
|
||||
|
||||
# Setup entity
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"input_select",
|
||||
{
|
||||
"input_select": {
|
||||
"test_select": {"options": VALID_OPTION_SET1, "initial": VALID_OPTION1}
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
# These calls should do nothing as entities already in desired state
|
||||
await hass.helpers.state.async_reproduce_state(
|
||||
[
|
||||
State(ENTITY, VALID_OPTION1),
|
||||
# Should not raise
|
||||
State("input_select.non_existing", VALID_OPTION1),
|
||||
],
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Test that entity is in desired state
|
||||
assert hass.states.get(ENTITY).state == VALID_OPTION1
|
||||
|
||||
# Try reproducing with different state
|
||||
await hass.helpers.state.async_reproduce_state(
|
||||
[
|
||||
State(ENTITY, VALID_OPTION3),
|
||||
# Should not raise
|
||||
State("input_select.non_existing", VALID_OPTION3),
|
||||
],
|
||||
blocking=True,
|
||||
)
|
||||
|
||||
# Test that we got the desired result
|
||||
assert hass.states.get(ENTITY).state == VALID_OPTION3
|
||||
|
||||
# Test setting state to invalid state
|
||||
await hass.helpers.state.async_reproduce_state(
|
||||
[State(ENTITY, INVALID_OPTION)], blocking=True
|
||||
)
|
||||
|
||||
# The entity state should be unchanged
|
||||
assert hass.states.get(ENTITY).state == VALID_OPTION3
|
||||
|
||||
# Test setting a different option set
|
||||
await hass.helpers.state.async_reproduce_state(
|
||||
[State(ENTITY, VALID_OPTION5, {"options": VALID_OPTION_SET2})], blocking=True
|
||||
)
|
||||
|
||||
# These should fail if options weren't changed to VALID_OPTION_SET2
|
||||
assert hass.states.get(ENTITY).attributes == {"options": VALID_OPTION_SET2}
|
||||
assert hass.states.get(ENTITY).state == VALID_OPTION5
|
Loading…
Add table
Reference in a new issue