[core] Only treat action names as actions in has_non_synchronous_actions() (#20524)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Frédéric Metrich
2026-10-11 09:40:27 -10:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 86192d89f4
commit 16a474cce5
2 changed files with 88 additions and 6 deletions
+18 -6
View File
@@ -1033,18 +1033,30 @@ def has_non_synchronous_actions(actions: ConfigType) -> bool:
Non-synchronous actions (delay, wait_until, script.wait, etc.) store
trigger args for later execution, making non-owning types like StringRef
unsafe.
Accepts an action list, a single action, or an automation. Only the keys of
action entries are looked up as action names; the options of an action are
only searched for nested action lists (``then:``, ``else:``, ...), so an
option that shares a name with an action, such as the ``delay:`` option of
``speaker_source.set_playlist_delay``, is not mistaken for that action.
"""
if isinstance(actions, list):
return any(has_non_synchronous_actions(item) for item in actions)
if isinstance(actions, dict):
for key in actions:
for key, options in actions.items():
if key in ACTION_REGISTRY and not ACTION_REGISTRY[key].synchronous:
return True
return any(
has_non_synchronous_actions(v)
for v in actions.values()
if isinstance(v, (list, dict))
)
if _options_have_non_synchronous_actions(options):
return True
return False
def _options_have_non_synchronous_actions(options: Any) -> bool:
"""Search an action's options for nested action lists, without treating option keys as actions."""
if isinstance(options, list):
return has_non_synchronous_actions(options)
if isinstance(options, dict):
return any(_options_have_non_synchronous_actions(v) for v in options.values())
return False
+70
View File
@@ -216,6 +216,76 @@ def test_has_non_synchronous_actions_dict_input(
assert has_non_synchronous_actions({"logger.log": "hello"}) is False
def test_has_non_synchronous_actions_option_named_like_action(
mock_registry: dict[str, RegistryEntry],
) -> None:
"""An option that shares a name with a deferring action is not that action."""
actions = [{"logger.log": {"message": "hello", "delay": "1s"}}]
assert has_non_synchronous_actions(actions) is False
def test_has_non_synchronous_actions_nested_option_named_like_action(
mock_registry: dict[str, RegistryEntry],
) -> None:
"""Option keys in nested option dicts are not looked up as action names either."""
actions = [{"logger.log": {"settings": {"wait_until": {"condition": {}}}}}]
assert has_non_synchronous_actions(actions) is False
def test_has_non_synchronous_actions_option_named_like_action_inside_then(
mock_registry: dict[str, RegistryEntry],
) -> None:
actions = [
{
"if": {
"condition": {},
"then": [{"logger.log": {"message": "hello", "delay": "1s"}}],
}
}
]
assert has_non_synchronous_actions(actions) is False
def test_has_non_synchronous_actions_in_else(
mock_registry: dict[str, RegistryEntry],
) -> None:
"""Action lists in any option of an action are searched, not only then:."""
actions = [
{
"if": {
"condition": {},
"then": [{"logger.log": "hello"}],
"else": [{"delay": "1s"}],
}
}
]
assert has_non_synchronous_actions(actions) is True
def test_has_non_synchronous_actions_automation_dict(
mock_registry: dict[str, RegistryEntry],
) -> None:
"""A single automation, as validate_automation(single=True) returns it."""
assert has_non_synchronous_actions({CONF_THEN: [{"delay": "1s"}]}) is True
assert has_non_synchronous_actions({CONF_THEN: [{"logger.log": "hi"}]}) is False
def test_has_non_synchronous_actions_set_playlist_delay() -> None:
"""speaker_source.set_playlist_delay is synchronous despite its delay: option."""
import esphome.components.speaker_source.media_player # noqa: F401 # registers the action
actions = [
{
"speaker_source.set_playlist_delay": {
CONF_ID: "player",
"pipeline": 0,
"delay": 1000,
}
}
]
assert has_non_synchronous_actions(actions) is False
def test_validate_no_deferring_actions_accepts_synchronous(
mock_registry: dict[str, RegistryEntry],
) -> None: