Coverage for extras/models/notifications.py: 70%

1from functools import cached_property 

2 

3from django.conf import settings 

4from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation 

5from django.core.exceptions import ValidationError 

6from django.db import models 

7from django.urls import reverse 

8from django.utils.translation import gettext_lazy as _ 

9 

10from extras.querysets import NotificationQuerySet 

11from netbox.models import ChangeLoggedModel 

12from netbox.models.features import has_feature 

13from netbox.registry import registry 

14from users.models import User 

15from utilities.choices import Choice 

16from utilities.querysets import RestrictedQuerySet 

17 

18__all__ = ( 

19 'Notification', 

20 'NotificationGroup', 

21 'Subscription', 

22) 

23 

24 

25def get_event_type_choices(): 

26 """ 

27 Compile a list of choices from all registered event types 

28 """ 

29 return [ 

30 Choice(name, event.text) 

31 for name, event in registry['event_types'].items() 

32 ] 

33 

34 

35class Notification(models.Model): 

36 """ 

37 A notification message for a User relating to a specific object in NetBox. 

38 """ 

39 created = models.DateTimeField( 

40 verbose_name=_('created'), 

41 auto_now_add=True 

42 ) 

43 read = models.DateTimeField( 

44 verbose_name=_('read'), 

45 null=True, 

46 blank=True 

47 ) 

48 user = models.ForeignKey( 

49 to=settings.AUTH_USER_MODEL, 

50 on_delete=models.CASCADE, 

51 related_name='notifications' 

52 ) 

53 object_type = models.ForeignKey( 

54 to='contenttypes.ContentType', 

55 on_delete=models.PROTECT 

56 ) 

57 object_id = models.PositiveBigIntegerField() 

58 object = GenericForeignKey( 

59 ct_field='object_type', 

60 fk_field='object_id' 

61 ) 

62 object_repr = models.CharField( 

63 max_length=200, 

64 editable=False 

65 ) 

66 event_type = models.CharField( 

67 verbose_name=_('event'), 

68 max_length=50, 

69 choices=get_event_type_choices 

70 ) 

71 

72 objects = NotificationQuerySet.as_manager() 

73 

74 class Meta: 

75 ordering = ('-created', 'pk') 

76 indexes = ( 

77 models.Index(fields=('-created', 'id')), # Default ordering 

78 models.Index(fields=('object_type', 'object_id')), 

79 ) 

80 constraints = ( 

81 models.UniqueConstraint( 

82 fields=('object_type', 'object_id', 'user'), 

83 name='%(app_label)s_%(class)s_unique_per_object_and_user' 

84 ), 

85 ) 

86 verbose_name = _('notification') 

87 verbose_name_plural = _('notifications') 

88 

89 def __str__(self): 

90 return self.object_repr 

91 

92 def get_absolute_url(self): 

93 return reverse('account:notifications') 

94 

95 def clean(self): 

96 super().clean() 

97 

98 # Validate the assigned object type 

99 if not has_feature(self.object_type, 'notifications'): 

100 raise ValidationError( 

101 _("Objects of this type ({type}) do not support notifications.").format(type=self.object_type) 

102 ) 

103 

104 def save(self, *args, **kwargs): 

105 # Record a string representation of the associated object 

106 if self.object: 

107 self.object_repr = self.get_object_repr(self.object) 

108 

109 super().save(*args, **kwargs) 

110 

111 @cached_property 

112 def event(self): 

113 """ 

114 Returns the registered Event which triggered this Notification. 

115 """ 

116 return registry['event_types'].get(self.event_type) 

117 

118 @classmethod 

119 def get_object_repr(cls, obj): 

120 return str(obj)[:200] 

121 

122 

123class NotificationGroup(ChangeLoggedModel): 

124 """ 

125 A collection of users and/or groups to be informed for certain notifications. 

126 """ 

127 name = models.CharField( 

128 verbose_name=_('name'), 

129 max_length=100, 

130 unique=True 

131 ) 

132 description = models.CharField( 

133 verbose_name=_('description'), 

134 max_length=200, 

135 blank=True 

136 ) 

137 groups = models.ManyToManyField( 

138 to='users.Group', 

139 verbose_name=_('groups'), 

140 blank=True, 

141 related_name='notification_groups' 

142 ) 

143 users = models.ManyToManyField( 

144 to='users.User', 

145 verbose_name=_('users'), 

146 blank=True, 

147 related_name='notification_groups' 

148 ) 

149 event_rules = GenericRelation( 

150 to='extras.EventRule', 

151 content_type_field='action_object_type', 

152 object_id_field='action_object_id', 

153 related_query_name='+' 

154 ) 

155 

156 objects = RestrictedQuerySet.as_manager() 

157 

158 class Meta: 

159 ordering = ('name',) 

160 verbose_name = _('notification group') 

161 verbose_name_plural = _('notification groups') 

162 

163 def __str__(self): 

164 return self.name 

165 

166 def get_absolute_url(self): 

167 return reverse('extras:notificationgroup', args=[self.pk]) 

168 

169 @cached_property 

170 def members(self): 

171 """ 

172 Return all Users who belong to this notification group. 

173 """ 

174 return self.users.union( 

175 User.objects.filter(groups__in=self.groups.all()) 

176 ).order_by('username') 

177 

178 def notify(self, object_type, object_id, **kwargs): 

179 """ 

180 Bulk-create Notifications for all members of this group. 

181 """ 

182 for user in self.members: 

183 Notification.objects.update_or_create( 

184 object_type=object_type, 

185 object_id=object_id, 

186 user=user, 

187 defaults=kwargs 

188 ) 

189 notify.alters_data = True 

190 

191 

192class Subscription(models.Model): 

193 """ 

194 A User's subscription to a particular object, to be notified of changes. 

195 """ 

196 created = models.DateTimeField( 

197 verbose_name=_('created'), 

198 auto_now_add=True 

199 ) 

200 user = models.ForeignKey( 

201 to=settings.AUTH_USER_MODEL, 

202 on_delete=models.CASCADE, 

203 related_name='subscriptions' 

204 ) 

205 object_type = models.ForeignKey( 

206 to='contenttypes.ContentType', 

207 on_delete=models.PROTECT 

208 ) 

209 object_id = models.PositiveBigIntegerField() 

210 object = GenericForeignKey( 

211 ct_field='object_type', 

212 fk_field='object_id' 

213 ) 

214 

215 objects = RestrictedQuerySet.as_manager() 

216 

217 class Meta: 

218 ordering = ('-created', 'user') 

219 indexes = ( 

220 models.Index(fields=('-created', 'user')), # Default ordering 

221 models.Index(fields=('object_type', 'object_id')), 

222 ) 

223 constraints = ( 

224 models.UniqueConstraint( 

225 fields=('object_type', 'object_id', 'user'), 

226 name='%(app_label)s_%(class)s_unique_per_object_and_user' 

227 ), 

228 ) 

229 verbose_name = _('subscription') 

230 verbose_name_plural = _('subscriptions') 

231 

232 def __str__(self): 

233 if self.object: 

234 return str(self.object) 

235 return super().__str__() 

236 

237 def get_absolute_url(self): 

238 return reverse('account:subscriptions') 

239 

240 def clean(self): 

241 super().clean() 

242 

243 # Validate the assigned object type 

244 if not has_feature(self.object_type, 'notifications'): 

245 raise ValidationError( 

246 _("Objects of this type ({type}) do not support notifications.").format(type=self.object_type) 

247 )