Refactor tests

homeassistant/components/media_player/icons.json

@@ -106,6 +106,9 @@
     }
   },
   "triggers": {
+    "muted": {
+      "trigger": "mdi:volume-mute"
+    },
     "stopped_playing": {
       "trigger": "mdi:stop"
     }

homeassistant/components/media_player/strings.json

@@ -380,6 +380,16 @@
   },
   "title": "Media player",
   "triggers": {
+    "muted": {
+      "description": "Triggers after one or more media players are muted.",
+      "fields": {
+        "behavior": {
+          "description": "[%key:component::media_player::common::trigger_behavior_description%]",
+          "name": "[%key:component::media_player::common::trigger_behavior_name%]"
+        }
+      },
+      "name": "Media player muted"
+    },
     "stopped_playing": {
       "description": "Triggers after one or more media players stop playing media.",
       "fields": {

homeassistant/components/media_player/trigger.py

@@ -1,12 +1,35 @@
 """Provides triggers for media players."""
 
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.trigger import Trigger, make_conditional_entity_state_trigger
+from homeassistant.core import HomeAssistant, State
+from homeassistant.helpers.trigger import (
+    EntityTriggerBase,
+    Trigger,
+    make_conditional_entity_state_trigger,
+)
 
-from . import MediaPlayerState
+from . import ATTR_MEDIA_VOLUME_LEVEL, ATTR_MEDIA_VOLUME_MUTED, MediaPlayerState
 from .const import DOMAIN
 
+
+class MediaPlayerMutedTrigger(EntityTriggerBase):
+    """Class for media player muted triggers."""
+
+    _domain: str = DOMAIN
+
+    def is_muted(self, state: State) -> bool:
+        """Check if the media player is muted."""
+        return (
+            state.attributes.get(ATTR_MEDIA_VOLUME_MUTED) is True
+            or state.attributes.get(ATTR_MEDIA_VOLUME_LEVEL) == 0
+        )
+
+    def is_to_state(self, state: State) -> bool:
+        """Check if the state matches the target state."""
+        return self.is_muted(state)
+
+
 TRIGGERS: dict[str, type[Trigger]] = {
+    "muted": MediaPlayerMutedTrigger,
     "stopped_playing": make_conditional_entity_state_trigger(
         DOMAIN,
         from_states={

homeassistant/components/media_player/triggers.yaml

@@ -1,4 +1,4 @@
-stopped_playing:
+.trigger_common: &trigger_common
   target:
     entity:
       domain: media_player
@@ -13,3 +13,6 @@ stopped_playing:
             - first
             - last
             - any
+
+muted: *trigger_common
+stopped_playing: *trigger_common

tests/components/media_player/test_trigger.py

@@ -5,7 +5,11 @@ from unittest.mock import patch
 
 import pytest
 
-from homeassistant.components.media_player import MediaPlayerState
+from homeassistant.components.media_player import (
+    ATTR_MEDIA_VOLUME_LEVEL,
+    ATTR_MEDIA_VOLUME_MUTED,
+    MediaPlayerState,
+)
 from homeassistant.const import ATTR_LABEL_ID, CONF_ENTITY_ID
 from homeassistant.core import HomeAssistant, ServiceCall
 from homeassistant.setup import async_setup_component
@@ -60,6 +64,52 @@ async def test_media_player_triggers_gated_by_labs_flag(
     ) in caplog.text
 
 
+def parametrize_muted_trigger_states() -> list[tuple[str, list[StateDescription]]]:
+    """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).
+    """
+    trigger = "media_player.muted"
+    return parametrize_trigger_states(
+        trigger=trigger,
+        target_states=[
+            # States with muted attribute
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: True}),
+            # States with volume attribute
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 0}),
+            # States with muted and volume attribute
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: True},
+            ),
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 0, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: True},
+            ),
+        ],
+        other_states=[
+            # States with muted attribute
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: False}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_MUTED: None}),
+            (MediaPlayerState.PLAYING, {}),  # Missing attribute
+            # States with volume attribute
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: 1}),
+            (MediaPlayerState.PLAYING, {ATTR_MEDIA_VOLUME_LEVEL: None}),
+            (MediaPlayerState.PLAYING, {}),  # Missing attribute
+            # States with muted and volume attribute
+            (
+                MediaPlayerState.PLAYING,
+                {ATTR_MEDIA_VOLUME_LEVEL: 1, ATTR_MEDIA_VOLUME_MUTED: False},
+            ),
+        ],
+    )
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -122,6 +172,56 @@ async def test_media_player_state_trigger_behavior_any(
         service_calls.clear()
 
 
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_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[StateDescription],
+) -> 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]["included"])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {}, trigger_target_config)
+
+    for state in states[1:]:
+        included_state = state["included"]
+        set_or_remove_state(hass, entity_id, included_state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == state["count"]
+        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, included_state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == (entities_in_target - 1) * state["count"]
+        service_calls.clear()
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -183,6 +283,60 @@ async def test_media_player_state_trigger_behavior_first(
         assert len(service_calls) == 0
 
 
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_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[StateDescription],
+) -> None:
+    """Test that the media player state trigger fires when the first 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]["included"])
+        await hass.async_block_till_done()
+
+    await arm_trigger(
+        hass,
+        trigger,
+        {"behavior": "first"},
+        trigger_target_config,
+    )
+
+    for state in states[1:]:
+        included_state = state["included"]
+        set_or_remove_state(hass, entity_id, included_state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == state["count"]
+        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, included_state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+
 @pytest.mark.usefixtures("enable_experimental_triggers_conditions")
 @pytest.mark.parametrize(
     ("trigger_target_config", "entity_id", "entities_in_target"),
@@ -241,3 +395,51 @@ async def test_media_player_state_trigger_behavior_last(
         for service_call in service_calls:
             assert service_call.data[CONF_ENTITY_ID] == entity_id
         service_calls.clear()
+
+
+@pytest.mark.usefixtures("enable_experimental_triggers_conditions")
+@pytest.mark.parametrize(
+    ("trigger_target_config", "entity_id", "entities_in_target"),
+    parametrize_target_entities("media_player"),
+)
+@pytest.mark.parametrize(
+    ("trigger", "states"),
+    [
+        *parametrize_muted_trigger_states(),
+    ],
+)
+async def test_media_player_state_attribute_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[StateDescription],
+) -> None:
+    """Test that the media player state trigger fires when the last 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]["included"])
+        await hass.async_block_till_done()
+
+    await arm_trigger(hass, trigger, {"behavior": "last"}, trigger_target_config)
+
+    for state in states[1:]:
+        included_state = state["included"]
+        for other_entity_id in other_entity_ids:
+            set_or_remove_state(hass, other_entity_id, included_state)
+            await hass.async_block_till_done()
+        assert len(service_calls) == 0
+
+        set_or_remove_state(hass, entity_id, included_state)
+        await hass.async_block_till_done()
+        assert len(service_calls) == state["count"]
+        for service_call in service_calls:
+            assert service_call.data[CONF_ENTITY_ID] == entity_id
+        service_calls.clear()