"""The tests for the trigger helper."""
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
import pytest
import voluptuous as vol
from homeassistant.core import Context, HomeAssistant, ServiceCall, callback
from homeassistant.helpers.trigger import (
DATA_PLUGGABLE_ACTIONS,
PluggableAction,
_async_get_trigger_platform,
async_initialize_triggers,
async_validate_trigger_config,
)
from homeassistant.setup import async_setup_component
from tests.common import async_mock_service
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_bad_trigger_platform(hass: HomeAssistant) -> None:
"""Test bad trigger platform."""
with pytest.raises(vol.Invalid) as ex:
await async_validate_trigger_config(hass, [{"platform": "not_a_platform"}])
assert "Invalid platform 'not_a_platform' specified" in str(ex)
async def test_trigger_subtype(hass: HomeAssistant) -> None:
"""Test trigger subtypes."""
with patch(
"homeassistant.helpers.trigger.async_get_integration", return_value=MagicMock()
) as integration_mock:
await _async_get_trigger_platform(hass, {"platform": "test.subtype"})
assert integration_mock.call_args == call(hass, "test")
async def test_trigger_variables(hass: HomeAssistant) -> None:
"""Test trigger variables."""
async def test_if_fires_on_event(hass: HomeAssistant, calls) -> None:
"""Test the firing of events."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": {
"platform": "event",
"event_type": "test_event",
"variables": {
"name": "Paulus",
"via_event": "{{ trigger.event.event_type }}",
},
},
"action": {
"service": "test.automation",
"data_template": {"hello": "{{ name }} + {{ via_event }}"},
},
}
},
)
hass.bus.async_fire("test_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["hello"] == "Paulus + test_event"
async def test_if_disabled_trigger_not_firing(
hass: HomeAssistant, calls: list[ServiceCall]
) -> None:
"""Test disabled triggers don't fire."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"platform": "event",
"event_type": "enabled_trigger_event",
},
{
"enabled": False,
"platform": "event",
"event_type": "disabled_trigger_event",
},
],
"action": {
"service": "test.automation",
},
}
},
)
hass.bus.async_fire("disabled_trigger_event")
await hass.async_block_till_done()
assert not calls
hass.bus.async_fire("enabled_trigger_event")
await hass.async_block_till_done()
assert len(calls) == 1
async def test_trigger_alias(
hass: HomeAssistant, calls: list[ServiceCall], caplog: pytest.LogCaptureFixture
) -> None:
"""Test triggers support aliases."""
assert await async_setup_component(
hass,
"automation",
{
"automation": {
"trigger": [
{
"alias": "My event",
"platform": "event",
"event_type": "trigger_event",
}
],
"action": {
"service": "test.automation",
"data_template": {"alias": "{{ trigger.alias }}"},
},
}
},
)
hass.bus.async_fire("trigger_event")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["alias"] == "My event"
assert (
"Automation trigger 'My event' triggered by event 'trigger_event'"
in caplog.text
)
async def test_async_initialize_triggers(
hass: HomeAssistant, calls: list[ServiceCall], caplog: pytest.LogCaptureFixture
) -> None:
"""Test async_initialize_triggers with different action types."""
log_cb = MagicMock()
action_calls = []
trigger_config = await async_validate_trigger_config(
hass,
[
{
"platform": "event",
"event_type": ["trigger_event"],
"variables": {
"name": "Paulus",
"via_event": "{{ trigger.event.event_type }}",
},
}
],
)
async def async_action(*args):
action_calls.append([*args])
@callback
def cb_action(*args):
action_calls.append([*args])
def non_cb_action(*args):
action_calls.append([*args])
for action in (async_action, cb_action, non_cb_action):
action_calls = []
unsub = await async_initialize_triggers(
hass,
trigger_config,
action,
"test",
"",
log_cb,
)
await hass.async_block_till_done()
hass.bus.async_fire("trigger_event")
await hass.async_block_till_done()
await hass.async_block_till_done()
assert len(action_calls) == 1
assert action_calls[0][0]["name"] == "Paulus"
assert action_calls[0][0]["via_event"] == "trigger_event"
log_cb.assert_called_once_with(ANY, "Initialized trigger")
log_cb.reset_mock()
unsub()
async def test_pluggable_action(hass: HomeAssistant, calls: list[ServiceCall]):
"""Test normal behavior of pluggable actions."""
update_1 = MagicMock()
update_2 = MagicMock()
action_1 = AsyncMock()
action_2 = AsyncMock()
trigger_1 = {"domain": "test", "device": "1"}
trigger_2 = {"domain": "test", "device": "2"}
variables_1 = {"source": "test 1"}
variables_2 = {"source": "test 2"}
context_1 = Context()
context_2 = Context()
plug_1 = PluggableAction(update_1)
plug_2 = PluggableAction(update_2)
# Verify plug is inactive without triggers
remove_plug_1 = plug_1.async_register(hass, trigger_1)
assert not plug_1
assert not plug_2
# Verify plug remain inactive with non matching trigger
remove_attach_2 = PluggableAction.async_attach_trigger(
hass, trigger_2, action_2, variables_2
)
assert not plug_1
assert not plug_2
update_1.assert_not_called()
update_2.assert_not_called()
# Verify plug is active, and update when matching trigger attaches
remove_attach_1 = PluggableAction.async_attach_trigger(
hass, trigger_1, action_1, variables_1
)
assert plug_1
assert not plug_2
update_1.assert_called()
update_1.reset_mock()
update_2.assert_not_called()
# Verify a non registered plug is inactive
remove_plug_1()
assert not plug_1
assert not plug_2
# Verify a plug registered to existing trigger is true
remove_plug_1 = plug_1.async_register(hass, trigger_1)
assert plug_1
assert not plug_2
remove_plug_2 = plug_2.async_register(hass, trigger_2)
assert plug_1
assert plug_2
# Verify no actions should have been triggered so far
action_1.assert_not_called()
action_2.assert_not_called()
# Verify action is triggered with correct data
await plug_1.async_run(hass, context_1)
await plug_2.async_run(hass, context_2)
action_1.assert_called_with(variables_1, context_1)
action_2.assert_called_with(variables_2, context_2)
# Verify plug goes inactive if trigger is removed
remove_attach_1()
assert not plug_1
# Verify registry is cleaned when no plugs nor triggers are attached
assert hass.data[DATA_PLUGGABLE_ACTIONS]
remove_plug_1()
remove_plug_2()
remove_attach_2()
assert not hass.data[DATA_PLUGGABLE_ACTIONS]
assert not plug_2