Fail schema validation for state conditions which do not support duration (#174083)

This commit is contained in:
Erik Montnemery
2026-09-13 09:40:55 +02:00
committed by GitHub
parent 3feccf4d6c
commit db80968d8f
3 changed files with 125 additions and 10 deletions
+1 -6
View File
@@ -9,7 +9,6 @@ from datetime import datetime, time as dt_time, timedelta
import functools as ft
import inspect
import logging
import re
import sys
from typing import (
TYPE_CHECKING,
@@ -151,10 +150,6 @@ _PLATFORM_ALIASES: dict[str | None, str | None] = {
"trigger": None,
}
INPUT_ENTITY_ID = re.compile(
r"^input_(?:select|text|number|boolean|datetime)\.(?!.+__)(?!_)[\da-z_]+(?<!_)$"
)
CONDITION_DESCRIPTION_CACHE: HassKey[dict[str, dict[str, Any] | None]] = HassKey(
"condition_description_cache"
@@ -1732,7 +1727,7 @@ def state(
state_value = req_state_value
if (
isinstance(req_state_value, str)
and INPUT_ENTITY_ID.match(req_state_value) is not None
and cv.INPUT_ENTITY_ID.match(req_state_value) is not None
):
if not (state_entity := hass.states.get(req_state_value)):
raise ConditionErrorMessage(
+23 -1
View File
@@ -1549,6 +1549,10 @@ NUMERIC_STATE_CONDITION_SCHEMA = vol.All(
has_at_least_one_key(CONF_BELOW, CONF_ABOVE),
)
INPUT_ENTITY_ID = re.compile(
r"^input_(?:select|text|number|boolean|datetime)\.(?!.+__)(?!_)[\da-z_]+(?<!_)$"
)
STATE_CONDITION_BASE_SCHEMA = {
**CONDITION_BASE_SCHEMA,
vol.Required(CONF_CONDITION): "state",
@@ -1585,7 +1589,25 @@ def STATE_CONDITION_SCHEMA(value: Any) -> dict[str, Any]:
else:
validated = STATE_CONDITION_STATE_SCHEMA(value)
return key_dependency("for", "state")(validated)
validated = key_dependency("for", "state")(validated)
if CONF_FOR in validated:
# `for` is anchored to the entity's last_changed, which only reflects a
# single current state. It therefore can't track an attribute, multiple
# states, or a state resolved from another entity.
if CONF_ATTRIBUTE in validated:
raise vol.Invalid("Cannot use 'for' with an attribute")
state = validated[CONF_STATE]
# A single-element list is just that one state; unwrap it so the
# input-entity check below also rejects `state: [input_select.x]`.
if isinstance(state, list):
if len(state) != 1:
raise vol.Invalid("Cannot use 'for' with a list of states")
state = state[0]
if INPUT_ENTITY_ID.match(state):
raise vol.Invalid("Cannot use 'for' with a state referencing an entity")
return validated
TEMPLATE_CONDITION_SCHEMA = vol.Schema(
+101 -3
View File
@@ -1385,15 +1385,25 @@ async def test_state_raises(hass: HomeAssistant) -> None:
test.async_check()
async def test_state_for(hass: HomeAssistant) -> None:
"""Test state with duration."""
@pytest.mark.parametrize(
"req_state",
[
pytest.param("100", id="scalar"),
pytest.param(["100"], id="single_item_list"),
],
)
async def test_state_for(hass: HomeAssistant, req_state: str | list[str]) -> None:
"""Test state with duration.
A single-element list `state` is equivalent to the scalar form.
"""
config = {
"condition": "and",
"conditions": [
{
"condition": "state",
"entity_id": ["sensor.temperature"],
"state": "100",
"state": req_state,
"for": {"seconds": 5},
},
],
@@ -1462,6 +1472,56 @@ async def test_state_for_invalid_template(
assert not test.async_check()
@pytest.mark.parametrize(
("extra_config", "error"),
[
pytest.param(
{"attribute": "battery_level"},
r"Cannot use 'for' with an attribute",
id="attribute",
),
pytest.param(
{"state": ["100", "200"]},
r"Cannot use 'for' with a list of states",
id="list_of_states",
),
pytest.param(
{"state": []},
r"Cannot use 'for' with a list of states",
id="empty_list",
),
pytest.param(
{"state": "input_number.threshold"},
r"Cannot use 'for' with a state referencing an entity",
id="state_from_entity",
),
pytest.param(
{"state": ["input_number.threshold"]},
r"Cannot use 'for' with a state referencing an entity",
id="single_item_list_from_entity",
),
],
)
def test_state_for_not_allowed(extra_config: dict[str, Any], error: str) -> None:
"""Test state condition rejects `for` with unsupported `state`/`attribute`.
`for` is anchored to the entity's last_changed, which reflects a single
current state. It therefore cannot be combined with an attribute, a list
that is not a single state, or a state resolved from another entity (even as
a single-element list). A single-element literal list behaves like the
scalar form (see `test_state_for`).
"""
config = {
"condition": "state",
"entity_id": "sensor.temperature",
"state": "100",
"for": {"seconds": 5},
**extra_config,
}
with pytest.raises(vol.Invalid, match=error):
cv.CONDITION_SCHEMA(config)
async def test_state_unknown_attribute(hass: HomeAssistant) -> None:
"""Test that state returns False on unknown attribute."""
# Unknown attribute
@@ -1494,6 +1554,44 @@ async def test_state_unknown_attribute(hass: HomeAssistant) -> None:
)
@pytest.mark.parametrize(
("req_state", "attribute_value", "expected"),
[
# A list `state` is matched as alternatives, so the attribute value must
# equal one of the items; the list itself is never compared as a whole.
pytest.param(["a", "b"], "a", True, id="item_in_list"),
pytest.param(["a", "b"], ["a", "b"], False, id="list_is_not_an_item"),
# Nesting the list makes the list value itself one of the items to match.
pytest.param([["a", "b"]], ["a", "b"], True, id="list_in_list_of_lists"),
pytest.param([["a", "b"]], "a", False, id="scalar_not_in_list_of_lists"),
],
)
async def test_state_attribute_list_matching(
hass: HomeAssistant,
req_state: list[Any],
attribute_value: str | list[str],
expected: bool,
) -> None:
"""Test how a state-attribute condition matches against a list `state`.
A list `state` is treated as alternatives (match any item), so a list-valued
attribute only matches when the list is nested as an item of `state`. This
documents the current behavior; the implementation is unchanged.
"""
config = {
"condition": "state",
"entity_id": "sensor.test",
"attribute": "options",
"state": req_state,
}
config = cv.CONDITION_SCHEMA(config)
config = await condition.async_validate_condition_config(hass, config)
test = await condition.async_from_config(hass, config)
hass.states.async_set("sensor.test", "on", {"options": attribute_value})
assert test.async_check() is expected
async def test_state_multiple_entities(hass: HomeAssistant) -> None:
"""Test with multiple entities in condition."""
config = {