Add fan triggers (#156831)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

homeassistant/components/alarm_control_panel/icons.json

@@ -36,5 +36,13 @@
     "alarm_trigger": {
       "service": "mdi:bell-ring"
     }
+  },
+  "triggers": {
+    "disarmed": {
+      "trigger": "mdi:shield-off"
+    },
+    "triggered": {
+      "trigger": "mdi:bell-ring"
+    }
   }
 }

homeassistant/components/alarm_control_panel/strings.json

@@ -71,6 +71,15 @@
       "message": "Arming requires a code but none was given for {entity_id}."
     }
   },
+  "selector": {
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
+    }
+  },
   "services": {
     "alarm_arm_away": {
       "description": "Arms the alarm in the away mode.",
@@ -143,5 +152,29 @@
       "name": "Trigger"
     }
   },
-  "title": "Alarm control panel"
+  "title": "Alarm control panel",
+  "triggers": {
+    "disarmed": {
+      "description": "Triggers when an alarm is disarmed.",
+      "description_configured": "[%key:component::alarm_control_panel::triggers::disarmed::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted alarms to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When an alarm is disarmed"
+    },
+    "triggered": {
+      "description": "Triggers when an alarm is triggered.",
+      "description_configured": "[%key:component::alarm_control_panel::triggers::triggered::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::alarm_control_panel::triggers::disarmed::fields::behavior::description%]",
+          "name": "[%key:component::alarm_control_panel::triggers::disarmed::fields::behavior::name%]"
+        }
+      },
+      "name": "When an alarm is triggered"
+    }
+  }
 }

homeassistant/components/alarm_control_panel/trigger.py

@@ -0,0 +1,16 @@
+"""Provides triggers for alarm control panels."""
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
+
+from .const import DOMAIN, AlarmControlPanelState
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "disarmed": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.DISARMED),
+    "triggered": make_entity_state_trigger(DOMAIN, AlarmControlPanelState.TRIGGERED),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for alarm control panels."""
+    return TRIGGERS

homeassistant/components/alarm_control_panel/triggers.yaml

@@ -0,0 +1,19 @@
+.trigger_common: &trigger_common
+  target:
+    entity:
+      domain: alarm_control_panel
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          options:
+            - first
+            - last
+            - any
+          translation_key: trigger_behavior
+
+disarmed: *trigger_common
+
+triggered: *trigger_common

homeassistant/components/assist_satellite/icons.json

@@ -14,5 +14,19 @@
     "start_conversation": {
       "service": "mdi:forum"
     }
+  },
+  "triggers": {
+    "idle": {
+      "trigger": "mdi:chat-sleep"
+    },
+    "listening": {
+      "trigger": "mdi:chat-question"
+    },
+    "processing": {
+      "trigger": "mdi:chat-processing"
+    },
+    "responding": {
+      "trigger": "mdi:chat-alert"
+    }
   }
 }

homeassistant/components/assist_satellite/strings.json

@@ -16,6 +16,13 @@
         "id": "Answer ID",
         "sentences": "Sentences"
       }
+    },
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
     }
   },
   "services": {
@@ -98,5 +105,51 @@
       "name": "Start conversation"
     }
   },
-  "title": "Assist satellite"
+  "title": "Assist satellite",
+  "triggers": {
+    "idle": {
+      "description": "Triggers when an assist satellite becomes idle.",
+      "description_configured": "[%key:component::assist_satellite::triggers::idle::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted assist satellites to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When an assist satellite becomes idle"
+    },
+    "listening": {
+      "description": "Triggers when an assist satellite starts listening.",
+      "description_configured": "[%key:component::assist_satellite::triggers::listening::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::assist_satellite::triggers::idle::fields::behavior::description%]",
+          "name": "[%key:component::assist_satellite::triggers::idle::fields::behavior::name%]"
+        }
+      },
+      "name": "When an assist satellite starts listening"
+    },
+    "processing": {
+      "description": "Triggers when an assist satellite is processing.",
+      "description_configured": "[%key:component::assist_satellite::triggers::processing::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::assist_satellite::triggers::idle::fields::behavior::description%]",
+          "name": "[%key:component::assist_satellite::triggers::idle::fields::behavior::name%]"
+        }
+      },
+      "name": "When an assist satellite is processing"
+    },
+    "responding": {
+      "description": "Triggers when an assist satellite is responding.",
+      "description_configured": "[%key:component::assist_satellite::triggers::responding::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::assist_satellite::triggers::idle::fields::behavior::description%]",
+          "name": "[%key:component::assist_satellite::triggers::idle::fields::behavior::name%]"
+        }
+      },
+      "name": "When an assist satellite is responding"
+    }
+  }
 }

homeassistant/components/assist_satellite/trigger.py

@@ -0,0 +1,19 @@
+"""Provides triggers for assist satellites."""
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
+
+from .const import DOMAIN
+from .entity import AssistSatelliteState
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "idle": make_entity_state_trigger(DOMAIN, AssistSatelliteState.IDLE),
+    "listening": make_entity_state_trigger(DOMAIN, AssistSatelliteState.LISTENING),
+    "processing": make_entity_state_trigger(DOMAIN, AssistSatelliteState.PROCESSING),
+    "responding": make_entity_state_trigger(DOMAIN, AssistSatelliteState.RESPONDING),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for assist satellites."""
+    return TRIGGERS

homeassistant/components/assist_satellite/triggers.yaml

@@ -0,0 +1,20 @@
+.trigger_common: &trigger_common
+  target:
+    entity:
+      domain: assist_satellite
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          options:
+            - first
+            - last
+            - any
+          translation_key: trigger_behavior
+
+idle: *trigger_common
+listening: *trigger_common
+processing: *trigger_common
+responding: *trigger_common

homeassistant/components/climate/icons.json

@@ -96,5 +96,13 @@
     "turn_on": {
       "service": "mdi:power-on"
     }
+  },
+  "triggers": {
+    "started_heating": {
+      "trigger": "mdi:fire"
+    },
+    "turned_off": {
+      "trigger": "mdi:power-off"
+    }
   }
 }

homeassistant/components/climate/strings.json

@@ -187,6 +187,13 @@
         "heat_cool": "Heat/cool",
         "off": "[%key:common::state::off%]"
       }
+    },
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
     }
   },
   "services": {
@@ -285,5 +292,29 @@
       "name": "[%key:common::action::turn_on%]"
     }
   },
-  "title": "Climate"
+  "title": "Climate",
+  "triggers": {
+    "started_heating": {
+      "description": "Triggers when a climate starts to heat.",
+      "description_configured": "[%key:component::climate::triggers::started_heating::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::climate::triggers::turned_off::fields::behavior::description%]",
+          "name": "[%key:component::climate::triggers::turned_off::fields::behavior::name%]"
+        }
+      },
+      "name": "When a climate starts to heat"
+    },
+    "turned_off": {
+      "description": "Triggers when a climate is turned off.",
+      "description_configured": "[%key:component::climate::triggers::turned_off::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted climates to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When a climate is turned off"
+    }
+  }
 }

homeassistant/components/climate/trigger.py

