Coverage for custom_components/supernotify/methods/generic.py: 100%

1import logging 

2from typing import Any 

3 

4from homeassistant.components.notify.const import ATTR_MESSAGE, ATTR_TITLE 

5from homeassistant.const import ( # ATTR_VARIABLES from script.const has import issues 

6 ATTR_ENTITY_ID, 

7 CONF_ACTION, 

8) 

9 

10from custom_components.supernotify import CONF_DATA, CONF_TARGETS_REQUIRED, METHOD_GENERIC 

11from custom_components.supernotify.delivery_method import DeliveryMethod 

12from custom_components.supernotify.envelope import Envelope 

13 

14_LOGGER = logging.getLogger(__name__) 

15 

16 

17class GenericDeliveryMethod(DeliveryMethod): 

18 """Call any service, including non-notify ones, like switch.turn_on or mqtt.publish""" 

19 

20 method = METHOD_GENERIC 

21 

22 def __init__(self, *args: Any, **kwargs: Any) -> None: 

23 kwargs.setdefault(CONF_TARGETS_REQUIRED, False) 

24 super().__init__(*args, **kwargs) 

25 

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

27 if action is not None and "." in action: 

28 return True 

29 _LOGGER.warning("SUPERNOTIFY generic method must have a qualified action name, e.g. notify.foo") 

30 return False 

31 

32 async def deliver(self, envelope: Envelope) -> bool: 

33 data = envelope.data or {} 

34 targets = envelope.targets or [] 

35 config = self.delivery_config(envelope.delivery_name) 

36 target_data: dict[str, Any] = {ATTR_ENTITY_ID: targets} if targets else {} 

37 

38 qualified_action = config.get(CONF_ACTION) 

39 if qualified_action and qualified_action.startswith("notify."): 

40 action_data = envelope.core_action_data() 

41 if data is not None: 

42 action_data[CONF_DATA] = data 

43 if qualified_action == "notify.send_message": 

44 allowed_keys: list[str] = [ATTR_MESSAGE, ATTR_TITLE, ATTR_ENTITY_ID] 

45 action_data = {k: v for k, v in action_data.items() if k in allowed_keys} 

46 

47 else: 

48 action_data = data 

49 

50 return await self.call_action(envelope, qualified_action, action_data=action_data, target_data=target_data)