Coverage for custom_components/supernotify/transports/html5.py: 98%

1"""HTML5 browser push transport for SuperNotify.

3Sends web push notifications to browsers registered with Home Assistant's

4`html5` integration, calling the modern `html5.send_message` entity service

5(one `notify.*` entity per registered browser). The legacy `notify.html5`

6platform is intentionally NOT used: its `ttl` and `priority` parameters are

7read from send_message kwargs that the notify service never populates, so

8urgency would silently always be "normal".

10Supported data keys (all optional):

11 html5_urgency str override web push urgency, one of

12 low | normal | high. Default is

13 mapped from the SuperNotify

14 priority: critical/high -> high,

15 medium -> normal,

16 low/minimum -> low

17 html5_tag str notification tag: notifications

18 sharing a tag replace each other

19 html5_actions list[dict] action buttons, each

20 {action, title, icon}. Clicks fire

21 `html5_notification.clicked`

22 events with the `action` value

23 html5_attach_image bool attach camera snapshot as `image`

24 URL (default: False). Uses the

25 shared media pipeline; the URL must

26 be reachable by the browser, so an

27 HTTPS external_url (or HA Cloud)

28 is usually required

29 html5_icon str icon URL

30 html5_badge str badge URL (Android status bar)

31 html5_url str URL opened when the notification

32 is clicked (sent as `data.url`)

33 html5_require_interaction bool keep the notification on screen

34 until the user interacts with it

35 html5_renotify bool alert again when a new notification

36 replaces an existing tag

37 html5_silent bool suppress sound/vibration. Mutually

38 exclusive with html5_vibrate in the

39 service schema (vol.Exclusive): when

40 both are supplied, a truthy silent

41 wins and vibrate is dropped, a falsy

42 silent is dropped in favour of

43 vibrate (warning either way)

44 html5_vibrate list[int] vibration pattern in milliseconds,

45 e.g. [200, 100, 200]. See

46 html5_silent for the exclusivity

47 rule

48 html5_ttl int/dict time-to-live: seconds or an HA

49 duration dict, forwarded as-is

50 html5_data dict extra keys merged into the custom

51 `data` field of the service call

52 (html5_url wins on `url` clashes)

54Notes on the HA `html5.send_message` service schema:

55- The schema is a strict whitelist of first-class fields: unknown keys at

56 the top level fail the whole call. Residual generic data keys are

57 therefore NOT merged into the payload (dropped with a debug log); the

58 `data` custom field, fed by `html5_url` / `html5_data`, is the explicit

59 passthrough for anything else.

60- `title` is REQUIRED by the schema; when the envelope has no title the

61 HA default "Home Assistant" is used.

62- Targets are `notify.*` entities created by browser push registrations

63 (html5 config entry with VAPID keys). Non-matching targets are dropped

64 with a debug log; no valid target fails the delivery.

65- Expired push subscriptions (410 GONE) are handled by the core, which

66 unregisters the browser and raises: call_action then returns False.

68Internal data keys filtered upstream by notification.py and NOT popped

69here: force_resend, spoken_message.

71References:

72- HTML5 push integration: https://www.home-assistant.io/integrations/html5/

74"""

76from __future__ import annotations

78import logging

79import re

80from typing import TYPE_CHECKING, Any

82from homeassistant.const import ATTR_ENTITY_ID

84from custom_components.supernotify.common import boolify

85from custom_components.supernotify.const import (

86 ATTR_DATA,

87 ATTR_MEDIA_SNAPSHOT_URL,

88 OPTION_TARGET_CATEGORIES,

89 OPTION_TARGET_SELECT,

90 TRANSPORT_HTML5,

91)

92from custom_components.supernotify.model import DebugTrace, TargetRequired, TransportConfig, TransportFeature

93from custom_components.supernotify.transport import Transport

95if TYPE_CHECKING:

96 from custom_components.supernotify.envelope import Envelope

98_LOGGER = logging.getLogger(__name__)

100RE_VALID_HTML5 = r"notify\.[A-Za-z0-9_]+"

101_HTML5_TARGET_RE = re.compile(r"^notify\.[A-Za-z0-9_]+$")

102

103# HA schema default for the required title field

