Coverage for custom_components/supernotify/transport.py: 98%

1from __future__ import annotations 

2 

3import datetime as dt 

4import logging 

5import time 

6import unicodedata 

7from abc import abstractmethod 

8from traceback import format_exception 

9from typing import TYPE_CHECKING, Any 

10from urllib.parse import urlparse 

11 

12from homeassistant.components.notify.const import ATTR_TARGET 

13from homeassistant.const import ( 

14 ATTR_ENTITY_ID, 

15 ATTR_FRIENDLY_NAME, 

16 ATTR_NAME, 

17) 

18from homeassistant.exceptions import IntegrationError 

19from homeassistant.util import dt as dt_util 

20 

21from custom_components.supernotify.model import ( 

22 DebugTrace, 

23 Target, 

24 TargetRequired, 

25 TransportConfig, 

26 TransportFeature, 

27) 

28 

29from .common import CallRecord 

30from .const import ( 

31 ATTR_ENABLED, 

32 CONF_DELIVERY_DEFAULTS, 

33) 

34from .model import DeliveryConfig, SuppressionReason 

35 

36if TYPE_CHECKING: 

37 from homeassistant.helpers.typing import ConfigType 

38 

39 from .context import Context 

40 from .delivery import Delivery, DeliveryRegistry 

41 from .hass_api import HomeAssistantAPI 

42 from .people import PeopleRegistry 

43 

44_LOGGER = logging.getLogger(__name__) 

45 

46 

47class Transport: 

48 """Base class for delivery transports. 

49 

50 Sub classes integrste with Home Assistant notification services 

51 or alternative notification mechanisms. 

52 """ 

53 

54 name: str 

55 

56 @abstractmethod 

57 def __init__(self, context: Context, transport_config: ConfigType | None = None) -> None: 

58 self.hass_api: HomeAssistantAPI = context.hass_api 

59 self.people_registry: PeopleRegistry = context.people_registry 

60 self.delivery_registry: DeliveryRegistry = context.delivery_registry 

61 self.context: Context = context 

62 transport_config = transport_config or {} 

63 self.transport_config = TransportConfig(transport_config, class_config=self.default_config) 

64 

65 self.delivery_defaults: DeliveryConfig = self.transport_config.delivery_defaults 

66 self.config_enabled = self.transport_config.enabled 

67 self.enabled = self.config_enabled 

68 self.alias = self.transport_config.alias 

69 self.last_error_at: dt.datetime | None = None 

70 self.last_error_in: str | None = None 

71 self.last_error_message: str | None = None 

72 self.error_count: int = 0 

73 self._unavailable: bool = False 

74 

75 async def initialize(self) -> None: 

76 """Async post-construction initialization""" 

77 if self.name is None: 

78 raise IntegrationError("Invalid nameless transport adaptor subclass") 

79 

80 def setup_delivery_options(self, options: dict[str, Any], delivery_name: str) -> dict[str, Any]: 

81 return {} 

82 

83 @property 

84 def supported_features(self) -> TransportFeature: 

85 return TransportFeature.MESSAGE | TransportFeature.TITLE 

86 

87 @property 

88 def targets(self) -> Target: 

89 return self.delivery_defaults.target if self.delivery_defaults.target is not None else Target() 

90 

91 @property 

92 def default_config(self) -> TransportConfig: 

93 return TransportConfig() 

94 

95 def auto_configure(self, hass_api: HomeAssistantAPI) -> DeliveryConfig | None: 

96 return None 

97 

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

99 """Override in subclass if transport has fixed action or doesn't require one""" 

100 return action == self.delivery_defaults.action 

101 

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

103 attrs: dict[str, Any] = { 

104 ATTR_NAME: self.name, 

105 ATTR_ENABLED: self.enabled, 

106 CONF_DELIVERY_DEFAULTS: self.delivery_defaults, 

107 } 

108 if self.alias: 

109 attrs[ATTR_FRIENDLY_NAME] = self.alias 

110 if self.last_error_at: 

111 attrs["last_error_at"] = self.last_error_at 

112 attrs["last_error_in"] = self.last_error_in 

113 attrs["last_error_message"] = self.last_error_message 

114 attrs["error_count"] = self.error_count 

115 attrs.update(self.extra_attributes()) 

116 return attrs 

117 

118 def extra_attributes(self) -> dict[str, Any]: 

119 return {} 

120 

121 @abstractmethod 

122 async def deliver(self, envelope: Envelope, debug_trace: DebugTrace | None = None) -> bool: # type: ignore # noqa: F821 

123 """Delivery implementation 

124 

125 Args: 

126 ---- 

127 envelope (Envelope): envelope to be delivered 

128 debug_trace (DebugTrace): debug info collector 

129 

130 """ 

131 

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

133 if data is not None: 

134 action_data[key] = data 

135 return action_data 

136 

