Coverage for custom_components/supernotify/delivery_method.py: 90%

1# mypy: disable-error-code="name-defined" 

2 

3import logging 

4import time 

5from abc import abstractmethod 

6from dataclasses import asdict 

7from traceback import format_exception 

8from typing import Any 

9from urllib.parse import urlparse 

10 

11from homeassistant.components.notify.const import ATTR_TARGET 

12from homeassistant.const import CONF_ACTION, CONF_CONDITION, CONF_DEFAULT, CONF_METHOD, CONF_NAME, CONF_OPTIONS, CONF_TARGET 

13from homeassistant.core import HomeAssistant 

14from homeassistant.helpers import condition 

15 

16from custom_components.supernotify.common import CallRecord 

17from custom_components.supernotify.configuration import Context 

18 

19from . import ( 

20 CONF_DATA, 

21 CONF_DEVICE_DISCOVERY, 

22 CONF_DEVICE_DOMAIN, 

23 CONF_TARGETS_REQUIRED, 

24 RESERVED_DELIVERY_NAMES, 

25 ConditionVariables, 

26 MessageOnlyPolicy, 

27) 

28 

29_LOGGER = logging.getLogger(__name__) 

30 

31OPTION_SIMPLIFY_TEXT = "simplify_text" 

32OPTION_STRIP_URLS = "strip_urls" 

33OPTION_MESSAGE_USAGE = "message_usage" 

34OPTIONS_WITH_DEFAULTS: dict[str, str | bool] = { 

35 OPTION_SIMPLIFY_TEXT: False, 

36 OPTION_STRIP_URLS: False, 

37 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

38} 

39 

40 

41class DeliveryMethod: 

42 """Base class for delivery methods. 

43 

44 Sub classes integrste with Home Assistant notification services 

45 or alternative notification mechanisms. 

46 """ 

47 

48 method: str 

49 

50 @abstractmethod 

51 def __init__( 

52 self, 

53 hass: HomeAssistant, 

54 context: Context, 

55 deliveries: dict[str, Any] | None = None, 

56 default: dict[str, Any] | None = None, 

57 targets_required: bool = True, 

58 device_domain: list[str] | None = None, 

59 device_discovery: bool = False, 

60 ) -> None: 

61 self.hass: HomeAssistant = hass 

62 self.context: Context = context 

63 self.default: dict[str, Any] = default or {} 

64 self.default_options: dict[str, Any] = self.default.get(CONF_OPTIONS) or {} 

65 self.default_action: str | None = self.default.get(CONF_ACTION) 

66 self.targets_required: bool = targets_required 

67 self.device_domain: list[str] = device_domain or [] 

68 self.device_discovery: bool = device_discovery 

69 

70 self.default_delivery: dict[str, Any] | None = None 

71 self.valid_deliveries: dict[str, dict[str, Any]] = {} 

72 self.method_deliveries: dict[str, dict[str, Any]] = ( 

73 {d: dc for d, dc in deliveries.items() if dc.get(CONF_METHOD) == self.method} if deliveries else {} 

74 ) 

75 

76 async def initialize(self) -> None: 

77 """Async post-construction initialization""" 

78 if self.method is None: 

79 raise OSError("No delivery method configured") 

80 self.valid_deliveries = await self.validate_deliveries() 

81 if self.device_discovery: 

82 self.default.setdefault(CONF_TARGET, []) 

83 for domain in self.device_domain: 

84 discovered: int = 0 

85 added: int = 0 

86 for d in self.context.discover_devices(domain): 

87 discovered += 1 

88 if d.id not in self.default[CONF_TARGET]: 

89 _LOGGER.info(f"SUPERNOTIFY Discovered device {d.name} for {domain}, id {d.id}") 

90 self.default[CONF_TARGET].append(d.id) 

91 added += 1 

92 

93 _LOGGER.info(f"SUPERNOTIFY device discovery for {domain} found {discovered} devices, added {added} new ones") 

94 

95 @property 

96 def targets(self) -> list[str]: 

97 return self.default.get(CONF_TARGET) or [] 

98 

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

100 """Override in subclass if delivery method has fixed action or doesn't require one""" 

101 return action is None or action.startswith("notify.") 

102 

103 async def validate_deliveries(self) -> dict[str, dict[str, Any]]: 

104 """Validate list of deliveries at startup for this method""" 

