diff --git a/homeassistant/components/persistent_notification/__init__.py b/homeassistant/components/persistent_notification/__init__.py index fe8849c7788..960d0a5ca59 100644 --- a/homeassistant/components/persistent_notification/__init__.py +++ b/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, diff --git a/homeassistant/components/persistent_notification/trigger.py b/homeassistant/components/persistent_notification/trigger.py new file mode 100644 index 00000000000..12f98083bdf --- /dev/null +++ b/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) diff --git a/tests/components/persistent_notification/conftest.py b/tests/components/persistent_notification/conftest.py new file mode 100644 index 00000000000..d665c0075b3 --- /dev/null +++ b/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, {}) diff --git a/tests/components/persistent_notification/test_init.py b/tests/components/persistent_notification/test_init.py index 4f0851dc477..71a0fcae917 100644 --- a/tests/components/persistent_notification/test_init.py +++ b/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) diff --git a/tests/components/persistent_notification/test_trigger.py b/tests/components/persistent_notification/test_trigger.py new file mode 100644 index 00000000000..3cf3655a3b6 --- /dev/null +++ b/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]