Use entity registry id in alarm_control_panel device actions (#95241)

pull/85886/head^2
Erik Montnemery 2023-06-26 16:59:43 +02:00 committed by GitHub
parent 8e2ba81995
commit 89c9e72768
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 205 additions and 50 deletions

View File

@ -5,6 +5,7 @@ from typing import Final
import voluptuous as vol
from homeassistant.components.device_automation import async_validate_entity_schema
from homeassistant.const import (
ATTR_CODE,
ATTR_ENTITY_ID,
@ -44,15 +45,22 @@ ACTION_TYPES: Final[set[str]] = {
"trigger",
}
ACTION_SCHEMA: Final = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
_ACTION_SCHEMA: Final = cv.DEVICE_ACTION_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): vol.In(ACTION_TYPES),
vol.Required(CONF_ENTITY_ID): cv.entity_domain(DOMAIN),
vol.Required(CONF_ENTITY_ID): cv.entity_id_or_uuid,
vol.Optional(CONF_CODE): cv.string,
}
)
async def async_validate_action_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
return async_validate_entity_schema(hass, config, _ACTION_SCHEMA)
async def async_get_actions(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
@ -70,7 +78,7 @@ async def async_get_actions(
base_action: dict = {
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
CONF_ENTITY_ID: entry.entity_id,
CONF_ENTITY_ID: entry.id,
}
# Add actions for each entity that belongs to this integration
@ -124,7 +132,9 @@ async def async_get_action_capabilities(
"""List action capabilities."""
# We need to refer to the state directly because ATTR_CODE_ARM_REQUIRED is not a
# capability attribute
state = hass.states.get(config[CONF_ENTITY_ID])
registry = er.async_get(hass)
entity_id = er.async_resolve_entity_id(registry, config[CONF_ENTITY_ID])
state = hass.states.get(entity_id) if entity_id else None
code_required = state.attributes.get(ATTR_CODE_ARM_REQUIRED) if state else False
if config[CONF_TYPE] == "trigger" or (

View File

@ -5,7 +5,7 @@ from typing import cast
import voluptuous as vol
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN
from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.typing import ConfigType
@ -25,7 +25,14 @@ STATIC_VALIDATOR = {
DeviceAutomationType.TRIGGER: "TRIGGER_SCHEMA",
}
TOGGLE_ENTITY_DOMAINS = {"fan", "humidifier", "light", "remote", "switch"}
ENTITY_PLATFORMS = {
Platform.ALARM_CONTROL_PANEL.value,
Platform.FAN.value,
Platform.HUMIDIFIER.value,
Platform.LIGHT.value,
Platform.REMOTE.value,
Platform.SWITCH.value,
}
async def async_validate_device_automation_config(
@ -45,10 +52,10 @@ async def async_validate_device_automation_config(
ConfigType, getattr(platform, STATIC_VALIDATOR[automation_type])(config)
)
# Bypass checks for toggle entity domains
# Bypass checks for entity platforms
if (
automation_type == DeviceAutomationType.ACTION
and validated_config[CONF_DOMAIN] in TOGGLE_ENTITY_DOMAINS
and validated_config[CONF_DOMAIN] in ENTITY_PLATFORMS
):
return cast(
ConfigType,

View File

@ -102,7 +102,7 @@ async def test_get_actions(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_registry.async_get_or_create(
entity_entry = entity_registry.async_get_or_create(
DOMAIN,
"test",
"5678",
@ -113,13 +113,12 @@ async def test_get_actions(
hass.states.async_set(
f"{DOMAIN}.test_5678", "attributes", {"supported_features": features_state}
)
expected_actions = []
expected_actions += [
expected_actions = [
{
"domain": DOMAIN,
"type": action,
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": entity_entry.id,
"metadata": {"secondary": False},
}
for action in expected_action_types
@ -153,7 +152,7 @@ async def test_get_actions_hidden_auxiliary(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_registry.async_get_or_create(
entity_entry = entity_registry.async_get_or_create(
DOMAIN,
"test",
"5678",
@ -168,7 +167,7 @@ async def test_get_actions_hidden_auxiliary(
"domain": DOMAIN,
"type": action,
"device_id": device_entry.id,
"entity_id": f"{DOMAIN}.test_5678",
"entity_id": entity_entry.id,
"metadata": {"secondary": True},
}
for action in ["disarm", "arm_away"]
@ -191,7 +190,7 @@ async def test_get_actions_arm_night_only(
config_entry_id=config_entry.entry_id,
connections={(dr.CONNECTION_NETWORK_MAC, "12:34:56:AB:CD:EF")},
)
entity_registry.async_get_or_create(
entity_entry = entity_registry.async_get_or_create(
DOMAIN, "test", "5678", device_id=device_entry.id
)
hass.states.async_set(
@ -202,14 +201,14 @@ async def test_get_actions_arm_night_only(
"domain": DOMAIN,
"type": "arm_night",
"device_id": device_entry.id,
"entity_id": "alarm_control_panel.test_5678",
"entity_id": entity_entry.id,
"metadata": {"secondary": False},
},
{
"domain": DOMAIN,
"type": "disarm",
"device_id": device_entry.id,
"entity_id": "alarm_control_panel.test_5678",
"entity_id": entity_entry.id,
"metadata": {"secondary": False},
},
]
@ -266,6 +265,54 @@ async def test_get_action_capabilities(
assert capabilities == expected_capabilities[action["type"]]
async def test_get_action_capabilities_legacy(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
enable_custom_integrations: None,
) -> None:
"""Test we get the expected capabilities from a sensor trigger."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
await hass.async_block_till_done()
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(
DOMAIN,
"test",
platform.ENTITIES["no_arm_code"].unique_id,
device_id=device_entry.id,
)
expected_capabilities = {
"arm_away": {"extra_fields": []},
"arm_home": {"extra_fields": []},
"arm_night": {"extra_fields": []},
"arm_vacation": {"extra_fields": []},
"disarm": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"trigger": {"extra_fields": []},
}
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert len(actions) == 6
assert {action["type"] for action in actions} == set(expected_capabilities)
for action in actions:
action["entity_id"] = entity_registry.async_get(action["entity_id"]).entity_id
capabilities = await async_get_device_automation_capabilities(
hass, DeviceAutomationType.ACTION, action
)
assert capabilities == expected_capabilities[action["type"]]
async def test_get_action_capabilities_arm_code(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
@ -321,11 +368,77 @@ async def test_get_action_capabilities_arm_code(
assert capabilities == expected_capabilities[action["type"]]
async def test_action(hass: HomeAssistant, enable_custom_integrations: None) -> None:
async def test_get_action_capabilities_arm_code_legacy(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
enable_custom_integrations: None,
) -> None:
"""Test we get the expected capabilities from a sensor trigger."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
await hass.async_block_till_done()
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(
DOMAIN,
"test",
platform.ENTITIES["arm_code"].unique_id,
device_id=device_entry.id,
)
expected_capabilities = {
"arm_away": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"arm_home": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"arm_night": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"arm_vacation": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"disarm": {
"extra_fields": [{"name": "code", "optional": True, "type": "string"}]
},
"trigger": {"extra_fields": []},
}
actions = await async_get_device_automations(
hass, DeviceAutomationType.ACTION, device_entry.id
)
assert len(actions) == 6
assert {action["type"] for action in actions} == set(expected_capabilities)
for action in actions:
action["entity_id"] = entity_registry.async_get(action["entity_id"]).entity_id
capabilities = await async_get_device_automation_capabilities(
hass, DeviceAutomationType.ACTION, action
)
assert capabilities == expected_capabilities[action["type"]]
async def test_action(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
enable_custom_integrations: None,
) -> None:
"""Test for turn_on and turn_off actions."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
entity_entry = entity_registry.async_get_or_create(
DOMAIN,
"test",
platform.ENTITIES["no_arm_code"].unique_id,
)
assert await async_setup_component(
hass,
automation.DOMAIN,
@ -339,7 +452,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "arm_away",
},
},
@ -351,7 +464,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "arm_home",
},
},
@ -363,7 +476,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "arm_night",
},
},
@ -375,7 +488,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "arm_vacation",
},
},
@ -384,7 +497,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "disarm",
"code": "1234",
},
@ -397,7 +510,7 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": "alarm_control_panel.alarm_no_arm_code",
"entity_id": entity_entry.id,
"type": "trigger",
},
},
@ -407,48 +520,73 @@ async def test_action(hass: HomeAssistant, enable_custom_integrations: None) ->
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state == STATE_UNKNOWN
)
assert hass.states.get(entity_entry.entity_id).state == STATE_UNKNOWN
hass.bus.async_fire("test_event_arm_away")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_ARMED_AWAY
)
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_ARMED_AWAY
hass.bus.async_fire("test_event_arm_home")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_ARMED_HOME
)
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_ARMED_HOME
hass.bus.async_fire("test_event_arm_vacation")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_ARMED_VACATION
)
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_ARMED_VACATION
hass.bus.async_fire("test_event_arm_night")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_ARMED_NIGHT
)
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_ARMED_NIGHT
hass.bus.async_fire("test_event_disarm")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_DISARMED
)
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_DISARMED
hass.bus.async_fire("test_event_trigger")
await hass.async_block_till_done()
assert (
hass.states.get("alarm_control_panel.alarm_no_arm_code").state
== STATE_ALARM_TRIGGERED
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_TRIGGERED
async def test_action_legacy(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
enable_custom_integrations: None,
) -> None:
"""Test for turn_on and turn_off actions."""
platform = getattr(hass.components, f"test.{DOMAIN}")
platform.init()
entity_entry = entity_registry.async_get_or_create(
DOMAIN,
"test",
platform.ENTITIES["no_arm_code"].unique_id,
)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": "event",
"event_type": "test_event_arm_away",
},
"action": {
"domain": DOMAIN,
"device_id": "abcdefgh",
"entity_id": entity_entry.entity_id,
"type": "arm_away",
},
},
]
},
)
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {CONF_PLATFORM: "test"}})
await hass.async_block_till_done()
assert hass.states.get(entity_entry.entity_id).state == STATE_UNKNOWN
hass.bus.async_fire("test_event_arm_away")
await hass.async_block_till_done()
assert hass.states.get(entity_entry.entity_id).state == STATE_ALARM_ARMED_AWAY