Coverage for custom_components/supernotify/switch.py: 100%

1"""Switch platform: enable or disable each scenario, recipient, delivery and transport. 

2 

3Forwarded to from async_setup_entry in __init__.py once the SupernotifyEngine (entry. 

4runtime_data) is fully initialized. 

5 

6A scenario's binary_sensor (binary_sensor.py) reports whether its *conditions* currently hold, so 

7it can't also be the control for enabling and disabling the scenario - writing its state used to do 

8both, and the two meanings fought each other. Recipient, delivery and transport binary_sensors had 

9no such conflict, but are treated the same way for consistency. These switches are the control, 

10and the binary_sensors are now read-only and kept only for backward compatibility. 

11 

12A delivery and its transport each have their own switch, and their own flag: switching a 

13transport off suppresses all of its deliveries, without changing the delivery switches. 

14 

15Each switch overrides the `enabled` value configured in YAML, and the override is kept across a 

16restart or reload (RestoreEntity) for as long as that configured value - its own `enabled`, or for 

17a delivery without one, its transport's - is unchanged. Editing it in YAML hands control back to 

18the configuration. 

19""" 

20 

21from __future__ import annotations 

22 

23import logging 

24from dataclasses import dataclass 

25from typing import TYPE_CHECKING, Any, Protocol 

26 

27from homeassistant.components.switch import SwitchEntity 

28from homeassistant.const import EntityCategory 

29from homeassistant.core import callback 

30from homeassistant.helpers.restore_state import ExtraStoredData, RestoreEntity 

31from homeassistant.util import slugify 

32 

33from . import DOMAIN 

34from .common import sanitize 

35from .const import ( 

36 DELIVERY_UNRECORDED_ATTRIBUTES, 

37 OVERRIDE_KIND_DELIVERY, 

38 OVERRIDE_KIND_RECIPIENT, 

39 OVERRIDE_KIND_SCENARIO, 

40 OVERRIDE_KIND_TRANSPORT, 

41 TRANSPORT_UNRECORDED_ATTRIBUTES, 

42) 

43from .hass_api import ha_device_info 

44 

45if TYPE_CHECKING: 

46 from homeassistant.core import HomeAssistant 

47 from homeassistant.helpers.device_registry import DeviceInfo 

48 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback 

49 

50 from . import SupernotifyConfigEntry 

51 from .delivery import Delivery, DeliveryRegistry 

52 from .people import PeopleRegistry, Recipient 

53 from .scenario import Scenario, ScenarioRegistry 

54 from .transport import Transport 

55 

56_LOGGER = logging.getLogger(__name__) 

57 

58 

59async def async_setup_entry( 

60 hass: HomeAssistant, 

61 entry: SupernotifyConfigEntry, 

62 async_add_entities: AddConfigEntryEntitiesCallback, 

63) -> None: 

64 """Add a switch for each scenario, recipient, and loaded transport and delivery.""" 

65 _ = hass 

66 context = entry.runtime_data.context 

67 switches = entry.runtime_data.override_switches 

68 device_info = ha_device_info(entry.entry_id) 

69 entities: list[SwitchEntity] = [ 

70 SupernotifyScenarioSwitch(scenario, context.scenario_registry, device_info, switches) 

71 for scenario in context.scenario_registry.scenarios.values() 

72 ] 

73 entities.extend( 

74 SupernotifyRecipientSwitch(recipient, context.people_registry, device_info, switches) 

75 for recipient in context.people_registry.people.values() 

76 ) 

77 # Only for what is loaded. The switch of a transport that isn't loaded this time, and of its 

78 # deliveries, is left in the entity registry rather than removed: a transport can be missing 

79 # just while what it depends on is still starting up 

80 delivery_registry = context.delivery_registry 

81 entities.extend( 

82 SupernotifyTransportSwitch(transport, delivery_registry, device_info, switches) 

83 for transport in delivery_registry.transports.values() 

84 ) 

85 entities.extend( 

86 SupernotifyDeliverySwitch(delivery, delivery_registry, device_info, switches) 

87 for delivery in delivery_registry.deliveries.values() 

88 ) 

89 async_add_entities(entities) 

90 

91 

92class Overridable(Protocol): 

93 """Anything with a configured enabled flag that a switch can override at runtime.""" 

94 

95 name: str 

96 enabled: bool 

97 config_enabled: bool 

98 

99 

100@dataclass(frozen=True) 

101class OverrideStoredData(ExtraStoredData): 

102 """What a switch keeps across a restart or reload: its state, and the configured value it 

103 was overriding, so that an override is dropped once the configuration changes.""" 

104 

105 enabled: bool 

106 config_enabled: bool 

107 

108 def as_dict(self) -> dict[str, Any]: 

109 return {"enabled": self.enabled, "config_enabled": self.config_enabled} 

110 

