blob: b7cd9b1406045ed393cb5a41fcf565c4ee34bb50 [file] [log] [blame]
# 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, [], [])