-
Notifications
You must be signed in to change notification settings - Fork 91
Expand file tree
/
Copy pathtest_notifications.py
More file actions
219 lines (168 loc) · 7.64 KB
/
test_notifications.py
File metadata and controls
219 lines (168 loc) · 7.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
# Patchwork - automated patch tracking system
# Copyright (C) 2011 Jeremy Kerr <jk@ozlabs.org>
#
# SPDX-License-Identifier: GPL-2.0-or-later
import datetime
from django.conf import settings
from django.core import mail
from django.test import TestCase
from django.utils import timezone as tz_utils
from patchwork.models import EmailOptout
from patchwork.models import PatchChangeNotification
from patchwork.notifications import send_notifications
from patchwork.tests.utils import create_patch
from patchwork.tests.utils import create_patches
from patchwork.tests.utils import create_project
from patchwork.tests.utils import create_state
class PatchNotificationModelTest(TestCase):
"""Tests for the creation and update of the PatchChangeNotifications."""
def setUp(self):
self.project = create_project(send_notifications=True)
def test_patch_creation(self):
"""Ensure we don't get a notification on create."""
create_patch(project=self.project)
self.assertEqual(PatchChangeNotification.objects.count(), 0)
def test_patch_uninteresting_change(self):
"""Ensure we don't get a notification for "uninteresting" changes"""
patch = create_patch(project=self.project)
patch.archived = True
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 0)
def test_patch_change(self):
"""Ensure we get a notification for interesting patch changes"""
patch = create_patch(project=self.project)
oldstate = patch.state
state = create_state()
patch.state = state
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 1)
notification = PatchChangeNotification.objects.all()[0]
self.assertEqual(notification.patch, patch)
self.assertEqual(notification.orig_state, oldstate)
def test_notification_cancelled(self):
"""Ensure we cancel notifications that are no longer valid"""
patch = create_patch(project=self.project)
oldstate = patch.state
state = create_state()
patch.state = state
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 1)
patch.state = oldstate
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 0)
def test_notification_updated(self):
"""Ensure we update notifications when the patch has a second change,
but keep the original patch details"""
patch = create_patch(project=self.project)
oldstate = patch.state
newstates = [create_state(), create_state()]
patch.state = newstates[0]
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 1)
notification = PatchChangeNotification.objects.all()[0]
self.assertEqual(notification.orig_state, oldstate)
orig_timestamp = notification.last_modified
patch.state = newstates[1]
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 1)
notification = PatchChangeNotification.objects.all()[0]
self.assertEqual(notification.orig_state, oldstate)
self.assertTrue(notification.last_modified >= orig_timestamp)
def test_notifications_disabled(self):
"""Ensure we don't see notifications created when a project is
configured not to send them"""
patch = create_patch() # don't use self.project
state = create_state()
patch.state = state
patch.save()
self.assertEqual(PatchChangeNotification.objects.count(), 0)
class PatchNotificationEmailTest(TestCase):
def setUp(self):
self.project = create_project(send_notifications=True)
def _expire_notifications(self, **kwargs):
timestamp = tz_utils.now() - datetime.timedelta(
minutes=settings.NOTIFICATION_DELAY_MINUTES + 1
)
qs = PatchChangeNotification.objects.all()
if kwargs:
qs = qs.filter(**kwargs)
qs.update(last_modified=timestamp)
def test_no_notifications(self):
self.assertEqual(send_notifications(), [])
def test_no_ready_notifications(self):
"""We shouldn't see immediate notifications."""
patch = create_patch(project=self.project)
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 0)
def test_notifications(self):
patch = create_patch(project=self.project)
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
self._expire_notifications()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 1)
msg = mail.outbox[0]
self.assertEqual(msg.to, [patch.submitter.email])
self.assertIn(patch.get_absolute_url(), msg.body)
def test_notification_escaping(self):
patch = create_patch(
name='Patch name with " character', project=self.project
)
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
self._expire_notifications()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 1)
msg = mail.outbox[0]
self.assertEqual(msg.to, [patch.submitter.email])
self.assertNotIn('"', msg.body)
def test_notification_optout(self):
"""Ensure opt-out addresses don't get notifications."""
patch = create_patch(project=self.project)
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
self._expire_notifications()
EmailOptout(email=patch.submitter.email).save()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 0)
def test_notification_merge(self):
"""Ensure only one summary email is delivered to each user."""
patches = create_patches(2, project=self.project)
for patch in patches:
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
self.assertEqual(PatchChangeNotification.objects.count(), len(patches))
self._expire_notifications()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 1)
msg = mail.outbox[0]
for patch in patches:
self.assertIn(patch.get_absolute_url(), msg.body)
def test_unexpired_notification_merge(self):
"""Test that when there are multiple pending notifications, with
at least one within the notification delay, that other notifications
are held"""
patches = create_patches(2, project=self.project)
for patch in patches:
patch.save()
PatchChangeNotification(patch=patch, orig_state=patch.state).save()
state = create_state()
self.assertEqual(PatchChangeNotification.objects.count(), len(patches))
self._expire_notifications()
# update one notification, to bring it out of the notification delay
patches[0].state = state
patches[0].save()
# the updated notification should prevent the other from being sent
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 0)
# expire the updated notification
self._expire_notifications()
errors = send_notifications()
self.assertEqual(errors, [])
self.assertEqual(len(mail.outbox), 1)
msg = mail.outbox[0]
for patch in patches:
self.assertIn(patch.get_absolute_url(), msg.body)