"""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
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,
mock_coro,
)
@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)
ha_device_registry = dr.async_get(hass)
reg_device = ha_device_registry.async_get_device({(DOMAIN, ieee_address)})
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 [
"select.fakemanufacturer_fakemodel_default_siren_level",
"select.fakemanufacturer_fakemodel_default_siren_tone",
"select.fakemanufacturer_fakemodel_default_strobe_level",
"select.fakemanufacturer_fakemodel_default_strobe",
]
]
)
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)
ha_device_registry = dr.async_get(hass)
inovelli_reg_device = ha_device_registry.async_get_device(
{(DOMAIN, inovelli_ieee_address)}
)
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": "button.inovelli_vzm31_sn_identify",
"metadata": {"secondary": True},
"type": "press",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"metadata": {"secondary": False},
"type": "turn_off",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"metadata": {"secondary": False},
"type": "turn_on",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"metadata": {"secondary": False},
"type": "toggle",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"metadata": {"secondary": False},
"type": "brightness_increase",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"metadata": {"secondary": False},
"type": "brightness_decrease",
},
{
"device_id": inovelli_reg_device.id,
"domain": Platform.LIGHT,
"entity_id": "light.inovelli_vzm31_sn_light",
"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)
ha_device_registry = dr.async_get(hass)
reg_device = ha_device_registry.async_get_device({(DOMAIN, ieee_address)})
inovelli_reg_device = ha_device_registry.async_get_device(
{(DOMAIN, inovelli_ieee_address)}
)
cluster = inovelli_zigpy_device.endpoints[1].in_clusters[0xFC31]
with patch(
"zigpy.zcl.Cluster.request",
return_value=mock_coro([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)