Coverage for website/pushnotifications/models.py: 84.26%

1import datetime 

2 

3from django.conf import settings 

4from django.db import models 

5from django.utils import timezone 

6from django.utils.translation import gettext_lazy as _ 

7 

8from firebase_admin import exceptions, messaging 

9 

10 

11class Category(models.Model): 

12 """Describes a Message category.""" 

13 

14 # These should be the keys of the categories that we automatically created 

15 # in the migrations (0012 to be specific) 

16 GENERAL = "general" 

17 PIZZA = "pizza" 

18 EVENT = "event" 

19 NEWSLETTER = "newsletter" 

20 PARTNER = "partner" 

21 PHOTO = "photo" 

22 BOARD = "board" 

23 THABLOID = "thabloid" 

24 

25 key = models.CharField(max_length=16, primary_key=True) 

26 

27 name = models.CharField( 

28 _("name"), 

29 max_length=32, 

30 ) 

31 

32 description = models.TextField(_("description"), default="") 

33 

34 def __str__(self): 

35 return self.name 

36 

37 

38def default_receive_category(): 

39 return Category.objects.filter(key=Category.GENERAL) 

40 

41 

42class Device(models.Model): 

43 """Describes a device.""" 

44 

45 DEVICE_TYPES = (("ios", "iOS"), ("android", "Android")) 

46 

47 registration_id = models.TextField( 

48 verbose_name=_("registration token"), unique=True 

49 ) 

50 type = models.CharField(choices=DEVICE_TYPES, max_length=10) 

51 active = models.BooleanField( 

52 verbose_name=_("active"), 

53 default=True, 

54 help_text=_("Inactive devices will not be sent notifications"), 

55 ) 

56 user = models.ForeignKey( 

57 settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=False, null=False 

58 ) 

59 date_created = models.DateTimeField( 

60 verbose_name=_("registration date"), auto_now_add=True, null=False 

61 ) 

62 

63 receive_category = models.ManyToManyField( 

64 Category, default=default_receive_category 

65 ) 

66 

67 class Meta: 

68 unique_together = ( 

69 "registration_id", 

70 "user", 

71 ) 

72 

73 def __str__(self): 

74 return _("{user}s {device_type} device").format( 

75 user=self.user, device_type=self.type 

76 ) 

77 

78 

79class NormalMessageManager(models.Manager): 

80 """Returns manual messages only.""" 

81 

82 def get_queryset(self): 

83 return super().get_queryset().filter(scheduledmessage__scheduled=None) 

84 

85 

86class MessageManager(models.Manager): 

87 """Returns all messages.""" 

88 

89 

90class Message(models.Model): 

91 """Describes a push notification.""" 

92 

93 objects = NormalMessageManager() 

94 all_objects = MessageManager() 

95 

96 users = models.ManyToManyField(settings.AUTH_USER_MODEL) 

97 title = models.CharField(max_length=150, verbose_name=_("title")) 

98 body = models.TextField(verbose_name=_("body")) 

99 url = models.CharField( 

100 verbose_name=_("url"), 

101 max_length=256, 

102 null=True, 

103 blank=True, 

104 ) 

105 category = models.ForeignKey( 

106 Category, 

107 on_delete=models.CASCADE, 

108 verbose_name=_("category"), 

109 default="general", 

110 ) 

111 sent = models.DateTimeField( 

112 verbose_name=_("sent"), 

113 null=True, 

114 ) 

115 failure = models.IntegerField( 

116 verbose_name=_("failure"), 

117 blank=True, 

118 null=True, 

119 ) 

120 success = models.IntegerField( 

121 verbose_name=_("success"), 

122 blank=True, 

123 null=True, 

124 ) 

125 

126 def __str__(self): 

127 return f"{self.title}: {self.body}" 

128 

129 def send(self, **kwargs): 

130 if self: 130 ↛ exitline 130 didn't return from function 'send' because the condition on line 130 was always true

131 success_total = 0 

132 failure_total = 0 

133 ttl = kwargs.get("ttl", 3600) 

134 

