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 

9 

10from homeassistant.components.notify.const import ATTR_TARGET 

11from homeassistant.const import CONF_ACTION, CONF_CONDITION, CONF_DEFAULT, CONF_METHOD, CONF_NAME 

12from homeassistant.core import HomeAssistant 

13from homeassistant.helpers import condition 

14 

15from custom_components.supernotify.common import CallRecord 

16from custom_components.supernotify.configuration import Context 

17 

18from . import CONF_DATA, CONF_OPTIONS, CONF_TARGETS_REQUIRED, RESERVED_DELIVERY_NAMES, ConditionVariables 

19 

20_LOGGER = logging.getLogger(__name__) 

21 

22 

23class DeliveryMethod: 

24 method: str 

25 

26 @abstractmethod 

27 def __init__( 

28 self, hass: HomeAssistant, context: Context, deliveries: dict[str, Any] | None = None, default_action: str | None = None 

29 ) -> None: 

30 self.hass: HomeAssistant = hass 

31 self.default_action: str | None = default_action 

32 self.context: Context = context 

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

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

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

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

37 ) 

38 

39 async def initialize(self) -> None: 

40 """Async post-construction initialization""" 

41 if self.method is None: 

42 raise OSError("No delivery method configured") 

43 self.valid_deliveries = await self.validate_deliveries() 

44 

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

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

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

48 

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

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

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

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

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

54 if d in RESERVED_DELIVERY_NAMES: 

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

56 continue 

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

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

59 continue 

60 delivery_condition = dc.get(CONF_CONDITION) 

61 if delivery_condition: 

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

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

64 continue 

65 

66 valid_deliveries[d] = dc 

67 dc[CONF_NAME] = d 

68 

69 if dc.get(CONF_DEFAULT): 

70 if self.default_delivery: 

71 _LOGGER.warning("SUPERNOTIFY Multiple default deliveries, skipping %s", d) 

72 else: 

73 self.default_delivery = dc 

74 

75 if not self.default_delivery: 

76 method_definition = self.context.method_defaults.get(self.method) 

77 if method_definition: 

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

79 self.default_delivery = method_definition 

80 else: 

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

82 

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

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

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

86 else: 

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

88 

89 _LOGGER.debug( 

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

91 self.method, 

92 self.default_delivery, 

93 self.default_action, 

94 valid_deliveries, 

95 ) 

96 return valid_deliveries 

97 

98 def attributes(self) -> dict[str, str | list[str] | dict[str, Any] | None]: 

99 return { 

100 CONF_METHOD: self.method, 

101 "default_action": self.default_action, 

102 "default_delivery": self.default_delivery, 

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

104 } 

105 

106 @abstractmethod 

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

108 """Delivery implementation 

109 

110 Args: 

111 ---- 

112 envelope (Envelope): envelope to be delivered 

113 

114 """ 

115 

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

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

118 

119 Args: 

120 ---- 

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

122 

123 """ 

124 return True 

125 

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

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

128 return [] 

129 

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

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

132 config = dict(config) 

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

134 return config 

135 

136 def combined_message(self, envelope: "Envelope", default_title_only: bool = True) -> str | None: # type: ignore # noqa: F821 

137 config = self.delivery_config(envelope.delivery_name) 

138 if config.get(CONF_OPTIONS, {}).get("title_only", default_title_only) and envelope.title: 

139 return f"{envelope.title}" 

140 if envelope.title: 

141 return f"{envelope.title} {envelope.message}" 

142 return f"{envelope.message}" 

143 

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

145 if data is not None: 

146 action_data[key] = data 

147 return action_data 

148 

149 async def evaluate_delivery_conditions( 

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

151 ) -> bool | None: 

152 if CONF_CONDITION not in delivery_config: 

153 return True 

154 cond_conf = delivery_config.get(CONF_CONDITION) 

155 if cond_conf is None: 

156 return True 

157 

158 try: 

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

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

161 except Exception as e: 

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

163 raise 

164 

165 async def call_action( 

166 self, 

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

168 qualified_action: str | None = None, 

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

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

171 ) -> bool: 

172 action_data = action_data or {} 

173 start_time = time.time() 

174 domain = service = None 

175 config = self.delivery_config(envelope.delivery_name) 

176 try: 

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

178 targets_required = config.get(CONF_TARGETS_REQUIRED, False) 

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

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

181 start_time = time.time() 

182 if target_data: 

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

184 else: 

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

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

187 envelope.delivered = 1 

188 else: 

189 _LOGGER.debug( 

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

191 qualified_action, 

192 action_data.get(ATTR_TARGET), 

193 ) 

194 envelope.skipped = 1 

195 return True 

196 except Exception as e: 

197 envelope.failed_calls.append(CallRecord(time.time() - start_time, domain, service, action_data, exception=str(e))) 

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

199 envelope.errored += 1 

200 envelope.delivery_error = format_exception(e) 

201 return False 

202 

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

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

205 if fragment: 

206 if fragment.startswith("http"): 

207 return fragment 

208 if fragment.startswith("/"): 

209 return base_url + fragment 

210 return base_url + "/" + fragment 

211 return None