@@ -0,0 +1,23 @@
+"""Provides triggers for climates."""
+
+from homeassistant.const import STATE_OFF
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.trigger import (
+    Trigger,
+    make_entity_state_attribute_trigger,
+    make_entity_state_trigger,
+)
+
+from .const import ATTR_HVAC_ACTION, DOMAIN, HVACAction
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
+    "started_heating": make_entity_state_attribute_trigger(
+        DOMAIN, ATTR_HVAC_ACTION, HVACAction.HEATING
+    ),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for climates."""
+    return TRIGGERS

homeassistant/components/climate/triggers.yaml

@@ -0,0 +1,19 @@
+.trigger_common: &trigger_common
+  target:
+    entity:
+      domain: climate
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: trigger_behavior
+          options:
+            - first
+            - last
+            - any
+
+started_heating: *trigger_common
+
+turned_off: *trigger_common

homeassistant/components/cover/icons.json

@@ -108,5 +108,10 @@
     "toggle_cover_tilt": {
       "service": "mdi:arrow-top-right-bottom-left"
     }
+  },
+  "triggers": {
+    "garage_opened": {
+      "trigger": "mdi:garage-open"
+    }
   }
 }

homeassistant/components/cover/strings.json

@@ -82,6 +82,15 @@
       "name": "Window"
     }
   },
+  "selector": {
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
+    }
+  },
   "services": {
     "close_cover": {
       "description": "Closes a cover.",
@@ -136,5 +145,22 @@
       "name": "Toggle tilt"
     }
   },
-  "title": "Cover"
+  "title": "Cover",
+  "triggers": {
+    "garage_opened": {
+      "description": "Triggers when a garage door opens.",
+      "description_configured": "[%key:component::cover::triggers::garage_opened::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted garage doors to trigger on.",
+          "name": "Behavior"
+        },
+        "fully_opened": {
+          "description": "Require the garage doors to be fully opened before triggering.",
+          "name": "Fully opened"
+        }
+      },
+      "name": "When a garage door opens"
+    }
+  }
 }

homeassistant/components/cover/trigger.py

@@ -0,0 +1,108 @@
+"""Provides triggers for covers."""
+
+from typing import Final
+
+import voluptuous as vol
+
+from homeassistant.const import CONF_OPTIONS
+from homeassistant.core import HomeAssistant, State
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity import get_device_class
+from homeassistant.helpers.trigger import (
+    ENTITY_STATE_TRIGGER_SCHEMA,
+    EntityTriggerBase,
+    Trigger,
+    TriggerConfig,
+)
+from homeassistant.helpers.typing import UNDEFINED, UndefinedType
+
+from . import ATTR_CURRENT_POSITION, CoverDeviceClass, CoverState
+from .const import DOMAIN
+
+ATTR_FULLY_OPENED: Final = "fully_opened"
+
+COVER_OPENED_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
+    {
+        vol.Required(CONF_OPTIONS): {
+            vol.Required(ATTR_FULLY_OPENED, default=False): bool,
+        },
+    }
+)
+
+
+def get_device_class_or_undefined(
+    hass: HomeAssistant, entity_id: str
+) -> str | None | UndefinedType:
+    """Get the device class of an entity or UNDEFINED if not found."""
+    try:
+        return get_device_class(hass, entity_id)
+    except HomeAssistantError:
+        return UNDEFINED
+
+
+class CoverOpenedClosedTrigger(EntityTriggerBase):
+    """Class for cover opened and closed triggers."""
+
+    _attribute: str = ATTR_CURRENT_POSITION
+    _attribute_value: int | None = None
+    _device_class: CoverDeviceClass | None
+    _domain: str = DOMAIN
+    _to_states: set[str]
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        if state.state not in self._to_states:
+            return False
+        if (
+            self._attribute_value is not None
+            and (value := state.attributes.get(self._attribute)) is not None
+            and value != self._attribute_value
+        ):
+            return False
+        return True
+
+    def entity_filter(self, entities: set[str]) -> set[str]:
+        """Filter entities of this domain."""
+        entities = super().entity_filter(entities)
+        return {
+            entity_id
+            for entity_id in entities
+            if get_device_class_or_undefined(self._hass, entity_id)
+            == self._device_class
+        }
+
+
+class CoverOpenedTrigger(CoverOpenedClosedTrigger):
+    """Class for cover opened triggers."""
+
+    _schema = COVER_OPENED_TRIGGER_SCHEMA
+    _to_states = {CoverState.OPEN, CoverState.OPENING}
+
+    def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
+        """Initialize the state trigger."""
+        super().__init__(hass, config)
+        if self._options.get(ATTR_FULLY_OPENED):
+            self._attribute_value = 100
+
+
+def make_cover_opened_trigger(
+    device_class: CoverDeviceClass | None,
+) -> type[CoverOpenedTrigger]:
+    """Create an entity state attribute trigger class."""
+
+    class CustomTrigger(CoverOpenedTrigger):
+        """Trigger for entity state changes."""
+
+        _device_class = device_class
+
+    return CustomTrigger
+
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "garage_opened": make_cover_opened_trigger(CoverDeviceClass.GARAGE),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for covers."""
+    return TRIGGERS

homeassistant/components/cover/triggers.yaml

@@ -0,0 +1,21 @@
+garage_opened:
+  target:
+    entity:
+      domain: cover
+      device_class: garage
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: trigger_behavior
+          options:
+            - first
+            - last
+            - any
+    fully_opened:
+      required: true
+      default: false
+      selector:
+        boolean:

homeassistant/components/fan/icons.json

@@ -47,5 +47,13 @@
     "turn_on": {
       "service": "mdi:fan"
     }
+  },
+  "triggers": {
+    "turned_off": {
+      "trigger": "mdi:fan-off"
+    },
+    "turned_on": {
+      "trigger": "mdi:fan"
+    }
   }
 }

homeassistant/components/fan/strings.json

@@ -66,6 +66,13 @@
         "forward": "Forward",
         "reverse": "Reverse"
       }
+    },
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
     }
   },
   "services": {
@@ -152,5 +159,29 @@
       "name": "[%key:common::action::turn_on%]"
     }
   },
-  "title": "Fan"
+  "title": "Fan",
+  "triggers": {
+    "turned_off": {
+      "description": "Triggers when a fan is turned off.",
+      "description_configured": "[%key:component::fan::triggers::turned_off::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted fans to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When a fan is turned off"
+    },
+    "turned_on": {
+      "description": "Triggers when a fan is turned on.",
+      "description_configured": "[%key:component::fan::triggers::turned_on::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::fan::triggers::turned_off::fields::behavior::description%]",
+          "name": "[%key:component::fan::triggers::turned_off::fields::behavior::name%]"
+        }
+      },
+      "name": "When a fan is turned on"
+    }
+  }
 }

homeassistant/components/fan/trigger.py

@@ -0,0 +1,17 @@
+"""Provides triggers for fans."""
+
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
+
+from . import DOMAIN
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
+    "turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for fans."""
+    return TRIGGERS

homeassistant/components/fan/triggers.yaml

@@ -0,0 +1,18 @@
+.trigger_common: &trigger_common
+  target:
+    entity:
+      domain: fan
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          options:
+            - first
+            - last
+            - any
+          translation_key: trigger_behavior
+
+turned_on: *trigger_common
+turned_off: *trigger_common

homeassistant/components/frontend/__init__.py

@@ -778,7 +778,7 @@ class ManifestJSONView(HomeAssistantView):
     {
         "type": "frontend/get_icons",
         vol.Required("category"): vol.In(
-            {"entity", "entity_component", "services", "triggers", "conditions"}
+            {"conditions", "entity", "entity_component", "services", "triggers"}
         ),
         vol.Optional("integration"): vol.All(cv.ensure_list, [str]),
     }

homeassistant/components/light/condition.py

@@ -0,0 +1,131 @@
+"""Provides conditions for lights."""
+
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any, Final, override
+
+import voluptuous as vol
+
+from homeassistant.const import CONF_OPTIONS, CONF_TARGET, STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant, split_entity_id
+from homeassistant.helpers import config_validation as cv, target
+from homeassistant.helpers.condition import (
+    Condition,
+    ConditionCheckerType,
+    ConditionConfig,
+    trace_condition_function,
+)
+from homeassistant.helpers.typing import ConfigType, TemplateVarsType
+
+from .const import DOMAIN
+
+ATTR_BEHAVIOR: Final = "behavior"
+BEHAVIOR_ANY: Final = "any"
+BEHAVIOR_ALL: Final = "all"
+
+
+STATE_CONDITION_VALID_STATES: Final = [STATE_ON, STATE_OFF]
+STATE_CONDITION_OPTIONS_SCHEMA: dict[vol.Marker, Any] = {
+    vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
+        [BEHAVIOR_ANY, BEHAVIOR_ALL]
+    ),
+}
+STATE_CONDITION_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
+        vol.Required(CONF_OPTIONS): STATE_CONDITION_OPTIONS_SCHEMA,
+    }
+)
+
+
+class StateConditionBase(Condition):
+    """State condition."""
+
+    @override
+    @classmethod
+    async def async_validate_config(
+        cls, hass: HomeAssistant, config: ConfigType
+    ) -> ConfigType:
+        """Validate config."""
+        return STATE_CONDITION_SCHEMA(config)  # type: ignore[no-any-return]
+
+    def __init__(
+        self, hass: HomeAssistant, config: ConditionConfig, state: str
+    ) -> None:
+        """Initialize condition."""
+        self._hass = hass
+        if TYPE_CHECKING:
+            assert config.target
+            assert config.options
+        self._target = config.target
+        self._behavior = config.options[ATTR_BEHAVIOR]
+        self._state = state
+
+    @override
+    async def async_get_checker(self) -> ConditionCheckerType:
+        """Get the condition checker."""
+
+        def check_any_match_state(states: list[str]) -> bool:
+            """Test if any entity match the state."""
+            return any(state == self._state for state in states)
+
+        def check_all_match_state(states: list[str]) -> bool:
+            """Test if all entities match the state."""
+            return all(state == self._state for state in states)
+
+        matcher: Callable[[list[str]], bool]
+        if self._behavior == BEHAVIOR_ANY:
+            matcher = check_any_match_state
+        elif self._behavior == BEHAVIOR_ALL:
+            matcher = check_all_match_state
+
+        @trace_condition_function
+        def test_state(hass: HomeAssistant, variables: TemplateVarsType = None) -> bool:
+            """Test state condition."""
+            selector_data = target.TargetSelectorData(self._target)
+            targeted_entities = target.async_extract_referenced_entity_ids(
+                hass, selector_data, expand_group=False
+            )
+            referenced_entity_ids = targeted_entities.referenced.union(
+                targeted_entities.indirectly_referenced
+            )
+            light_entity_ids = {
+                entity_id
+                for entity_id in referenced_entity_ids
+                if split_entity_id(entity_id)[0] == DOMAIN
+            }
+            light_entity_states = [
+                state.state
+                for entity_id in light_entity_ids
+                if (state := hass.states.get(entity_id))
+                and state.state in STATE_CONDITION_VALID_STATES
+            ]
+            return matcher(light_entity_states)
+
+        return test_state
+
+
+class IsOnCondition(StateConditionBase):
+    """Is on condition."""
+
+    def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
+        """Initialize condition."""
+        super().__init__(hass, config, STATE_ON)
+
+
+class IsOffCondition(StateConditionBase):
+    """Is off condition."""
+
+    def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
+        """Initialize condition."""
+        super().__init__(hass, config, STATE_OFF)
+
+
+CONDITIONS: dict[str, type[Condition]] = {
+    "is_off": IsOffCondition,
+    "is_on": IsOnCondition,
+}
+
+
+async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
+    """Return the light conditions."""
+    return CONDITIONS

