OLD | NEW |
(Empty) | |
| 1 # Copyright 2016 The Chromium Authors. All rights reserved. |
| 2 # Use of this source code is govered by a BSD-style |
| 3 # license that can be found in the LICENSE file or at |
| 4 # https://developers.google.com/open-source/licenses/bsd |
| 5 |
| 6 """Tests for notify.py.""" |
| 7 |
| 8 import os |
| 9 import unittest |
| 10 import urllib |
| 11 import webapp2 |
| 12 import webtest |
| 13 |
| 14 from google.appengine.api import taskqueue |
| 15 from google.appengine.ext import testbed |
| 16 |
| 17 from features import notify |
| 18 from framework import urls |
| 19 from proto import tracker_pb2 |
| 20 from services import service_manager |
| 21 from testing import fake |
| 22 from testing import testing_helpers |
| 23 from tracker import tracker_bizobj |
| 24 |
| 25 |
| 26 def MakeTestIssue(project_id, local_id, owner_id, reporter_id, is_spam=False): |
| 27 issue = tracker_pb2.Issue() |
| 28 issue.project_id = project_id |
| 29 issue.local_id = local_id |
| 30 issue.owner_id = owner_id |
| 31 issue.reporter_id = reporter_id |
| 32 issue.is_spam = is_spam |
| 33 return issue |
| 34 |
| 35 |
| 36 class SendNotificationTest(unittest.TestCase): |
| 37 |
| 38 def setUp(self): |
| 39 self.testbed = testbed.Testbed() |
| 40 self.testbed.activate() |
| 41 self.testbed.init_taskqueue_stub() |
| 42 self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME) |
| 43 self.taskqueue_stub._root_path = os.path.dirname( |
| 44 os.path.dirname(os.path.dirname( __file__ ))) |
| 45 |
| 46 def tearDown(self): |
| 47 self.testbed.deactivate() |
| 48 |
| 49 def testPrepareAndSendIssueChangeNotification(self): |
| 50 notify.PrepareAndSendIssueChangeNotification( |
| 51 project_id=789, |
| 52 local_id=1, |
| 53 hostport='testbed-test.appspotmail.com', |
| 54 commenter_id=1, |
| 55 seq_num=0, |
| 56 old_owner_id=2, |
| 57 send_email=True) |
| 58 |
| 59 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 60 url=urls.NOTIFY_ISSUE_CHANGE_TASK + '.do') |
| 61 self.assertEqual(1, len(tasks)) |
| 62 |
| 63 def testPrepareAndSendIssueBlockingNotification(self): |
| 64 notify.PrepareAndSendIssueBlockingNotification( |
| 65 project_id=789, |
| 66 hostport='testbed-test.appspotmail.com', |
| 67 local_id=1, |
| 68 delta_blocker_iids=[], |
| 69 commenter_id=1, |
| 70 send_email=True) |
| 71 |
| 72 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 73 url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do') |
| 74 self.assertEqual(0, len(tasks)) |
| 75 |
| 76 notify.PrepareAndSendIssueBlockingNotification( |
| 77 project_id=789, |
| 78 hostport='testbed-test.appspotmail.com', |
| 79 local_id=1, |
| 80 delta_blocker_iids=[2], |
| 81 commenter_id=1, |
| 82 send_email=True) |
| 83 |
| 84 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 85 url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do') |
| 86 self.assertEqual(1, len(tasks)) |
| 87 |
| 88 def testSendIssueBulkChangeNotification_CommentOnly(self): |
| 89 notify.SendIssueBulkChangeNotification( |
| 90 hostport='testbed-test.appspotmail.com', |
| 91 project_id=789, |
| 92 local_ids=[1], |
| 93 old_owner_ids=[2], |
| 94 comment_text='comment', |
| 95 commenter_id=1, |
| 96 amendments=[], |
| 97 send_email=True, |
| 98 users_by_id=2) |
| 99 |
| 100 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 101 url=urls.NOTIFY_BULK_CHANGE_TASK + '.do') |
| 102 self.assertEqual(1, len(tasks)) |
| 103 params = dict(urllib.unquote_plus(item).split('=') |
| 104 for item in tasks[0].payload.split('&')) |
| 105 self.assertEqual('comment', params['comment_text']) |
| 106 self.assertEqual('', params['amendments']) |
| 107 |
| 108 def testSendIssueBulkChangeNotification_Normal(self): |
| 109 notify.SendIssueBulkChangeNotification( |
| 110 hostport='testbed-test.appspotmail.com', |
| 111 project_id=789, |
| 112 local_ids=[1], |
| 113 old_owner_ids=[2], |
| 114 comment_text='comment', |
| 115 commenter_id=1, |
| 116 amendments=[ |
| 117 tracker_bizobj.MakeStatusAmendment('New', 'Old'), |
| 118 tracker_bizobj.MakeLabelsAmendment(['Added'], ['Removed']), |
| 119 tracker_bizobj.MakeStatusAmendment('New', 'Old'), |
| 120 ], |
| 121 send_email=True, |
| 122 users_by_id=2) |
| 123 |
| 124 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 125 url=urls.NOTIFY_BULK_CHANGE_TASK + '.do') |
| 126 self.assertEqual(1, len(tasks)) |
| 127 params = dict(urllib.unquote_plus(item).split('=') |
| 128 for item in tasks[0].payload.split('&')) |
| 129 self.assertEqual('comment', params['comment_text']) |
| 130 self.assertEqual( |
| 131 [' Status: New', |
| 132 ' Labels: -Removed Added'], |
| 133 params['amendments'].split('\n')) |
| 134 |
| 135 def testAddAllEmailTasks(self): |
| 136 notify.AddAllEmailTasks( |
| 137 tasks=[{'to': 'user'}, {'to': 'user2'}]) |
| 138 |
| 139 tasks = self.taskqueue_stub.get_filtered_tasks( |
| 140 url=urls.OUTBOUND_EMAIL_TASK + '.do') |
| 141 self.assertEqual(2, len(tasks)) |
| 142 |
| 143 |
| 144 class NotifyTaskHandleRequestTest(unittest.TestCase): |
| 145 |
| 146 def setUp(self): |
| 147 self.testbed = testbed.Testbed() |
| 148 self.testbed.activate() |
| 149 self.testbed.init_taskqueue_stub() |
| 150 self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME) |
| 151 self.taskqueue_stub._root_path = os.path.dirname( |
| 152 os.path.dirname(os.path.dirname( __file__ ))) |
| 153 self.services = service_manager.Services( |
| 154 user=fake.UserService(), |
| 155 usergroup=fake.UserGroupService(), |
| 156 project=fake.ProjectService(), |
| 157 config=fake.ConfigService(), |
| 158 issue=fake.IssueService(), |
| 159 issue_star=fake.IssueStarService(), |
| 160 features=fake.FeaturesService()) |
| 161 self.services.user.TestAddUser('requester@example.com', 1) |
| 162 self.services.user.TestAddUser('user@example.com', 2) |
| 163 self.services.project.TestAddProject( |
| 164 'test-project', owner_ids=[1], |
| 165 project_id=12345) |
| 166 issue1 = MakeTestIssue( |
| 167 project_id=12345, local_id=1, owner_id=2, reporter_id=1) |
| 168 self.services.issue.TestAddIssue(issue1) |
| 169 |
| 170 def VerifyParams(self, result, params): |
| 171 self.assertEqual( |
| 172 bool(params['send_email']), result['params']['send_email']) |
| 173 if 'id' in params: |
| 174 self.assertEqual(params['id'], result['params']['local_id']) |
| 175 if 'ids' in params: |
| 176 self.assertEqual([int(p) for p in params['ids'].split(',')], |
| 177 result['params']['local_ids']) |
| 178 self.assertEqual(params['project_id'], result['params']['project_id']) |
| 179 |
| 180 def testNotifyIssueChangeTask(self): |
| 181 task = notify.NotifyIssueChangeTask( |
| 182 request=None, response=None, services=self.services) |
| 183 params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0, |
| 184 'commenter_id': 2} |
| 185 mr = testing_helpers.MakeMonorailRequest( |
| 186 user_info={'user_id': 1}, |
| 187 params=params, |
| 188 method='POST', |
| 189 services=self.services) |
| 190 result = task.HandleRequest(mr) |
| 191 self.VerifyParams(result, params) |
| 192 |
| 193 def testNotifyIssueChangeTask_spam(self): |
| 194 issue = MakeTestIssue( |
| 195 project_id=12345, local_id=1, owner_id=1, reporter_id=1, |
| 196 is_spam=True) |
| 197 self.services.issue.TestAddIssue(issue) |
| 198 task = notify.NotifyIssueChangeTask( |
| 199 request=None, response=None, services=self.services) |
| 200 params = {'send_email': 0, 'project_id': 12345, 'id': 1, 'seq': 0, |
| 201 'commenter_id': 2} |
| 202 mr = testing_helpers.MakeMonorailRequest( |
| 203 user_info={'user_id': 1}, |
| 204 params=params, |
| 205 method='POST', |
| 206 services=self.services) |
| 207 result = task.HandleRequest(mr) |
| 208 self.assertEquals(0, len(result['notified'])) |
| 209 |
| 210 def testNotifyBlockingChangeTask(self): |
| 211 issue2 = MakeTestIssue( |
| 212 project_id=12345, local_id=2, owner_id=2, reporter_id=1) |
| 213 self.services.issue.TestAddIssue(issue2) |
| 214 task = notify.NotifyBlockingChangeTask( |
| 215 request=None, response=None, services=self.services) |
| 216 params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0, |
| 217 'delta_blocker_iids': 2, 'commenter_id': 1} |
| 218 mr = testing_helpers.MakeMonorailRequest( |
| 219 user_info={'user_id': 1}, |
| 220 params=params, |
| 221 method='POST', |
| 222 services=self.services) |
| 223 result = task.HandleRequest(mr) |
| 224 self.VerifyParams(result, params) |
| 225 |
| 226 def testNotifyBlockingChangeTask_spam(self): |
| 227 issue2 = MakeTestIssue( |
| 228 project_id=12345, local_id=2, owner_id=2, reporter_id=1, |
| 229 is_spam=True) |
| 230 self.services.issue.TestAddIssue(issue2) |
| 231 task = notify.NotifyBlockingChangeTask( |
| 232 request=None, response=None, services=self.services) |
| 233 params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0, |
| 234 'delta_blocker_iids': 2, 'commenter_id': 1} |
| 235 mr = testing_helpers.MakeMonorailRequest( |
| 236 user_info={'user_id': 1}, |
| 237 params=params, |
| 238 method='POST', |
| 239 services=self.services) |
| 240 result = task.HandleRequest(mr) |
| 241 self.assertEquals(0, len(result['notified'])) |
| 242 |
| 243 def testNotifyBulkChangeTask(self): |
| 244 issue2 = MakeTestIssue( |
| 245 project_id=12345, local_id=2, owner_id=2, reporter_id=1) |
| 246 self.services.issue.TestAddIssue(issue2) |
| 247 task = notify.NotifyBulkChangeTask( |
| 248 request=None, response=None, services=self.services) |
| 249 params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', 'seq': 0, |
| 250 'old_owner_ids': '1,1', 'commenter_id': 1} |
| 251 mr = testing_helpers.MakeMonorailRequest( |
| 252 user_info={'user_id': 1}, |
| 253 params=params, |
| 254 method='POST', |
| 255 services=self.services) |
| 256 result = task.HandleRequest(mr) |
| 257 self.VerifyParams(result, params) |
| 258 |
| 259 def testNotifyBulkChangeTask_spam(self): |
| 260 issue2 = MakeTestIssue( |
| 261 project_id=12345, local_id=2, owner_id=2, reporter_id=1, |
| 262 is_spam=True) |
| 263 self.services.issue.TestAddIssue(issue2) |
| 264 task = notify.NotifyBulkChangeTask( |
| 265 request=None, response=None, services=self.services) |
| 266 params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', 'seq': 0, |
| 267 'old_owner_ids': '1,1', 'commenter_id': 1} |
| 268 mr = testing_helpers.MakeMonorailRequest( |
| 269 user_info={'user_id': 1}, |
| 270 params=params, |
| 271 method='POST', |
| 272 services=self.services) |
| 273 result = task.HandleRequest(mr) |
| 274 self.assertEquals(1, len(result['notified'])) |
| 275 |
| 276 def testOutboundEmailTask(self): |
| 277 task = notify.OutboundEmailTask( |
| 278 request=None, response=None, services=self.services) |
| 279 params = { |
| 280 'from_addr': 'requester@example.com', |
| 281 'reply_to': 'user@example.com', |
| 282 'to': 'user@example.com', |
| 283 'subject': 'Test subject'} |
| 284 mr = testing_helpers.MakeMonorailRequest( |
| 285 user_info={'user_id': 1}, |
| 286 params=params, |
| 287 method='POST', |
| 288 services=self.services) |
| 289 result = task.HandleRequest(mr) |
| 290 self.assertEqual(params['from_addr'], result['sender']) |
| 291 self.assertEqual(params['subject'], result['subject']) |
OLD | NEW |