Coverage for custom_components/supernotify/delivery.py: 99%
160 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-21 23:31 +0000
1import logging
2import re
3from typing import Any
5from homeassistant.const import (
6 ATTR_FRIENDLY_NAME,
7 ATTR_NAME,
8 CONF_ACTION,
9 CONF_ALIAS,
10 CONF_CONDITION,
11 CONF_DEBUG,
12 CONF_ENABLED,
13 CONF_NAME,
14 CONF_OPTIONS,
15)
16from homeassistant.helpers.typing import ConfigType
18from custom_components.supernotify.model import ConditionVariables, DeliveryConfig, Target
19from custom_components.supernotify.transport import Transport
21from . import (
22 ATTR_ENABLED,
23 CONF_MESSAGE,
24 CONF_OCCUPANCY,
25 CONF_SELECTION,
26 CONF_TEMPLATE,
27 CONF_TITLE,
28 CONF_TRANSPORT,
29 OCCUPANCY_ALL,
30 OPTION_TARGET_CATEGORIES,
31 OPTION_TARGET_INCLUDE_RE,
32 RESERVED_DELIVERY_NAMES,
33 SELECTION_DEFAULT,
34 SELECTION_FALLBACK,
35 SELECTION_FALLBACK_ON_ERROR,
36)
37from .context import Context
39_LOGGER = logging.getLogger(__name__)
42class Delivery(DeliveryConfig):
43 def __init__(self, name: str, conf: ConfigType, transport: "Transport") -> None:
44 self.name: str = name
45 self.alias: str | None = conf.get(CONF_ALIAS)
46 self.transport: Transport = transport
47 transport_defaults: DeliveryConfig = self.transport.delivery_defaults
48 super().__init__(conf, delivery_defaults=transport_defaults)
49 self.template: str | None = conf.get(CONF_TEMPLATE)
50 self.message: str | None = conf.get(CONF_MESSAGE)
51 self.title: str | None = conf.get(CONF_TITLE)
52 self.enabled: bool = conf.get(CONF_ENABLED, self.transport.enabled)
53 self.occupancy: str = conf.get(CONF_OCCUPANCY, OCCUPANCY_ALL)
54 self.condition: ConfigType | None = conf.get(CONF_CONDITION)
56 async def validate(self, context: "Context") -> bool:
57 errors = 0
58 if self.name in RESERVED_DELIVERY_NAMES:
59 _LOGGER.warning("SUPERNOTIFY Delivery uses reserved word %s", self.name)
60 context.hass_api.raise_issue(
61 f"delivery_{self.name}_reserved_name",
62 issue_key="delivery_reserved_name",
63 issue_map={"delivery": self.name},
64 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
65 )
66 errors += 1
67 if not self.transport.validate_action(self.action):
68 _LOGGER.warning("SUPERNOTIFY Invalid action definition for delivery %s (%s)", self.name, self.action)
69 context.hass_api.raise_issue(
70 f"delivery_{self.name}_invalid_action",
71 issue_key="delivery_invalid_action",
72 issue_map={"delivery": self.name, "action": self.action or ""},
73 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
74 )
75 errors += 1
77 if self.condition:
78 try:
79 await context.hass_api.evaluate_condition(self.condition, validate=True, strict=True)
80 passed = True
81 exception = ""
82 except Exception as e:
83 passed = False
84 exception = str(e)
85 if not passed:
86 _LOGGER.warning("SUPERNOTIFY Invalid delivery condition for %s: %s", self.name, self.condition)
87 context.hass_api.raise_issue(
88 f"delivery_{self.name}_invalid_condition",
89 issue_key="delivery_invalid_condition",
90 issue_map={"delivery": self.name, "condition": str(self.condition), "exception": exception},
91 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
92 )
93 errors += 1
94 return errors == 0
96 def select_targets(self, target: Target) -> Target:
97 def selected(category: str, targets: list[str]) -> list[str]:
98 if OPTION_TARGET_CATEGORIES in self.options and category not in self.options[OPTION_TARGET_CATEGORIES]:
99 return []
100 if OPTION_TARGET_INCLUDE_RE in self.options:
101 return [t for t in targets if any(re.fullmatch(r, t) for r in self.options[OPTION_TARGET_INCLUDE_RE])]
102 return targets
104 filtered_target = Target({k: selected(k, v) for k, v in target.targets.items()}, target_data=target.target_data)
105 # TODO: in model class
106 if target.target_specific_data:
107 filtered_target.target_specific_data = {
108 (c, t): data
109 for (c, t), data in target.target_specific_data.items()
110 if c in target.targets and t in target.targets[c]
111 }
112 return filtered_target
114 async def evaluate_conditions(self, condition_variables: ConditionVariables | None) -> bool | None:
115 if not self.enabled:
116 return False
117 if self.condition is None:
118 return True
119 # TODO: reconsider hass_api injection
120 return await self.transport.hass_api.evaluate_condition(self.condition, condition_variables)
122 def option(self, option_name: str) -> str | bool:
123 """Get an option value from delivery config or transport default options"""
124 opt: str | bool | None = None
125 if option_name in self.options:
126 opt = self.options[option_name]
127 if opt is None:
128 _LOGGER.debug("SUPERNOTIFY No default in %s for option %s, setting to empty string", self.name, option_name)
129 opt = ""
130 return opt
132 def option_bool(self, option_name: str) -> bool:
133 return bool(self.option(option_name))
135 def option_str(self, option_name: str) -> str:
136 return str(self.option(option_name))
138 def as_dict(self) -> dict[str, Any]:
139 base = super().as_dict()
140 base.update({
141 CONF_NAME: self.name,
142 CONF_ALIAS: self.alias,
143 CONF_TRANSPORT: self.transport.name,
144 CONF_TEMPLATE: self.template,
145 CONF_MESSAGE: self.message,
146 CONF_TITLE: self.title,
147 CONF_ENABLED: self.enabled,
148 CONF_OCCUPANCY: self.occupancy,
149 CONF_CONDITION: self.condition,
150 })
151 return base
153 def attributes(self) -> dict[str, Any]:
154 """For exposure as entity state"""
155 attrs: dict[str, Any] = {
156 ATTR_NAME: self.name,
157 ATTR_ENABLED: self.enabled,
158 CONF_TRANSPORT: self.transport.name,
159 CONF_ACTION: self.action,
160 CONF_OPTIONS: self.options,
161 CONF_SELECTION: self.selection,
162 CONF_DEBUG: self.debug,
163 }
164 if self.alias:
165 attrs[ATTR_FRIENDLY_NAME] = self.alias
166 return attrs
169class DeliveryRegistry:
170 def __init__(
171 self,
172 deliveries: ConfigType | None = None,
173 transport_configs: ConfigType | None = None,
174 transport_types: list[type[Transport]] | None = None,
175 # for unit tests only
176 transport_instances: list[Transport] | None = None,
177 ) -> None:
178 # raw configured deliveries
179 self._deliveries: ConfigType = deliveries if isinstance(deliveries, dict) else {}
180 # validated deliveries
181 self.deliveries: dict[str, Delivery] = {}
182 self.transports: dict[str, Transport] = {}
183 self._transport_configs: ConfigType = transport_configs or {}
184 self._fallback_on_error: list[Delivery] = []
185 self._fallback_by_default: list[Delivery] = []
186 self._implicit_deliveries: list[Delivery] = []
187 self._transport_types: list[type[Transport]] = transport_types or []
188 # test harness support
189 self._transport_instances: list[Transport] | None = transport_instances
191 async def initialize(self, context: "Context") -> None:
192 await self.initialize_transports(context)
193 self.autogenerate_deliveries()
194 self.initialize_deliveries()
196 def initialize_deliveries(self) -> None:
197 for delivery in self.deliveries.values():
198 if delivery.enabled:
199 if SELECTION_FALLBACK_ON_ERROR in delivery.selection:
200 self._fallback_on_error.append(delivery)
201 if SELECTION_FALLBACK in delivery.selection:
202 self._fallback_by_default.append(delivery)
203 if SELECTION_DEFAULT in delivery.selection:
204 self._implicit_deliveries.append(delivery)
206 @property
207 def fallback_by_default_deliveries(self) -> list[Delivery]:
208 return [d for d in self._fallback_by_default if d.enabled]
210 @property
211 def fallback_on_error_deliveries(self) -> list[Delivery]:
212 return [d for d in self._fallback_on_error if d.enabled]
214 @property
215 def implicit_deliveries(self) -> list[Delivery]:
216 """Deliveries switched on all the time for implicit selection"""
217 return [d for d in self._implicit_deliveries if d.enabled]
219 async def initialize_transports(self, context: "Context") -> None:
220 """Use configure_for_tests() to set transports to mocks or manually created fixtures"""
221 if self._transport_instances:
222 for transport in self._transport_instances:
223 self.transports[transport.name] = transport
224 await transport.initialize()
225 await self.initialize_transport_deliveries(context, transport)
226 if self._transport_types:
227 for transport_class in self._transport_types:
228 transport_config: ConfigType = self._transport_configs.get(transport_class.name, {})
229 transport = transport_class(context, transport_config)
230 self.transports[transport_class.name] = transport
231 await transport.initialize()
232 await self.initialize_transport_deliveries(context, transport)
233 self.transports[transport_class.name] = transport
235 unconfigured_deliveries = [dc for d, dc in self._deliveries.items() if d not in self.deliveries]
236 for bad_del in unconfigured_deliveries:
237 # presumably there was no transport for these
238 context.hass_api.raise_issue(
239 f"delivery_{bad_del.get(CONF_NAME)}_for_transport_{bad_del.get(CONF_TRANSPORT)}_failed_to_configure",
240 issue_key="delivery_unknown_transport",
241 issue_map={"delivery": bad_del.get(CONF_NAME), "transport": bad_del.get(CONF_TRANSPORT)},
242 learn_more_url="https://supernotify.rhizomatics.org.uk/deliveries",
243 )
244 _LOGGER.info("SUPERNOTIFY configured deliveries %s", "; ".join(self.deliveries.keys()))
246 async def initialize_transport_deliveries(self, context: Context, transport: Transport) -> None:
247 """Validate and initialize deliveries at startup for this transport"""
248 validated_deliveries: dict[str, Delivery] = {}
249 deliveries_for_this_transport = {
250 d: dc for d, dc in self._deliveries.items() if dc.get(CONF_TRANSPORT) == transport.name
251 }
252 for d, dc in deliveries_for_this_transport.items():
253 # don't care about ENABLED here since disabled deliveries can be overridden later
254 delivery = Delivery(d, dc, transport)
255 if not await delivery.validate(context):
256 _LOGGER.error(f"SUPERNOTIFY Ignoring delivery {d} with errors")
257 else:
258 validated_deliveries[d] = delivery
260 self.deliveries.update(validated_deliveries)
262 _LOGGER.debug(
263 "SUPERNOTIFY Validated transport %s, default action %s, valid deliveries: %s",
264 transport.name,
265 transport.delivery_defaults.action,
266 [d for d in self.deliveries.values() if d.enabled and d.transport == transport],
267 )
269 def autogenerate_deliveries(self) -> None:
270 # If the config has no deliveries, check if a default delivery should be auto-generated
271 # where there is a empty config, supernotify can at least handle NotifyEntities sensibly
273 autogenerated: dict[str, Delivery] = {}
274 for transport in self.transports.values():
275 if any(dc for dc in self._deliveries.values() if dc.get(CONF_TRANSPORT) == transport.name):
276 continue
278 transport_definition: DeliveryConfig = transport.delivery_defaults
279 _LOGGER.debug(
280 "SUPERNOTIFY Building default delivery for %s from transport %s", transport.name, transport_definition
281 )
282 if transport.auto_configure and transport.enabled and transport.validate_action(transport_definition.action):
283 # auto generate a delivery that will be implicitly selected
284 default_delivery = Delivery(f"DEFAULT_{transport.name}", transport_definition.as_dict(), transport)
285 default_delivery.enabled = transport.enabled
286 autogenerated[default_delivery.name] = default_delivery
287 _LOGGER.info(
288 "SUPERNOTIFY Auto-generating a default delivery for %s from transport %s",
289 transport.name,
290 transport_definition,
291 )
292 else:
293 _LOGGER.debug("SUPERNOTIFY No default delivery or transport_definition for transport %s", transport.name)
294 if autogenerated:
295 self.deliveries.update(autogenerated)