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

1"""Gotify transport for SuperNotify. 

2 

3Sends push notifications via Gotify (self-hosted, privacy-first push server). 

4Requires the HACS custom integration 1RandomDev/homeassistant-gotify installed 

5and configured in configuration.yaml. The notify service name (e.g. notify.gotify) 

6depends on the user's HACS configuration - it MUST be specified as `action:` in 

7delivery.yaml (no default exists). 

8 

9Prerequisites: 

10 - Gotify server running and reachable 

11 - HACS integration 1RandomDev/homeassistant-gotify installed 

12 - Application token configured in configuration.yaml 

13 - `action: notify.<name>` set in delivery.yaml (REQUIRED) 

14 

15For snapshot camera / bigImageUrl: 

16 - `media_web_path` must be configured (PLATFORM_SCHEMA) for grab_image to produce a URL. 

17 - `external_url` must be configured in HA for the URL to be reachable outside the local network. 

18 

19Supported data: keys (all optional): 

20 gotify_priority int (0-10) Override priority (0=silent ... 10=max). 

21 Accepts string "7" -> cast to int. 

22 Out-of-range values are clamped. 

23 gotify_click str (URL) URL opened on tap of the notification. 

24 gotify_image_url str (URL) Direct URL for bigImageUrl (expanded image). 

25 Takes precedence over gotify_attach_image. 

26 gotify_attach_image bool Grab image via shared pipeline and use as bigImageUrl. 

27 Used only when no snapshot_url is already in media. 

28 Requires media_web_path configured and image saved within it. 

29 gotify_markdown bool Enable Markdown rendering (text/markdown). 

30 Accepts "true"/"false" YAML strings safely. 

31 gotify_intent_url str (URL) Android intent URL on message receive. 

32 Requires "Intent Action Permission" in Gotify app. 

33 

34Priority mapping (SuperNotify -> Gotify integer): 

35 critical -> 10 high -> 7 medium -> 5 low -> 2 minimum -> 0 

36""" 

37 

38from __future__ import annotations 

39 

40import logging 

41from typing import TYPE_CHECKING, Any 

42 

43from homeassistant.components.notify.const import ATTR_DATA 

44 

45from custom_components.supernotify.common import boolify 

46from custom_components.supernotify.const import ( 

47 ATTR_MEDIA_SNAPSHOT_URL, 

48 TRANSPORT_GOTIFY, 

49) 

50from custom_components.supernotify.model import ( 

51 DebugTrace, 

52 TargetRequired, 

53 TransportConfig, 

54 TransportFeature, 

55) 

56from custom_components.supernotify.transport import Transport 

57 

58if TYPE_CHECKING: 

59 from custom_components.supernotify.envelope import Envelope 

60 

61_LOGGER = logging.getLogger(__name__) 

62 

63_PRIORITY_MAP: dict[str, int] = { 

64 "critical": 10, 

65 "high": 7, 

66 "medium": 5, 

67 "low": 2, 

68 "minimum": 0, 

69} 

70 

71 

72def _build_extras( 

73 click_url: str | None, 

74 image_url: str | None, 

75 markdown: bool, 

76 intent_url: str | None, 

77) -> dict | None: 

78 """Build Gotify extras dict. Returns None if no extras are needed.""" 

79 extras: dict = {} 

80 

81 client_notification: dict = {} 

82 if click_url: 

83 client_notification["click"] = {"url": click_url} 

84 if image_url: 

85 client_notification["bigImageUrl"] = image_url 

86 if client_notification: 

87 extras["client::notification"] = client_notification 

88 

89 if markdown: 

90 extras["client::display"] = {"contentType": "text/markdown"} 

91 

92 if intent_url: 

93 extras["android::action"] = {"onReceive": {"intentUrl": intent_url}} 

94 

95 return extras if extras else None 

96 

97 

98class GotifyTransport(Transport): 

99 """Notify via Gotify self-hosted push notification server.""" 

100 

101 name = TRANSPORT_GOTIFY 

102 

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

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

105 

106 @property 

107 def supported_features(self) -> TransportFeature: 

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

109 

110 @property 

111 def default_config(self) -> TransportConfig: 

112 config = TransportConfig() 

113 config.delivery_defaults.target_required = TargetRequired.NEVER 

114 # No default action - user MUST specify action: notify.<name> in delivery.yaml 

115 return config 

116 

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

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

119 return True 

120 _LOGGER.warning( 

121 "SUPERNOTIFY gotify: action must be a notify.* service (e.g. notify.gotify), got: %r", 

122 action, 

123 ) 

124 return False 

125 

126 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # noqa: ARG002 

127 _LOGGER.debug("SUPERNOTIFY gotify %s", envelope.message) 

128 

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

130 

131 # --- Extract gotify_* keys (must not reach the notify service) --- 

132 priority_ovr_raw = raw_data.pop("gotify_priority", None) 

133 click_url = raw_data.pop("gotify_click", None) 

134 image_url: str | None = raw_data.pop("gotify_image_url", None) 

135 attach_image = boolify(raw_data.pop("gotify_attach_image", False), default=False) 

136 markdown = boolify(raw_data.pop("gotify_markdown", False), default=False) 

137 intent_url = raw_data.pop("gotify_intent_url", None) 

138 

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

140 priority_ovr: int | None = None 

141 if priority_ovr_raw is not None: 

142 try: 

143 priority_ovr = int(priority_ovr_raw) 

144 if not 0 <= priority_ovr <= 10: 

145 _LOGGER.warning( 

146 "SUPERNOTIFY gotify: gotify_priority %d out of range 0-10, clamping", 

147 priority_ovr, 

148 ) 

149 priority_ovr = max(0, min(10, priority_ovr)) 

150 except (TypeError, ValueError) as e: # py3.13 compat 

151 _LOGGER.warning("SUPERNOTIFY gotify: invalid gotify_priority %r, using auto mapping: %s", priority_ovr_raw, e) 

152 priority_ovr = None 

153 

154 gotify_priority: int = priority_ovr if priority_ovr is not None else _PRIORITY_MAP.get(envelope.priority or "medium", 5) 

155 

156 # --- Base action data --- 

157 action_data = envelope.core_action_data() 

158 

159 # --- Resolve image_url (bigImageUrl): explicit > snapshot_url > grab_image --- 

160 if not image_url and envelope.media: 

161 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) 

162 if snapshot_url: 

163 image_url = self.hass_api.abs_url(snapshot_url) 

164 elif attach_image: 

165 image_path = await envelope.grab_image() 

166 if image_path: 

167 image_url = await self.context.media_storage.object_url(image_path) 

168 

169 # --- Build nested payload_data --- 

170 payload_data: dict[str, Any] = {"priority": gotify_priority} 

171 

172 extras = _build_extras(click_url, image_url, markdown, intent_url) 

173 if extras: 

174 payload_data["extras"] = extras 

175 

176 action_data[ATTR_DATA] = payload_data 

177 

178 # raw_data residuo non passato - schema HACS e' fisso 

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