core/tests/components/zha/test_device_action.py

"""The test for ZHA device automation actions."""
from unittest.mock import call, patch

import pytest
from pytest_unordered import unordered
from zhaquirks.inovelli.VZM31SN import InovelliVZM31SNv11
import zigpy.profiles.zha
import zigpy.zcl.clusters.general as general
import zigpy.zcl.clusters.security as security
import zigpy.zcl.foundation as zcl_f

import homeassistant.components.automation as automation
from homeassistant.components.device_automation import DeviceAutomationType
from homeassistant.components.zha import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.setup import async_setup_component

from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_TYPE

from tests.common import async_get_device_automations, async_mock_service


@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."""


SHORT_PRESS = "remote_button_short_press"
COMMAND = "command"
COMMAND_SINGLE = "single"


@pytest.fixture(autouse=True)
def required_platforms_only():
    """Only set up the required platforms and required base platforms to speed up tests."""
    with patch(
        "homeassistant.components.zha.PLATFORMS",
        (
            Platform.BINARY_SENSOR,
            Platform.BUTTON,
            Platform.DEVICE_TRACKER,
            Platform.LIGHT,
            Platform.NUMBER,
            Platform.SELECT,
            Platform.SENSOR,
            Platform.SWITCH,
            Platform.SIREN,
        ),
    ):
        yield


@pytest.fixture
async def device_ias(hass, zigpy_device_mock, zha_device_joined_restored):
    """IAS device fixture."""

    clusters = [general.Basic, security.IasZone, security.IasWd]
    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [c.cluster_id for c in clusters],
                SIG_EP_OUTPUT: [general.OnOff.cluster_id],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.ON_OFF_SWITCH,
            }
        },
    )

    zha_device = await zha_device_joined_restored(zigpy_device)
    zha_device.update_available(True)
    await hass.async_block_till_done()
    return zigpy_device, zha_device


@pytest.fixture
async def device_inovelli(hass, zigpy_device_mock, zha_device_joined):
    """Inovelli device fixture."""

    zigpy_device = zigpy_device_mock(
        {
            1: {
                SIG_EP_INPUT: [
                    general.Basic.cluster_id,
                    general.Identify.cluster_id,
                    general.OnOff.cluster_id,
                    general.LevelControl.cluster_id,
                    0xFC31,
                ],
                SIG_EP_OUTPUT: [],
                SIG_EP_TYPE: zigpy.profiles.zha.DeviceType.DIMMABLE_LIGHT,
            }
        },
        ieee="00:1d:8f:08:0c:90:69:6b",
        manufacturer="Inovelli",
        model="VZM31-SN",
        quirk=InovelliVZM31SNv11,
    )

    zha_device = await zha_device_joined(zigpy_device)
    zha_device.update_available(True)
    await hass.async_block_till_done()
    return zigpy_device, zha_device


async def test_get_actions(hass: HomeAssistant, device_ias) -> None:
    """Test we get the expected actions from a ZHA device."""

    ieee_address = str(device_ias[0].ieee)

    device_registry = dr.async_get(hass)
    reg_device = device_registry.async_get_device(identifiers={(DOMAIN, ieee_address)})
    entity_registry = er.async_get(hass)
    siren_level_select = entity_registry.async_get(
        "select.fakemanufacturer_fakemodel_default_siren_level"
    )
    siren_tone_select = entity_registry.async_get(
        "select.fakemanufacturer_fakemodel_default_siren_tone"
    )
    strobe_level_select = entity_registry.async_get(
        "select.fakemanufacturer_fakemodel_default_strobe_level"
    )
    strobe_select = entity_registry.async_get(
        "select.fakemanufacturer_fakemodel_default_strobe"
    )

    actions = await async_get_device_automations(
        hass, DeviceAutomationType.ACTION, reg_device.id
    )

    expected_actions = [
        {
            "domain": DOMAIN,
            "type": "squawk",
            "device_id": reg_device.id,
            "metadata": {},
        },
        {"domain": DOMAIN, "type": "warn", "device_id": reg_device.id, "metadata": {}},
    ]
    expected_actions.extend(
        [
            {
                "domain": Platform.SELECT,
                "type": action,
                "device_id": reg_device.id,
                "entity_id": entity_id,
                "metadata": {"secondary": True},
            }
            for action in [
                "select_first",
                "select_last",
                "select_next",
                "select_option",
                "select_previous",
            ]
            for entity_id in [
                siren_level_select.id,
                siren_tone_select.id,
                strobe_level_select.id,
                strobe_select.id,
            ]
        ]
    )

    assert actions == unordered(expected_actions)


async def test_get_inovelli_actions(hass: HomeAssistant, device_inovelli) -> None:
    """Test we get the expected actions from a ZHA device."""

    inovelli_ieee_address = str(device_inovelli[0].ieee)
    device_registry = dr.async_get(hass)
    inovelli_reg_device = device_registry.async_get_device(
        identifiers={(DOMAIN, inovelli_ieee_address)}
    )
    entity_registry = er.async_get(hass)
    inovelli_button = entity_registry.async_get("button.inovelli_vzm31_sn_identify")
    inovelli_light = entity_registry.async_get("light.inovelli_vzm31_sn_light")

    actions = await async_get_device_automations(
        hass, DeviceAutomationType.ACTION, inovelli_reg_device.id
    )

    expected_actions = [
        {
            "device_id": inovelli_reg_device.id,
            "domain": DOMAIN,
            "metadata": {},
            "type": "issue_all_led_effect",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": DOMAIN,
            "metadata": {},
            "type": "issue_individual_led_effect",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.BUTTON,
            "entity_id": inovelli_button.id,
            "metadata": {"secondary": True},
            "type": "press",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "turn_off",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "turn_on",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "toggle",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "brightness_increase",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "brightness_decrease",
        },
        {
            "device_id": inovelli_reg_device.id,
            "domain": Platform.LIGHT,
            "entity_id": inovelli_light.id,
            "metadata": {"secondary": False},
            "type": "flash",
        },
    ]

    assert actions == unordered(expected_actions)


async def test_action(hass: HomeAssistant, device_ias, device_inovelli) -> None:
    """Test for executing a ZHA device action."""
    zigpy_device, zha_device = device_ias
    inovelli_zigpy_device, inovelli_zha_device = device_inovelli

    zigpy_device.device_automation_triggers = {
        (SHORT_PRESS, SHORT_PRESS): {COMMAND: COMMAND_SINGLE}
    }

    ieee_address = str(zha_device.ieee)
    inovelli_ieee_address = str(inovelli_zha_device.ieee)

    device_registry = dr.async_get(hass)
    reg_device = device_registry.async_get_device(identifiers={(DOMAIN, ieee_address)})
    inovelli_reg_device = device_registry.async_get_device(
        identifiers={(DOMAIN, inovelli_ieee_address)}
    )

    cluster = inovelli_zigpy_device.endpoints[1].in_clusters[0xFC31]

    with patch(
        "zigpy.zcl.Cluster.request",
        return_value=[0x00, zcl_f.Status.SUCCESS],
    ):
        assert await async_setup_component(
            hass,
            automation.DOMAIN,
            {
                automation.DOMAIN: [
                    {
                        "trigger": {
                            "device_id": reg_device.id,
                            "domain": "zha",
                            "platform": "device",
                            "type": SHORT_PRESS,
                            "subtype": SHORT_PRESS,
                        },
                        "action": [
                            {
                                "domain": DOMAIN,
                                "device_id": reg_device.id,
                                "type": "warn",
                            },
                            {
                                "domain": DOMAIN,
                                "device_id": inovelli_reg_device.id,
                                "type": "issue_all_led_effect",
                                "effect_type": "Open_Close",
                                "duration": 5,
                                "level": 10,
                                "color": 41,
                            },
                            {
                                "domain": DOMAIN,
                                "device_id": inovelli_reg_device.id,
                                "type": "issue_individual_led_effect",
                                "effect_type": "Falling",
                                "led_number": 1,
                                "duration": 5,
                                "level": 10,
                                "color": 41,
                            },
                        ],
                    }
                ]
            },
        )

        await hass.async_block_till_done()
        calls = async_mock_service(hass, DOMAIN, "warning_device_warn")

        cluster_handler = zha_device.endpoints[1].client_cluster_handlers["1:0x0006"]
        cluster_handler.zha_send_event(COMMAND_SINGLE, [])
        await hass.async_block_till_done()

        assert len(calls) == 1
        assert calls[0].domain == DOMAIN
        assert calls[0].service == "warning_device_warn"
        assert calls[0].data["ieee"] == ieee_address

        assert len(cluster.request.mock_calls) == 2
        assert (
            call(
                False,
                cluster.commands_by_name["led_effect"].id,
                cluster.commands_by_name["led_effect"].schema,
                6,
                41,
                10,
                5,
                expect_reply=False,
                manufacturer=4151,
                tsn=None,
            )
            in cluster.request.call_args_list
        )
        assert (
            call(
                False,
                cluster.commands_by_name["individual_led_effect"].id,
                cluster.commands_by_name["individual_led_effect"].schema,
                1,
                6,
                41,
                10,
                5,
                expect_reply=False,
                manufacturer=4151,
                tsn=None,
            )
            in cluster.request.call_args_list
        )


async def test_invalid_zha_event_type(hass: HomeAssistant, device_ias) -> None:
    """Test that unexpected types are not passed to `zha_send_event`."""
    zigpy_device, zha_device = device_ias
    cluster_handler = zha_device._endpoints[1].client_cluster_handlers["1:0x0006"]

    # `zha_send_event` accepts only zigpy responses, lists, and dicts
    with pytest.raises(TypeError):
        cluster_handler.zha_send_event(COMMAND_SINGLE, 123)