homeassistant/components/light/conditions.yaml

@@ -0,0 +1,28 @@
+is_off:
+  target:
+    entity:
+      domain: light
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: condition_behavior
+          options:
+            - all
+            - any
+is_on:
+  target:
+    entity:
+      domain: light
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: condition_behavior
+          options:
+            - all
+            - any

homeassistant/components/light/icons.json

@@ -1,4 +1,12 @@
 {
+  "conditions": {
+    "is_off": {
+      "condition": "mdi:lightbulb-off"
+    },
+    "is_on": {
+      "condition": "mdi:lightbulb-on"
+    }
+  },
   "entity_component": {
     "_": {
       "default": "mdi:lightbulb",
@@ -25,5 +33,13 @@
     "turn_on": {
       "service": "mdi:lightbulb-on"
     }
+  },
+  "triggers": {
+    "turned_off": {
+      "trigger": "mdi:lightbulb-off"
+    },
+    "turned_on": {
+      "trigger": "mdi:lightbulb-on"
+    }
   }
 }

homeassistant/components/light/strings.json

@@ -36,6 +36,30 @@
     "field_xy_color_name": "XY-color",
     "section_advanced_fields_name": "Advanced options"
   },
+  "conditions": {
+    "is_off": {
+      "description": "Test if a light is off.",
+      "description_configured": "[%key:component::light::conditions::is_off::description%]",
+      "fields": {
+        "behavior": {
+          "description": "How the state should match on the targeted lights.",
+          "name": "Behavior"
+        }
+      },
+      "name": "If a light is off"
+    },
+    "is_on": {
+      "description": "Test if a light is on.",
+      "description_configured": "[%key:component::light::conditions::is_on::description%]",
+      "fields": {
+        "behavior": {
+          "description": "How the state should match on the targeted lights.",
+          "name": "Behavior"
+        }
+      },
+      "name": "If a light is on"
+    }
+  },
   "device_automation": {
     "action_type": {
       "brightness_decrease": "Decrease {entity_name} brightness",
@@ -284,11 +308,30 @@
         "yellowgreen": "Yellow green"
       }
     },
+    "condition_behavior": {
+      "options": {
+        "all": "All",
+        "any": "Any"
+      }
+    },
     "flash": {
       "options": {
         "long": "Long",
         "short": "Short"
       }
+    },
+    "state": {
+      "options": {
+        "off": "[%key:common::state::off%]",
+        "on": "[%key:common::state::on%]"
+      }
+    },
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
     }
   },
   "services": {
@@ -462,5 +505,29 @@
       }
     }
   },
-  "title": "Light"
+  "title": "Light",
+  "triggers": {
+    "turned_off": {
+      "description": "Triggers when a light is turned off.",
+      "description_configured": "[%key:component::light::triggers::turned_off::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted lights to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When a light is turned off"
+    },
+    "turned_on": {
+      "description": "Triggers when a light is turned on.",
+      "description_configured": "[%key:component::light::triggers::turned_on::description%]",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::light::triggers::turned_off::fields::behavior::description%]",
+          "name": "[%key:component::light::triggers::turned_off::fields::behavior::name%]"
+        }
+      },
+      "name": "When a light is turned on"
+    }
+  }
 }

homeassistant/components/light/trigger.py

