Coverage for custom_components/supernotify/transports/pushover.py: 100%

111 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-11-08 17:42 +0000

1"""Pushover transport for SuperNotify. 

2 

3Sends push notifications via Pushover (https://pushover.net). 

4Requires the official HA Pushover integration, added from Settings > Devices & services 

5with an API token and user key. 

6 

7The notify service name (e.g. notify.pushover) is found automatically when the 

8integration is set up. If it's named differently, or there are several accounts, set it as 

9`action:` on the delivery. 

10 

11Priority mapping (SuperNotify -> Pushover integer): 

12 critical -> 2 (emergency: requires retry+expire, repeats until acknowledged) 

13 high -> 1 (high: bypasses user's quiet hours) 

14 medium -> 0 (normal: standard sound and vibration) 

15 low -> -1 (low: no sound and no vibration) 

16 minimum -> -2 (silent: only iOS badge, no visible notification) 

17 

18Note on emergency (priority=2): Pushover REQUIRES the `retry` and `expire` 

19parameters. If not provided, SuperNotify supplies sensible defaults 

20(retry=60s, expire=3600s) and logs them. 

21 

22Supported data keys (all optional unless noted): 

23 pushover_priority int (-2..2) Override priority; out-of-range -> auto-mapping. 

24 pushover_sound str Notification sound: "pushover", "bike", "siren", 

25 "vibrate", "none", "alien", "echo", etc. 

26 See https://pushover.net/api#sounds 

27 pushover_url str Supplementary URL attached to the notification. 

28 pushover_url_title str Title for the URL (max 100 chars). 

29 pushover_retry int Seconds between retries (min 30, default 60). 

30 Emergency only (priority=2). 

31 pushover_expire int Total seconds to keep retrying (max 10800, 

32 default 3600). Emergency only. 

33 pushover_callback str Public URL for emergency acknowledgment webhook 

34 (HA webhook endpoint). 

35 pushover_html bool Enable HTML formatting in the message 

36 (links, bold, italic). Default: false. 

37 pushover_ttl int Seconds before automatic deletion of the 

38 notification from the device. 

39 pushover_device str|list Send to specific devices (device names as 

40 configured in Pushover, e.g. "iphone"), 

41 passed as the service `target`. 

42 Default: all devices on the account. 

43 pushover_tags str|list Tags for the notification. An emergency (priority=2) 

44 with tags can be cancelled with the 

45 `pushover.cancel` action. 

46 pushover_attach_image bool Grab camera snapshot (uses 

47 media.camera_entity_id from the SuperNotify 

48 call) and attach it to the notification. 

49 The media folder must be listed in 

50 homeassistant.allowlist_external_dirs, or the 

51 integration rejects the attachment. 

52""" 

53 

54from __future__ import annotations 

55 

56import logging 

57from typing import TYPE_CHECKING, Any, ClassVar 

58 

59from homeassistant.components.notify.const import ATTR_DATA, ATTR_TARGET 

60 

61from custom_components.supernotify.common import boolify, ensure_list 

62from custom_components.supernotify.const import TRANSPORT_PUSHOVER 

63from custom_components.supernotify.model import ( 

64 DebugTrace, 

65 TransportConfig, 

66 TransportFeature, 

67) 

68from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

69from custom_components.supernotify.transport import Transport 

70 

71if TYPE_CHECKING: 

72 from custom_components.supernotify.envelope import Envelope 

73 

74_LOGGER = logging.getLogger(__name__) 

75 

76# SuperNotify priority -> Pushover integer (-2..2) 

77_PRIORITY_MAP: dict[str, int] = { 

78 "critical": 2, # emergency - repeats until acknowledged, requires retry+expire 

79 "high": 1, # high - bypasses user quiet hours 

80 "medium": 0, # normal - standard sound and vibration 

81 "low": -1, # low - no sound and no vibration 

82 "minimum": -2, # silent - only iOS badge, no visible notification 

83} 

84 

85_EMERGENCY_PRIORITY = 2 

86_EMERGENCY_RETRY_MIN = 30 # seconds (Pushover API limit) 

87_EMERGENCY_EXPIRE_MAX = 10800 # seconds (Pushover API limit = 3 hours) 

88_EMERGENCY_RETRY_DEFAULT = 60 # sensible default when not specified 

89_EMERGENCY_EXPIRE_DEFAULT = 3600 # sensible default when not specified (1 hour) 

90 

91 

92class PushoverTransport(Transport): 

93 """Notify via Pushover push notification service.""" 

94 

95 name = TRANSPORT_PUSHOVER 

96 integration_domain = "pushover" 

97 action_configured = True 

98 declared_options: ClassVar[list[DeliveryOption]] = [*MEDIA_OPTIONS] 

99 

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

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

102 

103 @property 

104 def supported_features(self) -> TransportFeature: 

105 return TransportFeature.MESSAGE | TransportFeature.TITLE | TransportFeature.IMAGES | TransportFeature.SNAPSHOT_IMAGE 

106 

107 @property 

108 def default_config(self) -> TransportConfig: 

109 config = super().default_config 

110 config.delivery_defaults.action = self.hass_api.find_service("notify", "homeassistant.components.pushover.notify") 

111 return config 

112 

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

114 return self.validate_notify_action(action) 

115 

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

117 _LOGGER.debug("SUPERNOTIFY pushover %s", envelope.message) 

118 

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

120 

121 # --- Pop pushover_* keys (must not be forwarded to the service) --- 

122 priority_ovr_raw = raw_data.pop("pushover_priority", None) 

123 sound = raw_data.pop("pushover_sound", None) 

124 url = raw_data.pop("pushover_url", None) 

