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

88 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-09-25 21:14 +0000

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 - auto_configure() discovers it via the 

7registered service's module, or it can be set manually as `action:` on a delivery. 

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 on a delivery only if auto-discovery doesn't apply 

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, ClassVar 

42 

43from homeassistant.components.notify.const import ATTR_DATA 

44from homeassistant.helpers.typing import ConfigType 

45 

46from custom_components.supernotify.common import boolify 

47from custom_components.supernotify.const import ATTR_MEDIA_SNAPSHOT_URL, TRANSPORT_GOTIFY 

48from custom_components.supernotify.model import ( 

49 DebugTrace, 

50 TargetRequired, 

51 TransportConfig, 

52 TransportFeature, 

53) 

54from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

55from custom_components.supernotify.transport import Transport 

56 

57if TYPE_CHECKING: 

58 from custom_components.supernotify.envelope import Envelope 

59 from custom_components.supernotify.hass_api import HomeAssistantAPI 

60 

61# Gotify's HACS notify platform module (no longer bundled with HA core) 

62HA_GOTIFY_MODULE = "custom_components.gotify.notify" 

63 

64_LOGGER = logging.getLogger(__name__) 

65 

66_PRIORITY_MAP: dict[str, int] = { 

67 "critical": 10, 

68 "high": 7, 

69 "medium": 5, 

70 "low": 2, 

71 "minimum": 0, 

72} 

73 

74 

75def _build_extras( 

76 click_url: str | None, 

77 image_url: str | None, 

78 markdown: bool, 

79 intent_url: str | None, 

80) -> dict | None: 

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

82 extras: dict = {} 

83 

84 client_notification: dict = {} 

85 if click_url: 

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

87 if image_url: 

88 client_notification["bigImageUrl"] = image_url 

89 if client_notification: 

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

91 

92 if markdown: 

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

94 

95 if intent_url: 

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

97 

98 return extras or None 

99 

100 

101class GotifyTransport(Transport): 

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

103 

104 name = TRANSPORT_GOTIFY 

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

106 

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

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

109 

110 @property 

111 def supported_features(self) -> TransportFeature: 

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

113 

114 @property 

115 def default_config(self) -> TransportConfig: 

116 config = TransportConfig() 

117 config.delivery_defaults.target_required = TargetRequired.NEVER 

118 config.delivery_defaults.inclusion = self.inclusion_mode 

119 config.delivery_defaults.action = self.hass_api.find_service("notify", HA_GOTIFY_MODULE) 

120 return config 

121 

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

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

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

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

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

127 return True 

128 

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

130 if self.delivery_defaults.action: 

131 return {self.name: {}} 

132 return {} 

133 

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

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

136 return True 

137 _LOGGER.warning( 

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

139 action, 

140 ) 

141 return False 

142 

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

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

145 

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

147 

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

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

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

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

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

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

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

155 

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

157 priority_ovr: int | None = None 

158 if priority_ovr_raw is not None: 

159 try: 

160 priority_ovr = int(priority_ovr_raw) 

161 if not 0 <= priority_ovr <= 10: 

162 _LOGGER.warning( 

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

164 priority_ovr, 

165 ) 

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

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

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

169 priority_ovr = None 

170 

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

172 

173 # --- Base action data --- 

174 action_data = envelope.core_action_data() 

175 

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

177 if not image_url and envelope.media: 

178 snapshot_url = envelope.media.get(ATTR_MEDIA_SNAPSHOT_URL) 

179 if snapshot_url: 

180 image_url = self.hass_api.abs_url(snapshot_url) 

181 elif attach_image: 

182 image_path = await envelope.grab_image() 

183 if image_path: 

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

185 

186 # --- Build nested payload_data --- 

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

188 

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

190 if extras: 

191 payload_data["extras"] = extras 

192 

193 action_data[ATTR_DATA] = payload_data 

194 

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

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