@@ -0,0 +1,17 @@
+"""Provides triggers for lights."""
+
+from homeassistant.const import STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.trigger import Trigger, make_entity_state_trigger
+
+from .const import DOMAIN
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "turned_off": make_entity_state_trigger(DOMAIN, STATE_OFF),
+    "turned_on": make_entity_state_trigger(DOMAIN, STATE_ON),
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for lights."""
+    return TRIGGERS

homeassistant/components/light/triggers.yaml

@@ -0,0 +1,18 @@
+.trigger_common: &trigger_common
+  target:
+    entity:
+      domain: light
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          options:
+            - first
+            - last
+            - any
+          translation_key: trigger_behavior
+
+turned_on: *trigger_common
+turned_off: *trigger_common

homeassistant/components/media_player/icons.json

@@ -104,5 +104,10 @@
     "volume_up": {
       "service": "mdi:volume-plus"
     }
+  },
+  "triggers": {
+    "stopped_playing": {
+      "trigger": "mdi:stop"
+    }
   }
 }

homeassistant/components/media_player/strings.json

@@ -177,6 +177,13 @@
         "off": "[%key:common::state::off%]",
         "one": "Repeat one"
       }
+    },
+    "trigger_behavior": {
+      "options": {
+        "any": "Any",
+        "first": "First",
+        "last": "Last"
+      }
     }
   },
   "services": {
@@ -367,5 +374,18 @@
       "name": "Turn up volume"
     }
   },
-  "title": "Media player"
+  "title": "Media player",
+  "triggers": {
+    "stopped_playing": {
+      "description": "Triggers when a media player stops playing.",
+      "description_configured": "[%key:component::media_player::triggers::stopped_playing::description%]",
+      "fields": {
+        "behavior": {
+          "description": "The behavior of the targeted media players to trigger on.",
+          "name": "Behavior"
+        }
+      },
+      "name": "When a media player stops playing"
+    }
+  }
 }

homeassistant/components/media_player/trigger.py

@@ -0,0 +1,41 @@
+"""Provides triggers for media players."""
+
+from homeassistant.core import HomeAssistant, State
+from homeassistant.helpers.trigger import EntityTriggerBase, Trigger
+
+from . import MediaPlayerState
+from .const import DOMAIN
+
+
+class MediaPlayerStoppedPlayingTrigger(EntityTriggerBase):
+    """Class for media player stopped playing trigger."""
+
+    _domain: str = DOMAIN
+    _from_states = {
+        MediaPlayerState.BUFFERING,
+        MediaPlayerState.PAUSED,
+        MediaPlayerState.PLAYING,
+    }
+    _to_states = {
+        MediaPlayerState.IDLE,
+        MediaPlayerState.OFF,
+        MediaPlayerState.ON,
+    }
+
+    def is_from_state(self, state: State) -> bool:
+        """Check if the state matches the origin state."""
+        return state.state in self._from_states
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        return state.state in self._to_states
+
+
+TRIGGERS: dict[str, type[Trigger]] = {
+    "stopped_playing": MediaPlayerStoppedPlayingTrigger,
+}
+
+
+async def async_get_triggers(hass: HomeAssistant) -> dict[str, type[Trigger]]:
+    """Return the triggers for media players."""
+    return TRIGGERS

homeassistant/components/media_player/triggers.yaml

@@ -0,0 +1,15 @@
+stopped_playing:
+  target:
+    entity:
+      domain: media_player
+  fields:
+    behavior:
+      required: true
+      default: any
+      selector:
+        select:
+          translation_key: trigger_behavior
+          options:
+            - first
+            - last
+            - any

homeassistant/helpers/target.py

@@ -289,7 +289,7 @@ class TargetStateChangeTracker:
         )
 
         tracked_entities = self._entity_filter(
-            selected.referenced.union(selected.indirectly_referenced)
+            selected.referenced | selected.indirectly_referenced
         )
 
         @callback

homeassistant/helpers/trigger.py

@@ -10,11 +10,12 @@ from dataclasses import dataclass, field
 import functools
 import inspect
 import logging
-from typing import TYPE_CHECKING, Any, Protocol, TypedDict, cast
+from typing import TYPE_CHECKING, Any, Final, Protocol, TypedDict, cast, override
 
 import voluptuous as vol
 
 from homeassistant.const import (
+    ATTR_ENTITY_ID,
     CONF_ALIAS,
     CONF_ENABLED,
     CONF_ID,
@@ -23,6 +24,8 @@ from homeassistant.const import (
     CONF_SELECTOR,
     CONF_TARGET,
     CONF_VARIABLES,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
 )
 from homeassistant.core import (
     CALLBACK_TYPE,
@@ -30,9 +33,11 @@ from homeassistant.core import (
     HassJob,
     HassJobType,
     HomeAssistant,
+    State,
     callback,
     get_hassjob_callable_job_type,
     is_callback,
+    split_entity_id,
 )
 from homeassistant.exceptions import HomeAssistantError, TemplateError
 from homeassistant.loader import (
@@ -49,6 +54,10 @@ from . import config_validation as cv, selector
 from .automation import get_absolute_description_key, get_relative_description_key
 from .integration_platform import async_process_integration_platforms
 from .selector import TargetSelector
+from .target import (
+    TargetStateChangedData,
+    async_track_target_selector_state_change_event,
+)
 from .template import Template
 from .typing import ConfigType, TemplateVarsType
 
@@ -108,6 +117,22 @@ _TRIGGERS_DESCRIPTION_SCHEMA = vol.Schema(
     }
 )
 
+ATTR_BEHAVIOR: Final = "behavior"
+BEHAVIOR_FIRST: Final = "first"
+BEHAVIOR_LAST: Final = "last"
+BEHAVIOR_ANY: Final = "any"
+
+ENTITY_STATE_TRIGGER_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_OPTIONS): {
+            vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
+                [BEHAVIOR_FIRST, BEHAVIOR_LAST, BEHAVIOR_ANY]
+            ),
+        },
+        vol.Required(CONF_TARGET): cv.TARGET_FIELDS,
+    }
+)
+
 
 async def async_setup(hass: HomeAssistant) -> None:
     """Set up the trigger helper."""
@@ -245,6 +270,170 @@ class Trigger(abc.ABC):
         """Attach the trigger to an action runner."""
 
 
+class EntityTriggerBase(Trigger):
+    """Trigger for entity state changes."""
+
+    _domain: str
+    _schema: vol.Schema = ENTITY_STATE_TRIGGER_SCHEMA
+
+    @override
+    @classmethod
+    async def async_validate_config(
+        cls, hass: HomeAssistant, config: ConfigType
+    ) -> ConfigType:
+        """Validate config."""
+        return cast(ConfigType, cls._schema(config))
+
+    def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
+        """Initialize the state trigger."""
+        super().__init__(hass, config)
+        if TYPE_CHECKING:
+            assert config.options is not None
+            assert config.target is not None
+        self._options = config.options
+        self._target = config.target
+
+    def is_from_state(self, state: State) -> bool:
+        """Check if the state matches the origin state."""
+        return not self.is_to_state(state)
+
+    @abc.abstractmethod
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+
+    def check_all_match(self, entity_ids: set[str]) -> bool:
+        """Check if all entity states match."""
+        return all(
+            self.is_to_state(state)
+            for entity_id in entity_ids
+            if (state := self._hass.states.get(entity_id)) is not None
+        )
+
+    def check_one_match(self, entity_ids: set[str]) -> bool:
+        """Check that only one entity state matches."""
+        return (
+            sum(
+                self.is_to_state(state)
+                for entity_id in entity_ids
+                if (state := self._hass.states.get(entity_id)) is not None
+            )
+            == 1
+        )
+
+    def entity_filter(self, entities: set[str]) -> set[str]:
+        """Filter entities of this domain."""
+        return {
+            entity_id
+            for entity_id in entities
+            if split_entity_id(entity_id)[0] == self._domain
+        }
+
+    @override
+    async def async_attach_runner(
+        self, run_action: TriggerActionRunner
+    ) -> CALLBACK_TYPE:
+        """Attach the trigger to an action runner."""
+
+        behavior = self._options.get(ATTR_BEHAVIOR)
+
+        @callback
+        def state_change_listener(
+            target_state_change_data: TargetStateChangedData,
+        ) -> None:
+            """Listen for state changes and call action."""
+            event = target_state_change_data.state_change_event
+            entity_id = event.data["entity_id"]
+            from_state = event.data["old_state"]
+            to_state = event.data["new_state"]
+
+            # The trigger should never fire if the previous state was not a valid state
+            if not from_state or from_state.state in (STATE_UNAVAILABLE, STATE_UNKNOWN):
+                return
+
+            # The trigger should never fire if the previous state was not the from state
+            if not self.is_from_state(from_state):
+                return
+
+            # The trigger should never fire if the new state is not the to state
+            if not to_state or not self.is_to_state(to_state):
+                return
+
+            if behavior == BEHAVIOR_LAST:
+                if not self.check_all_match(
+                    target_state_change_data.targeted_entity_ids
+                ):
+                    return
+            elif behavior == BEHAVIOR_FIRST:
+                if not self.check_one_match(
+                    target_state_change_data.targeted_entity_ids
+                ):
+                    return
+
+            run_action(
+                {
+                    ATTR_ENTITY_ID: entity_id,
+                    "from_state": from_state,
+                    "to_state": to_state,
+                },
+                f"state of {entity_id}",
+                event.context,
+            )
+
+        return async_track_target_selector_state_change_event(
+            self._hass, self._target, state_change_listener, self.entity_filter
+        )
+
+
+class EntityStateTriggerBase(EntityTriggerBase):
+    """Trigger for entity state changes."""
+
+    _to_state: str
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        return state.state == self._to_state
+
+
+class EntityStateAttributeTriggerBase(EntityTriggerBase):
+    """Trigger for entity state attribute changes."""
+
+    _attribute: str
+    _attribute_to_state: str
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        return state.attributes.get(self._attribute) == self._attribute_to_state
+
+
+def make_entity_state_trigger(
+    domain: str, to_state: str
+) -> type[EntityStateTriggerBase]:
+    """Create an entity state trigger class."""
+
+    class CustomTrigger(EntityStateTriggerBase):
+        """Trigger for entity state changes."""
+
+        _domain = domain
+        _to_state = to_state
+
+    return CustomTrigger
+
+
+def make_entity_state_attribute_trigger(
+    domain: str, attribute: str, to_state: str
+) -> type[EntityStateAttributeTriggerBase]:
+    """Create an entity state attribute trigger class."""
+
+    class CustomTrigger(EntityStateAttributeTriggerBase):
+        """Trigger for entity state changes."""
+
+        _domain = domain
+        _attribute = attribute
+        _attribute_to_state = to_state
+
+    return CustomTrigger
+
+
 class TriggerProtocol(Protocol):
     """Define the format of trigger modules.
 

tests/components/__init__.py

@@ -1 +1,294 @@
 """The tests for components."""
