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

1"""Pushover transport for SuperNotify. 

2 

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

4Requires the official HA Pushover integration configured in configuration.yaml: 

5 

6 notify: 

7 - name: pushover_home 

8 platform: pushover 

9 api_key: YOUR_PUSHOVER_API_KEY 

10 user_key: YOUR_PUSHOVER_USER_KEY 

11 

12The notify service name (e.g. notify.pushover_home) MUST be specified as 

13`action:` on the delivery - there is no default, since the name depends on 

14the user's configuration.yaml entry. 

15 

16Priority mapping (SuperNotify -> Pushover integer): 

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

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

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

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

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

22 

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

24parameters. If not provided, SuperNotify supplies sensible defaults 

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

26 

27Supported data keys (all optional unless noted): 

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

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

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

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

32 pushover_url str Supplementary URL attached to the notification. 

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

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

35 Emergency only (priority=2). 

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

37 default 3600). Emergency only. 

38 pushover_callback str Public URL for emergency acknowledgment webhook 

39 (HA webhook endpoint). 

40 pushover_html bool Enable HTML formatting in the message 

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

42 pushover_ttl int Seconds before automatic deletion of the 

43 notification from the device. 

44 pushover_device str Send to a specific device (device name as 

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

46 Default: all devices on the account. 

47 pushover_attach_image bool Grab camera snapshot (uses 

48 media.camera_entity_id from the SuperNotify 

49 call) and attach it to the notification. 

50 Requires TransportFeature.SNAPSHOT_IMAGE. 

51""" 

52 

53from __future__ import annotations 

54 

55import logging 

56from typing import TYPE_CHECKING, Any, ClassVar 

57 

58from homeassistant.components.notify.const import ATTR_DATA 

59from homeassistant.helpers.typing import ConfigType 

60 

61from custom_components.supernotify.common import boolify 

62from custom_components.supernotify.const import TRANSPORT_PUSHOVER 

63from custom_components.supernotify.model import ( 

64 DebugTrace, 

65 TargetRequired, 

66 TransportConfig, 

67 TransportFeature, 

68) 

69from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

70from custom_components.supernotify.transport import Transport 

71 

72if TYPE_CHECKING: 

73 from custom_components.supernotify.envelope import Envelope 

74 from custom_components.supernotify.hass_api import HomeAssistantAPI 

75 

76_LOGGER = logging.getLogger(__name__) 

77 

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

79_PRIORITY_MAP: dict[str, int] = { 

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

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

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

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

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

85} 

86 

87_EMERGENCY_PRIORITY = 2 

88_EMERGENCY_RETRY_MIN = 30 # seconds (Pushover API limit) 

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

90_EMERGENCY_RETRY_DEFAULT = 60 # sensible default when not specified 

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

92 

93 

94class PushoverTransport(Transport): 

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

96 

97 name = TRANSPORT_PUSHOVER 

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 = TransportConfig() 

110 config.delivery_defaults.target_required = TargetRequired.NEVER 

111 config.delivery_defaults.inclusion = self.inclusion_mode 

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

113 return config 

114 

115 def is_viable(self, hass_api: HomeAssistantAPI) -> bool: 

116 # a manually configured delivery can set its own action: notify.<name> regardless 

117 # of whether the service is discoverable here - is_viable() can't see delivery-level 

118 # config, so it can't rule that out; DeliveryRegistry prunes this transport entirely 

119 # once it's confirmed no delivery (explicit or auto) uses it 

120 return True 

121 

122 def build_standard_deliveries(self, hass_api: HomeAssistantAPI) -> dict[str, ConfigType]: 

123 if self.delivery_defaults.action: 

124 return {self.name: {}} 

125 return {} 

126 

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

128 if action and action.startswith("notify."): 

129 return True 

130 _LOGGER.warning( 

131 "SUPERNOTIFY pushover: action must be a notify.* service (e.g. notify.pushover_home), got: %r", 

132 action, 

133 ) 

134 return False 

135 

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

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

138 

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

140 

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

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

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

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

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

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

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

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

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

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

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

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

153 

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

155 priority_ovr: int | None = None 

156 if priority_ovr_raw is not None: 

157 try: 

158 priority_ovr = int(priority_ovr_raw) 

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

160 _LOGGER.warning( 

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

162 priority_ovr, 

163 ) 

164 priority_ovr = None 

165 except (TypeError, ValueError): # py3.13 compat 

166 _LOGGER.warning( 

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

168 priority_ovr_raw, 

169 ) 

170 priority_ovr = None 

171 

172 pushover_priority: int = ( 

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

174 ) 

175 

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

177 action_data = envelope.core_action_data() 

178 

179 # --- Pushover-specific data payload --- 

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

181 

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

183 if pushover_priority == _EMERGENCY_PRIORITY: 

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

185 if retry_raw is None: 

186 retry_val: int = _EMERGENCY_RETRY_DEFAULT 

187 else: 

188 try: 

189 retry_val = int(retry_raw) 

190 except (TypeError, ValueError): 

191 _LOGGER.warning( 

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

193 retry_raw, 

194 _EMERGENCY_RETRY_DEFAULT, 

195 ) 

196 retry_val = _EMERGENCY_RETRY_DEFAULT 

197 

198 # expire: same robust pattern 

199 if expire_raw is None: 

200 expire_val: int = _EMERGENCY_EXPIRE_DEFAULT 

201 else: 

202 try: 

203 expire_val = int(expire_raw) 

204 except (TypeError, ValueError): 

205 _LOGGER.warning( 

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

207 expire_raw, 

208 _EMERGENCY_EXPIRE_DEFAULT, 

209 ) 

210 expire_val = _EMERGENCY_EXPIRE_DEFAULT 

211 

212 if retry_val < _EMERGENCY_RETRY_MIN: 

213 _LOGGER.warning( 

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

215 retry_val, 

216 _EMERGENCY_RETRY_MIN, 

217 ) 

218 retry_val = _EMERGENCY_RETRY_MIN 

219 

220 if expire_val > _EMERGENCY_EXPIRE_MAX: 

221 _LOGGER.warning( 

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

223 expire_val, 

224 _EMERGENCY_EXPIRE_MAX, 

225 ) 

226 expire_val = _EMERGENCY_EXPIRE_MAX 

227 

228 push_data["retry"] = retry_val 

229 push_data["expire"] = expire_val 

230 

231 if callback: 

232 push_data["callback"] = callback 

233 

234 _LOGGER.debug( 

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

236 retry_val, 

237 expire_val, 

238 ) 

239 

240 # Optional fields - added only when present 

241 if sound: 

242 push_data["sound"] = sound 

243 if url: 

244 push_data["url"] = url 

245 if url_title: 

246 push_data["url_title"] = url_title 

247 if html_flag: 

248 push_data["html"] = 1 

249 if ttl_raw is not None: 

250 try: 

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

252 except (TypeError, ValueError): 

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

254 if device: 

255 push_data["device"] = device 

256 

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

258 if attach_image: 

259 try: 

260 image_path = await envelope.grab_image() 

261 if image_path: 

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

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

264 except Exception as e: 

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

266 

267 action_data[ATTR_DATA] = push_data 

268 

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

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