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

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

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 

75class MatrixTransport(Transport): 

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

77 

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

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

80 

81 name = TRANSPORT_MATRIX 

82 integration_domain = "matrix" 

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

84 

85 @property 

86 def supported_features(self) -> TransportFeature: 

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

88 

89 @property 

90 def default_config(self) -> TransportConfig: 

91 config = super().default_config 

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

93 return config 

94 

95 @property 

96 def unique_target_categories(self) -> list[str | TargetEntityCategory]: 

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

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

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

100 # the shape itself once it arrives. matrix_room is nonetheless a dedicated category 

101 # name no other transport uses, so a value explicitly qualified with it is 

102 # unambiguous - safe to auto-select on 

103 return [ATTR_MATRIX_ROOM] 

104 

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

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

107 return action == "matrix.send_message" 

108 

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

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

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

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

113 

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

115 return {self.name: {}} 

116 

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

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

119 

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

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

122 of being forwarded. Duplicates are removed preserving order. 

123 """ 

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

125 rooms: list[str] = [] 

126 for target in raw_targets: 

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

128 if target not in rooms: 

129 rooms.append(target) 

130 else: 

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

132 return rooms 

133 

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

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

136 

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

138 

139 # Pop Matrix-specific data keys 

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

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

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

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

144 

145 # Resolve and pre-validate room targets 

146 rooms = self.select_rooms(envelope) 

147 if not rooms: 

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

149 envelope.skipped = 1 

150 envelope.skip_reason = SuppressionReason.NO_TARGET 

151 return False 

152 

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

154 # composed in bold, plain text otherwise 

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

156 if format_override: 

157 fmt = str(format_override).lower() 

158 if fmt not in _VALID_FORMATS: 

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

160 fmt = default_format 

161 else: 

162 fmt = default_format 

163 

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

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

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

167 message_text = envelope.message or "" 

168 if envelope.title: 

169 if fmt == "html": 

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

171 else: 

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

173 

174 # Opt-in emoji prefix mapped from SuperNotify priority 

175 if priority_prefix: 

176 message_text = self.priority_prefix(envelope) + message_text 

177 

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

179 # matrix integration checks them against allowlist_external_dirs 

180 images: list[str] = [] 

181 if attach_image: 

182 image_path = await self.safe_grab_image(envelope) 

183 if image_path: 

184 images.append(str(image_path)) 

185 else: 

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

187 

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

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

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

191 action_data: dict[str, Any] = { 

192 "message": message_text, 

193 "target": rooms, 

194 } 

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

196 if thread_id: 

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

198 if images: 

199 service_data["images"] = images 

200 action_data[ATTR_DATA] = service_data 

201 

202 if raw_data: 

203 _LOGGER.debug( 

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

205 sorted(raw_data), 

206 ) 

207 

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