+
+from collections.abc import Iterable
+import itertools
+
+from homeassistant.const import (
+    ATTR_AREA_ID,
+    ATTR_DEVICE_ID,
+    ATTR_FLOOR_ID,
+    ATTR_LABEL_ID,
+    CONF_ENTITY_ID,
+    CONF_OPTIONS,
+    CONF_PLATFORM,
+    CONF_TARGET,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import (
+    area_registry as ar,
+    device_registry as dr,
+    entity_registry as er,
+    floor_registry as fr,
+    label_registry as lr,
+)
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry, mock_device_registry
+
+
+async def target_entities(hass: HomeAssistant, domain: str) -> None:
+    """Create multiple entities associated with different targets."""
+    await async_setup_component(hass, domain, {})
+
+    config_entry = MockConfigEntry(domain="test")
+    config_entry.add_to_hass(hass)
+
+    floor_reg = fr.async_get(hass)
+    floor = floor_reg.async_create("Test Floor")
+
+    area_reg = ar.async_get(hass)
+    area = area_reg.async_create("Test Area", floor_id=floor.floor_id)
+
+    label_reg = lr.async_get(hass)
+    label = label_reg.async_create("Test Label")
+
+    device = dr.DeviceEntry(id="test_device", area_id=area.id, labels={label.label_id})
+    mock_device_registry(hass, {device.id: device})
+
+    entity_reg = er.async_get(hass)
+    # Entity associated with area
+    entity_area = entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_area",
+        suggested_object_id=f"area_{domain}",
+    )
+    entity_reg.async_update_entity(entity_area.entity_id, area_id=area.id)
+
+    # Entity associated with device
+    entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_device",
+        suggested_object_id=f"device_{domain}",
+        device_id=device.id,
+    )
+
+    # Entity associated with label
+    entity_label = entity_reg.async_get_or_create(
+        domain=domain,
+        platform="test",
+        unique_id=f"{domain}_label",
+        suggested_object_id=f"label_{domain}",
+    )
+    entity_reg.async_update_entity(entity_label.entity_id, labels={label.label_id})
+
+    # Return all available entities
+    return [
+        f"{domain}.standalone_{domain}",
+        f"{domain}.label_{domain}",
+        f"{domain}.area_{domain}",
+        f"{domain}.device_{domain}",
+    ]
+
+
+def parametrize_target_entities(domain: str) -> list[tuple[dict, str, int]]:
+    """Parametrize target entities for different target types.
+
+    Meant to be used with target_entities.
+    """
+    return [
+        (
+            {CONF_ENTITY_ID: f"{domain}.standalone_{domain}"},
+            f"{domain}.standalone_{domain}",
+            1,
+        ),
+        ({ATTR_LABEL_ID: "test_label"}, f"{domain}.label_{domain}", 2),
+        ({ATTR_AREA_ID: "test_area"}, f"{domain}.area_{domain}", 2),
+        ({ATTR_FLOOR_ID: "test_floor"}, f"{domain}.area_{domain}", 2),
+        ({ATTR_LABEL_ID: "test_label"}, f"{domain}.device_{domain}", 2),
+        ({ATTR_AREA_ID: "test_area"}, f"{domain}.device_{domain}", 2),
+        ({ATTR_FLOOR_ID: "test_floor"}, f"{domain}.device_{domain}", 2),
+        ({ATTR_DEVICE_ID: "test_device"}, f"{domain}.device_{domain}", 1),
+    ]
+
+
+def parametrize_trigger_states(
+    trigger: str, target_states: Iterable[str], other_states: Iterable[str]
+) -> list[tuple[str, list[tuple[str | None, int]]]]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, list of states),
+    where states is a list of tuples (state to set, expected service call count).
+    """
+    return [
+        # Initial state None
+        (
+            trigger,
+            list(
+                itertools.chain.from_iterable(
+                    ((None, 0), (target_state, 0), (other_state, 0), (target_state, 1))
+                    for target_state in target_states
+                    for other_state in other_states
+                )
+            ),
+        ),
+        # Initial state different from target state
+        (
+            trigger,
+            # other_state,
+            list(
+                itertools.chain.from_iterable(
+                    (
+                        (other_state, 0),
+                        (target_state, 1),
+                        (other_state, 0),
+                        (target_state, 1),
+                    )
+                    for target_state in target_states
+                    for other_state in other_states
+                )
+            ),
+        ),
+        # Initial state same as target state
+        (
+            trigger,
+            list(
+                itertools.chain.from_iterable(
+                    (
+                        (target_state, 0),
+                        (target_state, 0),
+                        (other_state, 0),
+                        (target_state, 1),
+                    )
+                    for target_state in target_states
+                    for other_state in other_states
+                )
+            ),
+        ),
+        # Initial state unavailable / unknown
+        (
+            trigger,
+            list(
+                itertools.chain.from_iterable(
+                    (
+                        (STATE_UNAVAILABLE, 0),
+                        (target_state, 0),
+                        (other_state, 0),
+                        (target_state, 1),
+                    )
+                    for target_state in target_states
+                    for other_state in other_states
+                )
+            ),
+        ),
+        (
+            trigger,
+            list(
+                itertools.chain.from_iterable(
+                    (
+                        (STATE_UNKNOWN, 0),
+                        (target_state, 0),
+                        (other_state, 0),
+                        (target_state, 1),
+                    )
+                    for target_state in target_states
+                    for other_state in other_states
+                )
+            ),
+        ),
+    ]
+
+
+def parametrize_attribute_trigger_states(
+    trigger: str, state: str, attribute: str, target_state: str, other_state: str
+) -> list[
+    tuple[str, tuple[str | None, dict], list[tuple[tuple[str | None, dict], int]]]
+]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, initial_state, list of states),
+    where states is a list of tuples (state to set, expected service call count).
+
+    The initial_state and state to set are tuples of (state, {attribute: value}).
+    """
+    return [
+        # Initial state None
+        (
+            trigger,
+            (None, {}),
+            [
+                ((state, {attribute: target_state}), 0),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+        # No initial state attribute
+        (
+            trigger,
+            (state, {}),
+            [
+                ((state, {attribute: target_state}), 1),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+        # Initial state attribute different from target state
+        (
+            trigger,
+            (state, {attribute: other_state}),
+            [
+                ((state, {attribute: target_state}), 1),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+        # Initial state attribute same as target state
+        (
+            trigger,
+            (state, {attribute: target_state}),
+            [
+                ((state, {attribute: target_state}), 0),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+        # Initial state unavailable / unknown
+        (
+            trigger,
+            (STATE_UNAVAILABLE, {}),
+            [
+                ((state, {attribute: target_state}), 0),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+        (
+            trigger,
+            (STATE_UNKNOWN, {}),
+            [
+                ((state, {attribute: target_state}), 0),
+                ((state, {}), 0),
+                ((state, {attribute: target_state}), 1),
+            ],
+        ),
+    ]
+
+
+async def arm_trigger(
+    hass: HomeAssistant, trigger: str, trigger_options: dict, trigger_target: dict
+) -> None:
+    """Arm the specified trigger, call service test.automation when it triggers."""
+
+    # Local include to avoid importing the automation component unnecessarily
+    from homeassistant.components import automation  # noqa: PLC0415
+
+    await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: {
+                "trigger": {
+                    CONF_PLATFORM: trigger,
+                    CONF_OPTIONS: {**trigger_options},
+                    CONF_TARGET: {**trigger_target},
+                },
+                "action": {
+                    "service": "test.automation",
+                    "data_template": {CONF_ENTITY_ID: "{{ trigger.entity_id }}"},
+                },
+            }
+        },
+    )

tests/components/alarm_control_panel/test_trigger.py

@@ -0,0 +1,204 @@
+"""Test alarm control panel triggers."""
+
+import pytest
+
+from homeassistant.components.alarm_control_panel import AlarmControlPanelState
+from homeassistant.const import CONF_ENTITY_ID
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+ACP_STATES = {s.value for s in AlarmControlPanelState}
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_alarm_control_panels(hass: HomeAssistant) -> None:
+    """Create multiple alarm control panel entities associated with different targets."""
+    return await target_entities(hass, "alarm_control_panel")
+
+
+def set_or_remove_state(hass: HomeAssistant, entity_id: str, state: str | None) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("alarm_control_panel"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "alarm_control_panel.disarmed",
+            (AlarmControlPanelState.DISARMED,),
+            ACP_STATES - {AlarmControlPanelState.DISARMED},
+        ),
+        *parametrize_trigger_states(
+            "alarm_control_panel.triggered",
+            (AlarmControlPanelState.TRIGGERED,),
+            ACP_STATES - {AlarmControlPanelState.TRIGGERED},
+        ),
+    ],
+)
+async def test_alarm_control_panel_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_alarm_control_panels: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the alarm control panel state trigger fires when any alarm control panel state changes to a specific state."""
+    await async_setup_component(hass, "alarm_control_panel", {})
+
+    other_entity_ids = set(target_alarm_control_panels) - {entity_id}
+
+    # Set all alarm control panels, including the tested one, to the initial state
+    for eid in target_alarm_control_panels:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other alarm control panels also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("alarm_control_panel"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "alarm_control_panel.disarmed",
+            (AlarmControlPanelState.DISARMED,),
+            ACP_STATES - {AlarmControlPanelState.DISARMED},
+        ),
+        *parametrize_trigger_states(
+            "alarm_control_panel.triggered",
+            (AlarmControlPanelState.TRIGGERED,),
+            ACP_STATES - {AlarmControlPanelState.TRIGGERED},
+        ),
+    ],
+)
+async def test_alarm_control_panel_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_alarm_control_panels: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the alarm control panel state trigger fires when the first alarm control panel changes to a specific state."""
+    await async_setup_component(hass, "alarm_control_panel", {})
+
+    other_entity_ids = set(target_alarm_control_panels) - {entity_id}
+
+    # Set all alarm control panels, including the tested one, to the initial state
+    for eid in target_alarm_control_panels:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other alarm control panels should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("alarm_control_panel"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "alarm_control_panel.disarmed",
+            (AlarmControlPanelState.DISARMED,),
+            ACP_STATES - {AlarmControlPanelState.DISARMED},
+        ),
+        *parametrize_trigger_states(
+            "alarm_control_panel.triggered",
+            (AlarmControlPanelState.TRIGGERED,),
+            ACP_STATES - {AlarmControlPanelState.TRIGGERED},
+        ),
+    ],
+)
+async def test_alarm_control_panel_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_alarm_control_panels: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the alarm_control_panel state trigger fires when the last alarm_control_panel changes to a specific state."""
+    await async_setup_component(hass, "alarm_control_panel", {})
+
+    other_entity_ids = set(target_alarm_control_panels) - {entity_id}
+
+    # Set all alarm control panels, including the tested one, to the initial state
+    for eid in target_alarm_control_panels:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/assist_satellite/test_trigger.py

@@ -0,0 +1,234 @@
+"""Test assist satellite triggers."""
+
+import pytest
+
+from homeassistant.components.assist_satellite.entity import AssistSatelliteState
+from homeassistant.const import CONF_ENTITY_ID
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+ASSIST_SATELLITE_STATES = {s.value for s in AssistSatelliteState}
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_assist_satellites(hass: HomeAssistant) -> None:
+    """Create multiple assist satellite entities associated with different targets."""
+    return await target_entities(hass, "assist_satellite")
+
+
+def set_or_remove_state(hass: HomeAssistant, entity_id: str, state: str | None) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("assist_satellite"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "assist_satellite.idle",
+            (AssistSatelliteState.IDLE,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.IDLE},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.listening",
+            (AssistSatelliteState.LISTENING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.LISTENING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.processing",
+            (AssistSatelliteState.PROCESSING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.PROCESSING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.responding",
+            (AssistSatelliteState.RESPONDING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.RESPONDING},
+        ),
+    ],
+)
+async def test_assist_satellite_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_assist_satellites: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the assist satellite state trigger fires when any assist satellite state changes to a specific state."""
+    await async_setup_component(hass, "assist_satellite", {})
+
+    other_entity_ids = set(target_assist_satellites) - {entity_id}
+
+    # Set all assist satellites, including the tested one, to the initial state
+    for eid in target_assist_satellites:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other assist satellites also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("assist_satellite"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "assist_satellite.idle",
+            (AssistSatelliteState.IDLE,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.IDLE},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.listening",
+            (AssistSatelliteState.LISTENING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.LISTENING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.processing",
+            (AssistSatelliteState.PROCESSING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.PROCESSING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.responding",
+            (AssistSatelliteState.RESPONDING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.RESPONDING},
+        ),
+    ],
+)
+async def test_assist_satellite_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_assist_satellites: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the assist satellite state trigger fires when the first assist satellite changes to a specific state."""
+    await async_setup_component(hass, "assist_satellite", {})
+
+    other_entity_ids = set(target_assist_satellites) - {entity_id}
+
+    # Set all assist satellites, including the tested one, to the initial state
+    for eid in target_assist_satellites:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other assist satellites should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("assist_satellite"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "assist_satellite.idle",
+            (AssistSatelliteState.IDLE,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.IDLE},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.listening",
+            (AssistSatelliteState.LISTENING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.LISTENING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.processing",
+            (AssistSatelliteState.PROCESSING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.PROCESSING},
+        ),
+        *parametrize_trigger_states(
+            "assist_satellite.responding",
+            (AssistSatelliteState.RESPONDING,),
+            ASSIST_SATELLITE_STATES - {AssistSatelliteState.RESPONDING},
+        ),
+    ],
+)
+async def test_assist_satellite_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_assist_satellites: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the assist_satellite state trigger fires when the last assist_satellite changes to a specific state."""
+    await async_setup_component(hass, "assist_satellite", {})
+
+    other_entity_ids = set(target_assist_satellites) - {entity_id}
+
+    # Set all assist satellites, including the tested one, to the initial state
+    for eid in target_assist_satellites:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/climate/test_trigger.py

@@ -0,0 +1,353 @@
+"""Test climate trigger."""
+
+import pytest
+
+from homeassistant.components.climate.const import (
+    ATTR_HVAC_ACTION,
+    HVACAction,
+    HVACMode,
+)
+from homeassistant.const import CONF_ENTITY_ID
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_attribute_trigger_states,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_climates(hass: HomeAssistant) -> None:
+    """Create multiple climate entities associated with different targets."""
+    return await target_entities(hass, "climate")
+
+
+def set_or_remove_state(
+    hass: HomeAssistant,
+    entity_id: str,
+    state: str | None,
+    attributes: dict | None = None,
+) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, attributes, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "climate.turned_off", (HVACMode.OFF,), (HVACMode.HEAT,)
+        ),
+    ],
+)
+async def test_climate_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the climate state trigger fires when any climate state changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other climates also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_attribute_trigger_states(
+            "climate.started_heating",
+            HVACMode.OFF,
+            ATTR_HVAC_ACTION,
+            HVACAction.HEATING,
+            HVACAction.IDLE,
+        )
+    ],
+)
+async def test_climate_state_attribute_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the climate state trigger fires when any climate state changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, *initial_state)
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states:
+        set_or_remove_state(hass, entity_id, *state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other climates also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, *state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "climate.turned_off", (HVACMode.OFF,), (HVACMode.HEAT,)
+        ),
+    ],
+)
+async def test_climate_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entities_in_target: int,
+    entity_id: str,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the climate state trigger fires when the first climate changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other climates should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_attribute_trigger_states(
+            "climate.started_heating",
+            HVACMode.OFF,
+            ATTR_HVAC_ACTION,
+            HVACAction.HEATING,
+            HVACAction.IDLE,
+        )
+    ],
+)
+async def test_climate_state_attribute_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the climate state trigger fires when any climate state changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, *initial_state)
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states:
+        set_or_remove_state(hass, entity_id, *state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other climates should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, *state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "climate.turned_off", (HVACMode.OFF,), (HVACMode.HEAT,)
+        ),
+    ],
+)
+async def test_climate_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entities_in_target: int,
+    entity_id: str,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the climate state trigger fires when the last climate changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("climate"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "initial_state", "states"),
+    [
+        *parametrize_attribute_trigger_states(
+            "climate.started_heating",
+            HVACMode.OFF,
+            ATTR_HVAC_ACTION,
+            HVACAction.HEATING,
+            HVACAction.IDLE,
+        )
+    ],
+)
+async def test_climate_state_attribute_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_climates: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the climate state trigger fires when any climate state changes to a specific state."""
+    await async_setup_component(hass, "climate", {})
+
+    other_entity_ids = set(target_climates) - {entity_id}
+
+    # Set all climates, including the tested climate, to the initial state
+    for eid in target_climates:
+        set_or_remove_state(hass, eid, *initial_state)
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, *state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, *state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/cover/test_trigger.py

@@ -0,0 +1,430 @@
+"""Test cover trigger."""
+
+import pytest
+
+from homeassistant.components.cover import ATTR_CURRENT_POSITION, CoverState
+from homeassistant.const import CONF_ENTITY_ID, STATE_UNAVAILABLE, STATE_UNKNOWN
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import arm_trigger, parametrize_target_entities, target_entities
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_covers(hass: HomeAssistant) -> None:
+    """Create multiple cover entities associated with different targets."""
+    return await target_entities(hass, "cover")
+
+
+def set_or_remove_state(
+    hass: HomeAssistant,
+    entity_id: str,
+    state: str | None,
+    attributes: dict | None = None,
+) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, attributes, force_update=True)
+
+
+def parametrize_opened_closed_trigger_states(
+    trigger: str,
+    trigger_options: dict,
+    device_class: str,
+    target_state: tuple[str, dict],
+    other_state: tuple[str, dict],
+) -> list[
+    tuple[
+        str,
+        dict,
+        str,
+        tuple[str | None, dict],
+        list[tuple[tuple[str | None, dict], int]],
+    ]
+]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, trigger_options, device_class,
+    initial_state, list of states), where states is a list of tuples
+    (state to set, expected service call count).
+    """
+    return [
+        ## TODO: Check what happens if attribute is missing
+        # Initial state None
+        (
+            trigger,
+            trigger_options,
+            device_class,
+            (None, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                # This doesn't trigger because the device class is not set
+                # when the trigger is created. We need to teach TargetStateChangeTracker
+                # about device classes.
+                (target_state, 0),
+            ],
+        ),
+        # Initial state different from target state
+        (
+            trigger,
+            trigger_options,
+            device_class,
+            other_state,
+            [
+                (target_state, 1),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        # Initial state same as target state
+        (
+            trigger,
+            trigger_options,
+            device_class,
+            target_state,
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        # Initial state unavailable / unknown
+        (
+            trigger,
+            trigger_options,
+            device_class,
+            (STATE_UNAVAILABLE, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+        (
+            trigger,
+            trigger_options,
+            device_class,
+            (STATE_UNKNOWN, {}),
+            [
+                (target_state, 0),
+                (other_state, 0),
+                (target_state, 1),
+            ],
+        ),
+    ]
+
+
+def parametrize_opened_trigger_states(
+    trigger: str, device_class: str
+) -> list[
+    tuple[
+        str,
+        dict,
+        str,
+        tuple[str | None, dict],
+        list[tuple[tuple[str | None, dict], int]],
+    ]
+]:
+    """Parametrize states and expected service call counts.
+
+    Returns a list of tuples with (trigger, trigger_options, device_class,
+    initial_state, list of states), where states is a list of tuples
+    (state to set, expected service call count).
+    """
+    return [
+        # States without current position attribute
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {"fully_opened": True},
+            device_class,
+            (CoverState.OPEN, {}),
+            (CoverState.CLOSED, {}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {"fully_opened": True},
+            device_class,
+            (CoverState.OPENING, {}),
+            (CoverState.CLOSED, {}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {},
+            device_class,
+            (CoverState.OPEN, {}),
+            (CoverState.CLOSED, {}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {},
+            device_class,
+            (CoverState.OPENING, {}),
+            (CoverState.CLOSED, {}),
+        ),
+        # States with current position attribute
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {"fully_opened": True},
+            device_class,
+            (CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}),
+            (CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {"fully_opened": True},
+            device_class,
+            (CoverState.OPENING, {ATTR_CURRENT_POSITION: 100}),
+            (CoverState.OPENING, {ATTR_CURRENT_POSITION: 0}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {},
+            device_class,
+            (CoverState.OPEN, {ATTR_CURRENT_POSITION: 1}),
+            (CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),
+        ),
+        *parametrize_opened_closed_trigger_states(
+            trigger,
+            {},
+            device_class,
+            (CoverState.OPENING, {ATTR_CURRENT_POSITION: 1}),
+            (CoverState.CLOSED, {ATTR_CURRENT_POSITION: 0}),
+        ),
+    ]
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("cover"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "trigger_options", "device_class", "initial_state", "states"),
+    [
+        *parametrize_opened_trigger_states("cover.garage_opened", "garage"),
+        # No initial state attribute, doesn't trigger because it's already in target state.
+        (
+            "cover.garage_opened",
+            {"fully_opened": True},
+            "garage",
+            (CoverState.OPEN, {}),
+            [
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 1),
+            ],
+        ),
+    ],
+)
+async def test_cover_state_attribute_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_covers: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    trigger_options: dict,
+    device_class: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the cover state trigger fires when any cover state changes to a specific state."""
+    await async_setup_component(hass, "cover", {})
+
+    other_entity_ids = set(target_covers) - {entity_id}
+
+    # Set all covers, including the tested cover, to the initial state
+    for eid in target_covers:
+        set_or_remove_state(
+            hass,
+            eid,
+            initial_state[0],
+            initial_state[1] | {"device_class": device_class},
+        )
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, trigger_options, trigger_target_config)
+
+    for state, expected_calls in states:
+        set_or_remove_state(
+            hass, entity_id, state[0], state[1] | {"device_class": device_class}
+        )
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other covers also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(
+                hass,
+                other_entity_id,
+                state[0],
+                state[1] | {"device_class": device_class},
+            )
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("cover"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "trigger_options", "device_class", "initial_state", "states"),
+    [
+        *parametrize_opened_trigger_states("cover.garage_opened", "garage"),
+        # No initial state attribute, doesn't trigger because it's already in target state.
+        (
+            "cover.garage_opened",
+            {"fully_opened": True},
+            "garage",
+            (CoverState.OPEN, {}),
+            [
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 1),
+            ],
+        ),
+    ],
+)
+async def test_cover_state_attribute_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_covers: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    trigger_options: dict,
+    device_class: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the cover state trigger fires when the first cover state changes to a specific state."""
+    await async_setup_component(hass, "cover", {})
+
+    other_entity_ids = set(target_covers) - {entity_id}
+
+    # Set all covers, including the tested cover, to the initial state
+    for eid in target_covers:
+        set_or_remove_state(
+            hass,
+            eid,
+            initial_state[0],
+            initial_state[1] | {"device_class": device_class},
+        )
+        await hass.async_block_till_done()
+
+    await arm_trigger(
+        hass,
+        trigger,
+        {"behavior": "first"} | trigger_options,
+        trigger_target_config,
+    )
+
+    for state, expected_calls in states:
+        set_or_remove_state(
+            hass, entity_id, state[0], state[1] | {"device_class": device_class}
+        )
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other covers should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(
+                hass,
+                other_entity_id,
+                state[0],
+                state[1] | {"device_class": device_class},
+            )
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("cover"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "trigger_options", "device_class", "initial_state", "states"),
+    [
+        *parametrize_opened_trigger_states("cover.garage_opened", "garage"),
+        # No initial state attribute, doesn't trigger because it's already in target state.
+        (
+            "cover.garage_opened",
+            {"fully_opened": True},
+            "garage",
+            (CoverState.OPEN, {}),
+            [
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 0}), 0),
+                ((CoverState.OPEN, {ATTR_CURRENT_POSITION: 100}), 1),
+            ],
+        ),
+    ],
+)
+async def test_cover_state_attribute_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_covers: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    trigger_options: dict,
+    device_class: str,
+    initial_state: tuple[str | None, dict],
+    states: list[tuple[tuple[str, dict], int]],
+) -> None:
+    """Test that the cover state trigger fires when the last cover state changes to a specific state."""
+    await async_setup_component(hass, "cover", {})
+
+    other_entity_ids = set(target_covers) - {entity_id}
+
+    # Set all covers, including the tested cover, to the initial state
+    for eid in target_covers:
+        set_or_remove_state(
+            hass,
+            eid,
+            initial_state[0],
+            initial_state[1] | {"device_class": device_class},
+        )
+        await hass.async_block_till_done()
+
+    await arm_trigger(
+        hass, trigger, {"behavior": "last"} | trigger_options, trigger_target_config
+    )
+
+    for state, expected_calls in states:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(
+                hass,
+                other_entity_id,
+                state[0],
+                state[1] | {"device_class": device_class},
+            )
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(
+            hass, entity_id, state[0], state[1] | {"device_class": device_class}
+        )
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/fan/test_trigger.py

