core/tests/helpers/test_trigger.py

"""The tests for the trigger helper."""

from contextlib import AbstractContextManager, nullcontext as does_not_raise
import io
from typing import Any
from unittest.mock import ANY, AsyncMock, MagicMock, Mock, call, patch

import pytest
from pytest_unordered import unordered
import voluptuous as vol

from homeassistant.components import automation
from homeassistant.components.sun import DOMAIN as DOMAIN_SUN
from homeassistant.components.system_health import DOMAIN as DOMAIN_SYSTEM_HEALTH
from homeassistant.components.tag import DOMAIN as DOMAIN_TAG
from homeassistant.components.text import DOMAIN as DOMAIN_TEXT
from homeassistant.const import (
    CONF_ABOVE,
    CONF_BELOW,
    CONF_ENTITY_ID,
    CONF_OPTIONS,
    CONF_PLATFORM,
    CONF_TARGET,
    STATE_UNAVAILABLE,
    STATE_UNKNOWN,
)
from homeassistant.core import (
    CALLBACK_TYPE,
    Context,
    HomeAssistant,
    ServiceCall,
    callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, trigger
from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
from homeassistant.helpers.trigger import (
    CONF_LOWER_LIMIT,
    CONF_THRESHOLD_TYPE,
    CONF_UPPER_LIMIT,
    DATA_PLUGGABLE_ACTIONS,
    PluggableAction,
    ThresholdType,
    Trigger,
    TriggerActionRunner,
    _async_get_trigger_platform,
    async_initialize_triggers,
    async_validate_trigger_config,
    make_entity_numerical_state_attribute_changed_trigger,
    make_entity_numerical_state_attribute_crossed_threshold_trigger,
)
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import Integration, async_get_integration
from homeassistant.setup import async_setup_component
from homeassistant.util.yaml.loader import parse_yaml

from tests.common import MockModule, MockPlatform, mock_integration, mock_platform
from tests.typing import WebSocketGenerator


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 trigger '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(async_get_platform=AsyncMock()),
    ) as integration_mock:
        await _async_get_trigger_platform(hass, "test.subtype")
        assert integration_mock.call_args == call(hass, "test")


