]> git.ozlabs.org Git - patchwork/blob - apps/patchwork/tests/notifications.py
notifications: Add code to send notifications
[patchwork] / apps / patchwork / tests / notifications.py
1 # Patchwork - automated patch tracking system
2 # Copyright (C) 2011 Jeremy Kerr <jk@ozlabs.org>
3 #
4 # This file is part of the Patchwork package.
5 #
6 # Patchwork is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
10 #
11 # Patchwork is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14 # GNU General Public License for more details.
15 #
16 # You should have received a copy of the GNU General Public License
17 # along with Patchwork; if not, write to the Free Software
18 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19
20 import datetime
21 from django.test import TestCase
22 from django.core.urlresolvers import reverse
23 from django.core import mail
24 from django.conf import settings
25 from django.db.utils import IntegrityError
26 from patchwork.models import Patch, State, PatchChangeNotification
27 from patchwork.tests.utils import defaults, create_maintainer
28 from patchwork.utils import send_notifications
29
30 class PatchNotificationModelTest(TestCase):
31     """Tests for the creation & update of the PatchChangeNotification model"""
32
33     def setUp(self):
34         self.project = defaults.project
35         self.project.send_notifications = True
36         self.project.save()
37         self.submitter = defaults.patch_author_person
38         self.submitter.save()
39         self.patch = Patch(project = self.project, msgid = 'testpatch',
40                         name = 'testpatch', content = '',
41                         submitter = self.submitter)
42
43     def tearDown(self):
44         self.patch.delete()
45         self.submitter.delete()
46         self.project.delete()
47
48     def testPatchCreation(self):
49         """Ensure we don't get a notification on create"""
50         self.patch.save()
51         self.assertEqual(PatchChangeNotification.objects.count(), 0)
52
53     def testPatchUninterestingChange(self):
54         """Ensure we don't get a notification for "uninteresting" changes"""
55         self.patch.save()
56         self.patch.archived = True
57         self.patch.save()
58         self.assertEqual(PatchChangeNotification.objects.count(), 0)
59
60     def testPatchChange(self):
61         """Ensure we get a notification for interesting patch changes"""
62         self.patch.save()
63         oldstate = self.patch.state
64         state = State.objects.exclude(pk = oldstate.pk)[0]
65
66         self.patch.state = state
67         self.patch.save()
68         self.assertEqual(PatchChangeNotification.objects.count(), 1)
69         notification = PatchChangeNotification.objects.all()[0]
70         self.assertEqual(notification.patch, self.patch)
71         self.assertEqual(notification.orig_state, oldstate)
72
73     def testNotificationCancelled(self):
74         """Ensure we cancel notifications that are no longer valid"""
75         self.patch.save()
76         oldstate = self.patch.state
77         state = State.objects.exclude(pk = oldstate.pk)[0]
78
79         self.patch.state = state
80         self.patch.save()
81         self.assertEqual(PatchChangeNotification.objects.count(), 1)
82
83         self.patch.state = oldstate
84         self.patch.save()
85         self.assertEqual(PatchChangeNotification.objects.count(), 0)
86
87     def testNotificationUpdated(self):
88         """Ensure we update notifications when the patch has a second change,
89            but keep the original patch details"""
90         self.patch.save()
91         oldstate = self.patch.state
92         newstates = State.objects.exclude(pk = oldstate.pk)[:2]
93
94         self.patch.state = newstates[0]
95         self.patch.save()
96         self.assertEqual(PatchChangeNotification.objects.count(), 1)
97         notification = PatchChangeNotification.objects.all()[0]
98         self.assertEqual(notification.orig_state, oldstate)
99         orig_timestamp = notification.last_modified
100                          
101         self.patch.state = newstates[1]
102         self.patch.save()
103         self.assertEqual(PatchChangeNotification.objects.count(), 1)
104         notification = PatchChangeNotification.objects.all()[0]
105         self.assertEqual(notification.orig_state, oldstate)
106         self.assertTrue(notification.last_modified > orig_timestamp)
107
108     def testProjectNotificationsDisabled(self):
109         """Ensure we don't see notifications created when a project is
110            configured not to send them"""
111         self.project.send_notifications = False
112         self.project.save()
113
114         self.patch.save()
115         oldstate = self.patch.state
116         state = State.objects.exclude(pk = oldstate.pk)[0]
117
118         self.patch.state = state
119         self.patch.save()
120         self.assertEqual(PatchChangeNotification.objects.count(), 0)
121
122 class PatchNotificationEmailTest(TestCase):
123
124     def setUp(self):
125         self.project = defaults.project
126         self.project.send_notifications = True
127         self.project.save()
128         self.submitter = defaults.patch_author_person
129         self.submitter.save()
130         self.patch = Patch(project = self.project, msgid = 'testpatch',
131                         name = 'testpatch', content = '',
132                         submitter = self.submitter)
133         self.patch.save()
134
135     def tearDown(self):
136         self.patch.delete()
137         self.submitter.delete()
138         self.project.delete()
139
140     def _expireNotifications(self, **kwargs):
141         timestamp = datetime.datetime.now() - \
142                     datetime.timedelta(minutes =
143                             settings.NOTIFICATION_DELAY_MINUTES + 1)
144
145         qs = PatchChangeNotification.objects.all()
146         if kwargs:
147             qs = qs.filter(**kwargs)
148
149         qs.update(last_modified = timestamp)
150
151     def testNoNotifications(self):
152         self.assertEquals(send_notifications(), [])
153
154     def testNoReadyNotifications(self):
155         """ We shouldn't see immediate notifications"""
156         PatchChangeNotification(patch = self.patch,
157                                orig_state = self.patch.state).save()
158
159         errors = send_notifications()
160         self.assertEquals(errors, [])
161         self.assertEquals(len(mail.outbox), 0)
162
163     def testNotifications(self):
164         PatchChangeNotification(patch = self.patch,
165                                orig_state = self.patch.state).save()
166         self._expireNotifications()
167
168         errors = send_notifications()
169         self.assertEquals(errors, [])
170         self.assertEquals(len(mail.outbox), 1)
171         msg = mail.outbox[0]
172         self.assertEquals(msg.to, [self.submitter.email])
173         self.assertTrue(self.patch.get_absolute_url() in msg.body)
174
175     def testNotificationMerge(self):
176         patches = [self.patch,
177                    Patch(project = self.project, msgid = 'testpatch-2',
178                          name = 'testpatch 2', content = '',
179                          submitter = self.submitter)]
180
181         for patch in patches:
182             patch.save()
183             PatchChangeNotification(patch = patch,
184                                    orig_state = patch.state).save()
185
186         self.assertEquals(PatchChangeNotification.objects.count(), len(patches))
187         self._expireNotifications()
188         errors = send_notifications()
189         self.assertEquals(errors, [])
190         self.assertEquals(len(mail.outbox), 1)
191         msg = mail.outbox[0]
192         self.assertTrue(patches[0].get_absolute_url() in msg.body)
193         self.assertTrue(patches[1].get_absolute_url() in msg.body)
194
195     def testUnexpiredNotificationMerge(self):
196         """Test that when there are multiple pending notifications, with
197            at least one within the notification delay, that other notifications
198            are held"""
199         patches = [self.patch,
200                    Patch(project = self.project, msgid = 'testpatch-2',
201                          name = 'testpatch 2', content = '',
202                          submitter = self.submitter)]
203
204         for patch in patches:
205             patch.save()
206             PatchChangeNotification(patch = patch,
207                                    orig_state = patch.state).save()
208
209         self.assertEquals(PatchChangeNotification.objects.count(), len(patches))
210         self._expireNotifications()
211
212         # update one notification, to bring it out of the notification delay
213         patches[0].state = State.objects.exclude(pk = patches[0].state.pk)[0]
214         patches[0].save()
215
216         # the updated notification should prevent the other from being sent
217         errors = send_notifications()
218         self.assertEquals(errors, [])
219         self.assertEquals(len(mail.outbox), 0)
220
221         # expire the updated notification
222         self._expireNotifications()
223
224         errors = send_notifications()
225         self.assertEquals(errors, [])
226         self.assertEquals(len(mail.outbox), 1)
227         msg = mail.outbox[0]
228         self.assertTrue(patches[0].get_absolute_url() in msg.body)
229         self.assertTrue(patches[1].get_absolute_url() in msg.body)