105 valid_deliveries: dict[str, dict[str, Any]] = {} 

106 for d, dc in self.method_deliveries.items(): 

107 # don't care about ENABLED here since disabled deliveries can be overridden 

108 if d in RESERVED_DELIVERY_NAMES: 

109 _LOGGER.warning("SUPERNOTIFY Delivery uses reserved word %s", d) 

110 continue 

111 if not self.validate_action(dc.get(CONF_ACTION)): 

112 _LOGGER.warning("SUPERNOTIFY Invalid action definition for delivery %s (%s)", d, dc.get(CONF_ACTION)) 

113 continue 

114 delivery_condition = dc.get(CONF_CONDITION) 

115 if delivery_condition: 

116 if not await condition.async_validate_condition_config(self.hass, delivery_condition): 

117 _LOGGER.warning("SUPERNOTIFY Invalid delivery condition for %s: %s", d, delivery_condition) 

118 continue 

119 

120 valid_deliveries[d] = dc 

121 dc[CONF_NAME] = d 

122 

123 if dc.get(CONF_DEFAULT) and not self.default_delivery: 

124 # pick the first delivery with default flag set as the default 

125 self.default_delivery = dc 

126 elif dc.get(CONF_DEFAULT) and self.default_delivery and dc.get(CONF_NAME) != self.default_delivery.get(CONF_NAME): 

127 _LOGGER.debug("SUPERNOTIFY Multiple default deliveries, skipping %s", d) 

128 

129 if not self.default_delivery: 

130 method_definition = self.default 

131 if method_definition: 

132 _LOGGER.info("SUPERNOTIFY Building default delivery for %s from method %s", self.method, method_definition) 

133 self.default_delivery = method_definition 

134 else: 

135 _LOGGER.debug("SUPERNOTIFY No default delivery or method_definition for method %s", self.method) 

136 

137 if self.default_action is None and self.default_delivery: 

138 self.default_action = self.default_delivery.get(CONF_ACTION) 

139 _LOGGER.debug("SUPERNOTIFY Setting default action for method %s to %s", self.method, self.default_action) 

140 else: 

141 _LOGGER.debug("SUPERNOTIFY No default action for method %s", self.method) 

142 

143 _LOGGER.debug( 

144 "SUPERNOTIFY Validated method %s, default delivery %s, default action %s, valid deliveries: %s", 

145 self.method, 

146 self.default_delivery, 

147 self.default_action, 

148 valid_deliveries, 

149 ) 

150 return valid_deliveries 

151 

152 def attributes(self) -> dict[str, Any]: 

153 return { 

154 CONF_METHOD: self.method, 

155 CONF_TARGETS_REQUIRED: self.targets_required, 

156 CONF_DEVICE_DOMAIN: self.device_domain, 

157 CONF_DEVICE_DISCOVERY: self.device_discovery, 

158 CONF_DEFAULT: self.default, 

159 "default_delivery": self.default_delivery, 

160 "deliveries": list(self.valid_deliveries.keys()), 

161 } 

162 

163 @abstractmethod 

164 async def deliver(self, envelope: "Envelope") -> bool: # noqa: F821 # type: ignore 

165 """Delivery implementation 

166 

167 Args: 

168 ---- 

169 envelope (Envelope): envelope to be delivered 

170 

171 """ 

172 

173 def select_target(self, target: str) -> bool: # noqa: ARG002 

174 """Confirm if target appropriate for this delivery method 

175 

176 Args: 

177 ---- 

178 target (str): Target, typically an entity ID, or an email address, phone number 

179 

180 """ 

181 return True 

182 

183 def recipient_target(self, recipient: dict[str, Any]) -> list[str]: # noqa: ARG002 

184 """Pick out delivery appropriate target from a single person's (recipient) config""" 

185 return [] 

186 

187 def delivery_config(self, delivery_name: str) -> dict[str, Any]: 

188 config = self.context.deliveries.get(delivery_name) or self.default_delivery or {} 

189 config = dict(config) 

190 config[CONF_DATA] = dict(config.get(CONF_DATA) or {}) 

191 return config 

192 

193 def set_action_data(self, action_data: dict[str, Any], key: str, data: Any | None) -> Any: 

194 if data is not None: 

195 action_data[key] = data 

196 return action_data 

197 

198 def option(self, option_name: str, delivery_config: dict[str, Any]) -> str | bool: 

