Coverage for custom_components/supernotify/envelope.py: 97%

1# mypy: disable-error-code="name-defined" 

2 

3import copy 

4import logging 

5import time 

6import typing 

7from pathlib import Path 

8from typing import Any 

9 

10from . import ATTR_TIMESTAMP, CONF_MESSAGE, CONF_TITLE, PRIORITY_MEDIUM 

11from .media_grab import grab_image 

12from .model import Target 

13 

14if typing.TYPE_CHECKING: 

15 from custom_components.supernotify.common import CallRecord 

16 

17 from .delivery import Delivery 

18 from .notification import Notification 

19 

20_LOGGER = logging.getLogger(__name__) 

21 

22 

23class Envelope: 

24 """Wrap a notification with a specific set of targets and service data possibly customized for those targets""" 

25 

26 def __init__( 

27 self, 

28 delivery: "Delivery", 

29 notification: "Notification | None" = None, 

30 target: Target | None = None, # targets only for this delivery 

31 data: dict[str, Any] | None = None, # notification data customized for this delivery 

32 ) -> None: 

33 self.target: Target = target or Target() 

34 self.delivery_name: str = delivery.name 

35 self.delivery: Delivery = delivery 

36 self._notification = notification 

37 self.notification_id = None 

38 self.media = None 

39 self.action_groups = None 

40 self.priority = PRIORITY_MEDIUM 

41 self.message: str | None = None 

42 self.title: str | None = None 

43 self.message_html: str | None = None 

44 self.data: dict[str, Any] = {} 

45 self.actions: list[dict[str, Any]] = [] 

46 delivery_config_data: dict[str, Any] = {} 

47 if notification: 

48 self.notification_id = notification.id 

49 self.media = notification.media 

50 self.action_groups = notification.action_groups 

51 self.actions = notification.actions 

52 self.priority = notification.priority 

53 self.message = notification.message(delivery.name) 

54 self.message_html = notification.message_html 

55 self.title = notification.title(delivery.name) 

56 delivery_config_data = notification.delivery_data(delivery.name) 

57 

58 if data: 

59 self.data = copy.deepcopy(delivery_config_data) if delivery_config_data else {} 

60 self.data |= data 

61 else: 

62 self.data = delivery_config_data if delivery_config_data else {} 

63 

64 self.delivered: int = 0 

65 self.errored: int = 0 

66 self.skipped: int = 0 

67 self.calls: list[CallRecord] = [] 

68 self.failed_calls: list[CallRecord] = [] 

69 self.delivery_error: list[str] | None = None 

70 

71 async def grab_image(self) -> Path | None: 

72 """Grab an image from a camera, snapshot URL, MQTT Image etc""" 

73 image_path: Path | None = None 

74 if self._notification: 

75 image_path = await grab_image(self._notification, self.delivery_name, self._notification.context) 

76 return image_path 

77 

78 def core_action_data(self) -> dict[str, Any]: 

79 """Build the core set of `service_data` dict to pass to underlying notify service""" 

80 data: dict[str, Any] = {} 

81 # message is mandatory for notify platform 

82 data[CONF_MESSAGE] = self.message or "" 

83 timestamp = self.data.get(ATTR_TIMESTAMP) 

84 if timestamp: 

85 data[CONF_MESSAGE] = f"{data[CONF_MESSAGE]} [{time.strftime(timestamp, time.localtime())}]" 

86 if self.title: 

87 data[CONF_TITLE] = self.title 

88 return data 

89 

90 def contents(self, minimal: bool = True) -> dict[str, typing.Any]: 

91 exclude_attrs = ["_notification"] 

92 if minimal: 

93 exclude_attrs.extend("resolved") 

94 json_ready = {k: v for k, v in self.__dict__.items() if k not in exclude_attrs} 

95 json_ready["calls"] = [call.contents() for call in self.calls] 

96 json_ready["failedcalls"] = [call.contents() for call in self.failed_calls] 

97 return json_ready 

98 

99 def __eq__(self, other: Any | None) -> bool: 

100 """Specialized equality check for subset of attributes""" 

101 if other is None or not isinstance(other, Envelope): 

102 return False 

103 return bool( 

104 self.target == other.target 

105 and self.delivery_name == other.delivery_name 

106 and self.data == other.data 

107 and self.notification_id == other.notification_id 

108 ) 

109 

110 def __repr__(self) -> str: 

111 """Return a concise string representation of the Envelope. 

112 

113 The returned string includes the envelope's message, title, and delivery name 

114 in the form: Envelope(message=<message>,title=<title>,delivery=<delivery_name>). 

115 

116 Primarily intended for debugging and logging; note that attribute values are 

117 inserted directly and may not be quoted or escaped. 

118 """ 

119 return f"Envelope(message={self.message},title={self.title},delivery={self.delivery_name})"