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 SupernotificationConfiguration 

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 default_action: str | None = None 

26 

27 @abstractmethod 

28 def __init__(self, hass: HomeAssistant, context: SupernotificationConfiguration, deliveries: dict | None = None) -> None: 

29 self.hass: HomeAssistant = hass 

30 self.context: SupernotificationConfiguration = context 

31 self.default_delivery: dict | None = None 

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

33 self.method_deliveries: dict[str, dict] = ( 

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

35 ) 

36 

37 async def initialize(self) -> None: 

38 """Async post-construction initialization""" 

39 if self.method is None: 

40 raise OSError("No delivery method configured") 

41 self.valid_deliveries = await self.validate_deliveries() 

42 

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

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

45 return action is not None and action.startswith("notify.") 

46 

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

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

49 valid_deliveries: dict[str, dict] = {} 

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

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

52 if d in RESERVED_DELIVERY_NAMES: 

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

54 continue 

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

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

57 continue 

58 delivery_condition = dc.get(CONF_CONDITION) 

59 if delivery_condition: 

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

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

62 continue 

63 

64 valid_deliveries[d] = dc 

65 dc[CONF_NAME] = d 

66 

67 if dc.get(CONF_DEFAULT): 

68 if self.default_delivery: 

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

70 else: 

71 self.default_delivery = dc 

72 

73 if not self.default_delivery: 

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

75 if method_definition: 

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

77 self.default_delivery = method_definition 

78 

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

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

81 

82 _LOGGER.debug( 

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

84 self.method, 

85 self.default_delivery, 

86 self.default_action, 

87 valid_deliveries, 

88 ) 

89 return valid_deliveries 

90 

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

92 return { 

93 CONF_METHOD: self.method, 

94 "default_action": self.default_action, 

95 "default_delivery": self.default_delivery, 

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

97 } 

98 

99 @abstractmethod 

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

101 """Delivery implementation 

102 

103 Args: 

104 ---- 

105 envelope (Envelope): envelope to be delivered 

106 

107 """ 

108 

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

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

111 

112 Args: 

113 ---- 

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

115 

116 """ 

117 return True 

118 

119 def recipient_target(self, recipient: dict) -> list: # noqa: ARG002 

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

121 return [] 

122 

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

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

125 config = dict(config) 

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

127 return config 

128 

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

130 config = self.delivery_config(envelope.delivery_name) 

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

132 return envelope.title 

133 if envelope.title: 

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

135 return envelope.message 

136 

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

138 if data is not None: 

139 action_data[key] = data 

140 return action_data 

141 

142 async def evaluate_delivery_conditions( 

143 self, delivery_config: dict, condition_variables: ConditionVariables | None 

144 ) -> bool | None: 

145 if CONF_CONDITION not in delivery_config: 

146 return True 

147 cond_conf = delivery_config.get(CONF_CONDITION) 

148 if cond_conf is None: 

149 return True 

150 

151 try: 

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

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

154 except Exception as e: 

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

156 raise 

157 

158 async def call_action( 

159 self, 

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

161 qualified_action: str | None = None, 

162 action_data: dict | None = None, 

163 ) -> bool: 

164 action_data = action_data or {} 

165 start_time = time.time() 

166 domain = service = None 

167 config = self.delivery_config(envelope.delivery_name) 

168 try: 

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

170 targets_required = config.get(CONF_TARGETS_REQUIRED, False) 

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

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

173 start_time = time.time() 

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

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

176 envelope.delivered = 1 

177 else: 

178 _LOGGER.debug( 

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

180 qualified_action, 

181 action_data.get(ATTR_TARGET), 

182 ) 

183 envelope.skipped = 1 

184 return True 

185 except Exception as e: 

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

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

188 envelope.errored += 1 

189 envelope.delivery_error = format_exception(e) 

190 return False 

191 

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

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

194 if fragment: 

195 if fragment.startswith("http"): 

196 return fragment 

197 if fragment.startswith("/"): 

198 return base_url + fragment 

199 return base_url + "/" + fragment 

200 return None