137 async def call_action( 

138 self, 

139 envelope: Envelope, # type: ignore # noqa: F821 

140 qualified_action: str | None = None, 

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

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

143 implied_target: bool = False, # True if the qualified action implies a target 

144 ) -> bool: 

145 action_data = action_data or {} 

146 start_time = time.time() 

147 domain = service = None 

148 delivery: Delivery = envelope.delivery 

149 try: 

150 qualified_action = qualified_action or delivery.action 

151 if not qualified_action: 

152 _LOGGER.debug( 

153 "SUPERNOTIFY Skipping %s action call with no service, targets %s", 

154 envelope.delivery.name, 

155 action_data.get(ATTR_TARGET), 

156 ) 

157 envelope.skipped = 1 

158 envelope.skip_reason = SuppressionReason.NO_ACTION 

159 return False 

160 if ( 

161 delivery.target_required == TargetRequired.ALWAYS 

162 and not action_data.get(ATTR_TARGET) 

163 and not action_data.get(ATTR_ENTITY_ID) 

164 and not implied_target 

165 and not target_data 

166 ): 

167 _LOGGER.debug( 

168 "SUPERNOTIFY Skipping %s action call for service %s, missing targets", 

169 envelope.delivery.name, 

170 qualified_action, 

171 ) 

172 envelope.skipped = 1 

173 envelope.skip_reason = SuppressionReason.NO_TARGET 

174 return False 

175 

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

177 start_time = time.time() 

178 timestamp: dt.datetime | None = None 

179 if target_data: 

180 # home-assistant messes with the service_data passed by ref 

181 service_data_as_sent = dict(action_data) 

182 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone()) 

183 service_response = await self.hass_api.call_service( 

184 domain, service, service_data=action_data, target=target_data, debug=delivery.debug 

185 ) 

186 envelope.calls.append( 

187 CallRecord( 

188 timestamp, 

189 time.time() - start_time, 

190 domain, 

191 service, 

192 debug=delivery.debug, 

193 action_data=service_data_as_sent, 

194 target_data=target_data, 

195 service_response=service_response, 

196 ) 

197 ) 

198 else: 

199 service_data_as_sent = dict(action_data) 

200 timestamp = dt.datetime.now(tz=dt_util.get_default_time_zone()) 

201 service_response = await self.hass_api.call_service( 

202 domain, service, service_data=action_data, debug=delivery.debug 

203 ) 

204 envelope.calls.append( 

205 CallRecord( 

206 timestamp, 

207 time.time() - start_time, 

208 domain, 

209 service, 

210 debug=delivery.debug, 

211 action_data=service_data_as_sent, 

212 service_response=service_response, 

213 ) 

214 ) 

215 

216 envelope.delivered = 1 

217 self.log_delivery_recovered() 

218 return True 

219 except Exception as e: 

220 self.record_error(str(e), method="call_action") 

221 envelope.failed_calls.append( 

222 CallRecord( 

223 timestamp, 

224 time.time() - start_time, 

225 domain, 

226 service, 

227 action_data, 

228 target_data, 

229 exception=str(e), 

230 ) 

231 ) 

232 self.log_delivery_failure( 

233 e, "SUPERNOTIFY Failed to notify %s via %s, data=%s", self.name, qualified_action, action_data 

234 ) 

235 envelope.error_count += 1 

236 envelope.delivery_error = format_exception(e) 

237 return False 

238 

239 def record_error(self, message: str, method: str) -> None: 

240 self.last_error_at = dt_util.utcnow() 

241 self.last_error_message = message 

242 self.last_error_in = method 

243 self.error_count += 1 

244 

245 def log_delivery_failure(self, err: BaseException, message: str, *args: Any) -> None: 

246 """Log a delivery failure, passing the exception caught in the caller's except block. 

247 

248 Logged at ERROR (with traceback) the first time this transport becomes unavailable, 

249 then downgraded to DEBUG for consecutive failures until it recovers - avoids 

250 spamming the log every notification while an external service/device stays down. 

251 Call alongside record_error(), which keeps tracking the lifetime error count 

252 regardless of log level. 

253 """ 

254 if self._unavailable: 

255 _LOGGER.debug(message, *args, exc_info=err) 

256 else: 

257 _LOGGER.error(message, *args, exc_info=err) 

258 self._unavailable = True 

259 

260 def log_delivery_recovered(self) -> None: 

261 """Call on a successful delivery - logs once if this transport was previously 

262 flagged unavailable, then clears the flag.""" 

263 if self._unavailable: 

264 _LOGGER.info("SUPERNOTIFY %s transport recovered after prior delivery failures", self.name) 

265 self._unavailable = False 

266 

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

268 """Simplify text for delivery transports with speaking or plain text interfaces""" 

269 if not text: 

270 return None 

271 if strip_urls: 

272 words = text.split() 

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

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

275 text = "".join(c for c in text if unicodedata.category(c) not in ("So", "Sk", "Sm", "Mn")) 

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

277 return text