111 @classmethod 

112 def from_dict(cls, data: object) -> OverrideStoredData | None: 

113 """None unless data holds both values, as real booleans.""" 

114 if isinstance(data, dict) and isinstance(data.get("enabled"), bool) and isinstance(data.get("config_enabled"), bool): 

115 return cls(enabled=data["enabled"], config_enabled=data["config_enabled"]) 

116 return None 

117 

118 

119class SupernotifyOverridableSwitch(SwitchEntity, RestoreEntity): 

120 """A switch overriding the configured enabled flag of a scenario, recipient, delivery or 

121 transport, and keeping that override across a restart or reload while the configured value - 

122 its own `enabled`, or for a delivery without one, its transport's - is unchanged.""" 

123 

124 _attr_has_entity_name = True 

125 _attr_entity_category = EntityCategory.CONFIG 

126 _attr_should_poll = False 

127 

128 def __init__( 

129 self, kind: str, target: Overridable, device_info: DeviceInfo, switches: dict[str, SupernotifyOverridableSwitch] 

130 ) -> None: 

131 self._target = target 

132 self._switches = switches 

133 self._key = f"{kind}_{target.name}" 

134 self._attr_unique_id = self._key 

135 self._attr_device_info = device_info 

136 

137 @property 

138 def is_on(self) -> bool: 

139 return bool(self._target.enabled) 

140 

141 @property 

142 def extra_restore_state_data(self) -> OverrideStoredData: 

143 # kept trivial: some Home Assistant versions don't guard this getter against errors 

144 return OverrideStoredData(enabled=bool(self._target.enabled), config_enabled=bool(self._target.config_enabled)) 

145 

146 async def async_added_to_hass(self) -> None: 

147 await super().async_added_to_hass() 

148 self._switches[self._key] = self 

149 last = await self.async_get_last_extra_data() 

150 stored = OverrideStoredData.from_dict(last.as_dict()) if last else None 

151 if stored is None: 

152 return 

153 if stored.config_enabled != self._target.config_enabled: 

154 _LOGGER.info("SUPERNOTIFY Configuration of %s changed, discarding its previous override", self._key) 

155 return 

156 if stored.enabled != self._target.enabled: 

157 _LOGGER.info("SUPERNOTIFY Restoring override of %s to %s", self._key, "on" if stored.enabled else "off") 

158 # no state write here - Home Assistant writes the first state once this returns 

159 self._apply_enabled(stored.enabled) 

160 

161 async def async_will_remove_from_hass(self) -> None: 

162 self._switches.pop(self._key, None) 

163 await super().async_will_remove_from_hass() 

164 

165 async def async_turn_on(self, **kwargs: Any) -> None: 

166 self.async_set_enabled(True) 

167 

168 async def async_turn_off(self, **kwargs: Any) -> None: 

169 self.async_set_enabled(False) 

170 

171 @callback 

172 def async_set_enabled(self, enabled: bool) -> bool: 

173 """Change the flag and publish it, returning whether anything changed.""" 

174 if enabled == self._target.enabled: 

175 return False 

176 self._apply_enabled(enabled) 

177 self.async_write_ha_state() 

178 return True 

179 

180 def _apply_enabled(self, enabled: bool) -> None: 

181 self._target.enabled = enabled 

182 self._refresh_related() 

183 

184 def _refresh_related(self) -> None: 

185 """Re-publish any other entity whose state follows this flag.""" 

186 

187 

188class SupernotifyScenarioSwitch(SupernotifyOverridableSwitch): 

189 """Whether a scenario is enabled, and so able to apply to notifications.""" 

190 

191 _attr_translation_key = "scenario_enabled" 

192 

193 def __init__( 

194 self, 

195 scenario: Scenario, 

196 registry: ScenarioRegistry, 

197 device_info: DeviceInfo, 

198 switches: dict[str, SupernotifyOverridableSwitch], 

199 ) -> None: 

200 super().__init__(OVERRIDE_KIND_SCENARIO, scenario, device_info, switches) 

201 self._scenario = scenario 

202 self._registry = registry 

203 self._attr_translation_placeholders = {"scenario": scenario.alias or scenario.name} 

204 # Setting entity_id directly is not preferred Home Assistant practice - entities should 

205 # leave it to be derived from the device and entity name. It's done here so that a new 

206 # install gets the entity_id documented for scenarios (and used by the binary_sensor 

207 # this switch sits beside), which HA's own derivation would not produce. It has no effect 

208 # on an existing install: the entity registry entry found by unique_id always wins, so 

209 # nobody's entity_id, including one they have renamed, is changed by this. 

210 self.entity_id = f"switch.{DOMAIN}_scenario_{scenario.name}" 

211 

212 @property 

213 def extra_state_attributes(self) -> dict[str, object]: 

214 return sanitize(self._scenario.attributes(include_condition=False)) 