125 url_title = raw_data.pop("pushover_url_title", None) 

126 retry_raw = raw_data.pop("pushover_retry", None) 

127 expire_raw = raw_data.pop("pushover_expire", None) 

128 callback = raw_data.pop("pushover_callback", None) 

129 html_flag = boolify(raw_data.pop("pushover_html", False), default=False) 

130 ttl_raw = raw_data.pop("pushover_ttl", None) 

131 device = raw_data.pop("pushover_device", None) 

132 tags = raw_data.pop("pushover_tags", None) 

133 attach_image = boolify(raw_data.pop("pushover_attach_image", False), default=False) 

134 

135 # --- Priority: validate override or use auto-mapping --- 

136 priority_ovr: int | None = None 

137 if priority_ovr_raw is not None: 

138 try: 

139 priority_ovr = int(priority_ovr_raw) 

140 if not -2 <= priority_ovr <= 2: 

141 _LOGGER.warning( 

142 "SUPERNOTIFY pushover: pushover_priority %d out of range -2..2, falling back to auto mapping", 

143 priority_ovr, 

144 ) 

145 priority_ovr = None 

146 except TypeError, ValueError: # py3.13 compat 

147 _LOGGER.warning( 

148 "SUPERNOTIFY pushover: invalid pushover_priority %r, falling back to auto mapping", 

149 priority_ovr_raw, 

150 ) 

151 priority_ovr = None 

152 

153 pushover_priority: int = ( 

154 priority_ovr if priority_ovr is not None else _PRIORITY_MAP.get(envelope.priority or "medium", 0) 

155 ) 

156 

157 # --- Base action data (includes message and title) --- 

158 action_data = envelope.core_action_data() 

159 

160 # --- Pushover-specific data payload --- 

161 push_data: dict[str, Any] = {"priority": pushover_priority} 

162 

163 # Emergency (priority=2): Pushover REQUIRES retry and expire 

164 if pushover_priority == _EMERGENCY_PRIORITY: 

165 # retry: robust parse (YAML string or int) -> fallback default on error 

166 if retry_raw is None: 

167 retry_val: int = _EMERGENCY_RETRY_DEFAULT 

168 else: 

169 try: 

170 retry_val = int(retry_raw) 

171 except TypeError, ValueError: 

172 _LOGGER.warning( 

173 "SUPERNOTIFY pushover: invalid pushover_retry %r, using default %ds", 

174 retry_raw, 

175 _EMERGENCY_RETRY_DEFAULT, 

176 ) 

177 retry_val = _EMERGENCY_RETRY_DEFAULT 

178 

179 # expire: same robust pattern 

180 if expire_raw is None: 

181 expire_val: int = _EMERGENCY_EXPIRE_DEFAULT 

182 else: 

183 try: 

184 expire_val = int(expire_raw) 

185 except TypeError, ValueError: 

186 _LOGGER.warning( 

187 "SUPERNOTIFY pushover: invalid pushover_expire %r, using default %ds", 

188 expire_raw, 

189 _EMERGENCY_EXPIRE_DEFAULT, 

190 ) 

191 expire_val = _EMERGENCY_EXPIRE_DEFAULT 

192 

193 if retry_val < _EMERGENCY_RETRY_MIN: 

194 _LOGGER.warning( 

195 "SUPERNOTIFY pushover: emergency retry %ds < minimum %ds, clamping", 

196 retry_val, 

197 _EMERGENCY_RETRY_MIN, 

198 ) 

199 retry_val = _EMERGENCY_RETRY_MIN 

200 

201 if expire_val > _EMERGENCY_EXPIRE_MAX: 

202 _LOGGER.warning( 

203 "SUPERNOTIFY pushover: emergency expire %ds > maximum %ds, clamping", 

204 expire_val, 

205 _EMERGENCY_EXPIRE_MAX, 

206 ) 

207 expire_val = _EMERGENCY_EXPIRE_MAX 

208 

209 push_data["retry"] = retry_val 

210 push_data["expire"] = expire_val 

211 

212 if callback: 

213 push_data["callback_url"] = callback 

214 

215 _LOGGER.debug( 

216 "SUPERNOTIFY pushover: emergency mode - retry=%ds expire=%ds", 

217 retry_val, 

218 expire_val, 

219 ) 

220 

221 # Optional fields - added only when present 

222 if sound: 

223 push_data["sound"] = sound 

224 if url: 

225 push_data["url"] = url 

226 if url_title: 

227 push_data["url_title"] = url_title 

228 if html_flag: 

229 push_data["html"] = 1 

230 if ttl_raw is not None: 

231 try: 

232 push_data["ttl"] = int(ttl_raw) 

233 except TypeError, ValueError: 

234 _LOGGER.warning("SUPERNOTIFY pushover: invalid pushover_ttl %r, ignored", ttl_raw) 

235 if tags: 

236 push_data["tags"] = [str(t) for t in ensure_list(tags)] 

237 if device: 

238 # the integration takes devices as the notify target, and ignores a `device` in data 

239 action_data[ATTR_TARGET] = [str(d) for d in ensure_list(device)] 

240 

241 # --- Camera image attachment via envelope.grab_image() (v1.14.0+) --- 

242 if attach_image: 

243 image_path = await self.safe_grab_image(envelope) 

244 if image_path: 

245 push_data["attachment"] = str(image_path) 

246 _LOGGER.debug("SUPERNOTIFY pushover: attaching image %s", image_path) 

247 

248 action_data[ATTR_DATA] = push_data 

249 

250 # Remaining raw_data is NOT forwarded - Pushover HA service schema is fixed 

251 return await self.call_action(envelope, action_data=action_data)