Coverage for src/backend/InvenTree/plugin/builtin/integration/core_notifications.py: 73%

1"""Core set of Notifications as a Plugin.""" 

2 

3from django.contrib.auth.models import User 

4from django.db.models import Model 

5from django.template.loader import render_to_string 

6from django.utils.translation import gettext_lazy as _ 

7 

8import requests 

9import structlog 

10 

11import InvenTree.helpers_email 

12from common.settings import get_global_setting 

13from plugin import InvenTreePlugin 

14from plugin.mixins import NotificationMixin, SettingsMixin 

15 

16logger = structlog.get_logger('inventree') 

17 

18 

19class InvenTreeUINotifications(NotificationMixin, InvenTreePlugin): 

20 """Plugin mixin class for supporting UI notification methods.""" 

21 

22 NAME = 'InvenTreeUINotifications' 

23 TITLE = _('InvenTree UI Notifications') 

24 SLUG = 'inventree-ui-notification' 

25 AUTHOR = _('InvenTree contributors') 

26 DESCRIPTION = _('Integrated UI notification methods') 

27 VERSION = '1.0.0' 

28 

29 def send_notification( 

30 self, target: Model, category: str, users: list[User], context: dict 

31 ) -> bool: 

32 """Create a UI notification entry for specified users.""" 

33 from common.models import NotificationMessage 

34 

35 ctx = context if context else {} 

36 entries = [] 

37 

38 if not users: 

39 return False 

40 

41 # Ensure that there is always target object - see https://github.com/inventree/InvenTree/issues/10435 

42 if not target: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 target = self.plugin_config() 

44 

45 # Bulk create notification messages for all provided users 

46 for user in users: 

47 entries.append( 

48 NotificationMessage( 

49 target_object=target, 

50 source_object=user, 

51 user=user, 

52 category=category, 

53 name=ctx.get('name'), 

54 message=ctx.get('message'), 

55 ) 

56 ) 

57 

58 NotificationMessage.objects.bulk_create(entries, batch_size=250) 

59 

60 return True 

61 

62 

63class InvenTreeEmailNotifications(NotificationMixin, SettingsMixin, InvenTreePlugin): 

64 """Plugin mixin class for supporting email notification methods.""" 

65 

66 NAME = 'InvenTreeEmailNotifications' 

67 TITLE = _('InvenTree Email Notifications') 

68 SLUG = 'inventree-email-notification' 

69 AUTHOR = _('InvenTree contributors') 

70 DESCRIPTION = _('Integrated email notification methods') 

71 VERSION = '1.0.0' 

72 

73 USER_SETTINGS = { 

74 'NOTIFY_BY_EMAIL': { 

75 'name': _('Allow email notifications'), 

76 'description': _('Allow email notifications to be sent to this user'), 

77 'default': True, 

78 'validator': bool, 

79 } 

80 } 

81 

82 def send_notification( 

83 self, target: Model, category: str, users: list[User], context: dict 

84 ) -> bool: 

85 """Send notification to the specified targets.""" 

86 # Ignore if there is no template provided to render 

87 if not context.get('template'): 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 return False 

89 

90 html_message = render_to_string(context['template']['html'], context) 

91 

92 # Prefix the 'instance title' to the email subject 

93 instance_title = get_global_setting('INVENTREE_INSTANCE') 

94 subject = context['template'].get('subject', '') 

95 

96 if instance_title: 96 ↛ 99line 96 didn't jump to line 99 because the condition on line 96 was always true

97 subject = f'[{instance_title}] {subject}' 

98 

99 recipients = [] 

100 

101 for user in users: 101 ↛ 103line 101 didn't jump to line 103 because the loop on line 101 never started

102 # Skip if the user does not want to receive email notifications 

103 if not self.get_user_setting('NOTIFY_BY_EMAIL', user, backup_value=False): 

104 continue 

105 

106 if email := InvenTree.helpers_email.get_email_for_user(user): 

107 recipients.append(email) 

108 

109 if recipients: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 InvenTree.helpers_email.send_email( 

111 subject, '', recipients, html_message=html_message 

112 ) 

113 return True 

114 

115 # No recipients found, so we cannot send the email 

116 return False 

117 

118 

119class InvenTreeSlackNotifications(NotificationMixin, SettingsMixin, InvenTreePlugin): 

120 """Plugin mixin class for supporting Slack notification methods.""" 

121 

122 NAME = 'InvenTreeSlackNotifications' 

123 TITLE = _('InvenTree Slack Notifications') 

124 SLUG = 'inventree-slack-notification' 

125 AUTHOR = _('InvenTree contributors') 

126 DESCRIPTION = _('Integrated Slack notification methods') 

127 VERSION = '1.0.0' 

128 

129 SETTINGS = { 

130 'NOTIFICATION_SLACK_URL': { 

131 'name': _('Slack incoming webhook URL'), 

132 'description': _('URL that is used to send messages to a slack channel'), 

133 'protected': True, 

134 } 

135 } 

136 

137 def send_notification( 

138 self, target: Model, category: str, users: list[User], context: dict 

139 ) -> bool: 

140 """Send the notifications out via slack.""" 

141 url = self.get_setting('NOTIFICATION_SLACK_URL') 

142 

143 if not url: 

144 return False 

145 

146 ret = requests.post( 

147 url, 

148 json={ 

149 'text': str(context['message']), 

150 'blocks': [ 

151 { 

152 'type': 'section', 

153 'text': {'type': 'plain_text', 'text': str(context['name'])}, 

154 }, 

155 { 

156 'type': 'section', 

157 'text': {'type': 'mrkdwn', 'text': str(context['message'])}, 

158 'accessory': { 

159 'type': 'button', 

160 'text': { 

161 'type': 'plain_text', 

162 'text': str(_('Open link')), 

163 'emoji': True, 

164 }, 

165 'value': f'{category}_{target.pk}' if target else '', 

166 'url': context['link'], 

167 'action_id': 'button-action', 

168 }, 

169 }, 

170 ], 

171 }, 

172 ) 

173 

174 return ret.ok