104_DEFAULT_TITLE = "Home Assistant"

105

106_VALID_URGENCY = ("low", "normal", "high")

107

108# SuperNotify priority -> web push urgency

109_URGENCY_BY_PRIORITY = {

110 "critical": "high",

111 "high": "high",

112 "medium": "normal",

113 "low": "low",

114 "minimum": "low",

115}

116

117

118class HTML5Transport(Transport):

119 """Notify browsers via the Home Assistant html5 web push integration."""

120

121 def __init__(self, *args: Any, **kwargs: Any) -> None:

122 super().__init__(*args, **kwargs)

123

124 name = TRANSPORT_HTML5

125

126 @property

127 def supported_features(self) -> TransportFeature:

128 return (

129 TransportFeature.MESSAGE

130 | TransportFeature.TITLE

131 | TransportFeature.IMAGES

132 | TransportFeature.SNAPSHOT_IMAGE

133 | TransportFeature.ACTIONS

134 )

135

136 @property

137 def default_config(self) -> TransportConfig:

138 config = TransportConfig()

139 config.delivery_defaults.action = "html5.send_message"

140 config.delivery_defaults.target_required = TargetRequired.ALWAYS

141 config.delivery_defaults.options = {

142 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID],

143 OPTION_TARGET_SELECT: [RE_VALID_HTML5],

144 }

145 return config

146

147 def validate_action(self, action: str | None) -> bool:

148 """Validate that action is the html5 send_message service."""

149 return action == "html5.send_message"

150

151 def select_targets(self, envelope: Envelope) -> list[str]:

152 """Filter envelope targets down to html5 `notify.*` entity ids.

153

154 The service is entity-based: every target must be a notify entity

155 created by a browser push registration. Non-matching entries are

156 dropped with a debug log; duplicates are removed preserving order.

157 """

158 raw_targets: list[str] = envelope.target.resolved_targets() if envelope.target else []

159 targets: list[str] = []

160 for target in raw_targets:

161 if isinstance(target, str) and _HTML5_TARGET_RE.match(target):

162 if target not in targets:

163 targets.append(target)

164 else:

165 _LOGGER.debug("SUPERNOTIFY html5: skipping invalid target %r (expected notify.*)", target)

166 return targets

167

168 async def _resolve_image_url(self, envelope: Envelope) -> str | None:

169 """Resolve a browser-reachable snapshot URL.

170

171 Order of resolution:

172 1. snapshot URL already in envelope media, absolutised

173 2. envelope.grab_image() + media_storage.object_url() (shared

174 media pipeline; never a local path)

175 3. None

176 """

177 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) if envelope.media else None

178 if snapshot_url:

179 return self.hass_api.abs_url(snapshot_url)

180

181 image_path = None

182 try:

183 image_path = await envelope.grab_image()

184 except Exception as e:

185 _LOGGER.warning("SUPERNOTIFY html5: failed to grab image: %s", e)

186 if image_path:

187 try:

188 return await self.context.media_storage.object_url(image_path)

189 except Exception as e:

190 _LOGGER.debug("SUPERNOTIFY html5: object_url failed for %s: %s", image_path, e)

191 return None

192

193 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool:

194 _LOGGER.debug("SUPERNOTIFY html5 %s", envelope.message)

195

196 raw_data: dict[str, Any] = dict(envelope.data) if envelope.data else {}

197

198 # Pop html5-specific data keys

199 urgency_override = raw_data.pop("html5_urgency", None)

200 tag = raw_data.pop("html5_tag", None)

201 actions = raw_data.pop("html5_actions", None)

202 attach_image = boolify(raw_data.pop("html5_attach_image", False), default=False)

203 icon = raw_data.pop("html5_icon", None)

204 badge = raw_data.pop("html5_badge", None)

205 click_url = raw_data.pop("html5_url", None)

206 require_interaction_raw = raw_data.pop("html5_require_interaction", None)

207 renotify_raw = raw_data.pop("html5_renotify", None)

208 silent_raw = raw_data.pop("html5_silent", None)

209 vibrate = raw_data.pop("html5_vibrate", None)

210 ttl = raw_data.pop("html5_ttl", None)

211 custom_data = raw_data.pop("html5_data", None)

