core/tests/components/device_automation/test_init.py

"""The test for light device automation."""

from unittest.mock import AsyncMock, Mock, patch

import attr
import pytest
from pytest_unordered import unordered
import voluptuous as vol

from homeassistant import loader
from homeassistant.components import automation, device_automation
from homeassistant.components.device_automation import (
    InvalidDeviceAutomationConfig,
    toggle_entity,
)
from homeassistant.components.websocket_api import TYPE_RESULT
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import IntegrationNotFound
from homeassistant.requirements import RequirementsNotFound
from homeassistant.setup import async_setup_component

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


@attr.s(frozen=True, slots=True)
class MockDeviceEntry(dr.DeviceEntry):
    """Device Registry Entry with fixed UUID."""

    id: str = attr.ib(default="very_unique")


@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
def fake_integration(hass: HomeAssistant) -> None:
    """Set up a mock integration with device automation support."""
    DOMAIN = "fake_integration"

    hass.config.components.add(DOMAIN)

    async def _async_get_actions(
        hass: HomeAssistant, device_id: str
    ) -> list[dict[str, str]]:
        """List device actions."""
        return await toggle_entity.async_get_actions(hass, device_id, DOMAIN)

    async def _async_get_conditions(
        hass: HomeAssistant, device_id: str
    ) -> list[dict[str, str]]:
        """List device conditions."""
        return await toggle_entity.async_get_conditions(hass, device_id, DOMAIN)

    async def _async_get_triggers(
        hass: HomeAssistant, device_id: str
    ) -> list[dict[str, str]]:
        """List device triggers."""
        return await toggle_entity.async_get_triggers(hass, device_id, DOMAIN)

    mock_platform(
        hass,
        f"{DOMAIN}.device_action",
        Mock(
            ACTION_SCHEMA=toggle_entity.ACTION_SCHEMA.extend(
                {vol.Required("domain"): DOMAIN}
            ),
            async_get_actions=_async_get_actions,
            spec=["ACTION_SCHEMA", "async_get_actions"],
        ),
    )

    mock_platform(
        hass,
        f"{DOMAIN}.device_condition",
        Mock(
            CONDITION_SCHEMA=toggle_entity.CONDITION_SCHEMA.extend(
                {vol.Required("domain"): DOMAIN}
            ),
            async_get_conditions=_async_get_conditions,
            spec=["CONDITION_SCHEMA", "async_get_conditions"],
        ),
    )

    mock_platform(
        hass,
        f"{DOMAIN}.device_trigger",
        Mock(
            TRIGGER_SCHEMA=vol.All(
                toggle_entity.TRIGGER_SCHEMA,
                vol.Schema({vol.Required("domain"): DOMAIN}, extra=vol.ALLOW_EXTRA),
            ),
            async_get_triggers=_async_get_triggers,
            spec=["TRIGGER_SCHEMA", "async_get_triggers"],
        ),
    )


