diff --git a/homeassistant/components/shelly/__init__.py b/homeassistant/components/shelly/__init__.py index 34a09338c81..ad0ad5f4387 100644 --- a/homeassistant/components/shelly/__init__.py +++ b/homeassistant/components/shelly/__init__.py @@ -31,6 +31,7 @@ from .const import ( ATTR_CHANNEL, ATTR_CLICK_TYPE, ATTR_DEVICE, + ATTR_GENERATION, BATTERY_DEVICES_WITH_PERMANENT_CONNECTION, BLOCK, CONF_COAP_PORT, @@ -44,6 +45,7 @@ from .const import ( REST, REST_SENSORS_UPDATE_INTERVAL, RPC, + RPC_INPUTS_EVENTS_TYPES, RPC_RECONNECT_INTERVAL, SHBTN_MODELS, SLEEP_PERIOD_MULTIPLIER, @@ -250,8 +252,8 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator): self.entry = entry self.device = device - self._async_remove_device_updates_handler = self.async_add_listener( - self._async_device_updates_handler + entry.async_on_unload( + self.async_add_listener(self._async_device_updates_handler) ) self._last_input_events_count: dict = {} @@ -306,6 +308,7 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator): ATTR_DEVICE: self.device.settings["device"]["hostname"], ATTR_CHANNEL: channel, ATTR_CLICK_TYPE: INPUTS_EVENTS_DICT[event_type], + ATTR_GENERATION: 1, }, ) else: @@ -356,7 +359,6 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator): def shutdown(self) -> None: """Shutdown the wrapper.""" self.device.shutdown() - self._async_remove_device_updates_handler() @callback def _handle_ha_stop(self, _event: Event) -> None: @@ -435,27 +437,40 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return unload_ok -def get_device_wrapper( +def get_block_device_wrapper( hass: HomeAssistant, device_id: str -) -> BlockDeviceWrapper | RpcDeviceWrapper | None: - """Get a Shelly device wrapper for the given device id.""" +) -> BlockDeviceWrapper | None: + """Get a Shelly block device wrapper for the given device id.""" if not hass.data.get(DOMAIN): return None - for config_entry in hass.data[DOMAIN][DATA_CONFIG_ENTRY]: - block_wrapper: BlockDeviceWrapper | None = hass.data[DOMAIN][DATA_CONFIG_ENTRY][ - config_entry - ].get(BLOCK) + dev_reg = device_registry.async_get(hass) + if device := dev_reg.async_get(device_id): + for config_entry in device.config_entries: + if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry): + continue - if block_wrapper and block_wrapper.device_id == device_id: - return block_wrapper + if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(BLOCK): + return cast(BlockDeviceWrapper, wrapper) - rpc_wrapper: RpcDeviceWrapper | None = hass.data[DOMAIN][DATA_CONFIG_ENTRY][ - config_entry - ].get(RPC) + return None - if rpc_wrapper and rpc_wrapper.device_id == device_id: - return rpc_wrapper + +def get_rpc_device_wrapper( + hass: HomeAssistant, device_id: str +) -> RpcDeviceWrapper | None: + """Get a Shelly RPC device wrapper for the given device id.""" + if not hass.data.get(DOMAIN): + return None + + dev_reg = device_registry.async_get(hass) + if device := dev_reg.async_get(device_id): + for config_entry in device.config_entries: + if not hass.data[DOMAIN][DATA_CONFIG_ENTRY].get(config_entry): + continue + + if wrapper := hass.data[DOMAIN][DATA_CONFIG_ENTRY][config_entry].get(RPC): + return cast(RpcDeviceWrapper, wrapper) return None @@ -479,10 +494,42 @@ class RpcDeviceWrapper(update_coordinator.DataUpdateCoordinator): self.entry = entry self.device = device + entry.async_on_unload( + self.async_add_listener(self._async_device_updates_handler) + ) + self._last_event: dict[str, Any] | None = None + entry.async_on_unload( hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._handle_ha_stop) ) + @callback + def _async_device_updates_handler(self) -> None: + """Handle device updates.""" + if ( + not self.device.initialized + or not self.device.event + or self.device.event == self._last_event + ): + return + + self._last_event = self.device.event + + for event in self.device.event["events"]: + if event.get("event") not in RPC_INPUTS_EVENTS_TYPES: + continue + + self.hass.bus.async_fire( + EVENT_SHELLY_CLICK, + { + ATTR_DEVICE_ID: self.device_id, + ATTR_DEVICE: self.device.hostname, + ATTR_CHANNEL: event["id"] + 1, + ATTR_CLICK_TYPE: event["event"], + ATTR_GENERATION: 2, + }, + ) + async def _async_update_data(self) -> None: """Fetch data.""" if self.device.connected: diff --git a/homeassistant/components/shelly/const.py b/homeassistant/components/shelly/const.py index 917c10ff57c..3c9c24b1f7f 100644 --- a/homeassistant/components/shelly/const.py +++ b/homeassistant/components/shelly/const.py @@ -64,18 +64,28 @@ INPUTS_EVENTS_DICT: Final = { # List of battery devices that maintain a permanent WiFi connection BATTERY_DEVICES_WITH_PERMANENT_CONNECTION: Final = ["SHMOS-01"] +# Button/Click events for Block & RPC devices EVENT_SHELLY_CLICK: Final = "shelly.click" ATTR_CLICK_TYPE: Final = "click_type" ATTR_CHANNEL: Final = "channel" ATTR_DEVICE: Final = "device" +ATTR_GENERATION: Final = "generation" CONF_SUBTYPE: Final = "subtype" BASIC_INPUTS_EVENTS_TYPES: Final = {"single", "long"} SHBTN_INPUTS_EVENTS_TYPES: Final = {"single", "double", "triple", "long"} -SUPPORTED_INPUTS_EVENTS_TYPES: Final = { +RPC_INPUTS_EVENTS_TYPES: Final = { + "btn_down", + "btn_up", + "single_push", + "double_push", + "long_push", +} + +BLOCK_INPUTS_EVENTS_TYPES: Final = { "single", "double", "triple", @@ -84,9 +94,15 @@ SUPPORTED_INPUTS_EVENTS_TYPES: Final = { "long_single", } -SHIX3_1_INPUTS_EVENTS_TYPES = SUPPORTED_INPUTS_EVENTS_TYPES +SHIX3_1_INPUTS_EVENTS_TYPES = BLOCK_INPUTS_EVENTS_TYPES -INPUTS_EVENTS_SUBTYPES: Final = {"button": 1, "button1": 1, "button2": 2, "button3": 3} +INPUTS_EVENTS_SUBTYPES: Final = { + "button": 1, + "button1": 1, + "button2": 2, + "button3": 3, + "button4": 4, +} SHBTN_MODELS: Final = ["SHBTN-1", "SHBTN-2"] diff --git a/homeassistant/components/shelly/device_trigger.py b/homeassistant/components/shelly/device_trigger.py index 552d1d62032..f5abf76e8f2 100644 --- a/homeassistant/components/shelly/device_trigger.py +++ b/homeassistant/components/shelly/device_trigger.py @@ -25,28 +25,52 @@ from homeassistant.const import ( from homeassistant.core import CALLBACK_TYPE, HomeAssistant from homeassistant.helpers.typing import ConfigType -from . import RpcDeviceWrapper, get_device_wrapper +from . import get_block_device_wrapper, get_rpc_device_wrapper from .const import ( ATTR_CHANNEL, ATTR_CLICK_TYPE, + BLOCK_INPUTS_EVENTS_TYPES, CONF_SUBTYPE, DOMAIN, EVENT_SHELLY_CLICK, INPUTS_EVENTS_SUBTYPES, - SHBTN_INPUTS_EVENTS_TYPES, + RPC_INPUTS_EVENTS_TYPES, SHBTN_MODELS, - SUPPORTED_INPUTS_EVENTS_TYPES, ) -from .utils import get_input_triggers +from .utils import ( + get_block_input_triggers, + get_rpc_input_triggers, + get_shbtn_input_triggers, +) TRIGGER_SCHEMA: Final = DEVICE_TRIGGER_BASE_SCHEMA.extend( { - vol.Required(CONF_TYPE): vol.In(SUPPORTED_INPUTS_EVENTS_TYPES), + vol.Required(CONF_TYPE): vol.In( + RPC_INPUTS_EVENTS_TYPES | BLOCK_INPUTS_EVENTS_TYPES + ), vol.Required(CONF_SUBTYPE): vol.In(INPUTS_EVENTS_SUBTYPES), } ) +def append_input_triggers( + triggers: list[dict[str, Any]], + input_triggers: list[tuple[str, str]], + device_id: str, +) -> None: + """Add trigger to triggers list.""" + for trigger, subtype in input_triggers: + triggers.append( + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: trigger, + CONF_SUBTYPE: subtype, + } + ) + + async def async_validate_trigger_config( hass: HomeAssistant, config: dict[str, Any] ) -> dict[str, Any]: @@ -54,23 +78,29 @@ async def async_validate_trigger_config( config = TRIGGER_SCHEMA(config) # if device is available verify parameters against device capabilities - wrapper = get_device_wrapper(hass, config[CONF_DEVICE_ID]) - - if isinstance(wrapper, RpcDeviceWrapper): - return config - - if not wrapper or not wrapper.device.initialized: - return config - trigger = (config[CONF_TYPE], config[CONF_SUBTYPE]) - assert wrapper.device.blocks + if config[CONF_TYPE] in RPC_INPUTS_EVENTS_TYPES: + rpc_wrapper = get_rpc_device_wrapper(hass, config[CONF_DEVICE_ID]) + if not rpc_wrapper or not rpc_wrapper.device.initialized: + return config - for block in wrapper.device.blocks: - input_triggers = get_input_triggers(wrapper.device, block) + input_triggers = get_rpc_input_triggers(rpc_wrapper.device) if trigger in input_triggers: return config + elif config[CONF_TYPE] in BLOCK_INPUTS_EVENTS_TYPES: + block_wrapper = get_block_device_wrapper(hass, config[CONF_DEVICE_ID]) + if not block_wrapper or not block_wrapper.device.initialized: + return config + + assert block_wrapper.device.blocks + + for block in block_wrapper.device.blocks: + input_triggers = get_block_input_triggers(block_wrapper.device, block) + if trigger in input_triggers: + return config + raise InvalidDeviceAutomationConfig( f"Invalid ({CONF_TYPE},{CONF_SUBTYPE}): {trigger}" ) @@ -80,45 +110,28 @@ async def async_get_triggers( hass: HomeAssistant, device_id: str ) -> list[dict[str, Any]]: """List device triggers for Shelly devices.""" - wrapper = get_device_wrapper(hass, device_id) - if not wrapper: - raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}") + triggers: list[dict[str, Any]] = [] - if isinstance(wrapper, RpcDeviceWrapper): - return [] - - triggers = [] - - if wrapper.model in SHBTN_MODELS: - for trigger in SHBTN_INPUTS_EVENTS_TYPES: - triggers.append( - { - CONF_PLATFORM: "device", - CONF_DEVICE_ID: device_id, - CONF_DOMAIN: DOMAIN, - CONF_TYPE: trigger, - CONF_SUBTYPE: "button", - } - ) + if rpc_wrapper := get_rpc_device_wrapper(hass, device_id): + input_triggers = get_rpc_input_triggers(rpc_wrapper.device) + append_input_triggers(triggers, input_triggers, device_id) return triggers - assert wrapper.device.blocks + if block_wrapper := get_block_device_wrapper(hass, device_id): + if block_wrapper.model in SHBTN_MODELS: + input_triggers = get_shbtn_input_triggers() + append_input_triggers(triggers, input_triggers, device_id) + return triggers - for block in wrapper.device.blocks: - input_triggers = get_input_triggers(wrapper.device, block) + assert block_wrapper.device.blocks - for trigger, subtype in input_triggers: - triggers.append( - { - CONF_PLATFORM: "device", - CONF_DEVICE_ID: device_id, - CONF_DOMAIN: DOMAIN, - CONF_TYPE: trigger, - CONF_SUBTYPE: subtype, - } - ) + for block in block_wrapper.device.blocks: + input_triggers = get_block_input_triggers(block_wrapper.device, block) + append_input_triggers(triggers, input_triggers, device_id) - return triggers + return triggers + + raise InvalidDeviceAutomationConfig(f"Device not found: {device_id}") async def async_attach_trigger( @@ -137,6 +150,7 @@ async def async_attach_trigger( ATTR_CLICK_TYPE: config[CONF_TYPE], }, } + event_config = event_trigger.TRIGGER_SCHEMA(event_config) return await event_trigger.async_attach_trigger( hass, event_config, action, automation_info, platform_type="device" diff --git a/homeassistant/components/shelly/logbook.py b/homeassistant/components/shelly/logbook.py index d58691439cf..a1c8d5eceee 100644 --- a/homeassistant/components/shelly/logbook.py +++ b/homeassistant/components/shelly/logbook.py @@ -7,15 +7,17 @@ from homeassistant.const import ATTR_DEVICE_ID from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.typing import EventType -from . import RpcDeviceWrapper, get_device_wrapper +from . import get_block_device_wrapper, get_rpc_device_wrapper from .const import ( ATTR_CHANNEL, ATTR_CLICK_TYPE, ATTR_DEVICE, + BLOCK_INPUTS_EVENTS_TYPES, DOMAIN, EVENT_SHELLY_CLICK, + RPC_INPUTS_EVENTS_TYPES, ) -from .utils import get_block_device_name +from .utils import get_block_device_name, get_rpc_entity_name @callback @@ -27,23 +29,27 @@ def async_describe_events( @callback def async_describe_shelly_click_event(event: EventType) -> dict[str, str]: - """Describe shelly.click logbook event.""" - wrapper = get_device_wrapper(hass, event.data[ATTR_DEVICE_ID]) - - if isinstance(wrapper, RpcDeviceWrapper): - return {} - - if wrapper and wrapper.device.initialized: - device_name = get_block_device_name(wrapper.device) - else: - device_name = event.data[ATTR_DEVICE] - - channel = event.data[ATTR_CHANNEL] + """Describe shelly.click logbook event (block device).""" + device_id = event.data[ATTR_DEVICE_ID] click_type = event.data[ATTR_CLICK_TYPE] + channel = event.data[ATTR_CHANNEL] + input_name = f"{event.data[ATTR_DEVICE]} channel {channel}" + + if click_type in RPC_INPUTS_EVENTS_TYPES: + rpc_wrapper = get_rpc_device_wrapper(hass, device_id) + if rpc_wrapper and rpc_wrapper.device.initialized: + key = f"input:{channel-1}" + input_name = get_rpc_entity_name(rpc_wrapper.device, key) + + elif click_type in BLOCK_INPUTS_EVENTS_TYPES: + block_wrapper = get_block_device_wrapper(hass, device_id) + if block_wrapper and block_wrapper.device.initialized: + device_name = get_block_device_name(block_wrapper.device) + input_name = f"{device_name} channel {channel}" return { "name": "Shelly", - "message": f"'{click_type}' click event for {device_name} channel {channel} was fired.", + "message": f"'{click_type}' click event for {input_name} Input was fired.", } async_describe_event(DOMAIN, EVENT_SHELLY_CLICK, async_describe_shelly_click_event) diff --git a/homeassistant/components/shelly/strings.json b/homeassistant/components/shelly/strings.json index 85a1fa87d0c..43cae79f94a 100644 --- a/homeassistant/components/shelly/strings.json +++ b/homeassistant/components/shelly/strings.json @@ -33,7 +33,8 @@ "button": "Button", "button1": "First button", "button2": "Second button", - "button3": "Third button" + "button3": "Third button", + "button4": "Fourth button" }, "trigger_type": { "single": "{subtype} single clicked", @@ -41,7 +42,12 @@ "triple": "{subtype} triple clicked", "long": " {subtype} long clicked", "single_long": "{subtype} single clicked and then long clicked", - "long_single": "{subtype} long clicked and then single clicked" + "long_single": "{subtype} long clicked and then single clicked", + "btn_down": "{subtype} button down", + "btn_up": "{subtype} button up", + "single_push": "{subtype} single push", + "double_push": "{subtype} double push", + "long_push": " {subtype} long push" } } } diff --git a/homeassistant/components/shelly/translations/en.json b/homeassistant/components/shelly/translations/en.json index b60d9dfbe3e..2ed09356363 100644 --- a/homeassistant/components/shelly/translations/en.json +++ b/homeassistant/components/shelly/translations/en.json @@ -33,7 +33,8 @@ "button": "Button", "button1": "First button", "button2": "Second button", - "button3": "Third button" + "button3": "Third button", + "button4": "Fourth button" }, "trigger_type": { "double": "{subtype} double clicked", @@ -41,7 +42,12 @@ "long_single": "{subtype} long clicked and then single clicked", "single": "{subtype} single clicked", "single_long": "{subtype} single clicked and then long clicked", - "triple": "{subtype} triple clicked" + "triple": "{subtype} triple clicked", + "btn_down": "{subtype} button down", + "btn_up": "{subtype} button up", + "single_push": "{subtype} single push", + "double_push": "{subtype} double push", + "long_push": " {subtype} long push" } } } \ No newline at end of file diff --git a/homeassistant/components/shelly/utils.py b/homeassistant/components/shelly/utils.py index 4d3655829a7..6f24b4a64be 100644 --- a/homeassistant/components/shelly/utils.py +++ b/homeassistant/components/shelly/utils.py @@ -22,6 +22,7 @@ from .const import ( DEFAULT_COAP_PORT, DOMAIN, MAX_RPC_KEY_INSTANCES, + RPC_INPUTS_EVENTS_TYPES, SHBTN_INPUTS_EVENTS_TYPES, SHBTN_MODELS, SHIX3_1_INPUTS_EVENTS_TYPES, @@ -162,7 +163,9 @@ def get_device_uptime(uptime: float, last_uptime: str | None) -> str: return last_uptime -def get_input_triggers(device: BlockDevice, block: Block) -> list[tuple[str, str]]: +def get_block_input_triggers( + device: BlockDevice, block: Block +) -> list[tuple[str, str]]: """Return list of input triggers for block.""" if "inputEvent" not in block.sensor_ids or "inputEventCnt" not in block.sensor_ids: return [] @@ -191,6 +194,16 @@ def get_input_triggers(device: BlockDevice, block: Block) -> list[tuple[str, str return triggers +def get_shbtn_input_triggers() -> list[tuple[str, str]]: + """Return list of input triggers for SHBTN models.""" + triggers = [] + + for trigger_type in SHBTN_INPUTS_EVENTS_TYPES: + triggers.append((trigger_type, "button")) + + return triggers + + @singleton.singleton("shelly_coap") async def get_coap_context(hass: HomeAssistant) -> COAP: """Get CoAP context to be used in all Shelly devices.""" @@ -314,3 +327,21 @@ def is_rpc_channel_type_light(config: dict[str, Any], channel: int) -> bool: """Return true if rpc channel consumption type is set to light.""" con_types = config["sys"]["ui_data"].get("consumption_types") return con_types is not None and con_types[channel].lower().startswith("light") + + +def get_rpc_input_triggers(device: RpcDevice) -> list[tuple[str, str]]: + """Return list of input triggers for RPC device.""" + triggers = [] + + key_ids = get_rpc_key_ids(device.config, "input") + + for id_ in key_ids: + key = f"input:{id_}" + if not is_rpc_momentary_input(device.config, key): + continue + + for trigger_type in RPC_INPUTS_EVENTS_TYPES: + subtype = f"button{id_+1}" + triggers.append((trigger_type, subtype)) + + return triggers diff --git a/tests/components/shelly/conftest.py b/tests/components/shelly/conftest.py index e38dd252b3a..9dbba7732ac 100644 --- a/tests/components/shelly/conftest.py +++ b/tests/components/shelly/conftest.py @@ -56,6 +56,7 @@ MOCK_BLOCKS = [ ] MOCK_CONFIG = { + "input:0": {"id": 0, "type": "button"}, "switch:0": {"name": "test switch_0"}, "sys": {"ui_data": {}}, "wifi": { @@ -147,6 +148,7 @@ async def rpc_wrapper(hass): device = Mock( call_rpc=AsyncMock(), config=MOCK_CONFIG, + event={}, shelly=MOCK_SHELLY, status=MOCK_STATUS, firmware_version="some fw string", diff --git a/tests/components/shelly/test_device_trigger.py b/tests/components/shelly/test_device_trigger.py index bf1529e4aaf..67e4660d167 100644 --- a/tests/components/shelly/test_device_trigger.py +++ b/tests/components/shelly/test_device_trigger.py @@ -30,8 +30,8 @@ from tests.common import ( ) -async def test_get_triggers(hass, coap_wrapper): - """Test we get the expected triggers from a shelly.""" +async def test_get_triggers_block_device(hass, coap_wrapper): + """Test we get the expected triggers from a shelly block device.""" assert coap_wrapper expected_triggers = [ { @@ -57,6 +57,54 @@ async def test_get_triggers(hass, coap_wrapper): assert_lists_same(triggers, expected_triggers) +async def test_get_triggers_rpc_device(hass, rpc_wrapper): + """Test we get the expected triggers from a shelly RPC device.""" + assert rpc_wrapper + expected_triggers = [ + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "btn_down", + CONF_SUBTYPE: "button1", + }, + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "btn_up", + CONF_SUBTYPE: "button1", + }, + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "single_push", + CONF_SUBTYPE: "button1", + }, + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "double_push", + CONF_SUBTYPE: "button1", + }, + { + CONF_PLATFORM: "device", + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_DOMAIN: DOMAIN, + CONF_TYPE: "long_push", + CONF_SUBTYPE: "button1", + }, + ] + + triggers = await async_get_device_automations( + hass, "trigger", rpc_wrapper.device_id + ) + + assert_lists_same(triggers, expected_triggers) + + async def test_get_triggers_button(hass): """Test we get the expected triggers from a shelly button.""" await async_setup_component(hass, "shelly", {}) @@ -136,8 +184,8 @@ async def test_get_triggers_for_invalid_device_id(hass, device_reg, coap_wrapper await async_get_device_automations(hass, "trigger", invalid_device.id) -async def test_if_fires_on_click_event(hass, calls, coap_wrapper): - """Test for click_event trigger firing.""" +async def test_if_fires_on_click_event_block_device(hass, calls, coap_wrapper): + """Test for click_event trigger firing for block device.""" assert coap_wrapper await setup.async_setup_component(hass, "persistent_notification", {}) @@ -175,8 +223,47 @@ async def test_if_fires_on_click_event(hass, calls, coap_wrapper): assert calls[0].data["some"] == "test_trigger_single_click" -async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper): - """Test for click_event with no device.""" +async def test_if_fires_on_click_event_rpc_device(hass, calls, rpc_wrapper): + """Test for click_event trigger firing for rpc device.""" + assert rpc_wrapper + await setup.async_setup_component(hass, "persistent_notification", {}) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: rpc_wrapper.device_id, + CONF_TYPE: "single_push", + CONF_SUBTYPE: "button1", + }, + "action": { + "service": "test.automation", + "data_template": {"some": "test_trigger_single_push"}, + }, + }, + ] + }, + ) + + message = { + CONF_DEVICE_ID: rpc_wrapper.device_id, + ATTR_CLICK_TYPE: "single_push", + ATTR_CHANNEL: 1, + } + hass.bus.async_fire(EVENT_SHELLY_CLICK, message) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data["some"] == "test_trigger_single_push" + + +async def test_validate_trigger_block_device_not_ready(hass, calls, coap_wrapper): + """Test validate trigger config when block device is not ready.""" assert coap_wrapper await setup.async_setup_component(hass, "persistent_notification", {}) @@ -189,7 +276,7 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper): "trigger": { CONF_PLATFORM: "device", CONF_DOMAIN: DOMAIN, - CONF_DEVICE_ID: "no_device", + CONF_DEVICE_ID: "device_not_ready", CONF_TYPE: "single", CONF_SUBTYPE: "button1", }, @@ -201,7 +288,11 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper): ] }, ) - message = {CONF_DEVICE_ID: "no_device", ATTR_CLICK_TYPE: "single", ATTR_CHANNEL: 1} + message = { + CONF_DEVICE_ID: "device_not_ready", + ATTR_CLICK_TYPE: "single", + ATTR_CHANNEL: 1, + } hass.bus.async_fire(EVENT_SHELLY_CLICK, message) await hass.async_block_till_done() @@ -209,6 +300,44 @@ async def test_validate_trigger_config_no_device(hass, calls, coap_wrapper): assert calls[0].data["some"] == "test_trigger_single_click" +async def test_validate_trigger_rpc_device_not_ready(hass, calls, rpc_wrapper): + """Test validate trigger config when RPC device is not ready.""" + assert rpc_wrapper + await setup.async_setup_component(hass, "persistent_notification", {}) + + assert await async_setup_component( + hass, + automation.DOMAIN, + { + automation.DOMAIN: [ + { + "trigger": { + CONF_PLATFORM: "device", + CONF_DOMAIN: DOMAIN, + CONF_DEVICE_ID: "device_not_ready", + CONF_TYPE: "single_push", + CONF_SUBTYPE: "button1", + }, + "action": { + "service": "test.automation", + "data_template": {"some": "test_trigger_single_push"}, + }, + }, + ] + }, + ) + message = { + CONF_DEVICE_ID: "device_not_ready", + ATTR_CLICK_TYPE: "single_push", + ATTR_CHANNEL: 1, + } + hass.bus.async_fire(EVENT_SHELLY_CLICK, message) + await hass.async_block_till_done() + + assert len(calls) == 1 + assert calls[0].data["some"] == "test_trigger_single_push" + + async def test_validate_trigger_invalid_triggers(hass, coap_wrapper): """Test for click_event with invalid triggers.""" assert coap_wrapper diff --git a/tests/components/shelly/test_logbook.py b/tests/components/shelly/test_logbook.py index 9cfda9ddcaa..9ece9590cbb 100644 --- a/tests/components/shelly/test_logbook.py +++ b/tests/components/shelly/test_logbook.py @@ -13,8 +13,8 @@ from homeassistant.setup import async_setup_component from tests.components.logbook.test_init import MockLazyEventPartialState -async def test_humanify_shelly_click_event(hass, coap_wrapper): - """Test humanifying Shelly click event.""" +async def test_humanify_shelly_click_event_block_device(hass, coap_wrapper): + """Test humanifying Shelly click event for block device.""" assert coap_wrapper hass.config.components.add("recorder") assert await async_setup_component(hass, "logbook", {}) @@ -51,12 +51,63 @@ async def test_humanify_shelly_click_event(hass, coap_wrapper): assert event1["name"] == "Shelly" assert event1["domain"] == DOMAIN assert ( - event1["message"] == "'single' click event for Test name channel 1 was fired." + event1["message"] + == "'single' click event for Test name channel 1 Input was fired." ) assert event2["name"] == "Shelly" assert event2["domain"] == DOMAIN assert ( event2["message"] - == "'long' click event for shellyswitch25-12345678 channel 2 was fired." + == "'long' click event for shellyswitch25-12345678 channel 2 Input was fired." + ) + + +async def test_humanify_shelly_click_event_rpc_device(hass, rpc_wrapper): + """Test humanifying Shelly click event for rpc device.""" + assert rpc_wrapper + hass.config.components.add("recorder") + assert await async_setup_component(hass, "logbook", {}) + entity_attr_cache = logbook.EntityAttributeCache(hass) + + event1, event2 = list( + logbook.humanify( + hass, + [ + MockLazyEventPartialState( + EVENT_SHELLY_CLICK, + { + ATTR_DEVICE_ID: rpc_wrapper.device_id, + ATTR_DEVICE: "shellyplus1pm-12345678", + ATTR_CLICK_TYPE: "single_push", + ATTR_CHANNEL: 1, + }, + ), + MockLazyEventPartialState( + EVENT_SHELLY_CLICK, + { + ATTR_DEVICE_ID: "no_device_id", + ATTR_DEVICE: "shellypro4pm-12345678", + ATTR_CLICK_TYPE: "btn_down", + ATTR_CHANNEL: 2, + }, + ), + ], + entity_attr_cache, + {}, + ) + ) + + assert event1["name"] == "Shelly" + assert event1["domain"] == DOMAIN + assert ( + event1["message"] + == "'single_push' click event for test switch_0 Input was fired." + ) + + assert event2["name"] == "Shelly" + assert event2["domain"] == DOMAIN + assert ( + event2["message"] + == "'btn_down' click event for shellypro4pm-12345678 channel 2 Input was fired." )