215 

216 def _refresh_related(self) -> None: 

217 # the condition state of a disabled scenario is always off, so that changes with this 

218 self._registry.async_refresh_entity(self._scenario.name) 

219 

220 

221class SupernotifyRecipientSwitch(SupernotifyOverridableSwitch): 

222 """Whether a recipient is enabled, and so able to be notified.""" 

223 

224 _attr_translation_key = "recipient_enabled" 

225 

226 def __init__( 

227 self, 

228 recipient: Recipient, 

229 registry: PeopleRegistry, 

230 device_info: DeviceInfo, 

231 switches: dict[str, SupernotifyOverridableSwitch], 

232 ) -> None: 

233 super().__init__(OVERRIDE_KIND_RECIPIENT, recipient, device_info, switches) 

234 self._recipient = recipient 

235 self._registry = registry 

236 self._attr_translation_placeholders = {"recipient": recipient.alias or recipient.name} 

237 # Setting entity_id directly is not preferred Home Assistant practice - see 

238 # SupernotifyScenarioSwitch for why it's done anyway, and why it can't change the 

239 # entity_id of an existing install. 

240 self.entity_id = f"switch.{DOMAIN}_recipient_{recipient.name}" 

241 

242 @property 

243 def extra_state_attributes(self) -> dict[str, object]: 

244 return sanitize(self._recipient.attributes()) 

245 

246 def _refresh_related(self) -> None: 

247 # the deprecated binary_sensor mirrors enabled 

248 self._registry.async_refresh_entity(self._recipient.name) 

249 

250 

251class SupernotifyDeliverySwitch(SupernotifyOverridableSwitch): 

252 """Whether a delivery is enabled, and so able to be used for notifications.""" 

253 

254 _attr_translation_key = "delivery_enabled" 

255 _unrecorded_attributes = DELIVERY_UNRECORDED_ATTRIBUTES 

256 

257 def __init__( 

258 self, 

259 delivery: Delivery, 

260 registry: DeliveryRegistry, 

261 device_info: DeviceInfo, 

262 switches: dict[str, SupernotifyOverridableSwitch], 

263 ) -> None: 

264 super().__init__(OVERRIDE_KIND_DELIVERY, delivery, device_info, switches) 

265 self._delivery = delivery 

266 self._registry = registry 

267 self._attr_translation_placeholders = {"delivery": delivery.alias or delivery.name} 

268 # Setting entity_id directly is not preferred Home Assistant practice - see 

269 # SupernotifyScenarioSwitch for why it's done anyway. Slugified, as a delivery name can be 

270 # any string. 

271 self.entity_id = f"switch.{DOMAIN}_delivery_{slugify(delivery.name)}" 

272 

273 @property 

274 def extra_state_attributes(self) -> dict[str, object]: 

275 return sanitize(self._delivery.attributes()) 

276 

277 def _refresh_related(self) -> None: 

278 # the deprecated binary_sensor mirrors enabled 

279 self._registry.async_refresh_entity(self._key) 

280 

281 

282class SupernotifyTransportSwitch(SupernotifyOverridableSwitch): 

283 """Whether a transport is enabled - when off, none of its deliveries are used.""" 

284 

285 _attr_translation_key = "transport_enabled" 

286 _unrecorded_attributes = TRANSPORT_UNRECORDED_ATTRIBUTES 

287 

288 def __init__( 

289 self, 

290 transport: Transport, 

291 registry: DeliveryRegistry, 

292 device_info: DeviceInfo, 

293 switches: dict[str, SupernotifyOverridableSwitch], 

294 ) -> None: 

295 super().__init__(OVERRIDE_KIND_TRANSPORT, transport, device_info, switches) 

296 self._transport = transport 

297 self._registry = registry 

298 self._attr_translation_placeholders = {"transport": transport.alias or transport.name} 

299 # Setting entity_id directly is not preferred Home Assistant practice - see 

300 # SupernotifyScenarioSwitch and SupernotifyDeliverySwitch. 

301 self.entity_id = f"switch.{DOMAIN}_transport_{slugify(transport.name)}" 

302 

303 @property 

304 def extra_state_attributes(self) -> dict[str, object]: 

305 return sanitize(self._transport.attributes()) 

306 

307 def _refresh_related(self) -> None: 

308 # the deprecated binary_sensor mirrors enabled 

309 self._registry.async_refresh_entity(self._key) 

310 # and each of this transport's deliveries shows it, as transport_enabled 

311 for delivery in self._registry.deliveries.values(): 

312 if delivery.transport is not self._transport: 

313 continue 

314 key = f"{OVERRIDE_KIND_DELIVERY}_{delivery.name}" 

315 switch = self._switches.get(key) 

316 if switch is not None: 

317 switch.async_write_ha_state() 

318 self._registry.async_refresh_entity(key)