Rfxtrx device triggers and actions (#47909)
* Add helper * Add device actions * Add trigger * Just make use of standard command * Generalize code a bit * Switch tests to currently existing features * Add tests for capabilities * Don't check schema asserted value * Adjust strings somewhat * Directly expose action subtypes * Add a status event test * Switch to modern typing * Drop chime that is now part of command * Adjust strings a bit * Drop ability to set custom value * Adjust changed base schema * Validate triggers * Try fix typing for 3.8
This commit is contained in:
Joakim Plate
GitHub
parent
ecf4a7813a
commit
9b00e0cb7a
7 changed files with 646 additions and 2 deletions
99
homeassistant/components/rfxtrx/device_action.py
Normal file
99
homeassistant/components/rfxtrx/device_action.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
"""Provides device automations for RFXCOM RFXtrx."""
|
||||
from __future__ import annotations
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.device_automation.exceptions import (
|
||||
InvalidDeviceAutomationConfig,
|
||||
)
|
||||
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
|
||||
from homeassistant.core import Context, HomeAssistant
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
|
||||
from . import DATA_RFXOBJECT, DOMAIN
|
||||
from .helpers import async_get_device_object
|
||||
|
||||
CONF_DATA = "data"
|
||||
CONF_SUBTYPE = "subtype"
|
||||
|
||||
ACTION_TYPE_COMMAND = "send_command"
|
||||
ACTION_TYPE_STATUS = "send_status"
|
||||
|
||||
ACTION_TYPES = {
|
||||
ACTION_TYPE_COMMAND,
|
||||
ACTION_TYPE_STATUS,
|
||||
}
|
||||
|
||||
ACTION_SELECTION = {
|
||||
ACTION_TYPE_COMMAND: "COMMANDS",
|
||||
ACTION_TYPE_STATUS: "STATUS",
|
||||
}
|
||||
|
||||
ACTION_SCHEMA = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_TYPE): vol.In(ACTION_TYPES),
|
||||
vol.Required(CONF_SUBTYPE): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_get_actions(hass: HomeAssistant, device_id: str) -> list[dict]:
|
||||
"""List device actions for RFXCOM RFXtrx devices."""
|
||||
|
||||
try:
|
||||
device = async_get_device_object(hass, device_id)
|
||||
except ValueError:
|
||||
return []
|
||||
|
||||
actions = []
|
||||
for action_type in ACTION_TYPES:
|
||||
if hasattr(device, action_type):
|
||||
values = getattr(device, ACTION_SELECTION[action_type], {})
|
||||
for value in values.values():
|
||||
actions.append(
|
||||
{
|
||||
CONF_DEVICE_ID: device_id,
|
||||
CONF_DOMAIN: DOMAIN,
|
||||
CONF_TYPE: action_type,
|
||||
CONF_SUBTYPE: value,
|
||||
}
|
||||
)
|
||||
|
||||
return actions
|
||||
|
||||
|
||||
def _get_commands(hass, device_id, action_type):
|
||||
device = async_get_device_object(hass, device_id)
|
||||
send_fun = getattr(device, action_type)
|
||||
commands = getattr(device, ACTION_SELECTION[action_type], {})
|
||||
return commands, send_fun
|
||||
|
||||
|
||||
async def async_validate_action_config(hass, config):
|
||||
"""Validate config."""
|
||||
config = ACTION_SCHEMA(config)
|
||||
commands, _ = _get_commands(hass, config[CONF_DEVICE_ID], config[CONF_TYPE])
|
||||
sub_type = config[CONF_SUBTYPE]
|
||||
|
||||
if sub_type not in commands.values():
|
||||
raise InvalidDeviceAutomationConfig(
|
||||
f"Subtype {sub_type} not found in device commands {commands}"
|
||||
)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
async def async_call_action_from_config(
|
||||
hass: HomeAssistant, config: dict, variables: dict, context: Context | None
|
||||
) -> None:
|
||||
"""Execute a device action."""
|
||||
config = ACTION_SCHEMA(config)
|
||||
|
||||
rfx = hass.data[DOMAIN][DATA_RFXOBJECT]
|
||||
commands, send_fun = _get_commands(hass, config[CONF_DEVICE_ID], config[CONF_TYPE])
|
||||
sub_type = config[CONF_SUBTYPE]
|
||||
|
||||
for key, value in commands.items():
|
||||
if value == sub_type:
|
||||
await hass.async_add_executor_job(send_fun, rfx.transport, key)
|
||||
return
|
||||
110
homeassistant/components/rfxtrx/device_trigger.py
Normal file
110
homeassistant/components/rfxtrx/device_trigger.py
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
"""Provides device automations for RFXCOM RFXtrx."""
|
||||
from __future__ import annotations
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.automation import AutomationActionType
|
||||
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
|
||||
from homeassistant.components.device_automation.exceptions import (
|
||||
InvalidDeviceAutomationConfig,
|
||||
)
|
||||
from homeassistant.components.homeassistant.triggers import event as event_trigger
|
||||
from homeassistant.components.rfxtrx.const import EVENT_RFXTRX_EVENT
|
||||
from homeassistant.const import (
|
||||
ATTR_DEVICE_ID,
|
||||
CONF_DEVICE_ID,
|
||||
CONF_DOMAIN,
|
||||
CONF_PLATFORM,
|
||||
CONF_TYPE,
|
||||
)
|
||||
from homeassistant.core import CALLBACK_TYPE, HomeAssistant
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
from . import DOMAIN
|
||||
from .helpers import async_get_device_object
|
||||
|
||||
CONF_SUBTYPE = "subtype"
|
||||
|
||||
CONF_TYPE_COMMAND = "command"
|
||||
CONF_TYPE_STATUS = "status"
|
||||
|
||||
TRIGGER_SELECTION = {
|
||||
CONF_TYPE_COMMAND: "COMMANDS",
|
||||
CONF_TYPE_STATUS: "STATUS",
|
||||
}
|
||||
TRIGGER_TYPES = [
|
||||
CONF_TYPE_COMMAND,
|
||||
CONF_TYPE_STATUS,
|
||||
]
|
||||
TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend(
|
||||
{
|
||||
vol.Required(CONF_TYPE): vol.In(TRIGGER_TYPES),
|
||||
vol.Required(CONF_SUBTYPE): str,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def async_get_triggers(hass: HomeAssistant, device_id: str) -> list[dict]:
|
||||
"""List device triggers for RFXCOM RFXtrx devices."""
|
||||
device = async_get_device_object(hass, device_id)
|
||||
|
||||
triggers = []
|
||||
for conf_type in TRIGGER_TYPES:
|
||||
data = getattr(device, TRIGGER_SELECTION[conf_type], {})
|
||||
for command in data.values():
|
||||
triggers.append(
|
||||
{
|
||||
CONF_PLATFORM: "device",
|
||||
CONF_DEVICE_ID: device_id,
|
||||
CONF_DOMAIN: DOMAIN,
|
||||
CONF_TYPE: conf_type,
|
||||
CONF_SUBTYPE: command,
|
||||
}
|
||||
)
|
||||
return triggers
|
||||
|
||||
|
||||
async def async_validate_trigger_config(hass, config):
|
||||
"""Validate config."""
|
||||
config = TRIGGER_SCHEMA(config)
|
||||
|
||||
device = async_get_device_object(hass, config[CONF_DEVICE_ID])
|
||||
|
||||
action_type = config[CONF_TYPE]
|
||||
sub_type = config[CONF_SUBTYPE]
|
||||
commands = getattr(device, TRIGGER_SELECTION[action_type], {})
|
||||
if config[CONF_SUBTYPE] not in commands.values():
|
||||
raise InvalidDeviceAutomationConfig(
|
||||
f"Subtype {sub_type} not found in device triggers {commands}"
|
||||
)
|
||||
|
||||
return config
|
||||
|
||||
|
||||
async def async_attach_trigger(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
action: AutomationActionType,
|
||||
automation_info: dict,
|
||||
) -> CALLBACK_TYPE:
|
||||
"""Attach a trigger."""
|
||||
config = TRIGGER_SCHEMA(config)
|
||||
|
||||
event_data = {ATTR_DEVICE_ID: config[CONF_DEVICE_ID]}
|
||||
|
||||
if config[CONF_TYPE] == CONF_TYPE_COMMAND:
|
||||
event_data["values"] = {"Command": config[CONF_SUBTYPE]}
|
||||
elif config[CONF_TYPE] == CONF_TYPE_STATUS:
|
||||
event_data["values"] = {"Status": config[CONF_SUBTYPE]}
|
||||
|
||||
event_config = event_trigger.TRIGGER_SCHEMA(
|
||||
{
|
||||
event_trigger.CONF_PLATFORM: "event",
|
||||
event_trigger.CONF_EVENT_TYPE: EVENT_RFXTRX_EVENT,
|
||||
event_trigger.CONF_EVENT_DATA: event_data,
|
||||
}
|
||||
)
|
||||
|
||||
return await event_trigger.async_attach_trigger(
|
||||
hass, event_config, action, automation_info, platform_type="device"
|
||||
)
|
||||
22
homeassistant/components/rfxtrx/helpers.py
Normal file
22
homeassistant/components/rfxtrx/helpers.py
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
"""Provides helpers for RFXtrx."""
|
||||
|
||||
|
||||
from RFXtrx import get_device
|
||||
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
|
||||
@callback
|
||||
def async_get_device_object(hass: HomeAssistantType, device_id):
|
||||
"""Get a device for the given device registry id."""
|
||||
device_registry = dr.async_get(hass)
|
||||
registry_device = device_registry.async_get(device_id)
|
||||
if registry_device is None:
|
||||
raise ValueError(f"Device {device_id} not found")
|
||||
|
||||
device_tuple = list(list(registry_device.identifiers)[0])
|
||||
return get_device(
|
||||
int(device_tuple[1], 16), int(device_tuple[2], 16), device_tuple[3]
|
||||
)
|
||||
|
|
@ -70,5 +70,15 @@
|
|||
"invalid_input_off_delay": "Invalid input for off delay",
|
||||
"unknown": "[%key:common::config_flow::error::unknown%]"
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
"action_type": {
|
||||
"send_status": "Send status update: {subtype}",
|
||||
"send_command": "Send command: {subtype}"
|
||||
},
|
||||
"trigger_type": {
|
||||
"status": "Received status: {subtype}",
|
||||
"command": "Received command: {subtype}"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -70,5 +70,16 @@
|
|||
"title": "Configure device options"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"device_automation": {
|
||||
"action_type": {
|
||||
"send_status": "Send status update: {subtype}",
|
||||
"send_command": "Send command: {subtype}"
|
||||
},
|
||||
"trigger_type": {
|
||||
"status": "Received status: {subtype}",
|
||||
"command": "Received command: {subtype}"
|
||||
}
|
||||
},
|
||||
"title": "Rfxtrx"
|
||||
}
|
||||
|
|
|
|||
206
tests/components/rfxtrx/test_device_action.py
Normal file
206
tests/components/rfxtrx/test_device_action.py
Normal file
|
|
@ -0,0 +1,206 @@
|
|||
"""The tests for RFXCOM RFXtrx device actions."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
import RFXtrx
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.automation as automation
|
||||
from homeassistant.components.rfxtrx import DOMAIN
|
||||
from homeassistant.helpers.device_registry import DeviceRegistry
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import (
|
||||
MockConfigEntry,
|
||||
assert_lists_same,
|
||||
async_get_device_automations,
|
||||
async_mock_service,
|
||||
mock_device_registry,
|
||||
mock_registry,
|
||||
)
|
||||
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
|
||||
|
||||
|
||||
@pytest.fixture(name="device_reg")
|
||||
def device_reg_fixture(hass):
|
||||
"""Return an empty, loaded, registry."""
|
||||
return mock_device_registry(hass)
|
||||
|
||||
|
||||
@pytest.fixture(name="entity_reg")
|
||||
def entity_reg_fixture(hass):
|
||||
"""Return an empty, loaded, registry."""
|
||||
return mock_registry(hass)
|
||||
|
||||
|
||||
class DeviceTestData(NamedTuple):
|
||||
"""Test data linked to a device."""
|
||||
|
||||
code: str
|
||||
device_identifiers: set[tuple[str, str, str, str]]
|
||||
|
||||
|
||||
DEVICE_LIGHTING_1 = DeviceTestData("0710002a45050170", {("rfxtrx", "10", "0", "E5")})
|
||||
|
||||
DEVICE_BLINDS_1 = DeviceTestData(
|
||||
"09190000009ba8010100", {("rfxtrx", "19", "0", "009ba8:1")}
|
||||
)
|
||||
|
||||
DEVICE_TEMPHUM_1 = DeviceTestData(
|
||||
"0a52080705020095220269", {("rfxtrx", "52", "8", "05:02")}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("device", [DEVICE_LIGHTING_1, DEVICE_TEMPHUM_1])
|
||||
async def test_device_test_data(rfxtrx, device: DeviceTestData):
|
||||
"""Verify that our testing data remains correct."""
|
||||
pkt: RFXtrx.lowlevel.Packet = RFXtrx.lowlevel.parse(bytearray.fromhex(device.code))
|
||||
assert device.device_identifiers == {
|
||||
("rfxtrx", f"{pkt.packettype:x}", f"{pkt.subtype:x}", pkt.id_string)
|
||||
}
|
||||
|
||||
|
||||
async def setup_entry(hass, devices):
|
||||
"""Construct a config setup."""
|
||||
entry_data = create_rfx_test_cfg(devices=devices)
|
||||
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
|
||||
|
||||
def _get_expected_actions(data):
|
||||
for value in data.values():
|
||||
yield {"type": "send_command", "subtype": value}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"device,expected",
|
||||
[
|
||||
[
|
||||
DEVICE_LIGHTING_1,
|
||||
list(_get_expected_actions(RFXtrx.lowlevel.Lighting1.COMMANDS)),
|
||||
],
|
||||
[
|
||||
DEVICE_BLINDS_1,
|
||||
list(_get_expected_actions(RFXtrx.lowlevel.RollerTrol.COMMANDS)),
|
||||
],
|
||||
[DEVICE_TEMPHUM_1, []],
|
||||
],
|
||||
)
|
||||
async def test_get_actions(hass, device_reg: DeviceRegistry, device, expected):
|
||||
"""Test we get the expected actions from a rfxtrx."""
|
||||
await setup_entry(hass, {device.code: {"signal_repetitions": 1}})
|
||||
|
||||
device_entry = device_reg.async_get_device(device.device_identifiers, set())
|
||||
assert device_entry
|
||||
|
||||
actions = await async_get_device_automations(hass, "action", device_entry.id)
|
||||
actions = [action for action in actions if action["domain"] == DOMAIN]
|
||||
|
||||
expected_actions = [
|
||||
{"domain": DOMAIN, "device_id": device_entry.id, **action_type}
|
||||
for action_type in expected
|
||||
]
|
||||
|
||||
assert_lists_same(actions, expected_actions)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"device,config,expected",
|
||||
[
|
||||
[
|
||||
DEVICE_LIGHTING_1,
|
||||
{"type": "send_command", "subtype": "On"},
|
||||
"0710000045050100",
|
||||
],
|
||||
[
|
||||
DEVICE_LIGHTING_1,
|
||||
{"type": "send_command", "subtype": "Off"},
|
||||
"0710000045050000",
|
||||
],
|
||||
[
|
||||
DEVICE_BLINDS_1,
|
||||
{"type": "send_command", "subtype": "Stop"},
|
||||
"09190000009ba8010200",
|
||||
],
|
||||
],
|
||||
)
|
||||
async def test_action(
|
||||
hass, device_reg: DeviceRegistry, rfxtrx: RFXtrx.Connect, device, config, expected
|
||||
):
|
||||
"""Test for actions."""
|
||||
|
||||
await setup_entry(hass, {device.code: {"signal_repetitions": 1}})
|
||||
|
||||
device_entry = device_reg.async_get_device(device.device_identifiers, set())
|
||||
assert device_entry
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "event",
|
||||
"event_type": "test_event",
|
||||
},
|
||||
"action": {
|
||||
"domain": DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
**config,
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
hass.bus.async_fire("test_event")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
rfxtrx.transport.send.assert_called_once_with(bytearray.fromhex(expected))
|
||||
|
||||
|
||||
async def test_invalid_action(hass, device_reg: DeviceRegistry):
|
||||
"""Test for invalid actions."""
|
||||
device = DEVICE_LIGHTING_1
|
||||
notification_calls = async_mock_service(hass, "persistent_notification", "create")
|
||||
|
||||
await setup_entry(hass, {device.code: {"signal_repetitions": 1}})
|
||||
|
||||
device_identifers: Any = device.device_identifiers
|
||||
device_entry = device_reg.async_get_device(device_identifers, set())
|
||||
assert device_entry
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "event",
|
||||
"event_type": "test_event",
|
||||
},
|
||||
"action": {
|
||||
"domain": DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"type": "send_command",
|
||||
"subtype": "invalid",
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(notification_calls) == 1
|
||||
assert (
|
||||
"The following integrations and platforms could not be set up"
|
||||
in notification_calls[0].data["message"]
|
||||
)
|
||||
186
tests/components/rfxtrx/test_device_trigger.py
Normal file
186
tests/components/rfxtrx/test_device_trigger.py
Normal file
|
|
@ -0,0 +1,186 @@
|
|||
"""The tests for RFXCOM RFXtrx device triggers."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
import pytest
|
||||
|
||||
import homeassistant.components.automation as automation
|
||||
from homeassistant.components.rfxtrx import DOMAIN
|
||||
from homeassistant.helpers.device_registry import DeviceRegistry
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import (
|
||||
MockConfigEntry,
|
||||
assert_lists_same,
|
||||
async_get_device_automations,
|
||||
async_mock_service,
|
||||
mock_device_registry,
|
||||
)
|
||||
from tests.components.rfxtrx.conftest import create_rfx_test_cfg
|
||||
|
||||
|
||||
class EventTestData(NamedTuple):
|
||||
"""Test data linked to a device."""
|
||||
|
||||
code: str
|
||||
device_identifiers: set[tuple[str, str, str, str]]
|
||||
type: str
|
||||
subtype: str
|
||||
|
||||
|
||||
DEVICE_LIGHTING_1 = {("rfxtrx", "10", "0", "E5")}
|
||||
EVENT_LIGHTING_1 = EventTestData("0710002a45050170", DEVICE_LIGHTING_1, "command", "On")
|
||||
|
||||
DEVICE_ROLLERTROL_1 = {("rfxtrx", "19", "0", "009ba8:1")}
|
||||
EVENT_ROLLERTROL_1 = EventTestData(
|
||||
"09190000009ba8010100", DEVICE_ROLLERTROL_1, "command", "Down"
|
||||
)
|
||||
|
||||
DEVICE_FIREALARM_1 = {("rfxtrx", "20", "3", "a10900:32")}
|
||||
EVENT_FIREALARM_1 = EventTestData(
|
||||
"08200300a109000670", DEVICE_FIREALARM_1, "status", "Panic"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(name="device_reg")
|
||||
def device_reg_fixture(hass):
|
||||
"""Return an empty, loaded, registry."""
|
||||
return mock_device_registry(hass)
|
||||
|
||||
|
||||
async def setup_entry(hass, devices):
|
||||
"""Construct a config setup."""
|
||||
entry_data = create_rfx_test_cfg(devices=devices)
|
||||
mock_entry = MockConfigEntry(domain="rfxtrx", unique_id=DOMAIN, data=entry_data)
|
||||
|
||||
mock_entry.add_to_hass(hass)
|
||||
|
||||
await hass.config_entries.async_setup(mock_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"event,expected",
|
||||
[
|
||||
[
|
||||
EVENT_LIGHTING_1,
|
||||
[
|
||||
{"type": "command", "subtype": subtype}
|
||||
for subtype in [
|
||||
"Off",
|
||||
"On",
|
||||
"Dim",
|
||||
"Bright",
|
||||
"All/group Off",
|
||||
"All/group On",
|
||||
"Chime",
|
||||
"Illegal command",
|
||||
]
|
||||
],
|
||||
]
|
||||
],
|
||||
)
|
||||
async def test_get_triggers(hass, device_reg, event: EventTestData, expected):
|
||||
"""Test we get the expected triggers from a rfxtrx."""
|
||||
await setup_entry(hass, {event.code: {"signal_repetitions": 1}})
|
||||
|
||||
device_entry = device_reg.async_get_device(event.device_identifiers, set())
|
||||
|
||||
expected_triggers = [
|
||||
{"domain": DOMAIN, "device_id": device_entry.id, "platform": "device", **expect}
|
||||
for expect in expected
|
||||
]
|
||||
|
||||
triggers = await async_get_device_automations(hass, "trigger", device_entry.id)
|
||||
triggers = [value for value in triggers if value["domain"] == "rfxtrx"]
|
||||
assert_lists_same(triggers, expected_triggers)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"event",
|
||||
[
|
||||
EVENT_LIGHTING_1,
|
||||
EVENT_ROLLERTROL_1,
|
||||
EVENT_FIREALARM_1,
|
||||
],
|
||||
)
|
||||
async def test_firing_event(hass, device_reg: DeviceRegistry, rfxtrx, event):
|
||||
"""Test for turn_on and turn_off triggers firing."""
|
||||
|
||||
await setup_entry(hass, {event.code: {"fire_event": True, "signal_repetitions": 1}})
|
||||
|
||||
device_entry = device_reg.async_get_device(event.device_identifiers, set())
|
||||
assert device_entry
|
||||
|
||||
calls = async_mock_service(hass, "test", "automation")
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "device",
|
||||
"domain": DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"type": event.type,
|
||||
"subtype": event.subtype,
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {"some": ("{{trigger.platform}}")},
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await rfxtrx.signal(event.code)
|
||||
|
||||
assert len(calls) == 1
|
||||
assert calls[0].data["some"] == "device"
|
||||
|
||||
|
||||
async def test_invalid_trigger(hass, device_reg: DeviceRegistry):
|
||||
"""Test for invalid actions."""
|
||||
event = EVENT_LIGHTING_1
|
||||
notification_calls = async_mock_service(hass, "persistent_notification", "create")
|
||||
|
||||
await setup_entry(hass, {event.code: {"fire_event": True, "signal_repetitions": 1}})
|
||||
|
||||
device_identifers: Any = event.device_identifiers
|
||||
device_entry = device_reg.async_get_device(device_identifers, set())
|
||||
assert device_entry
|
||||
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
automation.DOMAIN,
|
||||
{
|
||||
automation.DOMAIN: [
|
||||
{
|
||||
"trigger": {
|
||||
"platform": "device",
|
||||
"domain": DOMAIN,
|
||||
"device_id": device_entry.id,
|
||||
"type": event.type,
|
||||
"subtype": "invalid",
|
||||
},
|
||||
"action": {
|
||||
"service": "test.automation",
|
||||
"data_template": {"some": ("{{trigger.platform}}")},
|
||||
},
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(notification_calls) == 1
|
||||
assert (
|
||||
"The following integrations and platforms could not be set up"
|
||||
in notification_calls[0].data["message"]
|
||||
)
|
||||
Loading…
Add table
Reference in a new issue