async def test_trigger_variables(
    hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
    """Test trigger variables."""
    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(service_calls) == 1
    assert service_calls[0].data["hello"] == "Paulus + test_event"


async def test_if_disabled_trigger_not_firing(
    hass: HomeAssistant, service_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 service_calls

    hass.bus.async_fire("enabled_trigger_event")
    await hass.async_block_till_done()
    assert len(service_calls) == 1


async def test_trigger_enabled_templates(
    hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
    """Test triggers enabled by template."""
    assert await async_setup_component(
        hass,
        "automation",
        {
            "automation": {
                "trigger": [
                    {
                        "enabled": "{{ 'some text' }}",
                        "platform": "event",
                        "event_type": "truthy_template_trigger_event",
                    },
                    {
                        "enabled": "{{ 3 == 4 }}",
                        "platform": "event",
                        "event_type": "falsy_template_trigger_event",
                    },
                    {
                        "enabled": False,  # eg. from a blueprints input defaulting to `false`
                        "platform": "event",
                        "event_type": "falsy_trigger_event",
                    },
                    {
                        "enabled": "some text",  # eg. from a blueprints input value
                        "platform": "event",
                        "event_type": "truthy_trigger_event",
                    },
                ],
                "action": {
                    "service": "test.automation",
                },
            }
        },
    )

    hass.bus.async_fire("falsy_template_trigger_event")
    await hass.async_block_till_done()
    assert not service_calls

    hass.bus.async_fire("falsy_trigger_event")
    await hass.async_block_till_done()
    assert not service_calls

    hass.bus.async_fire("truthy_template_trigger_event")
    await hass.async_block_till_done()
    assert len(service_calls) == 1

    hass.bus.async_fire("truthy_trigger_event")
    await hass.async_block_till_done()
    assert len(service_calls) == 2


async def test_nested_trigger_list(
    hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
    """Test triggers within nested list."""

    assert await async_setup_component(
        hass,
        "automation",
        {
            "automation": {
                "trigger": [
                    {
                        "triggers": {
                            "platform": "event",
                            "event_type": "trigger_1",
                        },
                    },
                    {
                        "platform": "event",
                        "event_type": "trigger_2",
                    },
                    {"triggers": []},
                    {"triggers": None},
                    {
                        "triggers": [
                            {
                                "platform": "event",
                                "event_type": "trigger_3",
                            },
                            {
                                "platform": "event",
                                "event_type": "trigger_4",
                            },
                        ],
                    },
                ],
                "action": {
                    "service": "test.automation",
                },
            }
        },
    )

    hass.bus.async_fire("trigger_1")
    await hass.async_block_till_done()
    assert len(service_calls) == 1

    hass.bus.async_fire("trigger_2")
    await hass.async_block_till_done()
    assert len(service_calls) == 2

    hass.bus.async_fire("trigger_none")
    await hass.async_block_till_done()
    assert len(service_calls) == 2

    hass.bus.async_fire("trigger_3")
    await hass.async_block_till_done()
    assert len(service_calls) == 3

    hass.bus.async_fire("trigger_4")
    await hass.async_block_till_done()
    assert len(service_calls) == 4


async def test_trigger_enabled_template_limited(
    hass: HomeAssistant,
    service_calls: list[ServiceCall],
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test triggers enabled invalid template."""
    assert await async_setup_component(
        hass,
        "automation",
        {
            "automation": {
                "trigger": [
                    {
                        "enabled": "{{ states('sensor.limited') }}",  # only limited template supported
                        "platform": "event",
                        "event_type": "test_event",
                    },
                ],
                "action": {
                    "service": "test.automation",
                },
            }
        },
    )

    hass.bus.async_fire("test_event")
    await hass.async_block_till_done()
    assert not service_calls
    assert "Error rendering enabled template" in caplog.text


async def test_trigger_alias(
    hass: HomeAssistant,
    service_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(service_calls) == 1
    assert service_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,
    service_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, service_calls: list[ServiceCall]
) -> None:
    """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


class TriggerActionFunctionTypeHelper:
    """Helper for testing different trigger action function types."""

    def __init__(self) -> None:
        """Init helper."""
        self.action_calls = []

    @callback
    def cb_action(self, *args):
        """Callback action."""
        self.action_calls.append([*args])

    def sync_action(self, *args):
        """Sync action."""
        self.action_calls.append([*args])

    async def async_action(self, *args):
        """Async action."""
        self.action_calls.append([*args])


@pytest.mark.parametrize("action_method", ["cb_action", "sync_action", "async_action"])
async def test_platform_multiple_triggers(
    hass: HomeAssistant, action_method: str
) -> None:
    """Test a trigger platform with multiple trigger."""

    class MockTrigger(Trigger):
        """Mock trigger."""

        @classmethod
        async def async_validate_config(
            cls, hass: HomeAssistant, config: ConfigType
        ) -> ConfigType:
            """Validate config."""
            return config

    class MockTrigger1(MockTrigger):
        """Mock trigger 1."""

        async def async_attach_runner(
            self, run_action: TriggerActionRunner
        ) -> CALLBACK_TYPE:
            """Attach a trigger."""
            run_action({"extra": "test_trigger_1"}, "trigger 1 desc")

    class MockTrigger2(MockTrigger):
        """Mock trigger 2."""

        async def async_attach_runner(
            self, run_action: TriggerActionRunner
        ) -> CALLBACK_TYPE:
            """Attach a trigger."""
            run_action({"extra": "test_trigger_2"}, "trigger 2 desc")

    async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
        return {
            "_": MockTrigger1,
            "trig_2": MockTrigger2,
        }

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))

    config_1 = [{"platform": "test"}]
    config_2 = [{"platform": "test.trig_2", "options": {"x": 1}}]
    config_3 = [{"platform": "test.unknown_trig"}]
    assert await async_validate_trigger_config(hass, config_1) == config_1
    assert await async_validate_trigger_config(hass, config_2) == config_2
    with pytest.raises(
        vol.Invalid, match="Invalid trigger 'test.unknown_trig' specified"
    ):
        await async_validate_trigger_config(hass, config_3)

    log_cb = MagicMock()

    action_helper = TriggerActionFunctionTypeHelper()
    action_method = getattr(action_helper, action_method)

    await async_initialize_triggers(hass, config_1, action_method, "test", "", log_cb)
    assert len(action_helper.action_calls) == 1
    assert action_helper.action_calls[0][0] == {
        "trigger": {
            "alias": None,
            "description": "trigger 1 desc",
            "extra": "test_trigger_1",
            "id": "0",
            "idx": "0",
            "platform": "test",
        }
    }
    action_helper.action_calls.clear()

    await async_initialize_triggers(hass, config_2, action_method, "test", "", log_cb)
    assert len(action_helper.action_calls) == 1
    assert action_helper.action_calls[0][0] == {
        "trigger": {
            "alias": None,
            "description": "trigger 2 desc",
            "extra": "test_trigger_2",
            "id": "0",
            "idx": "0",
            "platform": "test.trig_2",
        }
    }
    action_helper.action_calls.clear()

    with pytest.raises(KeyError):
        await async_initialize_triggers(
            hass, config_3, action_method, "test", "", log_cb
        )


async def test_platform_migrate_trigger(hass: HomeAssistant) -> None:
    """Test a trigger platform with a migration."""

    OPTIONS_SCHEMA_DICT = {
        vol.Required("option_1"): str,
        vol.Optional("option_2"): int,
    }

    class MockTrigger(Trigger):
        """Mock trigger."""

        @classmethod
        async def async_validate_complete_config(
            cls, hass: HomeAssistant, complete_config: ConfigType
        ) -> ConfigType:
            """Validate complete config."""
            complete_config = move_top_level_schema_fields_to_options(
                complete_config, OPTIONS_SCHEMA_DICT
            )
            return await super().async_validate_complete_config(hass, complete_config)

        @classmethod
        async def async_validate_config(
            cls, hass: HomeAssistant, config: ConfigType
        ) -> ConfigType:
            """Validate config."""
            return config

    async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
        return {
            "_": MockTrigger,
        }

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))

    config_1 = [{"platform": "test", "option_1": "value_1", "option_2": 2}]
    config_2 = [{"platform": "test", "option_1": "value_1"}]
    config_3 = [{"platform": "test", "options": {"option_1": "value_1", "option_2": 2}}]
    config_4 = [{"platform": "test", "options": {"option_1": "value_1"}}]

    assert await async_validate_trigger_config(hass, config_1) == config_3
    assert await async_validate_trigger_config(hass, config_2) == config_4
    assert await async_validate_trigger_config(hass, config_3) == config_3
    assert await async_validate_trigger_config(hass, config_4) == config_4


async def test_platform_backwards_compatibility_for_new_style_configs(
    hass: HomeAssistant,
) -> None:
    """Test backwards compatibility for old-style triggers with new-style configs."""

    class MockTriggerPlatform:
        """Mock trigger platform."""

        TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
            {
                vol.Required("option_1"): str,
                vol.Optional("option_2"): int,
            }
        )

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", MockTriggerPlatform())

    config_old_style = [{"platform": "test", "option_1": "value_1", "option_2": 2}]
    result = await async_validate_trigger_config(hass, config_old_style)
    assert result == config_old_style

    config_new_style = [
        {"platform": "test", "options": {"option_1": "value_1", "option_2": 2}}
    ]
    result = await async_validate_trigger_config(hass, config_new_style)
    assert result == config_old_style