async def test_websocket_get_actions(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get the expected actions through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_actions = [
        {
            "domain": "fake_integration",
            "type": "turn_off",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
        {
            "domain": "fake_integration",
            "type": "turn_on",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
        {
            "domain": "fake_integration",
            "type": "toggle",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
    ]

    client = await hass_ws_client(hass)
    await client.send_json(
        {"id": 1, "type": "device_automation/action/list", "device_id": device_entry.id}
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    actions = msg["result"]
    assert actions == unordered(expected_actions)


async def test_websocket_get_conditions(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get the expected conditions through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_conditions = [
        {
            "condition": "device",
            "domain": "fake_integration",
            "type": "is_off",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
        {
            "condition": "device",
            "domain": "fake_integration",
            "type": "is_on",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
    ]

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/condition/list",
            "device_id": device_entry.id,
        }
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    conditions = msg["result"]
    assert conditions == unordered(expected_conditions)


async def test_websocket_get_triggers(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get the expected triggers through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_triggers = [
        {
            "platform": "device",
            "domain": "fake_integration",
            "type": "changed_states",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
        {
            "platform": "device",
            "domain": "fake_integration",
            "type": "turned_off",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
        {
            "platform": "device",
            "domain": "fake_integration",
            "type": "turned_on",
            "device_id": device_entry.id,
            "entity_id": entity_entry.id,
            "metadata": {"secondary": False},
        },
    ]

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/trigger/list",
            "device_id": device_entry.id,
        }
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    triggers = msg["result"]
    assert triggers == unordered(expected_triggers)


async def test_websocket_get_action_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get the expected action capabilities through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_capabilities = {
        "turn_on": {
            "extra_fields": [{"type": "string", "name": "code", "optional": True}]
        },
        "turn_off": {"extra_fields": []},
        "toggle": {"extra_fields": []},
    }

    async def _async_get_action_capabilities(
        hass: HomeAssistant, config: ConfigType
    ) -> dict[str, vol.Schema]:
        """List action capabilities."""
        if config["type"] == "turn_on":
            return {"extra_fields": vol.Schema({vol.Optional("code"): str})}
        return {}

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_action"]
    module.async_get_action_capabilities = _async_get_action_capabilities

    client = await hass_ws_client(hass)
    await client.send_json(
        {"id": 1, "type": "device_automation/action/list", "device_id": device_entry.id}
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    actions = msg["result"]

    msg_id = 2
    assert len(actions) == 3
    for action in actions:
        await client.send_json(
            {
                "id": msg_id,
                "type": "device_automation/action/capabilities",
                "action": action,
            }
        )
        msg = await client.receive_json()
        assert msg["id"] == msg_id
        assert msg["type"] == TYPE_RESULT
        assert msg["success"]
        capabilities = msg["result"]
        assert capabilities == expected_capabilities[action["type"]]
        msg_id = msg_id + 1


async def test_websocket_get_action_capabilities_unknown_domain(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get no action capabilities for a non existing domain."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/action/capabilities",
            "action": {"domain": "beer"},
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_action_capabilities_no_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no action capabilities for a domain which has none.

    The tests tests a domain which has a device action platform, but no
    async_get_action_capabilities.
    """
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/action/capabilities",
            "action": {"domain": "fake_integration"},
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_action_capabilities_bad_action(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no action capabilities when there is an error."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_action"]
    module.async_get_action_capabilities = Mock(
        side_effect=InvalidDeviceAutomationConfig
    )

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/action/capabilities",
            "action": {"domain": "fake_integration"},
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities
    module.async_get_action_capabilities.assert_called_once()


async def test_websocket_get_condition_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get the expected condition capabilities through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_capabilities = {
        "extra_fields": [
            {"name": "for", "optional": True, "type": "positive_time_period_dict"}
        ]
    }

    async def _async_get_condition_capabilities(
        hass: HomeAssistant, config: ConfigType
    ) -> dict[str, vol.Schema]:
        """List condition capabilities."""
        return await toggle_entity.async_get_condition_capabilities(hass, config)

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_condition"]
    module.async_get_condition_capabilities = _async_get_condition_capabilities

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/condition/list",
            "device_id": device_entry.id,
        }
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    conditions = msg["result"]

    msg_id = 2
    assert len(conditions) == 2
    for condition in conditions:
        await client.send_json(
            {
                "id": msg_id,
                "type": "device_automation/condition/capabilities",
                "condition": condition,
            }
        )
        msg = await client.receive_json()
        assert msg["id"] == msg_id
        assert msg["type"] == TYPE_RESULT
        assert msg["success"]
        capabilities = msg["result"]
        assert capabilities == expected_capabilities
        msg_id = msg_id + 1


async def test_websocket_get_condition_capabilities_unknown_domain(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get no condition capabilities for a non existing domain."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/condition/capabilities",
            "condition": {"condition": "device", "domain": "beer", "device_id": "1234"},
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_condition_capabilities_no_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no condition capabilities for a domain which has none.

    The tests tests a domain which has a device condition platform, but no
    async_get_condition_capabilities.
    """
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/condition/capabilities",
            "condition": {
                "condition": "device",
                "device_id": "abcd",
                "domain": "fake_integration",
            },
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_condition_capabilities_bad_condition(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no condition capabilities when there is an error."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_condition"]
    module.async_get_condition_capabilities = Mock(
        side_effect=InvalidDeviceAutomationConfig
    )

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/condition/capabilities",
            "condition": {
                "condition": "device",
                "device_id": "abcd",
                "domain": "fake_integration",
            },
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities
    module.async_get_condition_capabilities.assert_called_once()


async def test_async_get_device_automations_single_device_trigger(
    hass: HomeAssistant,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get can fetch the triggers for a device id."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "light", "test", "5678", device_id=device_entry.id
    )
    result = await device_automation.async_get_device_automations(
        hass, device_automation.DeviceAutomationType.TRIGGER, [device_entry.id]
    )
    assert device_entry.id in result
    assert len(result[device_entry.id]) == 3


async def test_async_get_device_automations_all_devices_trigger(
    hass: HomeAssistant,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get can fetch all the triggers when no device id is passed."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "light", "test", "5678", device_id=device_entry.id
    )
    result = await device_automation.async_get_device_automations(
        hass, device_automation.DeviceAutomationType.TRIGGER
    )
    assert device_entry.id in result
    assert len(result[device_entry.id]) == 3  # toggled, turned_on, turned_off


async def test_async_get_device_automations_all_devices_condition(
    hass: HomeAssistant,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get can fetch all the conditions when no device id is passed."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "light", "test", "5678", device_id=device_entry.id
    )
    result = await device_automation.async_get_device_automations(
        hass, device_automation.DeviceAutomationType.CONDITION
    )
    assert device_entry.id in result
    assert len(result[device_entry.id]) == 2


async def test_async_get_device_automations_all_devices_action(
    hass: HomeAssistant,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get can fetch all the actions when no device id is passed."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "light", "test", "5678", device_id=device_entry.id
    )
    result = await device_automation.async_get_device_automations(
        hass, device_automation.DeviceAutomationType.ACTION
    )
    assert device_entry.id in result
    assert len(result[device_entry.id]) == 3


async def test_async_get_device_automations_all_devices_action_exception_throw(
    hass: HomeAssistant,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test we get can fetch all the actions when no device id is passed and can handle one throwing an exception."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "light", "test", "5678", device_id=device_entry.id
    )
    with patch(
        "homeassistant.components.light.device_trigger.async_get_triggers",
        side_effect=KeyError,
    ):
        result = await device_automation.async_get_device_automations(
            hass, device_automation.DeviceAutomationType.TRIGGER
        )
    assert device_entry.id in result
    assert len(result[device_entry.id]) == 0
    assert "KeyError" in caplog.text


@pytest.mark.parametrize(
    "trigger_key",
    ["trigger", "platform"],
)
async def test_websocket_get_trigger_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
    trigger_key: str,
) -> None:
    """Test we get the expected trigger capabilities through websocket."""
    await async_setup_component(hass, "device_automation", {})
    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )
    expected_capabilities = {
        "extra_fields": [
            {"name": "for", "optional": True, "type": "positive_time_period_dict"}
        ]
    }

    async def _async_get_trigger_capabilities(
        hass: HomeAssistant, config: ConfigType
    ) -> dict[str, vol.Schema]:
        """List trigger capabilities."""
        return await toggle_entity.async_get_trigger_capabilities(hass, config)

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_get_trigger_capabilities = _async_get_trigger_capabilities

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/trigger/list",
            "device_id": device_entry.id,
        }
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    triggers: dict = msg["result"]

    msg_id = 2
    assert len(triggers) == 3  # toggled, turned_on, turned_off
    for trigger in triggers:
        trigger[trigger_key] = trigger.pop("platform")
        await client.send_json(
            {
                "id": msg_id,
                "type": "device_automation/trigger/capabilities",
                "trigger": trigger,
            }
        )
        msg = await client.receive_json()
        assert msg["id"] == msg_id
        assert msg["type"] == TYPE_RESULT
        assert msg["success"]
        capabilities = msg["result"]
        assert capabilities == expected_capabilities
        msg_id = msg_id + 1


async def test_websocket_get_trigger_capabilities_unknown_domain(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test we get no trigger capabilities for a non existing domain."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/trigger/capabilities",
            "trigger": {"platform": "device", "domain": "beer", "device_id": "abcd"},
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_trigger_capabilities_no_capabilities(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no trigger capabilities for a domain which has none.

    The tests tests a domain which has a device trigger platform, but no
    async_get_trigger_capabilities.
    """
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/trigger/capabilities",
            "trigger": {
                "platform": "device",
                "device_id": "abcd",
                "domain": "fake_integration",
            },
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities


async def test_websocket_get_trigger_capabilities_bad_trigger(
    hass: HomeAssistant,
    hass_ws_client: WebSocketGenerator,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test we get no trigger capabilities when there is an error."""
    await async_setup_component(hass, "device_automation", {})
    expected_capabilities = {}

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_get_trigger_capabilities = Mock(
        side_effect=InvalidDeviceAutomationConfig
    )

    client = await hass_ws_client(hass)
    await client.send_json(
        {
            "id": 1,
            "type": "device_automation/trigger/capabilities",
            "trigger": {
                "platform": "device",
                "device_id": "abcd",
                "domain": "fake_integration",
            },
        }
    )
    msg = await client.receive_json()
    assert msg["id"] == 1
    assert msg["type"] == TYPE_RESULT
    assert msg["success"]
    capabilities = msg["result"]
    assert capabilities == expected_capabilities
    module.async_get_trigger_capabilities.assert_called_once()


async def test_automation_with_non_existing_integration(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test device automation trigger with non existing integration."""
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": "none",
                    "domain": "beer",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    assert "Integration 'beer' not found" in caplog.text


async def test_automation_with_device_action(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test automation with a device action."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_action"]
    module.async_call_action_from_config = AsyncMock()

    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "action": {
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                    "entity_id": entity_entry.id,
                    "type": "turn_on",
                },
            }
        },
    )

    module.async_call_action_from_config.assert_not_called()

    hass.bus.async_fire("test_event1")
    await hass.async_block_till_done()

    module.async_call_action_from_config.assert_awaited_once()


async def test_automation_with_dynamically_validated_action(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    fake_integration,
) -> None:
    """Test device automation with an action which is dynamically validated."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_action"]
    module.async_validate_action_config = AsyncMock()

    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "action": {"device_id": device_entry.id, "domain": "fake_integration"},
            }
        },
    )

    module.async_validate_action_config.assert_awaited_once()


async def test_automation_with_integration_without_device_action(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test device automation action with integration without device action support."""
    mock_integration(hass, MockModule(domain="test"))
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "action": {"device_id": "", "domain": "test"},
            }
        },
    )

    assert (
        "Integration 'test' does not support device automation actions" in caplog.text
    )


async def test_automation_with_device_condition(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test automation with a device condition."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_condition"]
    module.async_condition_from_config = Mock()

    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "condition": {
                    "condition": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                    "entity_id": entity_entry.id,
                    "type": "is_on",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_condition_from_config.assert_called_once()


async def test_automation_with_dynamically_validated_condition(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    fake_integration,
) -> None:
    """Test device automation with a condition which is dynamically validated."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_condition"]
    module.async_validate_condition_config = AsyncMock()

    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "condition": {
                    "condition": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_validate_condition_config.assert_awaited_once()


async def test_automation_with_integration_without_device_condition(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test device automation condition with integration without device condition support."""
    mock_integration(hass, MockModule(domain="test"))
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "condition": {
                    "condition": "device",
                    "device_id": "none",
                    "domain": "test",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    assert (
        "Integration 'test' does not support device automation conditions"
        in caplog.text
    )


async def test_automation_with_device_trigger(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test automation with a device trigger."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_attach_trigger = AsyncMock()

    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry = entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                    "entity_id": entity_entry.id,
                    "type": "turned_off",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_attach_trigger.assert_awaited_once()


async def test_automation_with_dynamically_validated_trigger(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    fake_integration,
) -> None:
    """Test device automation with a trigger which is dynamically validated."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_attach_trigger = AsyncMock()
    module.async_validate_trigger_config = AsyncMock(wraps=lambda hass, config: config)

    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_registry.async_get_or_create(
        "fake_integration", "test", "5678", device_id=device_entry.id
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_validate_trigger_config.assert_awaited_once()
    module.async_attach_trigger.assert_awaited_once()


async def test_automation_with_integration_without_device_trigger(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test device automation trigger with integration without device trigger support."""
    mock_integration(hass, MockModule(domain="test"))
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": "none",
                    "domain": "test",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    assert (
        "Integration 'test' does not support device automation triggers" in caplog.text
    )


BAD_AUTOMATIONS = [
    (
        {"device_id": "very_unique", "domain": "light"},
        "required key not provided @ data['entity_id']",
    ),
    (
        {"device_id": "wrong", "domain": "light"},
        "Unknown device 'wrong'",
    ),
    (
        {"device_id": "wrong"},
        "required key not provided @ data{path}['domain']",
    ),
    (
        {"device_id": "wrong", "domain": "light"},
        "Unknown device 'wrong'",
    ),
    (
        {"device_id": "very_unique", "domain": "light"},
        "required key not provided @ data['entity_id']",
    ),
    (
        {"device_id": "very_unique", "domain": "light", "entity_id": "wrong"},
        "Unknown entity 'wrong'",
    ),
]

BAD_TRIGGERS = BAD_CONDITIONS = [
    *BAD_AUTOMATIONS,
    ({"domain": "light"}, "required key not provided @ data{path}['device_id']"),
]


@patch("homeassistant.helpers.device_registry.DeviceEntry", MockDeviceEntry)
@pytest.mark.parametrize(("action", "expected_error"), BAD_AUTOMATIONS)
async def test_automation_with_bad_action(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    action: dict[str, str],
    expected_error: str,
) -> None:
    """Test automation with bad device action."""
    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "action": action,
            }
        },
    )

    assert expected_error.format(path="['actions'][0]") in caplog.text


@patch("homeassistant.helpers.device_registry.DeviceEntry", MockDeviceEntry)
@pytest.mark.parametrize(("condition", "expected_error"), BAD_CONDITIONS)
async def test_automation_with_bad_condition_action(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
    condition: dict[str, str],
    expected_error: str,
) -> None:
    """Test automation with bad device action."""
    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "action": {"condition": "device"} | condition,
            }
        },
    )

    assert expected_error.format(path="['actions'][0]") in caplog.text


@patch("homeassistant.helpers.device_registry.DeviceEntry", MockDeviceEntry)
@pytest.mark.parametrize(("condition", "expected_error"), BAD_CONDITIONS)
async def test_automation_with_bad_condition(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    condition: dict[str, str],
    expected_error: str,
) -> None:
    """Test automation with bad device condition."""
    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "condition": {"condition": "device"} | condition,
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    assert expected_error.format(path="['conditions'][0]") in caplog.text


async def test_automation_with_sub_condition(
    hass: HomeAssistant,
    service_calls: list[ServiceCall],
    device_registry: dr.DeviceRegistry,
    entity_registry: er.EntityRegistry,
) -> None:
    """Test automation with device condition under and/or conditions."""
    DOMAIN = "light"

    config_entry = MockConfigEntry(domain="test", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    entity_entry1 = entity_registry.async_get_or_create(
        "fake_integration", "test", "0001", device_id=device_entry.id
    )
    entity_entry2 = entity_registry.async_get_or_create(
        "fake_integration", "test", "0002", device_id=device_entry.id
    )

    hass.states.async_set(entity_entry1.entity_id, STATE_ON)
    hass.states.async_set(entity_entry2.entity_id, STATE_OFF)

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: [
                {
                    "trigger": {"platform": "event", "event_type": "test_event1"},
                    "condition": [
                        {
                            "condition": "and",
                            "conditions": [
                                {
                                    "condition": "device",
                                    "domain": DOMAIN,
                                    "device_id": device_entry.id,
                                    "entity_id": entity_entry1.id,
                                    "type": "is_on",
                                },
                                {
                                    "condition": "device",
                                    "domain": DOMAIN,
                                    "device_id": device_entry.id,
                                    "entity_id": entity_entry2.id,
                                    "type": "is_on",
                                },
                            ],
                        }
                    ],
                    "action": {
                        "service": "test.automation",
                        "data_template": {
                            "some": (
                                "and {{ trigger.platform }}"
                                " - {{ trigger.event.event_type }}"
                            )
                        },
                    },
                },
                {
                    "trigger": {"platform": "event", "event_type": "test_event1"},
                    "condition": [
                        {
                            "condition": "or",
                            "conditions": [
                                {
                                    "condition": "device",
                                    "domain": DOMAIN,
                                    "device_id": device_entry.id,
                                    "entity_id": entity_entry1.id,
                                    "type": "is_on",
                                },
                                {
                                    "condition": "device",
                                    "domain": DOMAIN,
                                    "device_id": device_entry.id,
                                    "entity_id": entity_entry2.id,
                                    "type": "is_on",
                                },
                            ],
                        }
                    ],
                    "action": {
                        "service": "test.automation",
                        "data_template": {
                            "some": (
                                "or {{ trigger.platform }}"
                                " - {{ trigger.event.event_type }}"
                            )
                        },
                    },
                },
            ]
        },
    )
    await hass.async_block_till_done()
    assert hass.states.get(entity_entry1.entity_id).state == STATE_ON
    assert hass.states.get(entity_entry2.entity_id).state == STATE_OFF
    assert len(service_calls) == 0

    hass.bus.async_fire("test_event1")
    await hass.async_block_till_done()
    assert len(service_calls) == 1
    assert service_calls[0].data["some"] == "or event - test_event1"

    hass.states.async_set(entity_entry1.entity_id, STATE_OFF)
    hass.bus.async_fire("test_event1")
    await hass.async_block_till_done()
    assert len(service_calls) == 1

    hass.states.async_set(entity_entry2.entity_id, STATE_ON)
    hass.bus.async_fire("test_event1")
    await hass.async_block_till_done()
    assert len(service_calls) == 2
    assert service_calls[1].data["some"] == "or event - test_event1"

    hass.states.async_set(entity_entry1.entity_id, STATE_ON)
    hass.bus.async_fire("test_event1")
    await hass.async_block_till_done()
    assert len(service_calls) == 4
    assert [service_calls[2].data["some"], service_calls[3].data["some"]] == unordered(
        ["or event - test_event1", "and event - test_event1"]
    )


@patch("homeassistant.helpers.device_registry.DeviceEntry", MockDeviceEntry)
@pytest.mark.parametrize(("condition", "expected_error"), BAD_CONDITIONS)
async def test_automation_with_bad_sub_condition(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    condition: dict[str, str],
    expected_error: str,
) -> None:
    """Test automation with bad device condition under and/or conditions."""
    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "event", "event_type": "test_event1"},
                "condition": {
                    "condition": "and",
                    "conditions": [{"condition": "device"} | condition],
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    path = "['conditions'][0]['conditions'][0]"
    assert expected_error.format(path=path) in caplog.text


@patch("homeassistant.helpers.device_registry.DeviceEntry", MockDeviceEntry)
@pytest.mark.parametrize(("trigger", "expected_error"), BAD_TRIGGERS)
async def test_automation_with_bad_trigger(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    trigger: dict[str, str],
    expected_error: str,
) -> None:
    """Test automation with bad device trigger."""
    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.mock_state(hass, ConfigEntryState.LOADED)
    config_entry.add_to_hass(hass)
    device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {"platform": "device"} | trigger,
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    assert expected_error.format(path="") in caplog.text


async def test_websocket_device_not_found(
    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
) -> None:
    """Test calling command with unknown device."""
    await async_setup_component(hass, "device_automation", {})
    client = await hass_ws_client(hass)
    await client.send_json(
        {"id": 1, "type": "device_automation/action/list", "device_id": "non-existing"}
    )
    msg = await client.receive_json()

    assert msg["id"] == 1
    assert not msg["success"]
    assert msg["error"] == {"code": "not_found", "message": "Device not found"}


async def test_automation_with_unknown_device(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture, fake_integration
) -> None:
    """Test device automation with a trigger with an unknown device."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_validate_trigger_config = AsyncMock()

    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": "no_such_device",
                    "domain": "fake_integration",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_validate_trigger_config.assert_not_awaited()
    assert (
        "Automation with alias 'hello' failed to setup triggers and has been disabled: "
        "Unknown device 'no_such_device'" in caplog.text
    )


async def test_automation_with_device_wrong_domain(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    fake_integration,
) -> None:
    """Test device automation where the device doesn't have the right config entry."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_validate_trigger_config = AsyncMock()

    source_config_entry = MockConfigEntry(domain="not_fake_integration")
    source_config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=source_config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_validate_trigger_config.assert_not_awaited()
    assert (
        "Automation with alias 'hello' failed to setup triggers and has been disabled: "
        f"Device '{device_entry.id}' has no config entry from domain 'fake_integration'"
        in caplog.text
    )


async def test_automation_with_device_component_not_loaded(
    hass: HomeAssistant,
    caplog: pytest.LogCaptureFixture,
    device_registry: dr.DeviceRegistry,
    fake_integration,
) -> None:
    """Test device automation where the device's config entry is not loaded."""

    module_cache = hass.data[loader.DATA_COMPONENTS]
    module = module_cache["fake_integration.device_trigger"]
    module.async_validate_trigger_config = AsyncMock()
    module.async_attach_trigger = AsyncMock()

    config_entry = MockConfigEntry(domain="fake_integration", data={})
    config_entry.add_to_hass(hass)
    device_entry = device_registry.async_get_or_create(
        config_entry_id=config_entry.entry_id,
        connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
    )
    assert await async_setup_component(
        hass,
        automation.DOMAIN,
        {
            automation.DOMAIN: {
                "alias": "hello",
                "trigger": {
                    "platform": "device",
                    "device_id": device_entry.id,
                    "domain": "fake_integration",
                },
                "action": {"service": "test.automation", "entity_id": "hello.world"},
            }
        },
    )

    module.async_validate_trigger_config.assert_not_awaited()


@pytest.mark.parametrize(
    "exc",
    [
        IntegrationNotFound("test"),
        RequirementsNotFound("test", []),
        ImportError("test"),
    ],
)
async def test_async_get_device_automations_platform_reraises_exceptions(
    hass: HomeAssistant, exc: Exception
) -> None:
    """Test InvalidDeviceAutomationConfig is raised when async_get_integration_with_requirements fails."""
    await async_setup_component(hass, "device_automation", {})
    with (
        patch(
            "homeassistant.components.device_automation.async_get_integration_with_requirements",
            side_effect=exc,
        ),
        pytest.raises(InvalidDeviceAutomationConfig),
    ):
        await device_automation.async_get_device_automation_platform(
            hass, "test", device_automation.DeviceAutomationType.TRIGGER
        )