199 """Get an option value from delivery config or method default options""" 

200 opt: str | bool | None = None 

201 if CONF_OPTIONS in delivery_config and option_name in delivery_config[CONF_OPTIONS]: 

202 opt = delivery_config[CONF_OPTIONS][option_name] 

203 if opt is None: 

204 opt = self.default_options.get(option_name) 

205 if opt is None: 

206 opt = OPTIONS_WITH_DEFAULTS.get(option_name) 

207 if opt is None: 

208 _LOGGER.warning("SUPERNOTIFY No default for option %s, setting to empty string", option_name) 

209 opt = "" 

210 return opt 

211 

212 def option_bool(self, option_name: str, delivery_config: dict[str, Any]) -> bool: 

213 return bool(self.option(option_name, delivery_config)) 

214 

215 def option_str(self, option_name: str, delivery_config: dict[str, Any]) -> str: 

216 return str(self.option(option_name, delivery_config)) 

217 

218 async def evaluate_delivery_conditions( 

219 self, delivery_config: dict[str, Any], condition_variables: ConditionVariables | None 

220 ) -> bool | None: 

221 if CONF_CONDITION not in delivery_config: 

222 return True 

223 cond_conf = delivery_config.get(CONF_CONDITION) 

224 if cond_conf is None: 

225 return True 

226 

227 try: 

228 test = await condition.async_from_config(self.hass, cond_conf) 

229 return test(self.hass, asdict(condition_variables) if condition_variables else None) 

230 except Exception as e: 

231 _LOGGER.error("SUPERNOTIFY Condition eval failed: %s", e) 

232 raise 

233 

234 async def call_action( 

235 self, 

236 envelope: "Envelope", # noqa: F821 # type: ignore 

237 qualified_action: str | None = None, 

238 action_data: dict[str, Any] | None = None, 

239 target_data: dict[str, Any] | None = None, 

240 ) -> bool: 

241 action_data = action_data or {} 

242 start_time = time.time() 

243 domain = service = None 

244 config = self.delivery_config(envelope.delivery_name) 

245 try: 

246 qualified_action = qualified_action or config.get(CONF_ACTION) or self.default_action 

247 targets_required: bool = config.get(CONF_TARGETS_REQUIRED, self.targets_required) 

248 if qualified_action and (action_data.get(ATTR_TARGET) or not targets_required or target_data): 

249 domain, service = qualified_action.split(".", 1) 

250 start_time = time.time() 

251 envelope.calls.append(CallRecord(time.time() - start_time, domain, service, action_data, target_data)) 

252 if target_data: 

253 await self.hass.services.async_call(domain, service, service_data=action_data, target=target_data) 

254 else: 

255 await self.hass.services.async_call(domain, service, service_data=action_data) 

256 envelope.delivered = 1 

257 else: 

258 _LOGGER.debug( 

259 "SUPERNOTIFY skipping action call for service %s, targets %s", 

260 qualified_action, 

261 action_data.get(ATTR_TARGET), 

262 ) 

263 envelope.skipped = 1 

264 return True 

265 except Exception as e: 

266 envelope.failed_calls.append( 

267 CallRecord(time.time() - start_time, domain, service, action_data, target_data, exception=str(e)) 

268 ) 

269 _LOGGER.error("SUPERNOTIFY Failed to notify %s via %s, data=%s : %s", self.method, qualified_action, action_data, e) 

270 envelope.errored += 1 

271 envelope.delivery_error = format_exception(e) 

272 return False 

273 

274 def abs_url(self, fragment: str | None, prefer_external: bool = True) -> str | None: 

275 base_url = self.context.hass_external_url if prefer_external else self.context.hass_internal_url 

276 if fragment: 

277 if fragment.startswith("http"): 

278 return fragment 

279 if fragment.startswith("/"): 

280 return base_url + fragment 

281 return base_url + "/" + fragment 

282 return None 

283 

284 def simplify(self, text: str | None, strip_urls: bool = False) -> str | None: 

285 """Simplify text for delivery methods with speaking or plain text interfaces""" 

286 if not text: 

287 return None 

288 if strip_urls: 

289 words = text.split() 

290 text = " ".join(word for word in words if not urlparse(word).scheme) 

291 text = text.translate(str.maketrans("_", " ", "()£$<>")) 

292 _LOGGER.debug("SUPERNOTIFY Simplified text to: %s", text) 

293 return text