@@ -0,0 +1,177 @@
+"""Test fan trigger."""
+
+import pytest
+
+from homeassistant.const import CONF_ENTITY_ID, STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_fans(hass: HomeAssistant) -> None:
+    """Create multiple fan entities associated with different targets."""
+    return await target_entities(hass, "fan")
+
+
+def set_or_remove_state(hass: HomeAssistant, entity_id: str, state: str | None) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("fan"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("fan.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("fan.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_fan_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_fans: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the fan state trigger fires when any fan state changes to a specific state."""
+    await async_setup_component(hass, "fan", {})
+
+    other_entity_ids = set(target_fans) - {entity_id}
+
+    # Set all fans, including the tested fan, to the initial state
+    for eid in target_fans:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other fans also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("fan"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("fan.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("fan.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_fan_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_fans: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the fan state trigger fires when the first fan changes to a specific state."""
+    await async_setup_component(hass, "fan", {})
+
+    other_entity_ids = set(target_fans) - {entity_id}
+
+    # Set all fans, including the tested fan, to the initial state
+    for eid in target_fans:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other fans should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("fan"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("fan.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("fan.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_fan_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_fans: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the fan state trigger fires when the last fan changes to a specific state."""
+    await async_setup_component(hass, "fan", {})
+
+    other_entity_ids = set(target_fans) - {entity_id}
+
+    # Set all fans, including the tested fan, to the initial state
+    for eid in target_fans:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/light/test_condition.py

@@ -0,0 +1,203 @@
+"""Test light conditions."""
+
+import pytest
+
+from homeassistant.components import automation
+from homeassistant.const import (
+    ATTR_LABEL_ID,
+    CONF_CONDITION,
+    CONF_OPTIONS,
+    CONF_TARGET,
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+)
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.helpers import entity_registry as er, label_registry as lr
+from homeassistant.setup import async_setup_component
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def label_entities(hass: HomeAssistant) -> list[str]:
+    """Create multiple entities associated with labels."""
+    await async_setup_component(hass, "light", {})
+
+    config_entry = MockConfigEntry(domain="test_labels")
+    config_entry.add_to_hass(hass)
+
+    label_reg = lr.async_get(hass)
+    label = label_reg.async_create("Test Label")
+
+    entity_reg = er.async_get(hass)
+
+    for i in range(3):
+        light_entity = entity_reg.async_get_or_create(
+            domain="light",
+            platform="test",
+            unique_id=f"label_light_{i}",
+            suggested_object_id=f"label_light_{i}",
+        )
+        entity_reg.async_update_entity(light_entity.entity_id, labels={label.label_id})
+
+    # Also create switches to test that they don't impact the conditions
+    for i in range(2):
+        switch_entity = entity_reg.async_get_or_create(
+            domain="switch",
+            platform="test",
+            unique_id=f"label_switch_{i}",
+            suggested_object_id=f"label_switch_{i}",
+        )
+        entity_reg.async_update_entity(switch_entity.entity_id, labels={label.label_id})
+
+    return [
+        "light.label_light_0",
+        "light.label_light_1",
+        "light.label_light_2",
+    ]
+
+
+async def setup_automation_with_light_condition(
+    hass: HomeAssistant,
+    *,
+    condition: str,
+    target: dict,
+    behavior: str,
+) -> None:
+    """Set up automation with light state condition."""
+    await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: {
+                "trigger": {"platform": "event", "event_type": "test_event"},
+                "condition": {
+                    CONF_CONDITION: condition,
+                    CONF_TARGET: target,
+                    CONF_OPTIONS: {"behavior": behavior},
+                },
+                "action": {
+                    "service": "test.automation",
+                },
+            }
+        },
+    )
+
+
+async def has_calls_after_trigger(
+    hass: HomeAssistant, service_calls: list[ServiceCall]
+) -> bool:
+    """Check if there are service calls after the trigger event."""
+    hass.bus.async_fire("test_event")
+    await hass.async_block_till_done()
+    has_calls = len(service_calls) == 1
+    service_calls.clear()
+    return has_calls
+
+
+@pytest.mark.parametrize(
+    ("condition", "state", "reverse_state"),
+    [("light.is_on", STATE_ON, STATE_OFF), ("light.is_off", STATE_OFF, STATE_ON)],
+)
+async def test_light_state_condition_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    label_entities: list[str],
+    condition: str,
+    state: str,
+    reverse_state: str,
+) -> None:
+    """Test the light state condition with the 'any' behavior."""
+    await async_setup_component(hass, "light", {})
+
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, reverse_state)
+
+    await setup_automation_with_light_condition(
+        hass,
+        condition=condition,
+        target={ATTR_LABEL_ID: "test_label", "entity_id": "light.nonexistent"},
+        behavior="any",
+    )
+
+    # Set state for two switches to ensure that they don't impact the condition
+    hass.states.async_set("switch.label_switch_1", STATE_OFF)
+    hass.states.async_set("switch.label_switch_2", STATE_ON)
+
+    # No lights on the condition state
+    assert not await has_calls_after_trigger(hass, service_calls)
+
+    # Set one light to the condition state -> condition pass
+    hass.states.async_set(label_entities[0], state)
+    assert await has_calls_after_trigger(hass, service_calls)
+
+    # Set all lights to the condition state -> condition pass
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, state)
+    assert await has_calls_after_trigger(hass, service_calls)
+
+    # Set one light to unavailable -> condition pass
+    hass.states.async_set(label_entities[0], STATE_UNAVAILABLE)
+    assert await has_calls_after_trigger(hass, service_calls)
+
+    # Set all lights to unavailable -> condition fail
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, STATE_UNAVAILABLE)
+    assert not await has_calls_after_trigger(hass, service_calls)
+
+
+@pytest.mark.parametrize(
+    ("condition", "state", "reverse_state"),
+    [("light.is_on", STATE_ON, STATE_OFF), ("light.is_off", STATE_OFF, STATE_ON)],
+)
+async def test_light_state_condition_behavior_all(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    label_entities: list[str],
+    condition: str,
+    state: str,
+    reverse_state: str,
+) -> None:
+    """Test the light state condition with the 'all' behavior."""
+    await async_setup_component(hass, "light", {})
+
+    # Set state for two switches to ensure that they don't impact the condition
+    hass.states.async_set("switch.label_switch_1", STATE_OFF)
+    hass.states.async_set("switch.label_switch_2", STATE_ON)
+
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, reverse_state)
+
+    await setup_automation_with_light_condition(
+        hass,
+        condition=condition,
+        target={ATTR_LABEL_ID: "test_label", "entity_id": "light.nonexistent"},
+        behavior="all",
+    )
+
+    # No lights on the condition state
+    assert not await has_calls_after_trigger(hass, service_calls)
+
+    # Set one light to the condition state -> condition fail
+    hass.states.async_set(label_entities[0], state)
+    assert not await has_calls_after_trigger(hass, service_calls)
+
+    # Set all lights to the condition state -> condition pass
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, state)
+    assert await has_calls_after_trigger(hass, service_calls)
+
+    # Set one light to unavailable -> condition still pass
+    hass.states.async_set(label_entities[0], STATE_UNAVAILABLE)
+    assert await has_calls_after_trigger(hass, service_calls)
+
+    # Set all lights to unavailable -> condition passes
+    for entity_id in label_entities:
+        hass.states.async_set(entity_id, STATE_UNAVAILABLE)
+    assert await has_calls_after_trigger(hass, service_calls)

tests/components/light/test_trigger.py

@@ -0,0 +1,177 @@
+"""Test light trigger."""
+
+import pytest
+
+from homeassistant.const import CONF_ENTITY_ID, STATE_OFF, STATE_ON
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_lights(hass: HomeAssistant) -> None:
+    """Create multiple light entities associated with different targets."""
+    return await target_entities(hass, "light")
+
+
+def set_or_remove_state(hass: HomeAssistant, entity_id: str, state: str | None) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("light"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("light.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("light.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_light_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_lights: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the light state trigger fires when any light state changes to a specific state."""
+    await async_setup_component(hass, "light", {})
+
+    other_entity_ids = set(target_lights) - {entity_id}
+
+    # Set all lights, including the tested light, to the initial state
+    for eid in target_lights:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other lights also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("light"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("light.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("light.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_light_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_lights: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the light state trigger fires when the first light changes to a specific state."""
+    await async_setup_component(hass, "light", {})
+
+    other_entity_ids = set(target_lights) - {entity_id}
+
+    # Set all lights, including the tested light, to the initial state
+    for eid in target_lights:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other lights should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("light"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states("light.turned_on", (STATE_ON,), (STATE_OFF,)),
+        *parametrize_trigger_states("light.turned_off", (STATE_OFF,), (STATE_ON,)),
+    ],
+)
+async def test_light_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_lights: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the light state trigger fires when the last light changes to a specific state."""
+    await async_setup_component(hass, "light", {})
+
+    other_entity_ids = set(target_lights) - {entity_id}
+
+    # Set all lights, including the tested light, to the initial state
+    for eid in target_lights:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/components/media_player/test_trigger.py

@@ -0,0 +1,204 @@
+"""Test media player trigger."""
+
+import pytest
+
+from homeassistant.components.media_player import MediaPlayerState
+from homeassistant.const import CONF_ENTITY_ID
+from homeassistant.core import HomeAssistant, ServiceCall
+from homeassistant.setup import async_setup_component
+
+from tests.components import (
+    arm_trigger,
+    parametrize_target_entities,
+    parametrize_trigger_states,
+    target_entities,
+)
+
+
+@pytest.fixture(autouse=True, name="stub_blueprint_populate")
+def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
+    """Stub copying the blueprints to the config folder."""
+
+
+@pytest.fixture
+async def target_media_players(hass: HomeAssistant) -> None:
+    """Create multiple media player entities associated with different targets."""
+    return await target_entities(hass, "media_player")
+
+
+def set_or_remove_state(
+    hass: HomeAssistant,
+    entity_id: str,
+    state: str | None,
+    attributes: dict | None = None,
+) -> None:
+    """Set or clear the state of an entity."""
+    if state is None:
+        hass.states.async_remove(entity_id)
+    else:
+        hass.states.async_set(entity_id, state, attributes, force_update=True)
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "media_player.stopped_playing",
+            (MediaPlayerState.IDLE, MediaPlayerState.OFF, MediaPlayerState.ON),
+            (
+                MediaPlayerState.BUFFERING,
+                MediaPlayerState.PAUSED,
+                MediaPlayerState.PLAYING,
+            ),
+        ),
+    ],
+)
+async def test_media_player_state_trigger_behavior_any(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the media player state trigger fires when any media player state changes to a specific state."""
+    await async_setup_component(hass, "media_player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Check if changing other media players also triggers
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * expected_calls
+        service_calls.clear()
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "media_player.stopped_playing",
+            (MediaPlayerState.IDLE, MediaPlayerState.OFF, MediaPlayerState.ON),
+            (
+                MediaPlayerState.BUFFERING,
+                MediaPlayerState.PAUSED,
+                MediaPlayerState.PLAYING,
+            ),
+        ),
+    ],
+)
+async def test_media_player_state_trigger_behavior_first(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int, list[str]]],
+) -> None:
+    """Test that the media player state trigger fires when the first media player changes to a specific state."""
+    await async_setup_component(hass, "media_player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "first"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()
+
+        # Triggering other media players should not cause the trigger to fire again
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_trigger_states(
+            "media_player.stopped_playing",
+            (MediaPlayerState.IDLE, MediaPlayerState.OFF, MediaPlayerState.ON),
+            (
+                MediaPlayerState.BUFFERING,
+                MediaPlayerState.PAUSED,
+                MediaPlayerState.PLAYING,
+            ),
+        ),
+    ],
+)
+async def test_media_player_state_trigger_behavior_last(
+    hass: HomeAssistant,
+    service_calls: list[ServiceCall],
+    target_media_players: list[str],
+    trigger_target_config: dict,
+    entity_id: str,
+    entities_in_target: int,
+    trigger: str,
+    states: list[tuple[str, int]],
+) -> None:
+    """Test that the media player state trigger fires when the last media player changes to a specific state."""
+    await async_setup_component(hass, "media_player", {})
+
+    other_entity_ids = set(target_media_players) - {entity_id}
+
+    # Set all media players, including the tested media player, to the initial state
+    for eid in target_media_players:
+        set_or_remove_state(hass, eid, states[0][0])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state, expected_calls in states[1:]:
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == expected_calls
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()

tests/helpers/test_translation.py

@@ -450,10 +450,10 @@ async def test_caching(hass: HomeAssistant) -> None:
         side_effect=translation.build_resources,
     ) as mock_build_resources:
         load1 = await translation.async_get_translations(hass, "en", "entity_component")
-        assert len(mock_build_resources.mock_calls) == 7
+        assert len(mock_build_resources.mock_calls) == 9
 
         load2 = await translation.async_get_translations(hass, "en", "entity_component")
-        assert len(mock_build_resources.mock_calls) == 7
+        assert len(mock_build_resources.mock_calls) == 9
 
         assert load1 == load2