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

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 

11 

12if typing.TYPE_CHECKING: 

13 from custom_components.supernotify.common import CallRecord 

14 

15_LOGGER = logging.getLogger(__name__) 

16 

17 

18class Envelope: 

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

20 

21 def __init__( 

22 self, 

23 delivery_name: str, 

24 notification: "Notification | None" = None, # noqa: F821 # type: ignore 

25 targets: list | None = None, 

26 data: dict | None = None, 

27 ) -> None: 

28 self.targets: list = targets or [] 

29 self.delivery_name: str = delivery_name 

30 self._notification = notification 

31 self.notification_id = None 

32 self.media = None 

33 self.action_groups = None 

34 self.priority = PRIORITY_MEDIUM 

35 self.message: str | None = None 

36 self.title: str | None = None 

37 self.message_html: str | None = None 

38 self.data: dict = {} 

39 self.actions: list[dict] = [] 

40 delivery_config_data: dict = {} 

41 if notification: 

42 self.notification_id = notification.id 

43 self.media = notification.media 

44 self.action_groups = notification.action_groups 

45 self.actions = notification.actions 

46 self.priority = notification.priority 

47 self.message = notification.message(delivery_name) 

48 self.message_html = notification.message_html 

49 self.title = notification.title(delivery_name) 

50 delivery_config_data = notification.delivery_data(delivery_name) 

51 

52 if data: 

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

54 self.data |= data 

55 else: 

56 self.data = delivery_config_data if delivery_config_data else {} 

57 

58 self.delivered: int = 0 

59 self.errored: int = 0 

60 self.skipped: int = 0 

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

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

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

64 

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

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

67 if self._notification: 

68 return await self._notification.grab_image(self.delivery_name) 

69 return None 

70 

71 def core_action_data(self) -> dict: 

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

73 data: dict = {} 

74 # message is mandatory for notify platform 

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

76 timestamp = self.data.get(ATTR_TIMESTAMP) 

77 if timestamp: 

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

79 if self.title: 

80 data[CONF_TITLE] = self.title 

81 return data 

82 

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

84 exclude_attrs = ["_notification"] 

85 if minimal: 

86 exclude_attrs.extend("resolved") 

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

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

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

90 return json_ready 

91 

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

93 """Specialized equality check for subset of attributesfl""" 

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

95 return False 

96 return ( 

97 self.targets == other.targets 

98 and self.delivery_name == other.delivery_name 

99 and self.data == other.data 

100 and self.notification_id == other.notification_id 

101 )