Coverage for custom_components/supernotify/transports/matrix.py: 99%

1"""Matrix transport for SuperNotify. 

2 

3Sends messages to Matrix rooms using Home Assistant's `matrix` integration, 

4calling the native `matrix.send_message` service (not the thin legacy notify 

5wrapper), for granular control over format, images and threads. 

6 

7Supported data keys (all optional): 

8 matrix_format str "text" | "html" (default: "html" when a 

9 title is present, otherwise "text") 

10 matrix_thread_id str Send the message into a Matrix thread 

11 matrix_attach_image bool Attach camera snapshot (default: False) 

12 matrix_priority_prefix bool Prefix message with an emoji derived from 

13 the SuperNotify priority (default: False): 

14 critical=siren, high=warning, 

15 low/minimum=small diamond, medium=none 

16 

17Notes on the HA `matrix.send_message` service schema: 

18- The `data` sub-dict is STRICT (no ALLOW_EXTRA): only `format`, `images` 

19 and `thread_id` are accepted, any other key makes the whole call fail 

20 with `vol.Invalid`. Residual generic data keys are therefore NOT merged 

21 into the payload (they are dropped with a debug log), unlike the standard 

22 transport pattern. 

23- `target` is required and every entry must match the room regex 

24 `^[!|#][^:]*:.*` (room ID `!abc:server` or alias `#name:server`). A single 

25 invalid entry rejects the whole call, so targets are pre-filtered here and 

26 invalid ones are dropped with a debug log. Prefer room IDs, or aliases 

27 listed in the `rooms:` config of the matrix integration. 

28- There is no `title` field: the title is composed into the message body 

29 (bold + line break for html, plain line break for text). 

30- With `format: html` the core sets both `formatted_body` and plain `body` 

31 to the same string (no markup strip), so clients without HTML support 

32 will show raw tags. This mirrors core behaviour and is accepted. 

33- `images` is a list of LOCAL paths and the core checks 

34 `hass.config.is_allowed_path()`: the SuperNotify media path must be listed 

35 in `homeassistant.allowlist_external_dirs`, otherwise the image is dropped 

36 by the integration (the text message is still sent first). 

37- Matrix has no native message priority: the only priority mapping offered 

38 is the opt-in emoji prefix above. 

39""" 

40 

41from __future__ import annotations 

42 

43import html 

44import logging 

45import re 

46from typing import TYPE_CHECKING, Any, ClassVar 

47 

48from homeassistant.helpers.typing import ConfigType 

49 

50from custom_components.supernotify.common import boolify 

51from custom_components.supernotify.const import ATTR_DATA, ATTR_MATRIX_ROOM, TRANSPORT_MATRIX 

52from custom_components.supernotify.model import ( 

53 DebugTrace, 

54 TargetRequired, 

55 TransportConfig, 

56 TransportFeature, 

57) 

58from custom_components.supernotify.options import MEDIA_OPTIONS, DeliveryOption 

59from custom_components.supernotify.target import TargetEntityCategory 

60from custom_components.supernotify.transport import Transport 

61 

62if TYPE_CHECKING: 

63 from custom_components.supernotify.envelope import Envelope 

64 from custom_components.supernotify.hass_api import HomeAssistantAPI 

65 

66_LOGGER = logging.getLogger(__name__) 

67 

68# Slightly stricter subset of the core service regex `^[!|#][^:]*:.*`: 

69# require a room ID (!) or alias (#) sigil and a non-empty server part. 

70_MATRIX_ROOM_RE = re.compile(r"^[!#][^:]*:.+") 

71 

72_VALID_FORMATS = ("text", "html") 

73 

74# Opt-in emoji prefix per SuperNotify priority (medium: no prefix) 

75_PRIORITY_PREFIX = { 

76 "critical": "\U0001f6a8 ", # police car light 

77 "high": "⚠️ ", # warning sign 

78 "low": "\U0001f539 ", # small blue diamond 

79 "minimum": "\U0001f539 ", # small blue diamond 

80} 

81 

82 

83class MatrixTransport(Transport): 

84 """Notify via Matrix rooms using Home Assistant matrix integration.""" 

85 

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

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

88 

89 name = TRANSPORT_MATRIX 

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

91 

92 @property 

93 def supported_features(self) -> TransportFeature: 

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

95 

96 @property 

97 def default_config(self) -> TransportConfig: 

98 config = TransportConfig() 

99 config.delivery_defaults.action = "matrix.send_message" 

100 config.delivery_defaults.target_required = TargetRequired.ALWAYS 

101 config.delivery_defaults.inclusion = self.inclusion_mode 

102 return config 

