Add get_services_for_target websocket command (home-assistant#157334)

Author: abmantis

homeassistant/components/websocket_api/automation.py

@@ -14,6 +14,9 @@
     get_device_class,
     get_supported_features,
 )
+from homeassistant.helpers.service import (
+    async_get_all_descriptions as async_get_all_service_descriptions,
+)
 from homeassistant.helpers.trigger import (
     async_get_all_descriptions as async_get_all_trigger_descriptions,
 )
@@ -194,3 +197,19 @@ async def async_get_triggers_for_target(
     return _async_get_automation_components_for_target(
         hass, target_selector, expand_group, descriptions
     )
+
+
+async def async_get_services_for_target(
+    hass: HomeAssistant, target_selector: ConfigType, expand_group: bool
+) -> set[str]:
+    """Get services for a target."""
+    descriptions = await async_get_all_service_descriptions(hass)
+    # Flatten dicts to be keyed by domain.name to match trigger/condition format
+    descriptions_flatten = {
+        f"{domain}.{service_name}": desc
+        for domain, services in descriptions.items()
+        for service_name, desc in services.items()
+    }
+    return _async_get_automation_components_for_target(
+        hass, target_selector, expand_group, descriptions_flatten
+    )

homeassistant/components/websocket_api/commands.py

@@ -87,7 +87,7 @@
 from homeassistant.util.json import format_unserializable_data
 
 from . import const, decorators, messages
-from .automation import async_get_triggers_for_target
+from .automation import async_get_services_for_target, async_get_triggers_for_target
 from .connection import ActiveConnection
 from .messages import construct_event_message, construct_result_message
 
@@ -108,11 +108,12 @@ def async_register_commands(
     async_reg(hass, handle_entity_source)
     async_reg(hass, handle_execute_script)
     async_reg(hass, handle_extract_from_target)
-    async_reg(hass, handle_get_triggers_for_target)
     async_reg(hass, handle_fire_event)
     async_reg(hass, handle_get_config)
     async_reg(hass, handle_get_services)
+    async_reg(hass, handle_get_services_for_target)
     async_reg(hass, handle_get_states)
+    async_reg(hass, handle_get_triggers_for_target)
     async_reg(hass, handle_manifest_get)
     async_reg(hass, handle_integration_setup_info)
     async_reg(hass, handle_manifest_list)
@@ -902,6 +903,29 @@ async def handle_get_triggers_for_target(
     connection.send_result(msg["id"], triggers)
 
 
+@decorators.websocket_command(
+    {
+        vol.Required("type"): "get_services_for_target",
+        vol.Required("target"): cv.TARGET_FIELDS,
+        vol.Optional("expand_group", default=True): bool,
+    }
+)
+@decorators.async_response
+async def handle_get_services_for_target(
+    hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
+) -> None:
+    """Handle get services for target command.
+
+    This command returns all services that can be used with any entities that are currently
+    part of a target.
+    """
+    services = await async_get_services_for_target(
+        hass, msg["target"], msg["expand_group"]
+    )
+
+    connection.send_result(msg["id"], services)
+
+
 @decorators.websocket_command(
     {
         vol.Required("type"): "subscribe_trigger",

tests/components/websocket_api/test_commands.py

@@ -3904,3 +3904,239 @@ async def assert_triggers(target: dict[str, list[str]], expected: list[str]) ->
             "switch.turned_on",
         ],
     )
+
+
+@pytest.mark.usefixtures("target_entities")
+@patch("annotatedyaml.loader.load_yaml")
+@patch.object(Integration, "has_services", return_value=True)
+async def test_get_services_for_target(
+    mock_has_services: Mock,
+    mock_load_yaml: Mock,
+    hass: HomeAssistant,
+    websocket_client: MockHAClientWebSocket,
+) -> None:
+    """Test get_services_for_target command with mixed target types."""
+
+    def get_common_service_descriptions(domain: str):
+        return f"""
+        turn_on:
+          target:
+            entity:
+              domain: {domain}
+          fields:
+            behavior:
+              required: true
+              default: any
+              selector:
+                select:
+                  options:
+                    - first
+                    - last
+                    - any
+        non_target_service:
+          fields:
+            behavior:
+              required: true
+              default: any
+              selector:
+                select:
+                  options:
+                    - first
+                    - last
+                    - any
+    """
+
+    component1_service_descriptions = """
+        light_message:
+          target:
+            entity:
+              - domain: light
+                integration: component1
+              - domain: sensor
+                integration: component1
+                device_class: illuminance
+        light_flash:
+          target:
+            entity:
+              domain: light
+              integration: component1
+              supported_features:
+                - light.LightEntityFeature.FLASH
+                - light.LightEntityFeature.EFFECT
+        light_dance:
+          target:
+            entity:
+              domain: light
+              integration: component1
+              supported_features: # both required features
+                - - light.LightEntityFeature.FLASH
+                  - light.LightEntityFeature.TRANSITION
+    """
+
+    component2_service_descriptions = """
+        match_all:
+          target:
+
+        other_integration_lights:
+          target:
+            entity:
+              - domain: light
+                supported_features:
+                  - light.LightEntityFeature.EFFECT
+              - integration: test
+                domain: light
+    """
+
+    def _load_yaml(fname, secrets=None):
+        if fname.endswith("component1/services.yaml"):
+            service_descriptions = component1_service_descriptions
+        elif fname.endswith("component2/services.yaml"):
+            service_descriptions = component2_service_descriptions
+        else:
+            service_descriptions = get_common_service_descriptions(fname.split("/")[-2])
+        with io.StringIO(service_descriptions) as file:
+            return parse_yaml(file)
+
+    mock_load_yaml.side_effect = _load_yaml
+    mock_integration(hass, MockModule("light"))
+    assert await async_setup_component(hass, "light", {})
+    mock_integration(hass, MockModule("switch"))
+    assert await async_setup_component(hass, "switch", {})
+    mock_integration(hass, MockModule("sensor"))
+    assert await async_setup_component(hass, "sensor", {})
+    mock_integration(hass, MockModule("component1"))
+    assert await async_setup_component(hass, "component1", {})
+    mock_integration(hass, MockModule("component2"))
+    assert await async_setup_component(hass, "component2", {})
+    await hass.async_block_till_done()
+
+    hass.services.async_register("light", "turn_on", lambda call: None)
+    hass.services.async_register("light", "non_target_service", lambda call: None)
+    hass.services.async_register("switch", "turn_on", lambda call: None)
+    hass.services.async_register("switch", "non_target_service", lambda call: None)
+    hass.services.async_register("sensor", "turn_on", lambda call: None)
+    hass.services.async_register("sensor", "non_target_service", lambda call: None)
+    hass.services.async_register("component1", "light_message", lambda call: None)
+    hass.services.async_register("component1", "light_flash", lambda call: None)
+    hass.services.async_register("component1", "light_dance", lambda call: None)
+    hass.services.async_register("component2", "match_all", lambda call: None)
+    hass.services.async_register(
+        "component2", "other_integration_lights", lambda call: None
+    )
+    await hass.async_block_till_done()
+
+    async def assert_services(target: dict[str, list[str]], expected: list[str]) -> Any:
+        """Call the command and assert expected services."""
+        await websocket_client.send_json_auto_id(
+            {"type": "get_services_for_target", "target": target}
+        )
+        msg = await websocket_client.receive_json()
+
+        assert msg["type"] == const.TYPE_RESULT
+        assert msg["success"]
+        assert sorted(msg["result"]) == sorted(expected)
+
+    # Test entity target - unknown entity
+    await assert_services({"entity_id": ["light.unknown_entity"]}, [])
+
+    # Test entity target - entity not in registry
+    await assert_services(
+        {"entity_id": ["light.not_registry"]},
+        [
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "light.turn_on",
+        ],
+    )
+
+    # Test entity targets
+    await assert_services(
+        {"entity_id": ["light.component1_light", "switch.component1_switch"]},
+        [
+            "component1.light_message",
+            "component2.match_all",
+            "light.turn_on",
+            "switch.turn_on",
+        ],
+    )
+    await assert_services(
+        {"entity_id": ["light.component1_flash_light"]},
+        [
+            "component1.light_flash",
+            "component1.light_message",
+            "component2.match_all",
+            "light.turn_on",
+        ],
+    )
+    await assert_services(
+        {"entity_id": ["light.component1_effect_flash_light"]},
+        [
+            "component1.light_flash",
+            "component1.light_message",
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "light.turn_on",
+        ],
+    )
+    await assert_services(
+        {"entity_id": ["light.component1_flash_transition_light"]},
+        [
+            "component1.light_dance",
+            "component1.light_flash",
+            "component1.light_message",
+            "component2.match_all",
+            "light.turn_on",
+        ],
+    )
+
+    # Test device target - multiple devices
+    await assert_services(
+        {"device_id": ["device1", "device2"]},
+        [
+            "component1.light_message",
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "light.turn_on",
+            "sensor.turn_on",
+            "switch.turn_on",
+        ],
+    )
+
+    # Test area target - multiple areas
+    await assert_services(
+        {"area_id": ["kitchen", "living_room"]},
+        [
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "light.turn_on",
+            "switch.turn_on",
+        ],
+    )
+
+    # Test label target - multiple labels
+    await assert_services(
+        {"label_id": ["label_1", "label_2"]},
+        [
+            "light.turn_on",
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "switch.turn_on",
+        ],
+    )
+    # Test mixed target types
+    await assert_services(
+        {
+            "entity_id": ["light.test1"],
+            "device_id": ["device2"],
+            "area_id": ["kitchen"],
+            "label_id": ["label_1"],
+        },
+        [
+            "component1.light_message",
+            "component2.match_all",
+            "component2.other_integration_lights",
+            "light.turn_on",
+            "sensor.turn_on",
+            "switch.turn_on",
+        ],
+    )