| # Copyright 2016 The Chromium Authors. All rights reserved. |
| # Use of this source code is governed 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 the API v1.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import datetime |
| import endpoints |
| import logging |
| from mock import Mock, patch, ANY |
| import time |
| import unittest |
| import webtest |
| |
| from google.appengine.api import oauth |
| from protorpc import messages |
| from protorpc import message_types |
| |
| from features import send_notifications |
| from framework import authdata |
| from framework import exceptions |
| from framework import framework_constants |
| from framework import permissions |
| from framework import profiler |
| from framework import template_helpers |
| from proto import api_pb2_v1 |
| from proto import project_pb2 |
| from proto import tracker_pb2 |
| from search import frontendsearchpipeline |
| from services import api_svc_v1 |
| from services import service_manager |
| from services import template_svc |
| from services import tracker_fulltext |
| from testing import fake |
| from testing import testing_helpers |
| from testing_utils import testing |
| from tracker import tracker_bizobj |
| from tracker import tracker_constants |
| |
| |
| def MakeFakeServiceManager(): |
| return 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(), |
| template=Mock(spec=template_svc.TemplateService), |
| cache_manager=fake.CacheManager()) |
| |
| |
| class FakeMonorailApiRequest(object): |
| |
| def __init__(self, request, services, perms=None): |
| self.profiler = profiler.Profiler() |
| self.cnxn = None |
| self.auth = authdata.AuthData.FromEmail( |
| self.cnxn, request['requester'], services) |
| self.me_user_id = self.auth.user_id |
| self.project_name = None |
| self.project = None |
| self.viewed_username = None |
| self.viewed_user_auth = None |
| self.config = None |
| if 'userId' in request: |
| self.viewed_username = request['userId'] |
| self.viewed_user_auth = authdata.AuthData.FromEmail( |
| self.cnxn, self.viewed_username, services) |
| else: |
| assert 'groupName' in request |
| self.viewed_username = request['groupName'] |
| try: |
| self.viewed_user_auth = authdata.AuthData.FromEmail( |
| self.cnxn, self.viewed_username, services) |
| except exceptions.NoSuchUserException: |
| self.viewed_user_auth = None |
| if 'projectId' in request: |
| self.project_name = request['projectId'] |
| self.project = services.project.GetProjectByName( |
| self.cnxn, self.project_name) |
| self.config = services.config.GetProjectConfig( |
| self.cnxn, self.project_id) |
| self.perms = perms or permissions.GetPermissions( |
| self.auth.user_pb, self.auth.effective_ids, self.project) |
| self.granted_perms = set() |
| |
| self.params = { |
| 'can': request.get('can', 1), |
| 'start': request.get('startIndex', 0), |
| 'num': request.get('maxResults', 100), |
| 'q': request.get('q', ''), |
| 'sort': request.get('sort', ''), |
| 'groupby': '', |
| 'projects': request.get('additionalProject', []) + [self.project_name]} |
| self.use_cached_searches = True |
| self.errors = template_helpers.EZTError() |
| self.mode = None |
| |
| self.query_project_names = self.GetParam('projects') |
| self.group_by_spec = self.GetParam('groupby') |
| self.sort_spec = self.GetParam('sort') |
| self.query = self.GetParam('q') |
| self.can = self.GetParam('can') |
| self.start = self.GetParam('start') |
| self.num = self.GetParam('num') |
| self.warnings = [] |
| |
| def CleanUp(self): |
| self.cnxn = None |
| |
| @property |
| def project_id(self): |
| return self.project.project_id if self.project else None |
| |
| def GetParam(self, query_param_name, default_value=None, |
| _antitamper_re=None): |
| return self.params.get(query_param_name, default_value) |
| |
| |
| class FakeFrontendSearchPipeline(object): |
| |
| def __init__(self): |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, owner_id=222, status='New', summary='sum') |
| issue2 = fake.MakeTestIssue( |
| project_id=12345, local_id=2, owner_id=222, status='New', summary='sum') |
| self.allowed_results = [issue1, issue2] |
| self.visible_results = [issue1] |
| self.total_count = len(self.allowed_results) |
| self.config = None |
| self.projectId = 0 |
| |
| def SearchForIIDs(self): |
| pass |
| |
| def MergeAndSortIssues(self): |
| pass |
| |
| def Paginate(self): |
| pass |
| |
| |
| class MonorailApiBadAuthTest(testing.EndpointsTestCase): |
| |
| api_service_cls = api_svc_v1.MonorailApi |
| |
| def setUp(self): |
| super(MonorailApiBadAuthTest, self).setUp() |
| self.requester = RequesterMock(email='requester@example.com') |
| self.mock(endpoints, 'get_current_user', lambda: None) |
| self.request = {'userId': 'user@example.com'} |
| |
| def testUsersGet_BadOAuth(self): |
| """The requester's token is invalid, e.g., because it expired.""" |
| oauth.get_current_user = Mock( |
| return_value=RequesterMock(email='test@example.com')) |
| oauth.get_current_user.side_effect = oauth.Error() |
| with self.assertRaises(webtest.AppError) as cm: |
| self.call_api('users_get', self.request) |
| self.assertTrue(cm.exception.message.startswith('Bad response: 401')) |
| |
| |
| class MonorailApiTest(testing.EndpointsTestCase): |
| |
| api_service_cls = api_svc_v1.MonorailApi |
| |
| def setUp(self): |
| super(MonorailApiTest, self).setUp() |
| # Load queue.yaml. |
| self.requester = RequesterMock(email='requester@example.com') |
| self.mock(endpoints, 'get_current_user', lambda: self.requester) |
| self.config = None |
| self.services = MakeFakeServiceManager() |
| self.mock(api_svc_v1.MonorailApi, '_services', self.services) |
| self.services.user.TestAddUser('requester@example.com', 111) |
| self.services.user.TestAddUser('user@example.com', 222) |
| self.services.user.TestAddUser('group@example.com', 123) |
| self.services.usergroup.TestAddGroupSettings(123, 'group@example.com') |
| self.request = { |
| 'userId': 'user@example.com', |
| 'ownerProjectsOnly': False, |
| 'requester': 'requester@example.com', |
| 'projectId': 'test-project', |
| 'issueId': 1} |
| self.mock(api_svc_v1.MonorailApi, 'mar_factory', |
| lambda x, y, z: FakeMonorailApiRequest( |
| self.request, self.services)) |
| |
| # api_base_checks is tested in AllBaseChecksTest, |
| # so mock it to reduce noise. |
| self.mock(api_svc_v1, 'api_base_checks', |
| lambda x, y, z, u, v, w: ('id', 'email')) |
| |
| self.mock(tracker_fulltext, 'IndexIssues', lambda x, y, z, u, v: None) |
| |
| def SetUpComponents( |
| self, project_id, component_id, component_name, component_doc='doc', |
| deprecated=False, admin_ids=None, cc_ids=None, created=100000, |
| creator=111): |
| admin_ids = admin_ids or [] |
| cc_ids = cc_ids or [] |
| self.config = self.services.config.GetProjectConfig( |
| 'fake cnxn', project_id) |
| self.services.config.StoreConfig('fake cnxn', self.config) |
| cd = tracker_bizobj.MakeComponentDef( |
| component_id, project_id, component_name, component_doc, deprecated, |
| admin_ids, cc_ids, created, creator, modifier_id=creator) |
| self.config.component_defs.append(cd) |
| |
| def SetUpFieldDefs( |
| self, field_id, project_id, field_name, field_type_int, |
| min_value=0, max_value=100, needs_member=False, docstring='doc', |
| approval_id=None, is_phase_field=False): |
| self.config = self.services.config.GetProjectConfig( |
| 'fake cnxn', project_id) |
| self.services.config.StoreConfig('fake cnxn', self.config) |
| fd = tracker_bizobj.MakeFieldDef( |
| field_id, project_id, field_name, field_type_int, '', |
| '', False, False, False, min_value, max_value, None, needs_member, |
| None, '', tracker_pb2.NotifyTriggers.NEVER, 'no_action', docstring, |
| False, approval_id=approval_id, is_phase_field=is_phase_field) |
| self.config.field_defs.append(fd) |
| |
| def testUsersGet_NoProject(self): |
| """The viewed user has no projects.""" |
| |
| self.services.project.TestAddProject( |
| 'public-project', owner_ids=[111]) |
| resp = self.call_api('users_get', self.request).json_body |
| expected = { |
| 'id': '222', |
| 'kind': 'monorail#user'} |
| self.assertEqual(expected, resp) |
| |
| def testUsersGet_PublicProject(self): |
| """The viewed user has one public project.""" |
| self.services.template.GetProjectTemplates.return_value = \ |
| testing_helpers.DefaultTemplates() |
| self.services.project.TestAddProject( |
| 'public-project', owner_ids=[222]) |
| resp = self.call_api('users_get', self.request).json_body |
| |
| self.assertEqual(1, len(resp['projects'])) |
| self.assertEqual('public-project', resp['projects'][0]['name']) |
| |
| def testUsersGet_PrivateProject(self): |
| """The viewed user has one project but the requester cannot view.""" |
| |
| self.services.project.TestAddProject( |
| 'private-project', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY) |
| resp = self.call_api('users_get', self.request).json_body |
| self.assertNotIn('projects', resp) |
| |
| def testUsersGet_OwnerProjectOnly(self): |
| """The viewed user has different roles of projects.""" |
| self.services.template.GetProjectTemplates.return_value = \ |
| testing_helpers.DefaultTemplates() |
| self.services.project.TestAddProject( |
| 'owner-project', owner_ids=[222]) |
| self.services.project.TestAddProject( |
| 'member-project', owner_ids=[111], committer_ids=[222]) |
| resp = self.call_api('users_get', self.request).json_body |
| self.assertEqual(2, len(resp['projects'])) |
| |
| self.request['ownerProjectsOnly'] = True |
| resp = self.call_api('users_get', self.request).json_body |
| self.assertEqual(1, len(resp['projects'])) |
| self.assertEqual('owner-project', resp['projects'][0]['name']) |
| |
| def testIssuesGet_GetIssue(self): |
| """Get the requested issue.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| self.SetUpFieldDefs(1, 12345, 'Field1', tracker_pb2.FieldTypes.INT_TYPE) |
| |
| fv = tracker_pb2.FieldValue( |
| field_id=1, |
| int_value=11) |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, owner_id=222, reporter_id=111, |
| status='New', summary='sum', component_ids=[1], field_values=[fv]) |
| self.services.issue.TestAddIssue(issue1) |
| |
| resp = self.call_api('issues_get', self.request).json_body |
| self.assertEqual(1, resp['id']) |
| self.assertEqual('New', resp['status']) |
| self.assertEqual('open', resp['state']) |
| self.assertFalse(resp['canEdit']) |
| self.assertTrue(resp['canComment']) |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertEqual('user@example.com', resp['owner']['name']) |
| self.assertEqual('API', resp['components'][0]) |
| self.assertEqual('Field1', resp['fieldValues'][0]['fieldName']) |
| self.assertEqual('11', resp['fieldValues'][0]['fieldValue']) |
| |
| def testIssuesInsert_BadRequest(self): |
| """The request does not specify summary or status.""" |
| |
| with self.assertRaises(webtest.AppError): |
| self.call_api('issues_insert', self.request) |
| |
| issue_dict = { |
| 'status': 'New', |
| 'summary': 'Test issue', |
| 'owner': {'name': 'notexist@example.com'}} |
| self.request.update(issue_dict) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| with self.call_should_fail(400): |
| self.call_api('issues_insert', self.request) |
| |
| # Invalid field value |
| self.SetUpFieldDefs(1, 12345, 'Field1', tracker_pb2.FieldTypes.INT_TYPE) |
| issue_dict = { |
| 'status': 'New', |
| 'summary': 'Test issue', |
| 'owner': {'name': 'requester@example.com'}, |
| 'fieldValues': [{'fieldName': 'Field1', 'fieldValue': '111'}]} |
| self.request.update(issue_dict) |
| with self.call_should_fail(400): |
| self.call_api('issues_insert', self.request) |
| |
| def testIssuesInsert_NoPermission(self): |
| """The requester has no permission to create issues.""" |
| |
| issue_dict = { |
| 'status': 'New', |
| 'summary': 'Test issue'} |
| self.request.update(issue_dict) |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=12345) |
| with self.call_should_fail(403): |
| self.call_api('issues_insert', self.request) |
| |
| @patch('framework.cloud_tasks_helpers.create_task') |
| def testIssuesInsert_CreateIssue(self, _create_task_mock): |
| """Create an issue as requested.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], committer_ids=[111], project_id=12345) |
| self.SetUpFieldDefs(1, 12345, 'Field1', tracker_pb2.FieldTypes.INT_TYPE) |
| |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, owner_id=222, reporter_id=111, |
| status='New', summary='Test issue') |
| self.services.issue.TestAddIssue(issue1) |
| |
| issue_dict = { |
| 'blockedOn': [{'issueId': 1}], |
| 'cc': [{'name': 'user@example.com'}, {'name': ''}, {'name': ' '}], |
| 'description': 'description', |
| 'labels': ['label1', 'label2'], |
| 'owner': {'name': 'requester@example.com'}, |
| 'status': 'New', |
| 'summary': 'Test issue', |
| 'fieldValues': [{'fieldName': 'Field1', 'fieldValue': '11'}]} |
| self.request.update(issue_dict) |
| |
| resp = self.call_api('issues_insert', self.request).json_body |
| self.assertEqual('New', resp['status']) |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertEqual('requester@example.com', resp['owner']['name']) |
| self.assertEqual('user@example.com', resp['cc'][0]['name']) |
| self.assertEqual(1, resp['blockedOn'][0]['issueId']) |
| self.assertEqual([u'label1', u'label2'], resp['labels']) |
| self.assertEqual('Test issue', resp['summary']) |
| self.assertEqual('Field1', resp['fieldValues'][0]['fieldName']) |
| self.assertEqual('11', resp['fieldValues'][0]['fieldValue']) |
| |
| new_issue = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 12345, resp['id']) |
| |
| starrers = self.services.issue_star.LookupItemStarrers( |
| 'fake cnxn', new_issue.issue_id) |
| self.assertIn(111, starrers) |
| |
| @patch('framework.cloud_tasks_helpers.create_task') |
| def testIssuesInsert_EmptyOwnerCcNames(self, _create_task_mock): |
| """Create an issue as requested.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpFieldDefs(1, 12345, 'Field1', tracker_pb2.FieldTypes.INT_TYPE) |
| |
| issue_dict = { |
| 'cc': [{'name': 'user@example.com'}, {'name': ''}], |
| 'description': 'description', |
| 'owner': {'name': ''}, |
| 'status': 'New', |
| 'summary': 'Test issue'} |
| self.request.update(issue_dict) |
| |
| resp = self.call_api('issues_insert', self.request).json_body |
| self.assertEqual('New', resp['status']) |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertTrue('owner' not in resp) |
| self.assertEqual('user@example.com', resp['cc'][0]['name']) |
| self.assertEqual(len(resp['cc']), 1) |
| self.assertEqual('Test issue', resp['summary']) |
| |
| new_issue = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 12345, resp['id']) |
| self.assertEqual(new_issue.owner_id, 0) |
| |
| def testIssuesList_NoPermission(self): |
| """No permission for additional projects.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=123456) |
| self.request['additionalProject'] = ['test-project2'] |
| with self.call_should_fail(403): |
| self.call_api('issues_list', self.request) |
| |
| def testIssuesList_SearchIssues(self): |
| """Find issues of one project.""" |
| |
| self.mock( |
| frontendsearchpipeline, |
| 'FrontendSearchPipeline', lambda cnxn, serv, auth, me, q, q_proj_names, |
| num, start, can, group_spec, sort_spec, warnings, errors, use_cache, |
| profiler, project: FakeFrontendSearchPipeline()) |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], # requester |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=12345) |
| resp = self.call_api('issues_list', self.request).json_body |
| self.assertEqual(2, int(resp['totalResults'])) |
| self.assertEqual(1, len(resp['items'])) |
| self.assertEqual(1, resp['items'][0]['id']) |
| |
| def testIssuesCommentsList_GetComments(self): |
| """Get comments of requested issue.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, summary='test summary', status='New', |
| issue_id=10001, owner_id=222, reporter_id=111) |
| self.services.issue.TestAddIssue(issue1) |
| |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=222, |
| content='this is a comment', |
| timestamp=1437700000) |
| self.services.issue.TestAddComment(comment, 1) |
| |
| resp = self.call_api('issues_comments_list', self.request).json_body |
| self.assertEqual(2, resp['totalResults']) |
| comment1 = resp['items'][0] |
| comment2 = resp['items'][1] |
| self.assertEqual('requester@example.com', comment1['author']['name']) |
| self.assertEqual('test summary', comment1['content']) |
| self.assertEqual('user@example.com', comment2['author']['name']) |
| self.assertEqual('this is a comment', comment2['content']) |
| |
| def testParseImportedReporter_Normal(self): |
| """Normal attempt to post a comment under the requester's name.""" |
| mar = FakeMonorailApiRequest(self.request, self.services) |
| container = api_pb2_v1.ISSUES_COMMENTS_INSERT_REQUEST_RESOURCE_CONTAINER |
| request = container.body_message_class() |
| |
| monorail_api = self.api_service_cls() |
| monorail_api._set_services(self.services) |
| reporter_id, timestamp = monorail_api.parse_imported_reporter(mar, request) |
| self.assertEqual(111, reporter_id) |
| self.assertIsNone(timestamp) |
| |
| # API users should not need to specify anything for author when posting |
| # as the signed-in user, but it is OK if they specify their own email. |
| request.author = api_pb2_v1.AtomPerson(name='requester@example.com') |
| request.published = datetime.datetime.now() # Ignored |
| monorail_api = self.api_service_cls() |
| monorail_api._set_services(self.services) |
| reporter_id, timestamp = monorail_api.parse_imported_reporter(mar, request) |
| self.assertEqual(111, reporter_id) |
| self.assertIsNone(timestamp) |
| |
| def testParseImportedReporter_Import_Allowed(self): |
| """User is importing a comment posted by a different user.""" |
| project = self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], contrib_ids=[111], |
| project_id=12345) |
| project.extra_perms = [project_pb2.Project.ExtraPerms( |
| member_id=111, perms=['ImportComment'])] |
| mar = FakeMonorailApiRequest(self.request, self.services) |
| container = api_pb2_v1.ISSUES_COMMENTS_INSERT_REQUEST_RESOURCE_CONTAINER |
| request = container.body_message_class() |
| request.author = api_pb2_v1.AtomPerson(name='user@example.com') |
| NOW = 1234567890 |
| request.published = datetime.datetime.utcfromtimestamp(NOW) |
| monorail_api = self.api_service_cls() |
| monorail_api._set_services(self.services) |
| |
| reporter_id, timestamp = monorail_api.parse_imported_reporter(mar, request) |
| |
| self.assertEqual(222, reporter_id) # that is user@ |
| self.assertEqual(NOW, timestamp) |
| |
| def testParseImportedReporter_Import_NotAllowed(self): |
| """User is importing a comment posted by a different user without perm.""" |
| mar = FakeMonorailApiRequest(self.request, self.services) |
| container = api_pb2_v1.ISSUES_COMMENTS_INSERT_REQUEST_RESOURCE_CONTAINER |
| request = container.body_message_class() |
| request.author = api_pb2_v1.AtomPerson(name='user@example.com') |
| NOW = 1234567890 |
| request.published = datetime.datetime.fromtimestamp(NOW) |
| monorail_api = self.api_service_cls() |
| monorail_api._set_services(self.services) |
| |
| with self.assertRaises(permissions.PermissionException): |
| monorail_api.parse_imported_reporter(mar, request) |
| |
| def testIssuesCommentsInsert_ApprovalFields(self): |
| """Attempts to update approval field values are blocked.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 2, issue_id=1234501) |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.SetUpFieldDefs( |
| 1, 12345, 'Field_int', tracker_pb2.FieldTypes.INT_TYPE) |
| self.SetUpFieldDefs( |
| 2, 12345, 'ApprovalChild', tracker_pb2.FieldTypes.STR_TYPE, |
| approval_id=1) |
| |
| self.request['updates'] = { |
| 'fieldValues': [{'fieldName': 'Field_int', 'fieldValue': '11'}, |
| {'fieldName': 'ApprovalChild', 'fieldValue': 'str'}]} |
| |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_NoCommentPermission(self): |
| """No permission to comment an issue.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 2) |
| self.services.issue.TestAddIssue(issue1) |
| |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_CommentPermissionOnly(self): |
| """User has permission to comment, even though they cannot edit.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[], project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222) |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['content'] = 'This is just a comment' |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertEqual('This is just a comment', resp['content']) |
| |
| def testIssuesCommentsInsert_TooLongComment(self): |
| """Too long of a comment to add.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[], project_id=12345) |
| |
| issue1 = fake.MakeTestIssue(12345, 1, 'Issue 1', 'New', 222) |
| self.services.issue.TestAddIssue(issue1) |
| |
| long_comment = ' ' + 'c' * tracker_constants.MAX_COMMENT_CHARS + ' ' |
| self.request['content'] = long_comment |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_Amendments_Normal(self): |
| """Insert comments with amendments.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| issue2 = fake.MakeTestIssue( |
| 12345, 2, 'Issue 2', 'New', 222, project_name='test-project') |
| issue3 = fake.MakeTestIssue( |
| 12345, 3, 'Issue 3', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.services.issue.TestAddIssue(issue2) |
| self.services.issue.TestAddIssue(issue3) |
| |
| self.request['updates'] = { |
| 'summary': 'new summary', |
| 'status': 'Started', |
| 'owner': 'requester@example.com', |
| 'cc': ['user@example.com'], |
| 'labels': ['add_label', '-remove_label'], |
| 'blockedOn': ['2'], |
| 'blocking': ['3'], |
| } |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertEqual('Started', resp['updates']['status']) |
| self.assertEqual(0, issue1.merged_into) |
| |
| def testIssuesCommentsInsert_Amendments_NoPerms(self): |
| """Can't insert comments using account that lacks permissions.""" |
| |
| project1 = self.services.project.TestAddProject( |
| 'test-project', owner_ids=[], project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['updates'] = { |
| 'summary': 'new summary', |
| } |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| project1.contributor_ids = [1] # Does not grant edit perm. |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_Amendments_BadOwner(self): |
| """Can't set owner to someone who is not a project member.""" |
| |
| _project1 = self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['updates'] = { |
| 'owner': 'user@example.com', |
| } |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| @patch('framework.cloud_tasks_helpers.create_task') |
| def testIssuesCommentsInsert_MergeInto(self, _create_task_mock): |
| """Insert comment that merges an issue into another issue.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], committer_ids=[111], |
| project_id=12345) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| issue2 = fake.MakeTestIssue( |
| 12345, 2, 'Issue 2', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.services.issue.TestAddIssue(issue2) |
| self.services.issue_star.SetStarsBatch( |
| 'cnxn', 'service', 'config', issue1.issue_id, [111, 222, 333], True) |
| self.services.issue_star.SetStarsBatch( |
| 'cnxn', 'service', 'config', issue2.issue_id, [555], True) |
| |
| self.request['updates'] = { |
| 'summary': 'new summary', |
| 'status': 'Duplicate', |
| 'owner': 'requester@example.com', |
| 'cc': ['user@example.com'], |
| 'labels': ['add_label', '-remove_label'], |
| 'mergedInto': '2', |
| } |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| self.assertEqual('requester@example.com', resp['author']['name']) |
| self.assertEqual('Duplicate', resp['updates']['status']) |
| self.assertEqual(issue2.issue_id, issue1.merged_into) |
| issue2_comments = self.services.issue.GetCommentsForIssue( |
| 'cnxn', issue2.issue_id) |
| self.assertEqual(2, len(issue2_comments)) # description and merge |
| source_starrers = self.services.issue_star.LookupItemStarrers( |
| 'cnxn', issue1.issue_id) |
| self.assertItemsEqual([111, 222, 333], source_starrers) |
| target_starrers = self.services.issue_star.LookupItemStarrers( |
| 'cnxn', issue2.issue_id) |
| self.assertItemsEqual([111, 222, 333, 555], target_starrers) |
| |
| def testIssuesCommentsInsert_CustomFields(self): |
| """Update custom field values.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.SetUpFieldDefs( |
| 1, 12345, 'Field_int', tracker_pb2.FieldTypes.INT_TYPE) |
| self.SetUpFieldDefs( |
| 2, 12345, 'Field_enum', tracker_pb2.FieldTypes.ENUM_TYPE) |
| |
| self.request['updates'] = { |
| 'fieldValues': [{'fieldName': 'Field_int', 'fieldValue': '11'}, |
| {'fieldName': 'Field_enum', 'fieldValue': 'str'}]} |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| self.assertEqual( |
| {'fieldName': 'Field_int', 'fieldValue': '11'}, |
| resp['updates']['fieldValues'][0]) |
| |
| def testIssuesCommentsInsert_IsDescription(self): |
| """Add a new issue description.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| # Note: the initially issue description will be "Issue 1". |
| |
| self.request['content'] = 'new desc' |
| self.request['updates'] = {'is_description': True} |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| self.assertEqual('new desc', resp['content']) |
| comments = self.services.issue.GetCommentsForIssue('cnxn', issue1.issue_id) |
| self.assertEqual(2, len(comments)) |
| self.assertTrue(comments[1].is_description) |
| self.assertEqual('new desc', comments[1].content) |
| |
| def testIssuesCommentsInsert_MoveToProject_NoPermsSrc(self): |
| """Don't move issue when user has no perms to edit issue.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, labels=[], |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[111], project_id=12346) |
| |
| # The user has no permission in test-project. |
| self.request['projectId'] = 'test-project' |
| self.request['updates'] = { |
| 'moveToProject': 'test-project2'} |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_MoveToProject_NoPermsDest(self): |
| """Don't move issue to a different project where user has no perms.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, labels=[], |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[], project_id=12346) |
| |
| # The user has no permission in test-project2. |
| self.request['projectId'] = 'test-project' |
| self.request['updates'] = { |
| 'moveToProject': 'test-project2'} |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_MoveToProject_NoSuchProject(self): |
| """Don't move issue to a different project that does not exist.""" |
| project1 = self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, labels=[], |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| # Project doesn't exist. |
| project1.owner_ids = [111, 222] |
| self.request['updates'] = { |
| 'moveToProject': 'not exist'} |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_MoveToProject_SameProject(self): |
| """Don't move issue to the project it is already in.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, labels=[], |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| # The issue is already in destination |
| self.request['updates'] = { |
| 'moveToProject': 'test-project'} |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_MoveToProject_Restricted(self): |
| """Don't move restricted issue to a different project.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, labels=['Restrict-View-Google'], |
| project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[111], |
| project_id=12346) |
| |
| # Issue has restrict labels, so it cannot move. |
| self.request['projectId'] = 'test-project' |
| self.request['updates'] = { |
| 'moveToProject': 'test-project2'} |
| with self.call_should_fail(400): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsInsert_MoveToProject_Normal(self): |
| """Move issue.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111, 222], |
| project_id=12345) |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[111, 222], |
| project_id=12346) |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| issue2 = fake.MakeTestIssue( |
| 12346, 1, 'Issue 1', 'New', 222, project_name='test-project2') |
| self.services.issue.TestAddIssue(issue2) |
| |
| self.request['updates'] = { |
| 'moveToProject': 'test-project2'} |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| |
| self.assertEqual( |
| 'Moved issue test-project:1 to now be issue test-project2:2.', |
| resp['content']) |
| |
| def testIssuesCommentsInsert_Import_Allowed(self): |
| """Post a comment attributed to another user, with permission.""" |
| project = self.services.project.TestAddProject( |
| 'test-project', committer_ids=[111, 222], project_id=12345) |
| project.extra_perms = [project_pb2.Project.ExtraPerms( |
| member_id=111, perms=['ImportComment'])] |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['author'] = {'name': 'user@example.com'} # 222 |
| self.request['content'] = 'a comment' |
| self.request['updates'] = { |
| 'owner': 'user@example.com', |
| } |
| |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| |
| self.assertEqual('a comment', resp['content']) |
| comments = self.services.issue.GetCommentsForIssue('cnxn', issue1.issue_id) |
| self.assertEqual(2, len(comments)) |
| self.assertEqual(222, comments[1].user_id) |
| self.assertEqual('a comment', comments[1].content) |
| |
| |
| def testIssuesCommentsInsert_Import_Self(self): |
| """Specifying the comment author is OK if it is the requester.""" |
| self.services.project.TestAddProject( |
| 'test-project', committer_ids=[111, 222], project_id=12345) |
| # Note: No ImportComment permission has been granted. |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['author'] = {'name': 'requester@example.com'} # 111 |
| self.request['content'] = 'a comment' |
| self.request['updates'] = { |
| 'owner': 'user@example.com', |
| } |
| |
| resp = self.call_api('issues_comments_insert', self.request).json_body |
| |
| self.assertEqual('a comment', resp['content']) |
| comments = self.services.issue.GetCommentsForIssue('cnxn', issue1.issue_id) |
| self.assertEqual(2, len(comments)) |
| self.assertEqual(111, comments[1].user_id) |
| self.assertEqual('a comment', comments[1].content) |
| |
| def testIssuesCommentsInsert_Import_Denied(self): |
| """Cannot post a comment attributed to another user without permission.""" |
| self.services.project.TestAddProject( |
| 'test-project', committer_ids=[111, 222], project_id=12345) |
| # Note: No ImportComment permission has been granted. |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, project_name='test-project') |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.request['author'] = {'name': 'user@example.com'} # 222 |
| self.request['content'] = 'a comment' |
| self.request['updates'] = { |
| 'owner': 'user@example.com', |
| } |
| |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_insert', self.request) |
| |
| def testIssuesCommentsDelete_NoComment(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, summary='test summary', |
| issue_id=10001, status='New', owner_id=222, reporter_id=222) |
| self.services.issue.TestAddIssue(issue1) |
| self.request['commentId'] = 1 |
| with self.call_should_fail(404): |
| self.call_api('issues_comments_delete', self.request) |
| |
| def testIssuesCommentsDelete_NoDeletePermission(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, summary='test summary', |
| issue_id=10001, status='New', owner_id=222, reporter_id=222) |
| self.services.issue.TestAddIssue(issue1) |
| self.request['commentId'] = 0 |
| with self.call_should_fail(403): |
| self.call_api('issues_comments_delete', self.request) |
| |
| def testIssuesCommentsDelete_DeleteUndelete(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| issue1 = fake.MakeTestIssue( |
| project_id=12345, local_id=1, summary='test summary', |
| issue_id=10001, status='New', owner_id=222, reporter_id=111) |
| self.services.issue.TestAddIssue(issue1) |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=111, |
| content='this is a comment', |
| timestamp=1437700000) |
| self.services.issue.TestAddComment(comment, 1) |
| self.request['commentId'] = 1 |
| |
| comments = self.services.issue.GetCommentsForIssue(None, 10001) |
| |
| self.call_api('issues_comments_delete', self.request) |
| self.assertEqual(111, comments[1].deleted_by) |
| |
| self.call_api('issues_comments_undelete', self.request) |
| self.assertIsNone(comments[1].deleted_by) |
| |
| def approvalRequest(self, approval, request_fields=None, comment=None, |
| issue_labels=None): |
| request = {'userId': 'user@example.com', |
| 'requester': 'requester@example.com', |
| 'projectId': 'test-project', |
| 'issueId': 1, |
| 'approvalName': 'Legal-Review', |
| 'sendEmail': False, |
| } |
| if request_fields: |
| request.update(request_fields) |
| |
| self.SetUpFieldDefs( |
| 1, 12345, 'Legal-Review', tracker_pb2.FieldTypes.APPROVAL_TYPE) |
| |
| issue1 = fake.MakeTestIssue( |
| 12345, 1, 'Issue 1', 'New', 222, approval_values=[approval], |
| labels=issue_labels) |
| self.services.issue.TestAddIssue(issue1) |
| |
| self.services.issue.DeltaUpdateIssueApproval = Mock(return_value=comment) |
| |
| self.mock(api_svc_v1.MonorailApi, 'mar_factory', |
| lambda x, y, z: FakeMonorailApiRequest( |
| request, self.services)) |
| return request, issue1 |
| |
| def getFakeComments(self): |
| return [ |
| tracker_pb2.IssueComment( |
| id=123, issue_id=1234501, project_id=12345, user_id=111, |
| content='1st comment', timestamp=1437700000, approval_id=1), |
| tracker_pb2.IssueComment( |
| id=223, issue_id=1234501, project_id=12345, user_id=111, |
| content='2nd comment', timestamp=1437700000, approval_id=2), |
| tracker_pb2.IssueComment( |
| id=323, issue_id=1234501, project_id=12345, user_id=111, |
| content='3rd comment', timestamp=1437700000, approval_id=1, |
| is_description=True), |
| tracker_pb2.IssueComment( |
| id=423, issue_id=1234501, project_id=12345, user_id=111, |
| content='4th comment', timestamp=1437700000)] |
| |
| def testApprovalsCommentsList_NoViewPermission(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest( |
| approval, issue_labels=['Restrict-View-Google']) |
| |
| with self.call_should_fail(403): |
| self.call_api('approvals_comments_list', request) |
| |
| def testApprovalsCommentsList_NoApprovalFound(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest(approval) |
| self.config.field_defs = [] # empty field_defs of approval fd |
| |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_list', request) |
| |
| def testApprovalsCommentsList(self): |
| """Get comments of requested issue approval.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| self.services.issue.GetCommentsForIssue = Mock( |
| return_value=self.getFakeComments()) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest(approval) |
| |
| response = self.call_api('approvals_comments_list', request).json_body |
| self.assertEqual(response['kind'], 'monorail#approvalCommentList') |
| self.assertEqual(response['totalResults'], 2) |
| self.assertEqual(len(response['items']), 2) |
| |
| def testApprovalsCommentsList_MaxResults(self): |
| """get comments of requested issue approval with maxResults.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| self.services.issue.GetCommentsForIssue = Mock( |
| return_value=self.getFakeComments()) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest( |
| approval, request_fields={'maxResults': 1}) |
| |
| response = self.call_api('approvals_comments_list', request).json_body |
| self.assertEqual(response['kind'], 'monorail#approvalCommentList') |
| self.assertEqual(response['totalResults'], 2) |
| self.assertEqual(len(response['items']), 1) |
| self.assertEqual(response['items'][0]['content'], '1st comment') |
| |
| @patch('testing.fake.IssueService.GetCommentsForIssue') |
| def testApprovalsCommentsList_StartIndex(self, mockGetComments): |
| """get comments of requested issue approval with maxResults.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| mockGetComments.return_value = self.getFakeComments() |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest( |
| approval, request_fields={'startIndex': 1}) |
| |
| response = self.call_api('approvals_comments_list', request).json_body |
| self.assertEqual(response['kind'], 'monorail#approvalCommentList') |
| self.assertEqual(response['totalResults'], 2) |
| self.assertEqual(len(response['items']), 1) |
| self.assertEqual(response['items'][0]['content'], '3rd comment') |
| |
| def testApprovalsCommentsInsert_NoCommentPermission(self): |
| """No permission to comment on an issue, including approvals.""" |
| |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY, |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest(approval) |
| |
| with self.call_should_fail(403): |
| self.call_api('approvals_comments_insert', request) |
| |
| def testApprovalsCommentsInsert_TooLongComment(self): |
| """Too long of a comment when comments on approvals.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest(approval) |
| |
| long_comment = ' ' + 'c' * tracker_constants.MAX_COMMENT_CHARS + ' ' |
| request['content'] = long_comment |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| def testApprovalsCommentsInsert_NoApprovalDefFound(self): |
| """No approval with approvalName found.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| request, _issue = self.approvalRequest(approval) |
| self.config.field_defs = [] |
| |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| # Test wrong field_type is also caught. |
| self.SetUpFieldDefs( |
| 1, 12345, 'Legal-Review', tracker_pb2.FieldTypes.STR_TYPE) |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| def testApprovalscommentsInsert_NoIssueFound(self): |
| """No issue found in project.""" |
| request = {'userId': 'user@example.com', |
| 'requester': 'requester@example.com', |
| 'projectId': 'test-project', |
| 'issueId': 1, |
| 'approvalName': 'Legal-Review', |
| } |
| # No issue created. |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| def testApprovalsCommentsInsert_NoIssueApprovalFound(self): |
| """No approval with the given name found in the issue.""" |
| |
| request = {'userId': 'user@example.com', |
| 'requester': 'requester@example.com', |
| 'projectId': 'test-project', |
| 'issueId': 1, |
| 'approvalName': 'Legal-Review', |
| 'sendEmail': False, |
| } |
| |
| self.SetUpFieldDefs( |
| 1, 12345, 'Legal-Review', tracker_pb2.FieldTypes.APPROVAL_TYPE) |
| |
| # issue 1 does not contain the Legal-Review approval. |
| issue1 = fake.MakeTestIssue(12345, 1, 'Issue 1', 'New', 222) |
| self.services.issue.TestAddIssue(issue1) |
| |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| def testApprovalsCommentsInsert_FieldValueChanges_NotFound(self): |
| """Approval's subfield value not found.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| approval = tracker_pb2.ApprovalValue(approval_id=1) |
| |
| request, _issue = self.approvalRequest( |
| approval, |
| request_fields={ |
| 'approvalUpdates': { |
| 'fieldValues': [ |
| {'fieldName': 'DoesNotExist', 'fieldValue': 'cow'}] |
| }, |
| }) |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| # Test field belongs to another approval |
| self.config.field_defs.append( |
| tracker_bizobj.MakeFieldDef( |
| 2, 12345, 'DoesNotExist', tracker_pb2.FieldTypes.STR_TYPE, |
| '', '', False, False, False, None, None, None, False, |
| None, '', tracker_pb2.NotifyTriggers.NEVER, 'no_action', |
| 'parent approval is wrong', False, approval_id=4)) |
| with self.call_should_fail(400): |
| self.call_api('approvals_comments_insert', request) |
| |
| @patch('time.time') |
| def testApprovalCommentsInsert_FieldValueChanges(self, mock_time): |
| """Field value changes are properly processed.""" |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=111, |
| content='cows moo', |
| timestamp=143770000) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[444]) |
| |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={'approvalUpdates': { |
| 'fieldValues': [ |
| {'fieldName': 'CowLayerName', 'fieldValue': 'cow'}, |
| {'fieldName': 'CowType', 'fieldValue': 'skim'}, |
| {'fieldName': 'CowType', 'fieldValue': 'milk'}, |
| {'fieldName': 'CowType', 'fieldValue': 'chocolate', |
| 'operator': 'remove'}] |
| }}, |
| comment=comment) |
| self.config.field_defs.extend( |
| [tracker_bizobj.MakeFieldDef( |
| 2, 12345, 'CowLayerName', tracker_pb2.FieldTypes.STR_TYPE, |
| '', '', False, False, False, None, None, None, False, |
| None, '', tracker_pb2.NotifyTriggers.NEVER, 'no_action', |
| 'sub field value of approval 1', False, approval_id=1), |
| tracker_bizobj.MakeFieldDef( |
| 3, 12345, 'CowType', tracker_pb2.FieldTypes.ENUM_TYPE, |
| '', '', False, False, True, None, None, None, False, |
| None, '', tracker_pb2.NotifyTriggers.NEVER, 'no_action', |
| 'enum sub field value of approval 1', False, approval_id=1)]) |
| |
| response = self.call_api('approvals_comments_insert', request).json_body |
| fvs_add = [tracker_bizobj.MakeFieldValue( |
| 2, None, 'cow', None, None, None, False)] |
| labels_add = ['CowType-skim', 'CowType-milk'] |
| labels_remove = ['CowType-chocolate'] |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| None, 111, [], [], fvs_add, [], [], |
| labels_add, labels_remove, set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content=None, is_description=None) |
| self.assertEqual(response['content'], comment.content) |
| |
| @patch('time.time') |
| def testApprovalsCommentsInsert_StatusChanges_Normal(self, mock_time): |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=111, # requester |
| content='this is a comment', |
| timestamp=1437700000, |
| amendments=[tracker_bizobj.MakeApprovalStatusAmendment( |
| tracker_pb2.ApprovalStatus.REVIEW_REQUESTED)]) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], project_id=12345) |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[444], |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={'approvalUpdates': {'status': 'reviewRequested'}}, |
| comment=comment) |
| response = self.call_api('approvals_comments_insert', request).json_body |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| tracker_pb2.ApprovalStatus.REVIEW_REQUESTED, 111, [], [], [], [], [], |
| [], [], set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content=None, is_description=None) |
| |
| self.assertEqual(response['author']['name'], 'requester@example.com') |
| self.assertEqual(response['content'], comment.content) |
| self.assertTrue(response['canDelete']) |
| self.assertEqual(response['approvalUpdates'], |
| {'kind': 'monorail#approvalCommentUpdate', |
| 'status': 'reviewRequested'}) |
| |
| def testApprovalsCommentsInsert_StatusChanges_NoPerms(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[444], |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, _issue = self.approvalRequest( |
| approval, |
| request_fields={'approvalUpdates': {'status': 'approved'}}) |
| with self.call_should_fail(403): |
| self.call_api('approvals_comments_insert', request) |
| |
| @patch('time.time') |
| def testApprovalsCommentsInsert_StatusChanges_ApproverPerms(self, mock_time): |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=1234501, |
| project_id=12345, user_id=111, |
| content='this is a comment', |
| timestamp=1437700000, |
| amendments=[tracker_bizobj.MakeApprovalStatusAmendment( |
| tracker_pb2.ApprovalStatus.NOT_APPROVED)]) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[111], # requester |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={'approvalUpdates': {'status': 'notApproved'}}, |
| comment=comment) |
| response = self.call_api('approvals_comments_insert', request).json_body |
| |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| tracker_pb2.ApprovalStatus.NOT_APPROVED, 111, [], [], [], [], [], |
| [], [], set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content=None, is_description=None) |
| self.assertEqual(response['author']['name'], 'requester@example.com') |
| self.assertEqual(response['content'], comment.content) |
| self.assertTrue(response['canDelete']) |
| self.assertEqual(response['approvalUpdates'], |
| {'kind': 'monorail#approvalCommentUpdate', |
| 'status': 'notApproved'}) |
| |
| def testApprovalsCommentsInsert_ApproverChanges_NoPerms(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[444], |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, _issue = self.approvalRequest( |
| approval, |
| request_fields={'approvalUpdates': {'approvers': 'someone@test.com'}}) |
| with self.call_should_fail(403): |
| self.call_api('approvals_comments_insert', request) |
| |
| @patch('time.time') |
| def testApprovalsCommentsInsert_ApproverChanges_ApproverPerms( |
| self, mock_time): |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=1234501, |
| project_id=12345, user_id=111, |
| content='this is a comment', |
| timestamp=1437700000, |
| amendments=[tracker_bizobj.MakeApprovalApproversAmendment( |
| [222], [123])]) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[111], # requester |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={ |
| 'approvalUpdates': |
| {'approvers': ['user@example.com', '-group@example.com']}}, |
| comment=comment) |
| response = self.call_api('approvals_comments_insert', request).json_body |
| |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| None, 111, [222], [123], [], [], [], [], [], set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content=None, is_description=None) |
| self.assertEqual(response['author']['name'], 'requester@example.com') |
| self.assertEqual(response['content'], comment.content) |
| self.assertTrue(response['canDelete']) |
| self.assertEqual(response['approvalUpdates'], |
| {'kind': 'monorail#approvalCommentUpdate', |
| 'approvers': ['user@example.com', '-group@example.com']}) |
| |
| @patch('time.time') |
| def testApprovalsCommentsInsert_IsSurvey(self, mock_time): |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=111, |
| content='this is a comment', |
| timestamp=1437700000) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[111], # requester |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={'content': 'updated survey', 'is_description': True}, |
| comment=comment) |
| response = self.call_api('approvals_comments_insert', request).json_body |
| |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| None, 111, [], [], [], [], [], [], [], set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content='updated survey', is_description=True) |
| self.assertEqual(response['author']['name'], 'requester@example.com') |
| self.assertTrue(response['canDelete']) |
| |
| @patch('time.time') |
| @patch('features.send_notifications.PrepareAndSendApprovalChangeNotification') |
| def testApprovalsCommentsInsert_SendEmail( |
| self, mockPrepareAndSend, mock_time,): |
| test_time = 6789 |
| mock_time.return_value = test_time |
| comment = tracker_pb2.IssueComment( |
| id=123, issue_id=10001, |
| project_id=12345, user_id=111, |
| content='this is a comment', |
| timestamp=1437700000) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| |
| approval = tracker_pb2.ApprovalValue( |
| approval_id=1, approver_ids=[111], # requester |
| status=tracker_pb2.ApprovalStatus.NOT_SET) |
| request, issue = self.approvalRequest( |
| approval, |
| request_fields={'content': comment.content, 'sendEmail': True}, |
| comment=comment) |
| |
| response = self.call_api('approvals_comments_insert', request).json_body |
| |
| mockPrepareAndSend.assert_called_with( |
| issue.issue_id, approval.approval_id, ANY, comment.id, send_email=True) |
| |
| approval_delta = tracker_bizobj.MakeApprovalDelta( |
| None, 111, [], [], [], [], [], [], [], set_on=test_time) |
| self.services.issue.DeltaUpdateIssueApproval.assert_called_with( |
| None, 111, self.config, issue, approval, approval_delta, |
| comment_content=comment.content, is_description=None) |
| self.assertEqual(response['author']['name'], 'requester@example.com') |
| self.assertTrue(response['canDelete']) |
| |
| def testGroupsSettingsList_AllSettings(self): |
| resp = self.call_api('groups_settings_list', self.request).json_body |
| all_settings = resp['groupSettings'] |
| self.assertEqual(1, len(all_settings)) |
| self.assertEqual('group@example.com', all_settings[0]['groupName']) |
| |
| def testGroupsSettingsList_ImportedSettings(self): |
| self.services.user.TestAddUser('imported@example.com', 234) |
| self.services.usergroup.TestAddGroupSettings( |
| 234, 'imported@example.com', external_group_type='mdb') |
| self.request['importedGroupsOnly'] = True |
| resp = self.call_api('groups_settings_list', self.request).json_body |
| all_settings = resp['groupSettings'] |
| self.assertEqual(1, len(all_settings)) |
| self.assertEqual('imported@example.com', all_settings[0]['groupName']) |
| |
| def testGroupsCreate_NoPermission(self): |
| self.request['groupName'] = 'group' |
| with self.call_should_fail(403): |
| self.call_api('groups_create', self.request) |
| |
| def SetUpGroupRequest(self, group_name, who_can_view_members='MEMBERS', |
| ext_group_type=None, perms=None, |
| requester='requester@example.com'): |
| request = { |
| 'groupName': group_name, |
| 'requester': requester, |
| 'who_can_view_members': who_can_view_members, |
| 'ext_group_type': ext_group_type} |
| self.request.pop("userId", None) |
| self.mock(api_svc_v1.MonorailApi, 'mar_factory', |
| lambda x, y, z: FakeMonorailApiRequest( |
| request, self.services, perms=perms)) |
| return request |
| |
| def testGroupsCreate_Normal(self): |
| request = self.SetUpGroupRequest('newgroup@example.com', 'MEMBERS', |
| 'MDB', permissions.ADMIN_PERMISSIONSET) |
| |
| resp = self.call_api('groups_create', request).json_body |
| self.assertIn('groupID', resp) |
| |
| def testGroupsGet_NoPermission(self): |
| request = self.SetUpGroupRequest('group@example.com') |
| with self.call_should_fail(403): |
| self.call_api('groups_get', request) |
| |
| def testGroupsGet_Normal(self): |
| request = self.SetUpGroupRequest('group@example.com', |
| perms=permissions.ADMIN_PERMISSIONSET) |
| self.services.usergroup.TestAddMembers(123, [111], 'member') |
| self.services.usergroup.TestAddMembers(123, [222], 'owner') |
| resp = self.call_api('groups_get', request).json_body |
| self.assertEqual(123, resp['groupID']) |
| self.assertEqual(['requester@example.com'], resp['groupMembers']) |
| self.assertEqual(['user@example.com'], resp['groupOwners']) |
| self.assertEqual('group@example.com', resp['groupSettings']['groupName']) |
| |
| def testGroupsUpdate_NoPermission(self): |
| request = self.SetUpGroupRequest('group@example.com') |
| with self.call_should_fail(403): |
| self.call_api('groups_update', request) |
| |
| def testGroupsUpdate_Normal(self): |
| request = self.SetUpGroupRequest('group@example.com') |
| request = self.SetUpGroupRequest('group@example.com', |
| perms=permissions.ADMIN_PERMISSIONSET) |
| request['last_sync_time'] = 123456789 |
| request['groupOwners'] = ['requester@example.com'] |
| request['groupMembers'] = ['user@example.com'] |
| resp = self.call_api('groups_update', request).json_body |
| self.assertFalse(resp.get('error')) |
| |
| def testComponentsList(self): |
| """Get components for a project.""" |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| resp = self.call_api('components_list', self.request).json_body |
| |
| self.assertEqual(1, len(resp['components'])) |
| cd = resp['components'][0] |
| self.assertEqual(1, cd['componentId']) |
| self.assertEqual('API', cd['componentPath']) |
| self.assertEqual(1, cd['componentId']) |
| self.assertEqual('test-project', cd['projectName']) |
| |
| def testComponentsCreate_NoPermission(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| |
| cd_dict = { |
| 'componentName': 'Test'} |
| self.request.update(cd_dict) |
| |
| with self.call_should_fail(403): |
| self.call_api('components_create', self.request) |
| |
| def testComponentsCreate_Invalid(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| |
| # Component with invalid name |
| cd_dict = { |
| 'componentName': 'c>d>e'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(400): |
| self.call_api('components_create', self.request) |
| |
| # Name already in use |
| cd_dict = { |
| 'componentName': 'API'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(400): |
| self.call_api('components_create', self.request) |
| |
| # Parent component does not exist |
| cd_dict = { |
| 'componentName': 'test', |
| 'parentPath': 'NotExist'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(404): |
| self.call_api('components_create', self.request) |
| |
| |
| def testComponentsCreate_Normal(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| |
| cd_dict = { |
| 'componentName': 'Test', |
| 'description': 'test comp', |
| 'cc': ['requester@example.com', ''] |
| } |
| self.request.update(cd_dict) |
| |
| resp = self.call_api('components_create', self.request).json_body |
| self.assertEqual('test comp', resp['description']) |
| self.assertEqual('requester@example.com', resp['creator']) |
| self.assertEqual([u'requester@example.com'], resp['cc']) |
| self.assertEqual('Test', resp['componentPath']) |
| |
| cd_dict = { |
| 'componentName': 'TestChild', |
| 'parentPath': 'API'} |
| self.request.update(cd_dict) |
| resp = self.call_api('components_create', self.request).json_body |
| |
| self.assertEqual('API>TestChild', resp['componentPath']) |
| |
| def testComponentsDelete_Invalid(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| |
| # Fail to delete a non-existent component |
| cd_dict = { |
| 'componentPath': 'NotExist'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(404): |
| self.call_api('components_delete', self.request) |
| |
| # The user has no permission to delete component |
| cd_dict = { |
| 'componentPath': 'API'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(403): |
| self.call_api('components_delete', self.request) |
| |
| # The user tries to delete component that had subcomponents |
| self.services.project.TestAddProject( |
| 'test-project2', owner_ids=[111], |
| project_id=123456) |
| self.SetUpComponents(123456, 1, 'Parent') |
| self.SetUpComponents(123456, 2, 'Parent>Child') |
| cd_dict = { |
| 'componentPath': 'Parent', |
| 'projectId': 'test-project2',} |
| self.request.update(cd_dict) |
| with self.call_should_fail(403): |
| self.call_api('components_delete', self.request) |
| |
| def testComponentsDelete_Normal(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| |
| cd_dict = { |
| 'componentPath': 'API'} |
| self.request.update(cd_dict) |
| _ = self.call_api('components_delete', self.request).json_body |
| self.assertEqual(0, len(self.config.component_defs)) |
| |
| def testComponentsUpdate_Invalid(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[222], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| self.SetUpComponents(12345, 2, 'Test', admin_ids=[111]) |
| |
| # Fail to update a non-existent component |
| cd_dict = { |
| 'componentPath': 'NotExist'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(404): |
| self.call_api('components_update', self.request) |
| |
| # The user has no permission to edit component |
| cd_dict = { |
| 'componentPath': 'API'} |
| self.request.update(cd_dict) |
| with self.call_should_fail(403): |
| self.call_api('components_update', self.request) |
| |
| # The user tries an invalid component name |
| cd_dict = { |
| 'componentPath': 'Test', |
| 'updates': [{'field': 'LEAF_NAME', 'leafName': 'c>e'}]} |
| self.request.update(cd_dict) |
| with self.call_should_fail(400): |
| self.call_api('components_update', self.request) |
| |
| # The user tries a name already in use |
| cd_dict = { |
| 'componentPath': 'Test', |
| 'updates': [{'field': 'LEAF_NAME', 'leafName': 'API'}]} |
| self.request.update(cd_dict) |
| with self.call_should_fail(400): |
| self.call_api('components_update', self.request) |
| |
| def testComponentsUpdate_Normal(self): |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], |
| project_id=12345) |
| self.SetUpComponents(12345, 1, 'API') |
| self.SetUpComponents(12345, 2, 'Parent') |
| self.SetUpComponents(12345, 3, 'Parent>Child') |
| |
| cd_dict = { |
| 'componentPath': 'API', |
| 'updates': [ |
| {'field': 'DESCRIPTION', 'description': ''}, |
| {'field': 'CC', 'cc': [ |
| 'requester@example.com', 'user@example.com', '', ' ']}, |
| {'field': 'DEPRECATED', 'deprecated': True}]} |
| self.request.update(cd_dict) |
| _ = self.call_api('components_update', self.request).json_body |
| component_def = tracker_bizobj.FindComponentDef( |
| 'API', self.config) |
| self.assertIsNotNone(component_def) |
| self.assertEqual('', component_def.docstring) |
| self.assertItemsEqual([111, 222], component_def.cc_ids) |
| self.assertTrue(component_def.deprecated) |
| |
| cd_dict = { |
| 'componentPath': 'Parent', |
| 'updates': [ |
| {'field': 'LEAF_NAME', 'leafName': 'NewParent'}]} |
| self.request.update(cd_dict) |
| _ = self.call_api('components_update', self.request).json_body |
| cd_parent = tracker_bizobj.FindComponentDef( |
| 'NewParent', self.config) |
| cd_child = tracker_bizobj.FindComponentDef( |
| 'NewParent>Child', self.config) |
| self.assertIsNotNone(cd_parent) |
| self.assertIsNotNone(cd_child) |
| |
| |
| class RequestMock(object): |
| |
| def __init__(self): |
| self.projectId = None |
| self.issueId = None |
| |
| |
| class RequesterMock(object): |
| |
| def __init__(self, email=None): |
| self._email = email |
| |
| def email(self): |
| return self._email |
| |
| |
| class AllBaseChecksTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.services = MakeFakeServiceManager() |
| self.services.user.TestAddUser('test@example.com', 111) |
| self.user_2 = self.services.user.TestAddUser('test@google.com', 222) |
| self.services.project.TestAddProject( |
| 'test-project', owner_ids=[111], project_id=123, |
| access=project_pb2.ProjectAccess.MEMBERS_ONLY) |
| self.auth_client_ids = ['123456789.apps.googleusercontent.com'] |
| oauth.get_client_id = Mock(return_value=self.auth_client_ids[0]) |
| oauth.get_current_user = Mock( |
| return_value=RequesterMock(email='test@example.com')) |
| oauth.get_authorized_scopes = Mock() |
| |
| def testUnauthorizedRequester(self): |
| with self.assertRaises(endpoints.UnauthorizedException): |
| api_svc_v1.api_base_checks(None, None, None, None, [], []) |
| |
| def testNoUser(self): |
| requester = RequesterMock(email='notexist@example.com') |
| with self.assertRaises(exceptions.NoSuchUserException): |
| api_svc_v1.api_base_checks( |
| None, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testAllowedDomain_MonorailScope(self): |
| oauth.get_authorized_scopes.return_value = [ |
| framework_constants.MONORAIL_SCOPE] |
| oauth.get_current_user.return_value = RequesterMock( |
| email=self.user_2.email) |
| allowlisted_client_ids = [] |
| allowlisted_emails = [] |
| client_id, email = api_svc_v1.api_base_checks( |
| None, None, self.services, None, allowlisted_client_ids, |
| allowlisted_emails) |
| self.assertEqual(client_id, self.auth_client_ids[0]) |
| self.assertEqual(email, self.user_2.email) |
| |
| def testAllowedDomain_NoMonorailScope(self): |
| oauth.get_authorized_scopes.return_value = [] |
| oauth.get_current_user.return_value = RequesterMock( |
| email=self.user_2.email) |
| allowlisted_client_ids = [] |
| allowlisted_emails = [] |
| with self.assertRaises(endpoints.UnauthorizedException): |
| api_svc_v1.api_base_checks( |
| None, None, self.services, None, allowlisted_client_ids, |
| allowlisted_emails) |
| |
| def testAllowedDomain_BadEmail(self): |
| oauth.get_authorized_scopes.return_value = [ |
| framework_constants.MONORAIL_SCOPE] |
| oauth.get_current_user.return_value = RequesterMock( |
| email='chicken@chicken.test') |
| allowlisted_client_ids = [] |
| allowlisted_emails = [] |
| self.services.user.TestAddUser('chicken@chicken.test', 333) |
| with self.assertRaises(endpoints.UnauthorizedException): |
| api_svc_v1.api_base_checks( |
| None, None, self.services, None, allowlisted_client_ids, |
| allowlisted_emails) |
| |
| def testNoOauthUser(self): |
| oauth.get_current_user.side_effect = oauth.Error() |
| with self.assertRaises(endpoints.UnauthorizedException): |
| api_svc_v1.api_base_checks( |
| None, None, self.services, None, [], []) |
| |
| def testBannedUser(self): |
| banned_email = 'banned@example.com' |
| self.services.user.TestAddUser(banned_email, 222, banned=True) |
| requester = RequesterMock(email=banned_email) |
| with self.assertRaises(permissions.BannedUserException): |
| api_svc_v1.api_base_checks( |
| None, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testNoProject(self): |
| request = RequestMock() |
| request.projectId = 'notexist-project' |
| requester = RequesterMock(email='test@example.com') |
| with self.assertRaises(exceptions.NoSuchProjectException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testNonLiveProject(self): |
| archived_project = 'archived-project' |
| self.services.project.TestAddProject( |
| archived_project, owner_ids=[111], |
| state=project_pb2.ProjectState.ARCHIVED) |
| request = RequestMock() |
| request.projectId = archived_project |
| requester = RequesterMock(email='test@example.com') |
| with self.assertRaises(permissions.PermissionException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testNoViewProjectPermission(self): |
| nonmember_email = 'nonmember@example.com' |
| self.services.user.TestAddUser(nonmember_email, 222) |
| requester = RequesterMock(email=nonmember_email) |
| request = RequestMock() |
| request.projectId = 'test-project' |
| with self.assertRaises(permissions.PermissionException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testAllPass(self): |
| requester = RequesterMock(email='test@example.com') |
| request = RequestMock() |
| request.projectId = 'test-project' |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testNoIssue(self): |
| requester = RequesterMock(email='test@example.com') |
| request = RequestMock() |
| request.projectId = 'test-project' |
| request.issueId = 12345 |
| with self.assertRaises(exceptions.NoSuchIssueException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testNoViewIssuePermission(self): |
| requester = RequesterMock(email='test@example.com') |
| request = RequestMock() |
| request.projectId = 'test-project' |
| request.issueId = 1 |
| issue1 = fake.MakeTestIssue( |
| project_id=123, local_id=1, summary='test summary', |
| status='New', owner_id=111, reporter_id=111) |
| issue1.deleted = True |
| self.services.issue.TestAddIssue(issue1) |
| with self.assertRaises(permissions.PermissionException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, self.auth_client_ids, []) |
| |
| def testAnonymousClients(self): |
| # Some clients specifically pass "anonymous" as the client ID. |
| oauth.get_client_id = Mock(return_value='anonymous') |
| requester = RequesterMock(email='test@example.com') |
| request = RequestMock() |
| request.projectId = 'test-project' |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, [], ['test@example.com']) |
| |
| # Any client_id is OK if the email is allowlisted. |
| oauth.get_client_id = Mock(return_value='anything') |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, [], ['test@example.com']) |
| |
| # Reject request when neither client ID nor email is allowlisted. |
| with self.assertRaises(endpoints.UnauthorizedException): |
| api_svc_v1.api_base_checks( |
| request, requester, self.services, None, [], []) |