Add trigger for persistent_notification (#94809)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/persistent_notification/__init__.py

@@ -1,7 +1,7 @@
 """Support for displaying persistent notifications."""
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from datetime import datetime
 import logging
 from typing import Any, Final, TypedDict
@@ -10,7 +10,7 @@ import voluptuous as vol
 
 from homeassistant.backports.enum import StrEnum
 from homeassistant.components import websocket_api
-from homeassistant.core import HomeAssistant, ServiceCall, callback
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, ServiceCall, callback
 from homeassistant.helpers import config_validation as cv, singleton
 from homeassistant.helpers.dispatcher import (
     async_dispatcher_connect,
@@ -63,6 +63,17 @@ _LOGGER = logging.getLogger(__name__)
 CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
 
 
+@callback
+def async_register_callback(
+    hass: HomeAssistant,
+    _callback: Callable[[UpdateType, dict[str, Notification]], None],
+) -> CALLBACK_TYPE:
+    """Register a callback."""
+    return async_dispatcher_connect(
+        hass, SIGNAL_PERSISTENT_NOTIFICATIONS_UPDATED, _callback
+    )
+
+
 @bind_hass
 def create(
     hass: HomeAssistant,

homeassistant/components/persistent_notification/trigger.py

@@ -0,0 +1,80 @@
+"""Offer persistent_notifications triggered automation rules."""
+from __future__ import annotations
+
+import logging
+from typing import Final
+
+import voluptuous as vol
+
+from homeassistant.const import CONF_PLATFORM
+from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
+import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.trigger import TriggerActionType, TriggerData, TriggerInfo
+from homeassistant.helpers.typing import ConfigType
+
+from . import Notification, UpdateType, async_register_callback
+
+_LOGGER = logging.getLogger(__name__)
+
+
+CONF_NOTIFICATION_ID: Final = "notification_id"
+CONF_UPDATE_TYPE: Final = "update_type"
+
+TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
+    {
+        vol.Required(CONF_PLATFORM): "persistent_notification",
+        vol.Optional(CONF_NOTIFICATION_ID): str,
+        vol.Optional(CONF_UPDATE_TYPE): vol.All(
+            cv.ensure_list, [vol.Coerce(UpdateType)]
+        ),
+    }
+)
+
+
+async def async_attach_trigger(
+    hass: HomeAssistant,
+    config: ConfigType,
+    action: TriggerActionType,
+    trigger_info: TriggerInfo,
+) -> CALLBACK_TYPE:
+    """Listen for state changes based on configuration."""
+    trigger_data: TriggerData = trigger_info["trigger_data"]
+    job = HassJob(action)
+
+    persistent_notification_id = config.get(CONF_NOTIFICATION_ID)
+    update_types = config.get(CONF_UPDATE_TYPE)
+
+    @callback
+    def persistent_notification_listener(
+        update_type: UpdateType, notifications: dict[str, Notification]
+    ) -> None:
+        """Listen for persistent_notification updates."""
+
+        for notification in notifications.values():
+            if update_types and update_type not in update_types:
+                continue
+            if (
+                persistent_notification_id
+                and notification[CONF_NOTIFICATION_ID] != persistent_notification_id
+            ):
+                continue
+
+            hass.async_run_hass_job(
+                job,
+                {
+                    "trigger": {
+                        **trigger_data,  # type: ignore[arg-type]  # https://github.com/python/mypy/issues/9117
+                        "platform": "persistent_notification",
+                        "update_type": update_type,
+                        "notification": notification,
+                    }
+                },
+            )
+
+    _LOGGER.debug(
+        "Attaching persistent_notification trigger for ID: '%s', update_types: %s",
+        persistent_notification_id,
+        update_types,
+    )
+
+    return async_register_callback(hass, persistent_notification_listener)

tests/components/persistent_notification/conftest.py

@@ -0,0 +1,12 @@
+"""The tests for the persistent notification component."""
+
+import pytest
+
+import homeassistant.components.persistent_notification as pn
+from homeassistant.setup import async_setup_component
+
+
+@pytest.fixture(autouse=True)
+async def setup_integration(hass):
+    """Set up persistent notification integration."""
+    assert await async_setup_component(hass, pn.DOMAIN, {})

tests/components/persistent_notification/test_init.py

@@ -1,5 +1,5 @@
 """The tests for the persistent notification component."""
-import pytest
+
 
 import homeassistant.components.persistent_notification as pn
 from homeassistant.components.websocket_api.const import TYPE_RESULT
@@ -9,12 +9,6 @@ from homeassistant.setup import async_setup_component
 from tests.typing import WebSocketGenerator
 
 
-@pytest.fixture(autouse=True)
-async def setup_integration(hass):
-    """Set up persistent notification integration."""
-    assert await async_setup_component(hass, pn.DOMAIN, {})
-
-
 async def test_create(hass: HomeAssistant) -> None:
     """Test creating notification without title or notification id."""
     notifications = pn._async_get_or_create_notifications(hass)

tests/components/persistent_notification/test_trigger.py

@@ -0,0 +1,101 @@
+"""The tests for the persistent notification component triggers."""
+from typing import Any
+
+import homeassistant.components.persistent_notification as pn
+from homeassistant.components.persistent_notification import trigger
+from homeassistant.core import Context, HomeAssistant, callback
+
+
+async def test_automation_with_pn_trigger(hass: HomeAssistant) -> None:
+    """Test automation with a persistent_notification trigger."""
+
+    result_any = []
+    result_dismissed = []
+    result_id = []
+
+    trigger_info = {"trigger_data": {}}
+
+    @callback
+    def trigger_callback_any(
+        run_variables: dict[str, Any], context: Context | None = None
+    ) -> None:
+        result_any.append(run_variables)
+
+    await trigger.async_attach_trigger(
+        hass,
+        {"platform": "persistent_notification"},
+        trigger_callback_any,
+        trigger_info,
+    )
+
+    @callback
+    def trigger_callback_dismissed(
+        run_variables: dict[str, Any], context: Context | None = None
+    ) -> None:
+        result_dismissed.append(run_variables)
+
+    await trigger.async_attach_trigger(
+        hass,
+        {"platform": "persistent_notification", "update_type": "removed"},
+        trigger_callback_dismissed,
+        trigger_info,
+    )
+
+    @callback
+    def trigger_callback_id(
+        run_variables: dict[str, Any], context: Context | None = None
+    ) -> None:
+        result_id.append(run_variables)
+
+    await trigger.async_attach_trigger(
+        hass,
+        {"platform": "persistent_notification", "notification_id": "42"},
+        trigger_callback_id,
+        trigger_info,
+    )
+
+    await hass.services.async_call(
+        pn.DOMAIN,
+        "create",
+        {"notification_id": "test_notification", "message": "test"},
+        blocking=True,
+    )
+
+    result = result_any[0].get("trigger")
+    assert result["platform"] == "persistent_notification"
+    assert result["update_type"] == pn.UpdateType.ADDED
+    assert result["notification"]["notification_id"] == "test_notification"
+    assert result["notification"]["message"] == "test"
+
+    assert len(result_dismissed) == 0
+    assert len(result_id) == 0
+
+    await hass.services.async_call(
+        pn.DOMAIN,
+        "dismiss",
+        {"notification_id": "test_notification"},
+        blocking=True,
+    )
+
+    result = result_any[1].get("trigger")
+    assert result["platform"] == "persistent_notification"
+    assert result["update_type"] == pn.UpdateType.REMOVED
+    assert result["notification"]["notification_id"] == "test_notification"
+    assert result["notification"]["message"] == "test"
+    assert result_any[1] == result_dismissed[0]
+
+    assert len(result_id) == 0
+
+    await hass.services.async_call(
+        pn.DOMAIN,
+        "create",
+        {"notification_id": "42", "message": "Forty Two"},
+        blocking=True,
+    )
+
+    result = result_any[2].get("trigger")
+    assert result["platform"] == "persistent_notification"
+    assert result["update_type"] == pn.UpdateType.ADDED
+    assert result["notification"]["notification_id"] == "42"
+    assert result["notification"]["message"] == "Forty Two"
+    assert result_any[2] == result_id[0]