@pytest.mark.parametrize(
    "sun_trigger_descriptions",
    [
        """
        _:
          fields:
            event:
              example: sunrise
              selector:
                select:
                  options:
                    - sunrise
                    - sunset
            offset:
              selector:
                time: null
        """,
        """
        .anchor: &anchor
          - sunrise
          - sunset
        _:
          fields:
            event:
              example: sunrise
              selector:
                select:
                  options: *anchor
            offset:
              selector:
                time: null
        """,
    ],
)
# Patch out binary sensor triggers, because loading sun triggers also loads
# binary sensor triggers and those are irrelevant for this test
@patch(
    "homeassistant.components.binary_sensor.trigger.async_get_triggers",
    new=AsyncMock(return_value={}),
)
async def test_async_get_all_descriptions(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    sun_trigger_descriptions: str,
) -> None:
    """Test async_get_all_descriptions."""
    tag_trigger_descriptions = """
        _:
          target:
            entity:
              domain: alarm_control_panel
        """
    text_trigger_descriptions = """
        changed:
          target:
            entity:
              domain: text
        """

    ws_client = await hass_ws_client(hass)

    assert await async_setup_component(hass, DOMAIN_SUN, {})
    assert await async_setup_component(hass, DOMAIN_SYSTEM_HEALTH, {})
    await hass.async_block_till_done()

    def _load_yaml(fname, secrets=None):
        if fname.endswith("sun/triggers.yaml"):
            trigger_descriptions = sun_trigger_descriptions
        elif fname.endswith("tag/triggers.yaml"):
            trigger_descriptions = tag_trigger_descriptions
        elif fname.endswith("text/triggers.yaml"):
            trigger_descriptions = text_trigger_descriptions
        with io.StringIO(trigger_descriptions) as file:
            return parse_yaml(file)

    with (
        patch(
            "homeassistant.helpers.trigger._load_triggers_files",
            side_effect=trigger._load_triggers_files,
        ) as proxy_load_triggers_files,
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        descriptions = await trigger.async_get_all_descriptions(hass)

    # Test we only load triggers.yaml for integrations with triggers,
    # system_health has no triggers
    assert proxy_load_triggers_files.mock_calls[0][1][0] == unordered(
        [
            await async_get_integration(hass, DOMAIN_SUN),
        ]
    )

    # system_health does not have triggers and should not be in descriptions
    expected_descriptions = {
        "sun": {
            "fields": {
                "event": {
                    "example": "sunrise",
                    "selector": {
                        "select": {
                            "custom_value": False,
                            "multiple": False,
                            "options": ["sunrise", "sunset"],
                            "sort": False,
                        }
                    },
                },
                "offset": {"selector": {"time": {}}},
            }
        }
    }

    assert descriptions == expected_descriptions

    # Verify the cache returns the same object
    assert await trigger.async_get_all_descriptions(hass) is descriptions

    # Load the tag integration and check a new cache object is created
    assert await async_setup_component(hass, DOMAIN_TAG, {})
    await hass.async_block_till_done()

    with (
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        new_descriptions = await trigger.async_get_all_descriptions(hass)
    assert new_descriptions is not descriptions
    # The tag trigger should now be present
    expected_descriptions |= {
        "tag": {
            "target": {
                "entity": [
                    {
                        "domain": ["alarm_control_panel"],
                    }
                ],
            },
            "fields": {},
        },
    }
    assert new_descriptions == expected_descriptions

    # Verify the cache returns the same object
    assert await trigger.async_get_all_descriptions(hass) is new_descriptions

    # Load the text integration and check a new cache object is created
    assert await async_setup_component(hass, DOMAIN_TEXT, {})
    await hass.async_block_till_done()

    with (
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        new_descriptions = await trigger.async_get_all_descriptions(hass)
    assert new_descriptions is not descriptions
    # No text triggers added, they are gated by the automation.new_triggers_conditions
    # labs flag
    assert new_descriptions == expected_descriptions

    # Verify the cache returns the same object
    assert await trigger.async_get_all_descriptions(hass) is new_descriptions

    # Enable the new_triggers_conditions flag and verify text triggers are loaded
    assert await async_setup_component(hass, "labs", {})

    await ws_client.send_json_auto_id(
        {
            "type": "labs/update",
            "domain": "automation",
            "preview_feature": "new_triggers_conditions",
            "enabled": True,
        }
    )

    msg = await ws_client.receive_json()
    assert msg["success"]
    await hass.async_block_till_done()

    with (
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        new_descriptions = await trigger.async_get_all_descriptions(hass)
    assert new_descriptions is not descriptions
    # The text triggers should now be present
    assert new_descriptions == expected_descriptions | {
        "text.changed": {
            "fields": {},
            "target": {
                "entity": [
                    {
                        "domain": [
                            "text",
                        ],
                    },
                ],
            },
        },
    }

    # Verify the cache returns the same object
    assert await trigger.async_get_all_descriptions(hass) is new_descriptions

    # Disable the new_triggers_conditions flag and verify text triggers are removed
    assert await async_setup_component(hass, "labs", {})

    await ws_client.send_json_auto_id(
        {
            "type": "labs/update",
            "domain": "automation",
            "preview_feature": "new_triggers_conditions",
            "enabled": False,
        }
    )

    msg = await ws_client.receive_json()
    assert msg["success"]
    await hass.async_block_till_done()

    with (
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        new_descriptions = await trigger.async_get_all_descriptions(hass)
    assert new_descriptions is not descriptions
    # The text triggers should no longer be present
    assert new_descriptions == expected_descriptions

    # Verify the cache returns the same object
    assert await trigger.async_get_all_descriptions(hass) is new_descriptions


@pytest.mark.parametrize(
    ("yaml_error", "expected_message"),
    [
        (
            FileNotFoundError("Blah"),
            "Unable to find triggers.yaml for the sun integration",
        ),
        (
            HomeAssistantError("Test error"),
            "Unable to parse triggers.yaml for the sun integration: Test error",
        ),
    ],
)
async def test_async_get_all_descriptions_with_yaml_error(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    yaml_error: Exception,
    expected_message: str,
) -> None:
    """Test async_get_all_descriptions."""
    assert await async_setup_component(hass, DOMAIN_SUN, {})
    await hass.async_block_till_done()

    def _load_yaml_dict(fname, secrets=None):
        raise yaml_error

    with (
        patch(
            "homeassistant.helpers.trigger.load_yaml_dict",
            side_effect=_load_yaml_dict,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        descriptions = await trigger.async_get_all_descriptions(hass)

    assert descriptions == {DOMAIN_SUN: None}

    assert expected_message in caplog.text


async def test_async_get_all_descriptions_with_bad_description(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test async_get_all_descriptions."""
    sun_service_descriptions = """
        _:
          fields: not_a_dict
    """

    assert await async_setup_component(hass, DOMAIN_SUN, {})
    await hass.async_block_till_done()

    def _load_yaml(fname, secrets=None):
        with io.StringIO(sun_service_descriptions) as file:
            return parse_yaml(file)

    with (
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=_load_yaml,
        ),
        patch.object(Integration, "has_triggers", return_value=True),
    ):
        descriptions = await trigger.async_get_all_descriptions(hass)

    assert descriptions == {DOMAIN_SUN: None}

    assert (
        "Unable to parse triggers.yaml for the sun integration: "
        "expected a dictionary for dictionary value @ data['_']['fields']"
    ) in caplog.text


async def test_invalid_trigger_platform(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test invalid trigger platform."""
    mock_integration(hass, MockModule("test", async_setup=AsyncMock(return_value=True)))
    mock_platform(hass, "test.trigger", MockPlatform())

    await async_setup_component(hass, "test", {})

    assert "Integration test does not provide trigger support, skipping" in caplog.text


@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
async def test_subscribe_triggers(
    mock_has_triggers: Mock,
    mock_load_yaml: Mock,
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test trigger.async_subscribe_platform_events."""
    sun_trigger_descriptions = """
        _: {}
        """

    def _load_yaml(fname, secrets=None):
        if fname.endswith("sun/triggers.yaml"):
            trigger_descriptions = sun_trigger_descriptions
        else:
            raise FileNotFoundError
        with io.StringIO(trigger_descriptions) as file:
            return parse_yaml(file)

    mock_load_yaml.side_effect = _load_yaml

    async def broken_subscriber(_):
        """Simulate a broken subscriber."""
        raise Exception("Boom!")  # noqa: TRY002

    trigger_events = []

    async def good_subscriber(new_triggers: set[str]):
        """Simulate a working subscriber."""
        trigger_events.append(new_triggers)

    trigger.async_subscribe_platform_events(hass, broken_subscriber)
    trigger.async_subscribe_platform_events(hass, good_subscriber)

    assert await async_setup_component(hass, "sun", {})
    assert trigger_events == [{"sun"}]
    assert "Error while notifying trigger platform listener" in caplog.text


@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
@pytest.mark.parametrize(
    ("new_triggers_conditions_enabled", "expected_events"),
    [
        (True, [{"light.turned_off", "light.turned_on"}]),
        (False, []),
    ],
)
async def test_subscribe_triggers_experimental_triggers(
    mock_has_triggers: Mock,
    mock_load_yaml: Mock,
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    caplog: pytest.LogCaptureFixture,
    new_triggers_conditions_enabled: bool,
    expected_events: list[set[str]],
) -> None:
    """Test trigger.async_subscribe_platform_events doesn't send events for disabled triggers."""
    # Return empty triggers.yaml for light integration, the actual trigger descriptions
    # are irrelevant for this test
    light_trigger_descriptions = ""

    def _load_yaml(fname, secrets=None):
        if fname.endswith("light/triggers.yaml"):
            trigger_descriptions = light_trigger_descriptions
        else:
            raise FileNotFoundError
        with io.StringIO(trigger_descriptions) as file:
            return parse_yaml(file)

    mock_load_yaml.side_effect = _load_yaml

    trigger_events = []

    async def good_subscriber(new_triggers: set[str]):
        """Simulate a working subscriber."""
        trigger_events.append(new_triggers)

    ws_client = await hass_ws_client(hass)

    assert await async_setup_component(hass, "labs", {})
    await ws_client.send_json_auto_id(
        {
            "type": "labs/update",
            "domain": "automation",
            "preview_feature": "new_triggers_conditions",
            "enabled": new_triggers_conditions_enabled,
        }
    )

    msg = await ws_client.receive_json()
    assert msg["success"]
    await hass.async_block_till_done()

    trigger.async_subscribe_platform_events(hass, good_subscriber)

    assert await async_setup_component(hass, "light", {})
    await hass.async_block_till_done()
    assert trigger_events == expected_events


@patch("annotatedyaml.loader.load_yaml")
@patch.object(Integration, "has_triggers", return_value=True)
@patch(
    "homeassistant.components.light.trigger.async_get_triggers",
    new=AsyncMock(return_value={}),
)
async def test_subscribe_triggers_no_triggers(
    mock_has_triggers: Mock,
    mock_load_yaml: Mock,
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test trigger.async_subscribe_platform_events doesn't send events for platforms without triggers."""
    # Return empty triggers.yaml for light integration, the actual trigger descriptions
    # are irrelevant for this test
    light_trigger_descriptions = ""

    def _load_yaml(fname, secrets=None):
        if fname.endswith("light/triggers.yaml"):
            trigger_descriptions = light_trigger_descriptions
        else:
            raise FileNotFoundError
        with io.StringIO(trigger_descriptions) as file:
            return parse_yaml(file)

    mock_load_yaml.side_effect = _load_yaml

    trigger_events = []

    async def good_subscriber(new_triggers: set[str]):
        """Simulate a working subscriber."""
        trigger_events.append(new_triggers)

    ws_client = await hass_ws_client(hass)

    assert await async_setup_component(hass, "labs", {})
    await ws_client.send_json_auto_id(
        {
            "type": "labs/update",
            "domain": "automation",
            "preview_feature": "new_triggers_conditions",
            "enabled": True,
        }
    )

    msg = await ws_client.receive_json()
    assert msg["success"]
    await hass.async_block_till_done()

    trigger.async_subscribe_platform_events(hass, good_subscriber)

    assert await async_setup_component(hass, "light", {})
    await hass.async_block_till_done()
    assert trigger_events == []


@pytest.mark.parametrize(
    ("trigger_options", "expected_result"),
    [
        # Test validating climate.target_temperature_changed
        # Valid configurations
        (
            {},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: 10},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: "sensor.test"},
            does_not_raise(),
        ),
        (
            {CONF_BELOW: 90},
            does_not_raise(),
        ),
        (
            {CONF_BELOW: "sensor.test"},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: 10, CONF_BELOW: 90},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: "sensor.test", CONF_BELOW: 90},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: 10, CONF_BELOW: "sensor.test"},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: "sensor.test", CONF_BELOW: "sensor.test"},
            does_not_raise(),
        ),
        # Test verbose choose selector options
        (
            {CONF_ABOVE: {"chosen_selector": "entity", "entity": "sensor.test"}},
            does_not_raise(),
        ),
        (
            {CONF_ABOVE: {"chosen_selector": "number", "number": 10}},
            does_not_raise(),
        ),
        (
            {CONF_BELOW: {"chosen_selector": "entity", "entity": "sensor.test"}},
            does_not_raise(),
        ),
        (
            {CONF_BELOW: {"chosen_selector": "number", "number": 90}},
            does_not_raise(),
        ),
        # Test invalid configurations
        (
            # Must be valid entity id
            {CONF_ABOVE: "cat", CONF_BELOW: "dog"},
            pytest.raises(vol.Invalid),
        ),
        (
            # Above must be smaller than below
            {CONF_ABOVE: 90, CONF_BELOW: 10},
            pytest.raises(vol.Invalid),
        ),
        (
            # Invalid choose selector option
            {CONF_BELOW: {"chosen_selector": "cat", "cat": 90}},
            pytest.raises(vol.Invalid),
        ),
    ],
)
async def test_numerical_state_attribute_changed_trigger_config_validation(
    hass: HomeAssistant,
    trigger_options: dict[str, Any],
    expected_result: AbstractContextManager,
) -> None:
    """Test numerical state attribute change trigger config validation."""

    async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
        return {
            "test_trigger": make_entity_numerical_state_attribute_changed_trigger(
                "test", "test_attribute"
            ),
        }

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))

    with expected_result:
        await async_validate_trigger_config(
            hass,
            [
                {
                    "platform": "test.test_trigger",
                    CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
                    CONF_OPTIONS: trigger_options,
                }
            ],
        )


async def test_numerical_state_attribute_changed_error_handling(
    hass: HomeAssistant, service_calls: list[ServiceCall]
) -> None:
    """Test numerical state attribute change error handling."""

    async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
        return {
            "attribute_changed": make_entity_numerical_state_attribute_changed_trigger(
                "test", "test_attribute"
            ),
        }

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))

    hass.states.async_set("test.test_entity", "on", {"test_attribute": 20})

    options = {
        CONF_OPTIONS: {CONF_ABOVE: "sensor.above", CONF_BELOW: "sensor.below"},
    }

    await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "trigger": {
                    CONF_PLATFORM: "test.attribute_changed",
                    CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
                }
                | options,
                "action": {
                    "service": "test.automation",
                    "data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
                },
            }
        },
    )

    assert len(service_calls) == 0

    # Test the trigger works
    hass.states.async_set("sensor.above", "10")
    hass.states.async_set("sensor.below", "90")
    hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
    await hass.async_block_till_done()
    assert len(service_calls) == 1
    service_calls.clear()

    # Test the trigger fires again when still within limits
    hass.states.async_set("test.test_entity", "on", {"test_attribute": 51})
    await hass.async_block_till_done()
    assert len(service_calls) == 1
    service_calls.clear()

    # Test the trigger does not fire when the from-state is unknown or unavailable
    for from_state in (STATE_UNKNOWN, STATE_UNAVAILABLE):
        hass.states.async_set("test.test_entity", from_state)
        hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
        await hass.async_block_till_done()
        assert len(service_calls) == 0

    # Test the trigger does not fire when the attribute value is outside the limits
    for value in (5, 95):
        hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
        await hass.async_block_till_done()
        assert len(service_calls) == 0

    # Test the trigger does not fire when the attribute value is missing
    hass.states.async_set("test.test_entity", "on", {})
    await hass.async_block_till_done()
    assert len(service_calls) == 0

    # Test the trigger does not fire when the attribute value is invalid
    for value in ("cat", None):
        hass.states.async_set("test.test_entity", "on", {"test_attribute": value})
        await hass.async_block_till_done()
        assert len(service_calls) == 0

    # Test the trigger does not fire when the above sensor does not exist
    hass.states.async_remove("sensor.above")
    hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
    hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
    await hass.async_block_till_done()
    assert len(service_calls) == 0

    # Test the trigger does not fire when the above sensor state is not numeric
    for invalid_value in ("cat", None):
        hass.states.async_set("sensor.above", invalid_value)
        hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
        hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
        await hass.async_block_till_done()
        assert len(service_calls) == 0

    # Reset the above sensor state to a valid numeric value
    hass.states.async_set("sensor.above", "10")

    # Test the trigger does not fire when the below sensor does not exist
    hass.states.async_remove("sensor.below")
    hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
    hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
    await hass.async_block_till_done()
    assert len(service_calls) == 0

    # Test the trigger does not fire when the below sensor state is not numeric
    for invalid_value in ("cat", None):
        hass.states.async_set("sensor.below", invalid_value)
        hass.states.async_set("test.test_entity", "on", {"test_attribute": None})
        hass.states.async_set("test.test_entity", "on", {"test_attribute": 50})
        await hass.async_block_till_done()
        assert len(service_calls) == 0


