From eaee8d5b7857644d806a7e9a957ed47a790bc1bd Mon Sep 17 00:00:00 2001 From: Shay Levy Date: Wed, 4 Sep 2024 18:34:11 +0300 Subject: [PATCH] Fix BTHome validate triggers for device with multiple buttons (#125183) * Fix BTHome validate triggers for device with multiple buttons * Remove None default --- .../components/bthome/device_trigger.py | 56 +++++--- .../components/bthome/test_device_trigger.py | 124 +++++++++++++++++- 2 files changed, 158 insertions(+), 22 deletions(-) diff --git a/homeassistant/components/bthome/device_trigger.py b/homeassistant/components/bthome/device_trigger.py index c49664b1146..c50ffc05900 100644 --- a/homeassistant/components/bthome/device_trigger.py +++ b/homeassistant/components/bthome/device_trigger.py @@ -7,6 +7,9 @@ from typing import Any import voluptuous as vol 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.const import ( CONF_DEVICE_ID, @@ -43,33 +46,46 @@ TRIGGERS_BY_EVENT_CLASS = { EVENT_CLASS_DIMMER: {"rotate_left", "rotate_right"}, } -SCHEMA_BY_EVENT_CLASS = { - EVENT_CLASS_BUTTON: DEVICE_TRIGGER_BASE_SCHEMA.extend( - { - vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_BUTTON]), - vol.Required(CONF_SUBTYPE): vol.In( - TRIGGERS_BY_EVENT_CLASS[EVENT_CLASS_BUTTON] - ), - } - ), - EVENT_CLASS_DIMMER: DEVICE_TRIGGER_BASE_SCHEMA.extend( - { - vol.Required(CONF_TYPE): vol.In([EVENT_CLASS_DIMMER]), - vol.Required(CONF_SUBTYPE): vol.In( - TRIGGERS_BY_EVENT_CLASS[EVENT_CLASS_DIMMER] - ), - } - ), -} +TRIGGER_SCHEMA = DEVICE_TRIGGER_BASE_SCHEMA.extend( + {vol.Required(CONF_TYPE): str, vol.Required(CONF_SUBTYPE): str} +) async def async_validate_trigger_config( hass: HomeAssistant, config: ConfigType ) -> ConfigType: """Validate trigger config.""" - return SCHEMA_BY_EVENT_CLASS.get(config[CONF_TYPE], DEVICE_TRIGGER_BASE_SCHEMA)( # type: ignore[no-any-return] - config + config = TRIGGER_SCHEMA(config) + event_class = config[CONF_TYPE] + event_type = config[CONF_SUBTYPE] + + device_registry = dr.async_get(hass) + device = device_registry.async_get(config[CONF_DEVICE_ID]) + assert device is not None + config_entries = [ + hass.config_entries.async_get_entry(entry_id) + for entry_id in device.config_entries + ] + bthome_config_entry = next( + iter(entry for entry in config_entries if entry and entry.domain == DOMAIN) ) + event_classes: list[str] = bthome_config_entry.data.get( + CONF_DISCOVERED_EVENT_CLASSES, [] + ) + + if event_class not in event_classes: + raise InvalidDeviceAutomationConfig( + f"BTHome trigger {event_class} is not valid for device " + f"{device} ({config[CONF_DEVICE_ID]})" + ) + + if event_type not in TRIGGERS_BY_EVENT_CLASS.get(event_class.split("_")[0], ()): + raise InvalidDeviceAutomationConfig( + f"BTHome trigger {event_type} is not valid for device " + f"{device} ({config[CONF_DEVICE_ID]})" + ) + + return config async def async_get_triggers( diff --git a/tests/components/bthome/test_device_trigger.py b/tests/components/bthome/test_device_trigger.py index 459654826f9..c4c900ef6e1 100644 --- a/tests/components/bthome/test_device_trigger.py +++ b/tests/components/bthome/test_device_trigger.py @@ -1,10 +1,19 @@ """Test BTHome BLE events.""" +import pytest + from homeassistant.components import automation from homeassistant.components.bluetooth import DOMAIN as BLUETOOTH_DOMAIN from homeassistant.components.bthome.const import CONF_SUBTYPE, DOMAIN from homeassistant.components.device_automation import DeviceAutomationType -from homeassistant.const import CONF_DEVICE_ID, CONF_DOMAIN, CONF_PLATFORM, CONF_TYPE +from homeassistant.const import ( + CONF_DEVICE_ID, + CONF_DOMAIN, + CONF_PLATFORM, + CONF_TYPE, + STATE_ON, + STATE_UNAVAILABLE, +) from homeassistant.core import HomeAssistant, ServiceCall, callback from homeassistant.helpers import device_registry as dr from homeassistant.setup import async_setup_component @@ -121,6 +130,117 @@ async def test_get_triggers_button( await hass.async_block_till_done() +async def test_get_triggers_multiple_buttons( + hass: HomeAssistant, device_registry: dr.DeviceRegistry +) -> None: + """Test that we get the expected triggers for multiple buttons device.""" + mac = "A4:C1:38:8D:18:B2" + entry = await _async_setup_bthome_device(hass, mac) + events = async_capture_events(hass, "bthome_ble_event") + + # Emit button_1 long press and button_2 press events + # so it creates the device in the registry + inject_bluetooth_service_info_bleak( + hass, + make_bthome_v2_adv(mac, b"\x40\x3a\x04\x3a\x01"), + ) + + # wait for the event + await hass.async_block_till_done() + assert len(events) == 2 + + device = device_registry.async_get_device(identifiers={get_device_id(mac)}) + assert device + expected_trigger1 = { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "button_1", + CONF_SUBTYPE: "long_press", + "metadata": {}, + } + expected_trigger2 = { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: "button_2", + CONF_SUBTYPE: "press", + "metadata": {}, + } + triggers = await async_get_device_automations( + hass, DeviceAutomationType.TRIGGER, device.id + ) + assert expected_trigger1 in triggers + assert expected_trigger2 in triggers + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + +@pytest.mark.parametrize( + ("event_class", "event_type", "expected"), + [ + ("button_1", "long_press", STATE_ON), + ("button_2", "press", STATE_ON), + ("button_3", "long_press", STATE_UNAVAILABLE), + ("button", "long_press", STATE_UNAVAILABLE), + ("button_1", "invalid_press", STATE_UNAVAILABLE), + ], +) +async def test_validate_trigger_config( + hass: HomeAssistant, + device_registry: dr.DeviceRegistry, + event_class: str, + event_type: str, + expected: str, +) -> None: + """Test unsupported trigger does not return a trigger config.""" + mac = "A4:C1:38:8D:18:B2" + entry = await _async_setup_bthome_device(hass, mac) + + # Emit button_1 long press and button_2 press events + # so it creates the device in the registry + inject_bluetooth_service_info_bleak( + hass, + make_bthome_v2_adv(mac, b"\x40\x3a\x04\x3a\x01"), + ) + + # wait for the event + await hass.async_block_till_done() + + device = device_registry.async_get_device(identifiers={get_device_id(mac)}) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: device.id, + CONF_TYPE: event_class, + CONF_SUBTYPE: event_type, + }, + "action": { + "service": "test.automation", + "data_template": {"some": "test_trigger_button_long_press"}, + }, + }, + ] + }, + ) + await hass.async_block_till_done() + + automations = hass.states.async_entity_ids(automation.DOMAIN) + assert len(automations) == 1 + assert hass.states.get(automations[0]).state == expected + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + async def test_get_triggers_dimmer( hass: HomeAssistant, device_registry: dr.DeviceRegistry ) -> None: @@ -235,7 +355,7 @@ async def test_if_fires_on_motion_detected( make_bthome_v2_adv(mac, b"\x40\x3a\x03"), ) - # # wait for the event + # wait for the event await hass.async_block_till_done() device = device_registry.async_get_device(identifiers={get_device_id(mac)})