Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(172)

Unified Diff: appengine/monorail/features/test/notify_test.py

Issue 1868553004: Open Source Monorail (Closed) Base URL: https://chromium.googlesource.com/infra/infra.git@master
Patch Set: Rebase Created 4 years, 8 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: appengine/monorail/features/test/notify_test.py
diff --git a/appengine/monorail/features/test/notify_test.py b/appengine/monorail/features/test/notify_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..f6a0a1a2a89070fc80e4232ca51f0db5d628ce2b
--- /dev/null
+++ b/appengine/monorail/features/test/notify_test.py
@@ -0,0 +1,291 @@
+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is govered by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""Tests for notify.py."""
+
+import os
+import unittest
+import urllib
+import webapp2
+import webtest
+
+from google.appengine.api import taskqueue
+from google.appengine.ext import testbed
+
+from features import notify
+from framework import urls
+from proto import tracker_pb2
+from services import service_manager
+from testing import fake
+from testing import testing_helpers
+from tracker import tracker_bizobj
+
+
+def MakeTestIssue(project_id, local_id, owner_id, reporter_id, is_spam=False):
+ issue = tracker_pb2.Issue()
+ issue.project_id = project_id
+ issue.local_id = local_id
+ issue.owner_id = owner_id
+ issue.reporter_id = reporter_id
+ issue.is_spam = is_spam
+ return issue
+
+
+class SendNotificationTest(unittest.TestCase):
+
+ def setUp(self):
+ self.testbed = testbed.Testbed()
+ self.testbed.activate()
+ self.testbed.init_taskqueue_stub()
+ self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
+ self.taskqueue_stub._root_path = os.path.dirname(
+ os.path.dirname(os.path.dirname( __file__ )))
+
+ def tearDown(self):
+ self.testbed.deactivate()
+
+ def testPrepareAndSendIssueChangeNotification(self):
+ notify.PrepareAndSendIssueChangeNotification(
+ project_id=789,
+ local_id=1,
+ hostport='testbed-test.appspotmail.com',
+ commenter_id=1,
+ seq_num=0,
+ old_owner_id=2,
+ send_email=True)
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.NOTIFY_ISSUE_CHANGE_TASK + '.do')
+ self.assertEqual(1, len(tasks))
+
+ def testPrepareAndSendIssueBlockingNotification(self):
+ notify.PrepareAndSendIssueBlockingNotification(
+ project_id=789,
+ hostport='testbed-test.appspotmail.com',
+ local_id=1,
+ delta_blocker_iids=[],
+ commenter_id=1,
+ send_email=True)
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
+ self.assertEqual(0, len(tasks))
+
+ notify.PrepareAndSendIssueBlockingNotification(
+ project_id=789,
+ hostport='testbed-test.appspotmail.com',
+ local_id=1,
+ delta_blocker_iids=[2],
+ commenter_id=1,
+ send_email=True)
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.NOTIFY_BLOCKING_CHANGE_TASK + '.do')
+ self.assertEqual(1, len(tasks))
+
+ def testSendIssueBulkChangeNotification_CommentOnly(self):
+ notify.SendIssueBulkChangeNotification(
+ hostport='testbed-test.appspotmail.com',
+ project_id=789,
+ local_ids=[1],
+ old_owner_ids=[2],
+ comment_text='comment',
+ commenter_id=1,
+ amendments=[],
+ send_email=True,
+ users_by_id=2)
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.NOTIFY_BULK_CHANGE_TASK + '.do')
+ self.assertEqual(1, len(tasks))
+ params = dict(urllib.unquote_plus(item).split('=')
+ for item in tasks[0].payload.split('&'))
+ self.assertEqual('comment', params['comment_text'])
+ self.assertEqual('', params['amendments'])
+
+ def testSendIssueBulkChangeNotification_Normal(self):
+ notify.SendIssueBulkChangeNotification(
+ hostport='testbed-test.appspotmail.com',
+ project_id=789,
+ local_ids=[1],
+ old_owner_ids=[2],
+ comment_text='comment',
+ commenter_id=1,
+ amendments=[
+ tracker_bizobj.MakeStatusAmendment('New', 'Old'),
+ tracker_bizobj.MakeLabelsAmendment(['Added'], ['Removed']),
+ tracker_bizobj.MakeStatusAmendment('New', 'Old'),
+ ],
+ send_email=True,
+ users_by_id=2)
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.NOTIFY_BULK_CHANGE_TASK + '.do')
+ self.assertEqual(1, len(tasks))
+ params = dict(urllib.unquote_plus(item).split('=')
+ for item in tasks[0].payload.split('&'))
+ self.assertEqual('comment', params['comment_text'])
+ self.assertEqual(
+ [' Status: New',
+ ' Labels: -Removed Added'],
+ params['amendments'].split('\n'))
+
+ def testAddAllEmailTasks(self):
+ notify.AddAllEmailTasks(
+ tasks=[{'to': 'user'}, {'to': 'user2'}])
+
+ tasks = self.taskqueue_stub.get_filtered_tasks(
+ url=urls.OUTBOUND_EMAIL_TASK + '.do')
+ self.assertEqual(2, len(tasks))
+
+
+class NotifyTaskHandleRequestTest(unittest.TestCase):
+
+ def setUp(self):
+ self.testbed = testbed.Testbed()
+ self.testbed.activate()
+ self.testbed.init_taskqueue_stub()
+ self.taskqueue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)
+ self.taskqueue_stub._root_path = os.path.dirname(
+ os.path.dirname(os.path.dirname( __file__ )))
+ self.services = service_manager.Services(
+ user=fake.UserService(),
+ usergroup=fake.UserGroupService(),
+ project=fake.ProjectService(),
+ config=fake.ConfigService(),
+ issue=fake.IssueService(),
+ issue_star=fake.IssueStarService(),
+ features=fake.FeaturesService())
+ self.services.user.TestAddUser('requester@example.com', 1)
+ self.services.user.TestAddUser('user@example.com', 2)
+ self.services.project.TestAddProject(
+ 'test-project', owner_ids=[1],
+ project_id=12345)
+ issue1 = MakeTestIssue(
+ project_id=12345, local_id=1, owner_id=2, reporter_id=1)
+ self.services.issue.TestAddIssue(issue1)
+
+ def VerifyParams(self, result, params):
+ self.assertEqual(
+ bool(params['send_email']), result['params']['send_email'])
+ if 'id' in params:
+ self.assertEqual(params['id'], result['params']['local_id'])
+ if 'ids' in params:
+ self.assertEqual([int(p) for p in params['ids'].split(',')],
+ result['params']['local_ids'])
+ self.assertEqual(params['project_id'], result['params']['project_id'])
+
+ def testNotifyIssueChangeTask(self):
+ task = notify.NotifyIssueChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0,
+ 'commenter_id': 2}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.VerifyParams(result, params)
+
+ def testNotifyIssueChangeTask_spam(self):
+ issue = MakeTestIssue(
+ project_id=12345, local_id=1, owner_id=1, reporter_id=1,
+ is_spam=True)
+ self.services.issue.TestAddIssue(issue)
+ task = notify.NotifyIssueChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 0, 'project_id': 12345, 'id': 1, 'seq': 0,
+ 'commenter_id': 2}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.assertEquals(0, len(result['notified']))
+
+ def testNotifyBlockingChangeTask(self):
+ issue2 = MakeTestIssue(
+ project_id=12345, local_id=2, owner_id=2, reporter_id=1)
+ self.services.issue.TestAddIssue(issue2)
+ task = notify.NotifyBlockingChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0,
+ 'delta_blocker_iids': 2, 'commenter_id': 1}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.VerifyParams(result, params)
+
+ def testNotifyBlockingChangeTask_spam(self):
+ issue2 = MakeTestIssue(
+ project_id=12345, local_id=2, owner_id=2, reporter_id=1,
+ is_spam=True)
+ self.services.issue.TestAddIssue(issue2)
+ task = notify.NotifyBlockingChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 1, 'project_id': 12345, 'id': 1, 'seq': 0,
+ 'delta_blocker_iids': 2, 'commenter_id': 1}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.assertEquals(0, len(result['notified']))
+
+ def testNotifyBulkChangeTask(self):
+ issue2 = MakeTestIssue(
+ project_id=12345, local_id=2, owner_id=2, reporter_id=1)
+ self.services.issue.TestAddIssue(issue2)
+ task = notify.NotifyBulkChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', 'seq': 0,
+ 'old_owner_ids': '1,1', 'commenter_id': 1}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.VerifyParams(result, params)
+
+ def testNotifyBulkChangeTask_spam(self):
+ issue2 = MakeTestIssue(
+ project_id=12345, local_id=2, owner_id=2, reporter_id=1,
+ is_spam=True)
+ self.services.issue.TestAddIssue(issue2)
+ task = notify.NotifyBulkChangeTask(
+ request=None, response=None, services=self.services)
+ params = {'send_email': 1, 'project_id': 12345, 'ids': '1,2', 'seq': 0,
+ 'old_owner_ids': '1,1', 'commenter_id': 1}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.assertEquals(1, len(result['notified']))
+
+ def testOutboundEmailTask(self):
+ task = notify.OutboundEmailTask(
+ request=None, response=None, services=self.services)
+ params = {
+ 'from_addr': 'requester@example.com',
+ 'reply_to': 'user@example.com',
+ 'to': 'user@example.com',
+ 'subject': 'Test subject'}
+ mr = testing_helpers.MakeMonorailRequest(
+ user_info={'user_id': 1},
+ params=params,
+ method='POST',
+ services=self.services)
+ result = task.HandleRequest(mr)
+ self.assertEqual(params['from_addr'], result['sender'])
+ self.assertEqual(params['subject'], result['subject'])
« no previous file with comments | « appengine/monorail/features/test/notify_helpers_test.py ('k') | appengine/monorail/features/test/prettify_test.py » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698