@pytest.mark.parametrize(
    ("trigger_options", "expected_result"),
    [
        # Valid configurations
        (
            {CONF_THRESHOLD_TYPE: ThresholdType.ABOVE, CONF_LOWER_LIMIT: 10},
            does_not_raise(),
        ),
        (
            {CONF_THRESHOLD_TYPE: ThresholdType.ABOVE, CONF_LOWER_LIMIT: "sensor.test"},
            does_not_raise(),
        ),
        (
            {CONF_THRESHOLD_TYPE: ThresholdType.BELOW, CONF_UPPER_LIMIT: 90},
            does_not_raise(),
        ),
        (
            {CONF_THRESHOLD_TYPE: ThresholdType.BELOW, CONF_UPPER_LIMIT: "sensor.test"},
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_LOWER_LIMIT: 10,
                CONF_UPPER_LIMIT: 90,
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_LOWER_LIMIT: 10,
                CONF_UPPER_LIMIT: "sensor.test",
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_LOWER_LIMIT: "sensor.test",
                CONF_UPPER_LIMIT: 90,
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_LOWER_LIMIT: "sensor.test",
                CONF_UPPER_LIMIT: "sensor.test",
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
                CONF_LOWER_LIMIT: 10,
                CONF_UPPER_LIMIT: 90,
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
                CONF_LOWER_LIMIT: 10,
                CONF_UPPER_LIMIT: "sensor.test",
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
                CONF_LOWER_LIMIT: "sensor.test",
                CONF_UPPER_LIMIT: 90,
            },
            does_not_raise(),
        ),
        (
            {
                CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE,
                CONF_LOWER_LIMIT: "sensor.test",
                CONF_UPPER_LIMIT: "sensor.test",
            },
            does_not_raise(),
        ),
        # Test verbose choose selector options
        # Test invalid configurations
        (
            # Missing threshold type
            {},
            pytest.raises(vol.Invalid),
        ),
        (
            # Invalid threshold type
            {CONF_THRESHOLD_TYPE: "cat"},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide lower limit for ABOVE
            {CONF_THRESHOLD_TYPE: ThresholdType.ABOVE},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide lower limit for ABOVE
            {CONF_THRESHOLD_TYPE: ThresholdType.ABOVE, CONF_UPPER_LIMIT: 90},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper limit for BELOW
            {CONF_THRESHOLD_TYPE: ThresholdType.BELOW},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper limit for BELOW
            {CONF_THRESHOLD_TYPE: ThresholdType.BELOW, CONF_LOWER_LIMIT: 10},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for BETWEEN
            {CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for BETWEEN
            {CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN, CONF_LOWER_LIMIT: 10},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for BETWEEN
            {CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN, CONF_UPPER_LIMIT: 90},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for OUTSIDE
            {CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for OUTSIDE
            {CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE, CONF_LOWER_LIMIT: 10},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must provide upper and lower limits for OUTSIDE
            {CONF_THRESHOLD_TYPE: ThresholdType.OUTSIDE, CONF_UPPER_LIMIT: 90},
            pytest.raises(vol.Invalid),
        ),
        (
            # Must be valid entity id
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_ABOVE: "cat",
                CONF_BELOW: "dog",
            },
            pytest.raises(vol.Invalid),
        ),
        (
            # Above must be smaller than below
            {
                CONF_THRESHOLD_TYPE: ThresholdType.BETWEEN,
                CONF_ABOVE: 90,
                CONF_BELOW: 10,
            },
            pytest.raises(vol.Invalid),
        ),
    ],
)
async def test_numerical_state_attribute_crossed_threshold_trigger_config_validation(
    hass: HomeAssistant,
    trigger_options: dict[str, Any],
    expected_result: AbstractContextManager,
) -> None:
    """Test numerical state attribute change trigger config validation."""

    async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
        return {
            "test_trigger": make_entity_numerical_state_attribute_crossed_threshold_trigger(
                "test", "test_attribute"
            ),
        }

    mock_integration(hass, MockModule("test"))
    mock_platform(hass, "test.trigger", Mock(async_get_triggers=async_get_triggers))

    with expected_result:
        await async_validate_trigger_config(
            hass,
            [
                {
                    "platform": "test.test_trigger",
                    CONF_TARGET: {CONF_ENTITY_ID: "test.test_entity"},
                    CONF_OPTIONS: trigger_options,
                }
            ],
        )