Support receiving long-press events from WeMo devices (#45503)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

tests/components/wemo/conftest.py

@@ -43,13 +43,15 @@ def pywemo_registry_fixture():
 @pytest.fixture(name="pywemo_device")
 def pywemo_device_fixture(pywemo_registry, pywemo_model):
     """Fixture for WeMoDevice instances."""
-    device = create_autospec(getattr(pywemo, pywemo_model), instance=True)
+    cls = getattr(pywemo, pywemo_model)
+    device = create_autospec(cls, instance=True)
     device.host = MOCK_HOST
     device.port = MOCK_PORT
     device.name = MOCK_NAME
     device.serialnumber = MOCK_SERIAL_NUMBER
     device.model_name = pywemo_model
     device.get_state.return_value = 0  # Default to Off
+    device.supports_long_press.return_value = cls.supports_long_press()
 
     url = f"http://{MOCK_HOST}:{MOCK_PORT}/setup.xml"
     with patch("pywemo.setup_url_for_address", return_value=url), patch(

tests/components/wemo/entity_test_helpers.py

@@ -13,19 +13,31 @@ from homeassistant.components.homeassistant import (
     DOMAIN as HA_DOMAIN,
     SERVICE_UPDATE_ENTITY,
 )
+from homeassistant.components.wemo.const import SIGNAL_WEMO_STATE_PUSH
 from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_UNAVAILABLE
 from homeassistant.core import callback
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
 from homeassistant.setup import async_setup_component
 
 
 def _perform_registry_callback(hass, pywemo_registry, pywemo_device):
     """Return a callable method to trigger a state callback from the device."""
 
-    @callback
-    def async_callback():
+    async def async_callback():
+        event = asyncio.Event()
+
+        async def event_callback(e, *args):
+            event.set()
+
+        stop_dispatcher_listener = async_dispatcher_connect(
+            hass, SIGNAL_WEMO_STATE_PUSH, event_callback
+        )
         # Cause a state update callback to be triggered by the device.
-        pywemo_registry.callbacks[pywemo_device.name](pywemo_device, "", "")
-        return hass.async_block_till_done()
+        await hass.async_add_executor_job(
+            pywemo_registry.callbacks[pywemo_device.name], pywemo_device, "", ""
+        )
+        await event.wait()
+        stop_dispatcher_listener()
 
     return async_callback
 
@@ -63,8 +75,10 @@ async def _async_multiple_call_helper(
     """
     # get_state is called outside the event loop. Use non-async Python Event.
     event = threading.Event()
+    waiting = asyncio.Event()
 
     def get_update(force_update=True):
+        hass.add_job(waiting.set)
         event.wait()
 
     update_polling_method = update_polling_method or pywemo_device.get_state
@@ -77,6 +91,7 @@ async def _async_multiple_call_helper(
     )
 
     # Allow the blocked call to return.
+    await waiting.wait()
     event.set()
     if pending:
         await asyncio.wait(pending)

tests/components/wemo/test_device_trigger.py

@@ -0,0 +1,98 @@
+"""Verify that WeMo device triggers work as expected."""
+import pytest
+from pywemo.subscribe import EVENT_TYPE_LONG_PRESS
+
+from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN
+from homeassistant.components.light import DOMAIN as LIGHT_DOMAIN
+from homeassistant.components.wemo.const import DOMAIN, WEMO_SUBSCRIPTION_EVENT
+from homeassistant.const import (
+    CONF_DEVICE_ID,
+    CONF_DOMAIN,
+    CONF_ENTITY_ID,
+    CONF_PLATFORM,
+    CONF_TYPE,
+)
+from homeassistant.setup import async_setup_component
+
+from tests.common import (
+    assert_lists_same,
+    async_get_device_automations,
+    async_mock_service,
+)
+
+MOCK_DEVICE_ID = "some-device-id"
+DATA_MESSAGE = {"message": "service-called"}
+
+
+@pytest.fixture
+def pywemo_model():
+    """Pywemo Dimmer models use the light platform (WemoDimmer class)."""
+    return "Dimmer"
+
+
+async def setup_automation(hass, device_id, trigger_type):
+    """Set up an automation trigger for testing triggering."""
+    return await async_setup_component(
+        hass,
+        AUTOMATION_DOMAIN,
+        {
+            AUTOMATION_DOMAIN: [
+                {
+                    "trigger": {
+                        CONF_PLATFORM: "device",
+                        CONF_DOMAIN: DOMAIN,
+                        CONF_DEVICE_ID: device_id,
+                        CONF_TYPE: trigger_type,
+                    },
+                    "action": {
+                        "service": "test.automation",
+                        "data": DATA_MESSAGE,
+                    },
+                },
+            ]
+        },
+    )
+
+
+async def test_get_triggers(hass, wemo_entity):
+    """Test that the triggers appear for a supported device."""
+    assert wemo_entity.device_id is not None
+
+    expected_triggers = [
+        {
+            CONF_DEVICE_ID: wemo_entity.device_id,
+            CONF_DOMAIN: DOMAIN,
+            CONF_PLATFORM: "device",
+            CONF_TYPE: EVENT_TYPE_LONG_PRESS,
+        },
+        {
+            CONF_DEVICE_ID: wemo_entity.device_id,
+            CONF_DOMAIN: LIGHT_DOMAIN,
+            CONF_ENTITY_ID: wemo_entity.entity_id,
+            CONF_PLATFORM: "device",
+            CONF_TYPE: "turned_off",
+        },
+        {
+            CONF_DEVICE_ID: wemo_entity.device_id,
+            CONF_DOMAIN: LIGHT_DOMAIN,
+            CONF_ENTITY_ID: wemo_entity.entity_id,
+            CONF_PLATFORM: "device",
+            CONF_TYPE: "turned_on",
+        },
+    ]
+    triggers = await async_get_device_automations(
+        hass, "trigger", wemo_entity.device_id
+    )
+    assert_lists_same(triggers, expected_triggers)
+
+
+async def test_fires_on_long_press(hass):
+    """Test wemo long press trigger firing."""
+    assert await setup_automation(hass, MOCK_DEVICE_ID, EVENT_TYPE_LONG_PRESS)
+    calls = async_mock_service(hass, "test", "automation")
+
+    message = {CONF_DEVICE_ID: MOCK_DEVICE_ID, CONF_TYPE: EVENT_TYPE_LONG_PRESS}
+    hass.bus.async_fire(WEMO_SUBSCRIPTION_EVENT, message)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert calls[0].data == DATA_MESSAGE

tests/components/wemo/test_init.py

@@ -110,6 +110,7 @@ async def test_discovery(hass, pywemo_registry):
         device.serialnumber = f"{MOCK_SERIAL_NUMBER}_{counter}"
         device.model_name = "Motion"
         device.get_state.return_value = 0  # Default to Off
+        device.supports_long_press.return_value = False
         return device
 
     pywemo_devices = [create_device(0), create_device(1)]

tests/components/wemo/test_wemo_device.py

@@ -0,0 +1,40 @@
+"""Tests for wemo_device.py."""
+from unittest.mock import patch
+
+import pytest
+from pywemo import PyWeMoException
+
+from homeassistant.components.wemo import CONF_DISCOVERY, CONF_STATIC, wemo_device
+from homeassistant.components.wemo.const import DOMAIN
+from homeassistant.helpers import device_registry
+from homeassistant.setup import async_setup_component
+
+from .conftest import MOCK_HOST
+
+
+@pytest.fixture
+def pywemo_model():
+    """Pywemo Dimmer models use the light platform (WemoDimmer class)."""
+    return "Dimmer"
+
+
+async def test_async_register_device_longpress_fails(hass, pywemo_device):
+    """Device is still registered if ensure_long_press_virtual_device fails."""
+    with patch.object(pywemo_device, "ensure_long_press_virtual_device") as elp:
+        elp.side_effect = PyWeMoException
+        assert await async_setup_component(
+            hass,
+            DOMAIN,
+            {
+                DOMAIN: {
+                    CONF_DISCOVERY: False,
+                    CONF_STATIC: [MOCK_HOST],
+                },
+            },
+        )
+        await hass.async_block_till_done()
+    dr = device_registry.async_get(hass)
+    device_entries = list(dr.devices.values())
+    assert len(device_entries) == 1
+    device_wrapper = wemo_device.async_get_device(hass, device_entries[0].id)
+    assert device_wrapper.supports_long_press is False