103 

104 @property 

105 def target_categories(self) -> list[str | TargetEntityCategory]: 

106 # a Matrix room ID/alias has no shape distinct enough for automatic matching, so 

107 # it's only ever reachable here via explicit qualification (prefix, mapping, or 

108 # this transport's/a delivery's own name) - select_rooms() below still validates 

109 # the shape itself once it arrives 

110 return [ATTR_MATRIX_ROOM] 

111 

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

113 """Validate that action is the matrix send_message service.""" 

114 return action == "matrix.send_message" 

115 

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

117 # matrix is YAML-configured (no config entry); the service only registers once 

118 # the bot has connected, so check for it directly 

119 return hass_api.has_service("matrix", "send_message") 

120 

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

122 return {self.name: {}} 

123 

124 def select_rooms(self, envelope: Envelope) -> list[str]: 

125 """Filter envelope targets down to valid Matrix room IDs or aliases. 

126 

127 The service rejects the whole call if any target fails the room 

128 regex, so invalid entries are dropped here (with a debug log) instead 

129 of being forwarded. Duplicates are removed preserving order. 

130 """ 

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

132 rooms: list[str] = [] 

133 for target in raw_targets: 

134 if isinstance(target, str) and _MATRIX_ROOM_RE.match(target): 

135 if target not in rooms: 

136 rooms.append(target) 

137 else: 

138 _LOGGER.debug("SUPERNOTIFY matrix: skipping invalid room target %r", target) 

139 return rooms 

140 

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

142 _LOGGER.debug("SUPERNOTIFY matrix %s", envelope.message) 

143 

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

145 

146 # Pop Matrix-specific data keys 

147 format_override = raw_data.pop("matrix_format", None) 

148 thread_id = raw_data.pop("matrix_thread_id", None) 

149 attach_image = boolify(raw_data.pop("matrix_attach_image", False), default=False) 

150 priority_prefix = boolify(raw_data.pop("matrix_priority_prefix", False), default=False) 

151 

152 # Resolve and pre-validate room targets 

153 rooms = self.select_rooms(envelope) 

154 if not rooms: 

155 _LOGGER.warning("SUPERNOTIFY matrix: no valid room targets (expected !room:server or #alias:server)") 

156 self.record_error("no valid Matrix room targets", "deliver") 

157 return False 

158 

159 # Resolve format: explicit override, or html when a title must be 

160 # composed in bold, plain text otherwise 

161 default_format = "html" if envelope.title else "text" 

162 if format_override: 

163 fmt = str(format_override).lower() 

164 if fmt not in _VALID_FORMATS: 

165 _LOGGER.warning("SUPERNOTIFY matrix: invalid matrix_format '%s', using '%s'", format_override, default_format) 

166 fmt = default_format 

167 else: 

168 fmt = default_format 

169 

170 # Compose title into the message body (the service has no title field). 

171 # Only the title is escaped in html mode: the body may already contain 

172 # HTML the user wrote intentionally (consistent with core behaviour). 

173 message_text = envelope.message or "" 

174 if envelope.title: 

175 if fmt == "html": 

176 message_text = f"<b>{html.escape(envelope.title)}</b><br>{message_text}" 

177 else: 

178 message_text = f"{envelope.title}\n{message_text}" 

179 

180 # Opt-in emoji prefix mapped from SuperNotify priority 

181 if priority_prefix: 

182 message_text = _PRIORITY_PREFIX.get(envelope.priority or "medium", "") + message_text 

183 

184 # Grab camera snapshot if requested; images are local paths and the 

185 # matrix integration checks them against allowlist_external_dirs 

186 images: list[str] = [] 

187 if attach_image: 

188 image_path = None 

189 try: 

190 image_path = await envelope.grab_image() 

191 except Exception as e: 

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

193 if image_path: 

194 images.append(str(image_path)) 

195 else: 

196 _LOGGER.debug("SUPERNOTIFY matrix: no image available, sending text only") 

197 

198 # Build the payload. The service data sub-dict is whitelist-only 

199 # (format / images / thread_id): residual generic data keys are NOT 

200 # merged, they would fail the whole call as extra keys. 

201 action_data: dict[str, Any] = { 

202 "message": message_text, 

203 "target": rooms, 

204 } 

205 service_data: dict[str, Any] = {"format": fmt} 

206 if thread_id: 

207 service_data["thread_id"] = str(thread_id) 

208 if images: 

209 service_data["images"] = images 

210 action_data[ATTR_DATA] = service_data 

211 

212 if raw_data: 

213 _LOGGER.debug( 

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

215 sorted(raw_data), 

216 ) 

217 

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