212

213 # Resolve and pre-validate notify entity targets

214 targets = self.select_targets(envelope)

215 if not targets:

216 _LOGGER.warning("SUPERNOTIFY html5: no valid targets (expected notify.* entities)")

217 self.record_error("no valid html5 notify entity targets", "deliver")

218 return False

219

220 # Resolve urgency: explicit valid override, else mapped from priority

221 urgency = _URGENCY_BY_PRIORITY.get(envelope.priority or "medium", "normal")

222 if urgency_override is not None:

223 candidate = str(urgency_override).lower()

224 if candidate in _VALID_URGENCY:

225 urgency = candidate

226 else:

227 _LOGGER.warning(

228 "SUPERNOTIFY html5: invalid html5_urgency %r (valid: %s), using '%s'",

229 urgency_override,

230 _VALID_URGENCY,

231 urgency,

232 )

233

234 # The service schema declares silent and vibrate as mutually

235 # exclusive (vol.Exclusive shares the "silent_xor_vibrate" group):

236 # sending both keys fails the whole call, whatever their values

237 if silent_raw is not None and vibrate is not None:

238 if boolify(silent_raw, default=False):

239 _LOGGER.warning("SUPERNOTIFY html5: html5_silent and html5_vibrate are mutually exclusive, dropping vibrate")

240 vibrate = None

241 else:

242 _LOGGER.warning(

243 "SUPERNOTIFY html5: html5_silent and html5_vibrate are mutually exclusive, dropping falsy silent"

244 )

245 silent_raw = None

246

247 # Build the payload: title is REQUIRED by the service schema

248 action_data: dict[str, Any] = {

249 "title": envelope.title or _DEFAULT_TITLE,

250 "message": envelope.message or "",

251 "urgency": urgency,

252 }

253 if icon:

254 action_data["icon"] = str(icon)

255 if badge:

256 action_data["badge"] = str(badge)

257 if tag:

258 action_data["tag"] = str(tag)

259 if actions is not None:

260 if isinstance(actions, list):

261 action_data["actions"] = actions

262 else:

263 _LOGGER.warning("SUPERNOTIFY html5: html5_actions must be a list of dicts, dropping %r", actions)

264 if renotify_raw is not None:

265 action_data["renotify"] = boolify(renotify_raw, default=False)

266 if silent_raw is not None:

267 action_data["silent"] = boolify(silent_raw, default=False)

268 if require_interaction_raw is not None:

269 action_data["require_interaction"] = boolify(require_interaction_raw, default=False)

270 if vibrate is not None:

271 if isinstance(vibrate, list):

272 action_data["vibrate"] = vibrate

273 else:

274 _LOGGER.warning("SUPERNOTIFY html5: html5_vibrate must be a list of ints, dropping %r", vibrate)

275 if ttl is not None:

276 action_data["ttl"] = ttl

277

278 # Attach camera snapshot as browser-reachable URL (never a local path)

279 if attach_image:

280 image_url = await self._resolve_image_url(envelope)

281 if image_url:

282 action_data["image"] = str(image_url)

283 else:

284 _LOGGER.debug("SUPERNOTIFY html5: no image URL available, sending without image")

285

286 # Custom `data` field: the only passthrough the strict schema allows.

287 # html5_data is merged first so the explicit html5_url wins on `url`.

288 data_field: dict[str, Any] = {}

289 if isinstance(custom_data, dict):

290 data_field.update(custom_data)

291 elif custom_data is not None:

292 _LOGGER.warning("SUPERNOTIFY html5: html5_data must be a dict, dropping %r", custom_data)

293 if click_url:

294 data_field["url"] = str(click_url)

295 if data_field:

296 action_data[ATTR_DATA] = data_field

297

298 # Residual generic keys are NOT merged: the service schema is a

299 # strict whitelist and any extra top-level key fails the whole call.

300 if raw_data:

301 _LOGGER.debug(

302 "SUPERNOTIFY html5: dropping data keys not supported by the strict service schema: %s",

303 sorted(raw_data),

304 )

305

306 target_data = {ATTR_ENTITY_ID: targets}

307 return await self.call_action(envelope, action_data=action_data, target_data=target_data)