135 reg_ids = list( 

136 Device.objects.filter( 

137 user__in=self.users.all(), 

138 receive_category__key=self.category_id, 

139 active=True, 

140 ).values_list("registration_id", flat=True) 

141 ) 

142 

143 data = kwargs.get("data", {}) 

144 if self.url is not None: 144 ↛ 146line 144 didn't jump to line 146 because the condition on line 144 was always true

145 data["url"] = self.url 

146 data["title"] = self.title 

147 data["body"] = str(self.body) 

148 

149 message = messaging.Message( 

150 notification=messaging.Notification( 

151 title=data["title"], 

152 body=data["body"], 

153 ), 

154 data=data, 

155 android=messaging.AndroidConfig( 

156 ttl=datetime.timedelta(seconds=ttl), 

157 priority="normal", 

158 notification=messaging.AndroidNotification( 

159 color="#E62272", 

160 sound="default", 

161 ), 

162 ), 

163 ) 

164 

165 for reg_id in reg_ids: 165 ↛ 166line 165 didn't jump to line 166 because the loop on line 165 never started

166 message.token = reg_id 

167 try: 

168 messaging.send(message, dry_run=kwargs.get("dry_run", False)) 

169 success_total += 1 

170 except messaging.UnregisteredError: 

171 failure_total += 1 

172 Device.objects.filter(registration_id=reg_id).delete() 

173 except exceptions.InvalidArgumentError: 

174 failure_total += 1 

175 Device.objects.filter(registration_id=reg_id).update(active=False) 

176 except exceptions.FirebaseError: 

177 failure_total += 1 

178 

179 self.sent = timezone.now() 

180 self.success = success_total 

181 self.failure = failure_total 

182 self.save() 

183 

184 

185class ScheduledMessageManager(models.Manager): 

186 """Returns scheduled messages only.""" 

187 

188 

189class ScheduledMessage(Message): 

190 """Describes a scheduled push notification.""" 

191 

192 objects = ScheduledMessageManager() 

193 

194 scheduled = models.BooleanField(default=True) 

195 time = models.DateTimeField() 

196 executed = models.DateTimeField(null=True) 

197 

198 

199class NewAlbumMessageManager(models.Manager): 

200 """Returns new album messages only.""" 

201 

202 

203class NewAlbumMessage(ScheduledMessage): 

204 """A scheduled message to notify users of a new album.""" 

205 

206 objects = NewAlbumMessageManager() 

207 

208 album = models.OneToOneField( 

209 "photos.Album", 

210 related_name="new_album_notification", 

211 on_delete=models.deletion.CASCADE, 

212 ) 

213 

214 

215class FoodOrderReminderMessageManager(models.Manager): 

216 """Returns food event order end messages only.""" 

217 

218 

219class FoodOrderReminderMessage(ScheduledMessage): 

220 """A scheduled message to notify users of a food order reminder.""" 

221 

222 objects = FoodOrderReminderMessageManager() 

223 

224 food_event = models.OneToOneField( 

225 "pizzas.FoodEvent", 

226 related_name="end_reminder", 

227 on_delete=models.deletion.CASCADE, 

228 ) 

229 

230 

231class RegistrationReminderMessageManager(models.Manager): 

232 """Returns event registration reminders only.""" 

233 

234 

235class RegistrationReminderMessage(ScheduledMessage): 

236 """A scheduled message to notify users of something related to an event.""" 

237 

238 objects = RegistrationReminderMessageManager() 

239 

240 event = models.OneToOneField( 

241 "events.Event", 

242 related_name="registration_reminder", 

243 on_delete=models.deletion.CASCADE, 

244 ) 

245 

246 

247class EventStartReminderMessageManager(models.Manager): 

248 """Returns event start reminders only.""" 

249 

250 

251class EventStartReminderMessage(ScheduledMessage): 

252 """A scheduled message to notify users of an event start.""" 

253 

254 objects = EventStartReminderMessageManager() 

255 

256 event = models.OneToOneField( 

257 "events.Event", 

258 related_name="start_reminder", 

259 on_delete=models.deletion.CASCADE, 

260 )