mirror of
https://github.com/esphome/esphome.git
synced 2026-10-03 01:35:59 +00:00
[automation] Add CallbackAutomation dataclass and build_callback_automations helper (#15246)
This commit is contained in:
J. Nick Koston
@@ -1,14 +1,16 @@
|
||||
"""Tests for esphome.automation module."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock, call, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from esphome.automation import (
|
||||
CallbackAutomation,
|
||||
TriggerForwarder,
|
||||
TriggerOnFalseForwarder,
|
||||
TriggerOnTrueForwarder,
|
||||
build_callback_automations,
|
||||
has_non_synchronous_actions,
|
||||
)
|
||||
from esphome.cpp_generator import MockObj, RawExpression
|
||||
@@ -254,3 +256,222 @@ def test_trigger_forwarder_custom_type() -> None:
|
||||
custom = MockObj("MyForwarder", "")
|
||||
result = _build_forwarder("auto_1", [], forwarder=custom)
|
||||
assert result == "MyForwarder{auto_1}"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_build_callback() -> Generator[AsyncMock]:
|
||||
"""Patch build_callback_automation to capture calls."""
|
||||
with patch(
|
||||
"esphome.automation.build_callback_automation", new_callable=AsyncMock
|
||||
) as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_empty_entries(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""No entries means no calls."""
|
||||
parent = MockObj("var", "->")
|
||||
await build_callback_automations(parent, {}, ())
|
||||
mock_build_callback.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_missing_config_key(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Entry present but config key missing -- no calls."""
|
||||
parent = MockObj("var", "->")
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
{},
|
||||
(CallbackAutomation("on_state", "add_on_state_callback", [(bool, "x")]),),
|
||||
)
|
||||
mock_build_callback.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_single_entry(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Single entry with one config triggers one call."""
|
||||
parent = MockObj("var", "->")
|
||||
conf: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {"on_state": [conf]}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(CallbackAutomation("on_state", "add_on_state_callback", [(bool, "x")]),),
|
||||
)
|
||||
mock_build_callback.assert_called_once_with(
|
||||
parent, "add_on_state_callback", [(bool, "x")], conf, forwarder=None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_multiple_configs(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Single entry with multiple configs triggers multiple calls."""
|
||||
parent = MockObj("var", "->")
|
||||
conf1: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
conf2: dict[str, object] = {"automation_id": "auto_2", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {"on_state": [conf1, conf2]}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(CallbackAutomation("on_state", "add_on_state_callback", [(bool, "x")]),),
|
||||
)
|
||||
assert mock_build_callback.call_count == 2
|
||||
mock_build_callback.assert_any_call(
|
||||
parent, "add_on_state_callback", [(bool, "x")], conf1, forwarder=None
|
||||
)
|
||||
mock_build_callback.assert_any_call(
|
||||
parent, "add_on_state_callback", [(bool, "x")], conf2, forwarder=None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_multiple_entries(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Multiple entries each with one config."""
|
||||
parent = MockObj("var", "->")
|
||||
conf_a: dict[str, object] = {"automation_id": "auto_a", "then": []}
|
||||
conf_b: dict[str, object] = {"automation_id": "auto_b", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {
|
||||
"on_value": [conf_a],
|
||||
"on_raw_value": [conf_b],
|
||||
}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(
|
||||
CallbackAutomation("on_value", "add_on_value_callback", [(float, "x")]),
|
||||
CallbackAutomation(
|
||||
"on_raw_value", "add_on_raw_value_callback", [(float, "x")]
|
||||
),
|
||||
),
|
||||
)
|
||||
assert mock_build_callback.call_count == 2
|
||||
assert mock_build_callback.call_args_list == [
|
||||
call(parent, "add_on_value_callback", [(float, "x")], conf_a, forwarder=None),
|
||||
call(
|
||||
parent, "add_on_raw_value_callback", [(float, "x")], conf_b, forwarder=None
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_with_forwarder(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Entry with forwarder passes it through."""
|
||||
parent = MockObj("var", "->")
|
||||
conf: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {"on_press": [conf]}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(
|
||||
CallbackAutomation(
|
||||
"on_press", "add_on_state_callback", forwarder=TriggerOnTrueForwarder
|
||||
),
|
||||
),
|
||||
)
|
||||
mock_build_callback.assert_called_once_with(
|
||||
parent, "add_on_state_callback", [], conf, forwarder=TriggerOnTrueForwarder
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_mixed_entries(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Mix of entries with args, forwarders, and defaults."""
|
||||
parent = MockObj("var", "->")
|
||||
conf_state: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
conf_press: dict[str, object] = {"automation_id": "auto_2", "then": []}
|
||||
conf_release: dict[str, object] = {"automation_id": "auto_3", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {
|
||||
"on_state": [conf_state],
|
||||
"on_press": [conf_press],
|
||||
"on_release": [conf_release],
|
||||
}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(
|
||||
CallbackAutomation("on_state", "add_on_state_callback", [(bool, "x")]),
|
||||
CallbackAutomation(
|
||||
"on_press", "add_on_state_callback", forwarder=TriggerOnTrueForwarder
|
||||
),
|
||||
CallbackAutomation(
|
||||
"on_release", "add_on_state_callback", forwarder=TriggerOnFalseForwarder
|
||||
),
|
||||
),
|
||||
)
|
||||
assert mock_build_callback.call_count == 3
|
||||
assert mock_build_callback.call_args_list == [
|
||||
call(
|
||||
parent, "add_on_state_callback", [(bool, "x")], conf_state, forwarder=None
|
||||
),
|
||||
call(
|
||||
parent,
|
||||
"add_on_state_callback",
|
||||
[],
|
||||
conf_press,
|
||||
forwarder=TriggerOnTrueForwarder,
|
||||
),
|
||||
call(
|
||||
parent,
|
||||
"add_on_state_callback",
|
||||
[],
|
||||
conf_release,
|
||||
forwarder=TriggerOnFalseForwarder,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_skips_missing_keys(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Entries whose config keys are absent are silently skipped."""
|
||||
parent = MockObj("var", "->")
|
||||
conf: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {"on_press": [conf]}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(
|
||||
CallbackAutomation(
|
||||
"on_press", "add_on_state_callback", forwarder=TriggerOnTrueForwarder
|
||||
),
|
||||
CallbackAutomation(
|
||||
"on_release", "add_on_state_callback", forwarder=TriggerOnFalseForwarder
|
||||
),
|
||||
),
|
||||
)
|
||||
mock_build_callback.assert_called_once_with(
|
||||
parent, "add_on_state_callback", [], conf, forwarder=TriggerOnTrueForwarder
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_callback_automations_defaults(
|
||||
mock_build_callback: AsyncMock,
|
||||
) -> None:
|
||||
"""Verify CallbackAutomation with only required fields defaults args=[] and forwarder=None."""
|
||||
parent = MockObj("var", "->")
|
||||
conf: dict[str, object] = {"automation_id": "auto_1", "then": []}
|
||||
config: dict[str, list[dict[str, object]]] = {"on_press": [conf]}
|
||||
await build_callback_automations(
|
||||
parent,
|
||||
config,
|
||||
(CallbackAutomation("on_press", "add_on_press_callback"),),
|
||||
)
|
||||
mock_build_callback.assert_called_once_with(
|
||||
parent, "add_on_press_callback", [], conf, forwarder=None
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user