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

1import logging 

2from typing import Any 

3 

4from homeassistant.components.notify.const import ATTR_TARGET 

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

6 

7from custom_components.supernotify import ( 

8 CONF_DATA, 

9 OPTION_MESSAGE_USAGE, 

10 OPTION_SIMPLIFY_TEXT, 

11 OPTION_STRIP_URLS, 

12 OPTION_TARGET_CATEGORIES, 

13 TRANSPORT_GENERIC, 

14) 

15from custom_components.supernotify.common import ensure_list 

16from custom_components.supernotify.envelope import Envelope 

17from custom_components.supernotify.model import MessageOnlyPolicy, TargetRequired, TransportConfig 

18from custom_components.supernotify.transport import ( 

19 Transport, 

20) 

21 

22_LOGGER = logging.getLogger(__name__) 

23 

24 

25class GenericTransport(Transport): 

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

27 

28 name = TRANSPORT_GENERIC 

29 

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

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

32 

33 @property 

34 def default_config(self) -> TransportConfig: 

35 config = TransportConfig() 

36 config.delivery_defaults.target_required = TargetRequired.OPTIONAL 

37 config.delivery_defaults.options = { 

38 OPTION_SIMPLIFY_TEXT: False, 

39 OPTION_STRIP_URLS: False, 

40 OPTION_MESSAGE_USAGE: MessageOnlyPolicy.STANDARD, 

41 OPTION_TARGET_CATEGORIES: [ATTR_ENTITY_ID], 

42 } 

43 return config 

44 

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

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

47 return True 

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

49 return False 

50 

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

52 data = envelope.data or {} 

53 

54 qualified_action = envelope.delivery.action 

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

56 action_data = envelope.core_action_data() 

57 if data and qualified_action != "notify.send_message": 

58 action_data[CONF_DATA] = data 

59 else: 

60 action_data = data 

61 

62 target_data: dict[str, Any] = {} 

63 if envelope.delivery.action == "notify.send_message": 

64 # amongst the wild west of notifty handling, at least care for the modern core one 

65 target_data = {ATTR_ENTITY_ID: envelope.target.entity_ids} 

66 else: 

67 all_targets: list[str] = [] 

68 for category in ensure_list(envelope.delivery.option(OPTION_TARGET_CATEGORIES)): 

69 all_targets.extend(envelope.target.for_category(category)) 

70 if all_targets: 

71 action_data[ATTR_TARGET] = all_targets 

72 

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