Coverage for custom_components/supernotify/notify.py: 97%
100 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1from __future__ import annotations
3import logging
4from functools import partial
5from typing import TYPE_CHECKING, Any, cast
7from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN
8from homeassistant.components.notify import NotifyEntity, NotifyEntityFeature
9from homeassistant.components.notify.const import NOTIFY_SERVICE_SCHEMA
10from homeassistant.components.notify.legacy import BaseNotificationService
11from homeassistant.const import (
12 CONF_NAME,
13 CONF_TARGET,
14)
15from homeassistant.core import (
16 HomeAssistant,
17 ServiceCall,
18)
19from homeassistant.helpers import entity_registry as er
20from homeassistant.util import slugify
22from . import DOMAIN, NOTIFY_SERVICE_NAME
23from .common import ensure_list
24from .engine import SupernotifyEngine
25from .model import NotifyEntityPlatform
27if TYPE_CHECKING:
28 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
29 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
31 from . import SupernotifyConfigEntry
32from .const import (
33 ATTR_DATA,
34 ATTR_DELIVERY,
35 ATTR_SCENARIOS_APPLY,
36 CONF_MESSAGE,
37 CONF_TITLE,
38)
40_LOGGER = logging.getLogger(__name__)
42PARALLEL_UPDATES = 0
45class SuperNotificationService(BaseNotificationService):
46 """Legacy notify-platform compatibility shim.
48 The only reason this exists is to satisfy HA core's notify/legacy.py
49 BaseNotificationService contract, so notify.supernotify and notify.<target> keep working.
50 It adds nothing but that glue on top of SupernotifyEngine - nothing else in this
51 integration (SupernotifyEntity, RecipientNotifyEntity, the supplemental actions, tests
52 targeting the engine) references this class or BaseNotificationService. If/when HA core
53 drops BaseNotificationService, delete this class and the matching async_setup/
54 async_register_services/async_unregister_services calls in __init__.py; everything else
55 keeps working unchanged.
56 """
58 def __init__(self, engine: SupernotifyEngine, *args: Any, **kwargs: Any) -> None:
59 super().__init__()
60 self.engine = engine
62 async def async_unregister_services(self) -> None:
63 _LOGGER.info("SUPERNOTIFY Unregistering notify service")
64 return await super().async_unregister_services()
66 async def _async_notify_message_service(self, service: ServiceCall) -> None:
67 """Override of BaseNotificationService._async_notify_message_service (notify/legacy.py)
68 to forward the calling action's Context through to async_send_message. HA core's
69 implementation builds its own kwargs from service.data and drops service.context
70 entirely, which is why every notify.supernotify/notify.<target> call otherwise loses
71 its link back to the triggering automation, showing up downstream (e.g. in the mobile
72 app's notification history) as having no recorded cause.
73 """
74 kwargs: dict[str, Any] = {}
75 message: str = service.data[CONF_MESSAGE]
76 if title := service.data.get(CONF_TITLE):
77 kwargs[CONF_TITLE] = title
78 if self.registered_targets.get(service.service) is not None:
79 kwargs[CONF_TARGET] = [self.registered_targets[service.service]]
80 elif service.data.get(CONF_TARGET) is not None:
81 kwargs[CONF_TARGET] = service.data.get(CONF_TARGET)
82 kwargs[CONF_MESSAGE] = message
83 kwargs[ATTR_DATA] = service.data.get(ATTR_DATA)
84 kwargs["context"] = service.context
86 await self.engine.async_send_message(**kwargs, via_notify=f"{NOTIFY_DOMAIN}.{service.service}")
89class ExposedNotifyEntity(NotifyEntity):
90 """A scenario or delivery exposed as its own `notify.<notify_entity>` entity.
92 Sending to it is a plain notification, to the default recipients, with only the scenario
93 applied, or the delivery selected, as `data` gives. Per HA's notify entity service schema,
94 only message/title are ever passed in here.
95 """
97 _attr_supported_features = NotifyEntityFeature.TITLE
99 def __init__(self, unique_id: str, entity_id: str, name: str, engine: SupernotifyEngine, data: dict[str, Any]) -> None:
100 self._attr_unique_id = unique_id
101 self._attr_name = name
102 self.entity_id = entity_id
103 self._engine = engine
104 self._data = data
106 async def async_send_message(self, message: str, title: str | None = None) -> None:
107 await self._engine.async_send_message(
108 message, title=title, context=self._context, data=dict(self._data), via_notify=self.entity_id
109 )
112def exposed_notify_entities(
113 hass: HomeAssistant, entry_id: str, engine: SupernotifyEngine, reserved: str = NOTIFY_SERVICE_NAME
114) -> list[ExposedNotifyEntity]:
115 """An entity for each scenario and delivery with a `notify_entity`, other than where its
116 `notify.<notify_entity>` is already taken, by an entity or a notify action, which gets a
117 repair instead. An entity left from a `notify_entity` no longer configured is removed.
119 `reserved` is the name of the main notify action, which isn't registered until after this.
120 """
121 registry = er.async_get(hass)
122 context = engine.context
123 # recipients are added alongside, so may not be registered yet
124 taken: set[str] = {f"{NOTIFY_DOMAIN}.{reserved}"} | {
125 f"{NOTIFY_DOMAIN}.recipient_{r.name}" for r in context.people_registry.people.values()
126 }
127 entities: dict[str, ExposedNotifyEntity] = {}
128 for kind, data_key, configured in (
129 ("scenario", ATTR_SCENARIOS_APPLY, context.scenario_registry.scenarios),
130 ("delivery", ATTR_DELIVERY, context.delivery_registry.deliveries),
131 ):
132 for name, exposable in configured.items():
133 issue_id = f"notify_entity_{kind}_{name}"
134 exposable.notify_entity_id = None
135 if not exposable.notify_entity:
136 context.hass_api.delete_issue(issue_id)
137 continue
138 unique_id = f"{entry_id}_{kind}_{name}"
139 entity_id = f"{NOTIFY_DOMAIN}.{exposable.notify_entity}"
140 registered: str | None = registry.async_get_entity_id(NOTIFY_DOMAIN, DOMAIN, unique_id)
141 if (
142 entity_id in taken
143 or hass.services.has_service(NOTIFY_DOMAIN, exposable.notify_entity)
144 or (
145 registered != entity_id
146 and (registry.async_get(entity_id) is not None or hass.states.get(entity_id) is not None)
147 )
148 ):
149 _LOGGER.warning("SUPERNOTIFY %s already exists, so %s %s can't use it", entity_id, kind, name)
150 context.hass_api.raise_issue(
151 issue_id,
152 issue_key="notify_entity_exists",
153 issue_map={"entity_id": entity_id, "kind": kind, "name": name},
154 learn_more_url=f"https://supernotify.rhizomatics.org.uk/configuration/{'scenarios' if kind == 'scenario' else 'deliveries'}/",
155 )
156 continue
157 context.hass_api.delete_issue(issue_id)
158 if registered is not None and registered != entity_id:
159 # notify_entity has changed, and the registry would otherwise keep the old entity_id
160 registry.async_update_entity(registered, new_entity_id=entity_id)
161 taken.add(entity_id)
162 exposable.notify_entity_id = entity_id
163 entities[unique_id] = ExposedNotifyEntity(unique_id, entity_id, exposable.alias or name, engine, {data_key: [name]})
164 for registered_entry in er.async_entries_for_config_entry(registry, entry_id):
165 if (
166 registered_entry.domain == NOTIFY_DOMAIN
167 and registered_entry.unique_id.startswith((f"{entry_id}_scenario_", f"{entry_id}_delivery_"))
168 and registered_entry.unique_id not in entities
169 ):
170 registry.async_remove(registered_entry.entity_id)
171 return list(entities.values())
174def async_register_entity_actions(
175 hass: HomeAssistant, entry: SupernotifyConfigEntry, engine: SupernotifyEngine, targets: dict[str, str]
176) -> None:
177 """A legacy `notify.<name>` action alongside each notify entity, by action name and the target
178 it stands for, for callers like the Alert integration that only know notify actions.
180 The target goes through the same short circuit as when given to supernotify.notify, see
181 convert_notify_entities() and convert_exposed_notify_entities() in notification.py.
182 """
183 for name, target in targets.items():
184 if hass.services.has_service(NOTIFY_DOMAIN, name):
185 _LOGGER.warning("SUPERNOTIFY notify.%s is already registered - not registering it again", name)
186 continue
188 async def notify(call: ServiceCall, target: str = target, name: str = name) -> None:
189 await engine.async_send_message(
190 call.data[CONF_MESSAGE],
191 title=call.data.get(CONF_TITLE),
192 target=[target, *ensure_list(call.data.get(CONF_TARGET))],
193 context=call.context,
194 data=call.data.get(ATTR_DATA),
195 via_notify=f"{NOTIFY_DOMAIN}.{name}",
196 )
198 hass.services.async_register(NOTIFY_DOMAIN, name, notify, schema=NOTIFY_SERVICE_SCHEMA)
199 entry.async_on_unload(partial(hass.services.async_remove, NOTIFY_DOMAIN, name))
202async def async_get_service(
203 hass: HomeAssistant,
204 config: ConfigType,
205 discovery_info: DiscoveryInfoType | None = None,
206) -> None:
207 """Legacy `notify: - platform: supernotify` entrypoint - see async_setup_legacy in legacy
208 BaseNotificationService.
210 The config entry is now the sole, unconditional owner of notify.supernotify (see
211 async_setup_entry in __init__.py), so this leftover legacy YAML block never builds or
212 registers a service any more - it only raises a fixable repair pointing at the migration
213 (see repairs.py) and declines to set up (returning None is HA's supported "decline" path for
214 a legacy notify platform - a clean one-line log, no exception).
216 A `name:` in this leftover block still gets synced onto the owning entry every load though
217 (not gated behind that repair), and likewise for its template_path/media_path/etc and
218 archive/dupe_check/housekeeping settings - otherwise an entry auto-bootstrapped blank by
219 async_setup (see __init__.py), which happens before anyone gets around to opening and
220 confirming the migration repair, would keep running on defaults with nothing configured,
221 silently breaking automations, template/media paths and archiving on every restart until the
222 repair is manually confirmed. That repair is only ever needed for delivery/transports/
223 scenarios/etc - a "simple" install with none of that has no reason to see it at all, so this
224 core migration must not depend on it.
225 """
226 _ = discovery_info
228 from .repairs import async_create_legacy_yaml_issue, async_sync_entry_from_legacy_config
230 legacy_config = dict(config)
231 async_sync_entry_from_legacy_config(hass, legacy_config)
232 async_create_legacy_yaml_issue(hass, legacy_config)
235async def async_setup_entry(
236 hass: HomeAssistant,
237 entry: SupernotifyConfigEntry,
238 async_add_entities: AddConfigEntryEntitiesCallback,
239) -> None:
240 """Expose each configured recipient as its own notify entity, and each scenario or delivery
241 with a `notify_entity`, all with a legacy notify action of the same name.
243 Forwarded to from async_setup_entry in __init__.py once the SupernotifyEngine (entry.
244 runtime_data) is fully initialized, so people_registry is already populated.
245 """
246 engine: SupernotifyEngine = entry.runtime_data
247 exposed = exposed_notify_entities(
248 hass, entry.entry_id, engine, reserved=slugify(entry.data.get(CONF_NAME) or NOTIFY_SERVICE_NAME)
249 )
250 async_add_entities(exposed)
251 people_registry = engine.context.people_registry
252 people_registry.expose_notify_entities(
253 entry.entry_id,
254 async_add_entities,
255 cast(NotifyEntityPlatform, engine),
256 people_registry.subentry_ids(entry.subentries),
257 )
258 async_register_entity_actions(
259 hass,
260 entry,
261 engine,
262 {f"recipient_{r.name}": r.entity_id for r in engine.context.people_registry.people.values()}
263 | {e.entity_id.partition(".")[2]: e.entity_id for e in exposed},
264 )