| # Copyright 2022 The Chromium Authors |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| """Unittest for the tracker helpers module.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import copy |
| import mock |
| import unittest |
| import io |
| import six |
| |
| import settings |
| |
| from businesslogic import work_env |
| from framework import exceptions |
| from framework import framework_constants |
| from framework import framework_helpers |
| from framework import permissions |
| from framework import template_helpers |
| from framework import urls |
| from mrproto import project_pb2 |
| from mrproto import tracker_pb2 |
| from mrproto import user_pb2 |
| from services import service_manager |
| from testing import fake |
| from testing import testing_helpers |
| from tracker import tracker_bizobj |
| from tracker import tracker_constants |
| from tracker import tracker_helpers |
| from werkzeug.datastructures import FileStorage |
| |
| TEST_ID_MAP = { |
| 'a@example.com': 1, |
| 'b@example.com': 2, |
| 'c@example.com': 3, |
| 'd@example.com': 4, |
| } |
| |
| |
| def _Issue(project_name, local_id, summary='', status='', project_id=789): |
| issue = tracker_pb2.Issue() |
| issue.project_name = project_name |
| issue.project_id = project_id |
| issue.local_id = local_id |
| issue.issue_id = 100000 + local_id |
| issue.summary = summary |
| issue.status = status |
| return issue |
| |
| |
| def _MakeConfig(): |
| config = tracker_pb2.ProjectIssueConfig() |
| config.well_known_statuses.append(tracker_pb2.StatusDef( |
| means_open=True, status='New', deprecated=False)) |
| config.well_known_statuses.append(tracker_pb2.StatusDef( |
| status='Old', means_open=False, deprecated=False)) |
| config.well_known_statuses.append(tracker_pb2.StatusDef( |
| status='StatusThatWeDontUseAnymore', means_open=False, deprecated=True)) |
| |
| return config |
| |
| |
| class HelpersTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.services = service_manager.Services( |
| project=fake.ProjectService(), |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| user=fake.UserService(), |
| usergroup=fake.UserGroupService()) |
| |
| for email, user_id in TEST_ID_MAP.items(): |
| self.services.user.TestAddUser(email, user_id) |
| |
| self.services.project.TestAddProject('testproj', project_id=789) |
| self.issue1 = fake.MakeTestIssue(789, 1, 'one', 'New', 111) |
| self.issue1.project_name = 'testproj' |
| self.services.issue.TestAddIssue(self.issue1) |
| self.issue2 = fake.MakeTestIssue(789, 2, 'two', 'New', 111) |
| self.issue2.project_name = 'testproj' |
| self.services.issue.TestAddIssue(self.issue2) |
| self.issue3 = fake.MakeTestIssue(789, 3, 'three', 'New', 111) |
| self.issue3.project_name = 'testproj' |
| self.services.issue.TestAddIssue(self.issue3) |
| self.cnxn = 'fake connextion' |
| self.errors = template_helpers.EZTError() |
| self.default_colspec_param = 'colspec=%s' % ( |
| tracker_constants.DEFAULT_COL_SPEC.replace(' ', '%20')) |
| self.services.usergroup.TestAddGroupSettings(999, 'group@example.com') |
| |
| def testParseIssueRequest_Empty(self): |
| post_data = fake.PostData() |
| errors = template_helpers.EZTError() |
| parsed = tracker_helpers.ParseIssueRequest( |
| 'fake cnxn', post_data, self.services, errors, 'proj') |
| self.assertEqual('', parsed.summary) |
| self.assertEqual('', parsed.comment) |
| self.assertEqual('', parsed.status) |
| self.assertEqual('', parsed.users.owner_username) |
| self.assertEqual(0, parsed.users.owner_id) |
| self.assertEqual([], parsed.users.cc_usernames) |
| self.assertEqual([], parsed.users.cc_usernames_remove) |
| self.assertEqual([], parsed.users.cc_ids) |
| self.assertEqual([], parsed.users.cc_ids_remove) |
| self.assertEqual('', parsed.template_name) |
| self.assertEqual([], parsed.labels) |
| self.assertEqual([], parsed.labels_remove) |
| self.assertEqual({}, parsed.fields.vals) |
| self.assertEqual({}, parsed.fields.vals_remove) |
| self.assertEqual([], parsed.fields.fields_clear) |
| self.assertEqual('', parsed.blocked_on.entered_str) |
| self.assertEqual([], parsed.blocked_on.iids) |
| |
| def testParseIssueRequest_Normal(self): |
| post_data = fake.PostData({ |
| 'summary': ['some summary'], |
| 'comment': ['some comment'], |
| 'status': ['SomeStatus'], |
| 'template_name': ['some template'], |
| 'label': ['lab1', '-lab2'], |
| 'custom_123': ['field1123a', 'field1123b'], |
| }) |
| errors = template_helpers.EZTError() |
| parsed = tracker_helpers.ParseIssueRequest( |
| 'fake cnxn', post_data, self.services, errors, 'proj') |
| self.assertEqual('some summary', parsed.summary) |
| self.assertEqual('some comment', parsed.comment) |
| self.assertEqual('SomeStatus', parsed.status) |
| self.assertEqual('', parsed.users.owner_username) |
| self.assertEqual(0, parsed.users.owner_id) |
| self.assertEqual([], parsed.users.cc_usernames) |
| self.assertEqual([], parsed.users.cc_usernames_remove) |
| self.assertEqual([], parsed.users.cc_ids) |
| self.assertEqual([], parsed.users.cc_ids_remove) |
| self.assertEqual('some template', parsed.template_name) |
| self.assertEqual(['lab1'], parsed.labels) |
| self.assertEqual(['lab2'], parsed.labels_remove) |
| self.assertEqual({123: ['field1123a', 'field1123b']}, parsed.fields.vals) |
| self.assertEqual({}, parsed.fields.vals_remove) |
| self.assertEqual([], parsed.fields.fields_clear) |
| |
| def testMarkupDescriptionOnInput(self): |
| content = 'What?\nthat\nWhy?\nidk\nWhere?\n' |
| tmpl_txt = 'What?\nWhy?\nWhere?\nWhen?' |
| desc = '<b>What?</b>\nthat\n<b>Why?</b>\nidk\n<b>Where?</b>\n' |
| self.assertEqual(tracker_helpers.MarkupDescriptionOnInput( |
| content, tmpl_txt), desc) |
| |
| def testMarkupDescriptionLineOnInput(self): |
| line = 'What happened??' |
| tmpl_lines = ['What happened??','Why?'] |
| self.assertEqual(tracker_helpers._MarkupDescriptionLineOnInput( |
| line, tmpl_lines), '<b>What happened??</b>') |
| |
| line = 'Something terrible!!!' |
| self.assertEqual(tracker_helpers._MarkupDescriptionLineOnInput( |
| line, tmpl_lines), 'Something terrible!!!') |
| |
| def testClassifyPlusMinusItems(self): |
| add, remove = tracker_helpers._ClassifyPlusMinusItems([]) |
| self.assertEqual([], add) |
| self.assertEqual([], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['', ' ', ' \t', '-']) |
| six.assertCountEqual(self, [], add) |
| six.assertCountEqual(self, [], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['a', 'b', 'c']) |
| six.assertCountEqual(self, ['a', 'b', 'c'], add) |
| six.assertCountEqual(self, [], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['a-a-a', 'b-b', 'c-']) |
| six.assertCountEqual(self, ['a-a-a', 'b-b', 'c-'], add) |
| six.assertCountEqual(self, [], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['-a']) |
| six.assertCountEqual(self, [], add) |
| six.assertCountEqual(self, ['a'], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['-a', 'b', 'c-c']) |
| six.assertCountEqual(self, ['b', 'c-c'], add) |
| six.assertCountEqual(self, ['a'], remove) |
| |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['-a', '-b-b', '-c-']) |
| six.assertCountEqual(self, [], add) |
| six.assertCountEqual(self, ['a', 'b-b', 'c-'], remove) |
| |
| # We dedup, but we don't cancel out items that are both added and removed. |
| add, remove = tracker_helpers._ClassifyPlusMinusItems( |
| ['a', 'a', '-a']) |
| six.assertCountEqual(self, ['a'], add) |
| six.assertCountEqual(self, ['a'], remove) |
| |
| def testParseIssueRequestFields(self): |
| parsed_fields = tracker_helpers._ParseIssueRequestFields(fake.PostData({ |
| 'custom_1': ['https://hello.com'], |
| 'custom_12': ['https://blah.com'], |
| 'custom_14': ['https://remove.com'], |
| 'custom_15_goats': ['2', '3'], |
| 'custom_15_sheep': ['3', '5'], |
| 'custom_16_sheep': ['yarn'], |
| 'op_custom_14': ['remove'], |
| 'op_custom_12': ['clear'], |
| 'op_custom_16_sheep': ['remove'], |
| 'ignore': 'no matter',})) |
| self.assertEqual( |
| parsed_fields, |
| tracker_helpers.ParsedFields( |
| { |
| 1: ['https://hello.com'], |
| 12: ['https://blah.com'] |
| }, {14: ['https://remove.com']}, [12], |
| {15: { |
| 'goats': ['2', '3'], |
| 'sheep': ['3', '5'] |
| }}, {16: { |
| 'sheep': ['yarn'] |
| }})) |
| |
| def testParseIssueRequestAttachments(self): |
| file1 = FileStorage( |
| stream=io.BytesIO(b'hello world'), |
| filename='hello.c', |
| ) |
| file2 = FileStorage( |
| stream=io.BytesIO(b'Welcome to our project'), |
| filename='README', |
| ) |
| |
| file3 = FileStorage( |
| stream=io.BytesIO(b'Abort, Retry, or Fail?'), |
| filename='c:\\dir\\subdir\\FILENAME.EXT', |
| ) |
| |
| # Browsers send this if FILE field was not filled in. |
| file4 = FileStorage( |
| stream=io.BytesIO(b''), |
| filename='', |
| ) |
| |
| attachments = tracker_helpers._ParseIssueRequestAttachments({}) |
| self.assertEqual([], attachments) |
| |
| attachments = tracker_helpers._ParseIssueRequestAttachments(fake.PostData({ |
| 'file1': [file1], |
| })) |
| self.assertEqual([('hello.c', b'hello world', 'text/plain')], attachments) |
| file1.seek(0) |
| |
| attachments = tracker_helpers._ParseIssueRequestAttachments(fake.PostData({ |
| 'file1': [file1], |
| 'file2': [file2], |
| })) |
| self.assertEqual( |
| [ |
| ('hello.c', b'hello world', 'text/plain'), |
| ('README', b'Welcome to our project', 'text/plain') |
| ], attachments) |
| file1.seek(0) |
| file2.seek(0) |
| |
| attachments = tracker_helpers._ParseIssueRequestAttachments(fake.PostData({ |
| 'file3': [file3], |
| })) |
| self.assertEqual( |
| [ |
| ( |
| 'FILENAME.EXT', b'Abort, Retry, or Fail?', |
| 'application/octet-stream') |
| ], attachments) |
| file3.seek(0) |
| |
| attachments = tracker_helpers._ParseIssueRequestAttachments(fake.PostData({ |
| 'file1': [file4], # Does not appear in result |
| 'file3': [file3], |
| 'file4': [file4], # Does not appear in result |
| })) |
| self.assertEqual( |
| [ |
| ( |
| 'FILENAME.EXT', b'Abort, Retry, or Fail?', |
| 'application/octet-stream') |
| ], attachments) |
| file3.seek(0) |
| |
| def testParseIssueRequestKeptAttachments(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testParseIssueRequestUsers(self): |
| post_data = {} |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('', parsed_users.owner_username) |
| self.assertEqual( |
| framework_constants.NO_USER_SPECIFIED, parsed_users.owner_id) |
| self.assertEqual([], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'owner': [''], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('', parsed_users.owner_username) |
| self.assertEqual( |
| framework_constants.NO_USER_SPECIFIED, parsed_users.owner_id) |
| self.assertEqual([], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'owner': [' \t'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('', parsed_users.owner_username) |
| self.assertEqual( |
| framework_constants.NO_USER_SPECIFIED, parsed_users.owner_id) |
| self.assertEqual([], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'owner': ['b@example.com'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('b@example.com', parsed_users.owner_username) |
| self.assertEqual(TEST_ID_MAP['b@example.com'], parsed_users.owner_id) |
| self.assertEqual([], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'owner': ['b@example.com'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('b@example.com', parsed_users.owner_username) |
| self.assertEqual(TEST_ID_MAP['b@example.com'], parsed_users.owner_id) |
| self.assertEqual([], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'cc': ['b@example.com'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('', parsed_users.owner_username) |
| self.assertEqual( |
| framework_constants.NO_USER_SPECIFIED, parsed_users.owner_id) |
| self.assertEqual(['b@example.com'], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| self.assertEqual([TEST_ID_MAP['b@example.com']], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'cc': ['-b@example.com, c@example.com,,' |
| 'a@example.com,'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('', parsed_users.owner_username) |
| self.assertEqual( |
| framework_constants.NO_USER_SPECIFIED, parsed_users.owner_id) |
| six.assertCountEqual( |
| self, ['c@example.com', 'a@example.com'], parsed_users.cc_usernames) |
| self.assertEqual(['b@example.com'], parsed_users.cc_usernames_remove) |
| six.assertCountEqual( |
| self, [TEST_ID_MAP['c@example.com'], TEST_ID_MAP['a@example.com']], |
| parsed_users.cc_ids) |
| self.assertEqual([TEST_ID_MAP['b@example.com']], |
| parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'owner': ['fuhqwhgads@example.com'], |
| 'cc': ['c@example.com, fuhqwhgads@example.com'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| self.assertEqual('fuhqwhgads@example.com', parsed_users.owner_username) |
| gen_uid = framework_helpers.MurmurHash3_x86_32(parsed_users.owner_username) |
| self.assertEqual(gen_uid, parsed_users.owner_id) # autocreated user |
| six.assertCountEqual( |
| self, ['c@example.com', 'fuhqwhgads@example.com'], |
| parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| six.assertCountEqual( |
| self, [TEST_ID_MAP['c@example.com'], gen_uid], parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| post_data = fake.PostData({ |
| 'cc': ['C@example.com, b@exAmple.cOm'], |
| }) |
| parsed_users = tracker_helpers._ParseIssueRequestUsers( |
| 'fake connection', post_data, self.services) |
| six.assertCountEqual( |
| self, ['c@example.com', 'b@example.com'], parsed_users.cc_usernames) |
| self.assertEqual([], parsed_users.cc_usernames_remove) |
| six.assertCountEqual( |
| self, [TEST_ID_MAP['c@example.com'], TEST_ID_MAP['b@example.com']], |
| parsed_users.cc_ids) |
| self.assertEqual([], parsed_users.cc_ids_remove) |
| |
| def testParseBlockers_BlockedOnNothing(self): |
| """Was blocked on nothing, still nothing.""" |
| post_data = {tracker_helpers.BLOCKED_ON: ''} |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKED_ON) |
| |
| self.assertEqual('', parsed_blockers.entered_str) |
| self.assertEqual([], parsed_blockers.iids) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKING)) |
| |
| def testParseBlockers_BlockedOnAdded(self): |
| """Was blocked on nothing; now 1, 2, 3.""" |
| post_data = {tracker_helpers.BLOCKED_ON: '1, 2, 3'} |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKED_ON) |
| |
| self.assertEqual('1, 2, 3', parsed_blockers.entered_str) |
| self.assertEqual([100001, 100002, 100003], parsed_blockers.iids) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKING)) |
| |
| def testParseBlockers_BlockedOnDuplicateRef(self): |
| """Was blocked on nothing; now just 2, but repeated in input.""" |
| post_data = {tracker_helpers.BLOCKED_ON: '2, 2, 2'} |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKED_ON) |
| |
| self.assertEqual('2, 2, 2', parsed_blockers.entered_str) |
| self.assertEqual([100002], parsed_blockers.iids) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKING)) |
| |
| def testParseBlockers_Missing(self): |
| """Parsing an input field that was not in the POST.""" |
| post_data = {} |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKED_ON) |
| |
| self.assertEqual('', parsed_blockers.entered_str) |
| self.assertEqual([], parsed_blockers.iids) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKING)) |
| |
| def testParseBlockers_SameIssueNoProject(self): |
| """Adding same issue as blocker should modify the errors object.""" |
| post_data = {'id': '2', tracker_helpers.BLOCKING: '2, 3'} |
| |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('2, 3', parsed_blockers.entered_str) |
| self.assertEqual([], parsed_blockers.iids) |
| self.assertEqual( |
| getattr(self.errors, tracker_helpers.BLOCKING), |
| 'Cannot be blocking the same issue') |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| |
| def testParseBlockers_SameIssueSameProject(self): |
| """Adding same issue as blocker should modify the errors object.""" |
| post_data = {'id': '2', tracker_helpers.BLOCKING: 'testproj:2, 3'} |
| |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('testproj:2, 3', parsed_blockers.entered_str) |
| self.assertEqual([], parsed_blockers.iids) |
| self.assertEqual( |
| getattr(self.errors, tracker_helpers.BLOCKING), |
| 'Cannot be blocking the same issue') |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| |
| def testParseBlockers_SameIssueDifferentProject(self): |
| """Adding different blocker issue should not modify the errors object.""" |
| post_data = {'id': '2', tracker_helpers.BLOCKING: 'testproj:2'} |
| |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testprojB', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('testproj:2', parsed_blockers.entered_str) |
| self.assertEqual([100002], parsed_blockers.iids) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKING)) |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| |
| def testParseBlockers_Invalid(self): |
| """Input fields with invalid values should modify the errors object.""" |
| post_data = {tracker_helpers.BLOCKING: '2, foo', |
| tracker_helpers.BLOCKED_ON: '3, bar'} |
| |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('2, foo', parsed_blockers.entered_str) |
| self.assertEqual([100002], parsed_blockers.iids) |
| self.assertEqual( |
| getattr(self.errors, tracker_helpers.BLOCKING), 'Invalid issue ID foo') |
| self.assertIsNone(getattr(self.errors, tracker_helpers.BLOCKED_ON)) |
| |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKED_ON) |
| self.assertEqual('3, bar', parsed_blockers.entered_str) |
| self.assertEqual([100003], parsed_blockers.iids) |
| self.assertEqual( |
| getattr(self.errors, tracker_helpers.BLOCKED_ON), |
| 'Invalid issue ID bar') |
| |
| def testParseBlockers_Dangling(self): |
| """A ref to a sanctioned projected should be allowed.""" |
| post_data = {'id': '2', tracker_helpers.BLOCKING: 'otherproj:2'} |
| real_codesite_projects = settings.recognized_codesite_projects |
| settings.recognized_codesite_projects = ['otherproj'] |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('otherproj:2', parsed_blockers.entered_str) |
| self.assertEqual([('otherproj', 2)], parsed_blockers.dangling_refs) |
| settings.recognized_codesite_projects = real_codesite_projects |
| |
| def testParseBlockers_FederatedReferences(self): |
| """Should parse and return FedRefs.""" |
| post_data = {'id': '9', tracker_helpers.BLOCKING: '2, b/123, 3, b/789'} |
| parsed_blockers = tracker_helpers._ParseBlockers( |
| self.cnxn, post_data, self.services, self.errors, 'testproj', |
| tracker_helpers.BLOCKING) |
| self.assertEqual('2, b/123, 3, b/789', parsed_blockers.entered_str) |
| self.assertEqual([100002, 100003], parsed_blockers.iids) |
| self.assertEqual(['b/123', 'b/789'], parsed_blockers.federated_ref_strings) |
| |
| def testIsValidIssueOwner(self): |
| project = project_pb2.Project() |
| project.owner_ids.extend([1, 2]) |
| project.committer_ids.extend([3]) |
| project.contributor_ids.extend([4, 999]) |
| |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, framework_constants.NO_USER_SPECIFIED, |
| self.services) |
| self.assertTrue(valid) |
| |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 1, |
| self.services) |
| self.assertTrue(valid) |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 2, |
| self.services) |
| self.assertTrue(valid) |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 3, |
| self.services) |
| self.assertTrue(valid) |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 4, |
| self.services) |
| self.assertTrue(valid) |
| |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 7, |
| self.services) |
| self.assertFalse(valid) |
| |
| valid, _ = tracker_helpers.IsValidIssueOwner( |
| 'fake cnxn', project, 999, |
| self.services) |
| self.assertFalse(valid) |
| |
| # MakeViewsForUsersInIssuesTest is tested in MakeViewsForUsersInIssuesTest. |
| |
| def testGetAllowedOpenedAndClosedIssues(self): |
| pass # TOOD(jrobbins): Write this test. |
| |
| def testFormatIssueListURL_JumpedToIssue(self): |
| """If we jumped to issue 123, the list is can=1&q=id-123.""" |
| config = tracker_pb2.ProjectIssueConfig() |
| path = '/p/proj/issues/detail?id=123&q=123' |
| mr = testing_helpers.MakeMonorailRequest( |
| path=path, headers={'Host': 'code.google.com'}) |
| mr.ComputeColSpec(config) |
| |
| absolute_base_url = 'http://code.google.com' |
| |
| url_1 = tracker_helpers.FormatIssueListURL(mr, config) |
| self.assertEqual( |
| '%s/p/proj/issues/list?can=1&%s&q=id%%3D123' % ( |
| absolute_base_url, self.default_colspec_param), |
| url_1) |
| |
| def testFormatIssueListURL_NoCurrentState(self): |
| config = tracker_pb2.ProjectIssueConfig() |
| path = '/p/proj/issues/detail?id=123' |
| mr = testing_helpers.MakeMonorailRequest( |
| path=path, headers={'Host': 'code.google.com'}) |
| mr.ComputeColSpec(config) |
| |
| absolute_base_url = 'http://code.google.com' |
| |
| url_1 = tracker_helpers.FormatIssueListURL(mr, config) |
| self.assertEqual( |
| '%s/p/proj/issues/list?%s&q=' % ( |
| absolute_base_url, self.default_colspec_param), |
| url_1) |
| |
| url_2 = tracker_helpers.FormatIssueListURL( |
| mr, config, foo=123) |
| self.assertEqual( |
| '%s/p/proj/issues/list?%s&foo=123&q=' % ( |
| absolute_base_url, self.default_colspec_param), |
| url_2) |
| |
| url_3 = tracker_helpers.FormatIssueListURL( |
| mr, config, foo=123, bar='abc') |
| self.assertEqual( |
| '%s/p/proj/issues/list?bar=abc&%s&foo=123&q=' % ( |
| absolute_base_url, self.default_colspec_param), |
| url_3) |
| |
| url_4 = tracker_helpers.FormatIssueListURL( |
| mr, config, baz='escaped+encoded&and100% "safe"') |
| self.assertEqual( |
| '%s/p/proj/issues/list?' |
| 'baz=escaped%%2Bencoded%%26and100%%25%%20%%22safe%%22&%s&q=' % ( |
| absolute_base_url, self.default_colspec_param), |
| url_4) |
| |
| def testFormatIssueListURL_KeepCurrentState(self): |
| config = tracker_pb2.ProjectIssueConfig() |
| path = '/p/proj/issues/detail?id=123&sort=aa&colspec=a b c&groupby=d' |
| mr = testing_helpers.MakeMonorailRequest( |
| path=path, headers={'Host': 'localhost:8080'}) |
| mr.ComputeColSpec(config) |
| |
| absolute_base_url = 'http://localhost:8080' |
| |
| url_1 = tracker_helpers.FormatIssueListURL(mr, config) |
| self.assertEqual( |
| '%s/p/proj/issues/list?colspec=a%%20b%%20c' |
| '&groupby=d&q=&sort=aa' % absolute_base_url, |
| url_1) |
| |
| url_2 = tracker_helpers.FormatIssueListURL( |
| mr, config, foo=123) |
| self.assertEqual( |
| '%s/p/proj/issues/list?' |
| 'colspec=a%%20b%%20c&foo=123&groupby=d&q=&sort=aa' % absolute_base_url, |
| url_2) |
| |
| url_3 = tracker_helpers.FormatIssueListURL( |
| mr, config, colspec='X Y Z') |
| self.assertEqual( |
| '%s/p/proj/issues/list?colspec=a%%20b%%20c' |
| '&groupby=d&q=&sort=aa' % absolute_base_url, |
| url_3) |
| |
| def testFormatRelativeIssueURL(self): |
| self.assertEqual( |
| '/p/proj/issues/attachment', |
| tracker_helpers.FormatRelativeIssueURL( |
| 'proj', urls.ISSUE_ATTACHMENT)) |
| |
| self.assertEqual( |
| '/p/proj/issues/detail?id=123', |
| tracker_helpers.FormatRelativeIssueURL( |
| 'proj', urls.ISSUE_DETAIL, id=123)) |
| |
| @mock.patch('google.appengine.api.app_identity.get_application_id') |
| def testFormatCrBugURL_Prod(self, mock_get_app_id): |
| mock_get_app_id.return_value = 'monorail-prod' |
| self.assertEqual( |
| 'https://crbug.com/proj/123', |
| tracker_helpers.FormatCrBugURL('proj', 123)) |
| self.assertEqual( |
| 'https://crbug.com/123456', |
| tracker_helpers.FormatCrBugURL('chromium', 123456)) |
| |
| @mock.patch('google.appengine.api.app_identity.get_application_id') |
| def testFormatCrBugURL_NonProd(self, mock_get_app_id): |
| mock_get_app_id.return_value = 'monorail-staging' |
| self.assertEqual( |
| '/p/proj/issues/detail?id=123', |
| tracker_helpers.FormatCrBugURL('proj', 123)) |
| self.assertEqual( |
| '/p/chromium/issues/detail?id=123456', |
| tracker_helpers.FormatCrBugURL('chromium', 123456)) |
| |
| @mock.patch('tracker.tracker_constants.ISSUE_ATTACHMENTS_QUOTA_HARD', 1) |
| def testComputeNewQuotaBytesUsed_ProjectQuota(self): |
| upload_1 = framework_helpers.AttachmentUpload( |
| 'matter not', b'three men make a tiger', 'matter not') |
| upload_2 = framework_helpers.AttachmentUpload( |
| 'matter not', b'chicken', 'matter not') |
| attachments = [upload_1, upload_2] |
| |
| project = fake.Project() |
| project.attachment_bytes_used = 10 |
| project.attachment_quota = project.attachment_bytes_used + len( |
| upload_1.contents + upload_2.contents) + 1 |
| |
| actual_new = tracker_helpers.ComputeNewQuotaBytesUsed(project, attachments) |
| expected_new = project.attachment_quota - 1 |
| self.assertEqual(actual_new, expected_new) |
| |
| upload_3 = framework_helpers.AttachmentUpload( |
| 'matter not', b'donut', 'matter not') |
| attachments.append(upload_3) |
| with self.assertRaises(exceptions.OverAttachmentQuota): |
| tracker_helpers.ComputeNewQuotaBytesUsed(project, attachments) |
| |
| @mock.patch( |
| 'tracker.tracker_constants.ISSUE_ATTACHMENTS_QUOTA_HARD', len('tiger')) |
| def testComputeNewQuotaBytesUsed_GeneralQuota(self): |
| upload_1 = framework_helpers.AttachmentUpload( |
| 'matter not', b'tiger', 'matter not') |
| attachments = [upload_1] |
| |
| project = fake.Project() |
| |
| actual_new = tracker_helpers.ComputeNewQuotaBytesUsed(project, attachments) |
| expected_new = len(upload_1.contents) |
| self.assertEqual(actual_new, expected_new) |
| |
| upload_2 = framework_helpers.AttachmentUpload( |
| 'matter not', b'donut', 'matter not') |
| attachments.append(upload_2) |
| with self.assertRaises(exceptions.OverAttachmentQuota): |
| tracker_helpers.ComputeNewQuotaBytesUsed(project, attachments) |
| |
| upload_3 = framework_helpers.AttachmentUpload( |
| 'matter not', b'donut', 'matter not') |
| attachments.append(upload_3) |
| with self.assertRaises(exceptions.OverAttachmentQuota): |
| tracker_helpers.ComputeNewQuotaBytesUsed(project, attachments) |
| |
| def testIsUnderSoftAttachmentQuota(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| # GetAllIssueProjects is tested in GetAllIssueProjectsTest. |
| |
| def testGetPermissionsInAllProjects(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| # FilterOutNonViewableIssues is tested in FilterOutNonViewableIssuesTest. |
| |
| def testMeansOpenInProject(self): |
| config = _MakeConfig() |
| |
| # ensure open means open |
| self.assertTrue(tracker_helpers.MeansOpenInProject('New', config)) |
| self.assertTrue(tracker_helpers.MeansOpenInProject('new', config)) |
| |
| # ensure an unrecognized status means open |
| self.assertTrue(tracker_helpers.MeansOpenInProject( |
| '_undefined_status_', config)) |
| |
| # ensure closed means closed |
| self.assertFalse(tracker_helpers.MeansOpenInProject('Old', config)) |
| self.assertFalse(tracker_helpers.MeansOpenInProject('old', config)) |
| self.assertFalse(tracker_helpers.MeansOpenInProject( |
| 'StatusThatWeDontUseAnymore', config)) |
| |
| def testIsNoisy(self): |
| self.assertTrue(tracker_helpers.IsNoisy(778, 320)) |
| self.assertFalse(tracker_helpers.IsNoisy(20, 500)) |
| self.assertFalse(tracker_helpers.IsNoisy(500, 20)) |
| self.assertFalse(tracker_helpers.IsNoisy(1, 1)) |
| |
| def testMergeCCsAndAddComment(self): |
| target_issue = fake.MakeTestIssue( |
| 789, 10, 'Target issue', 'New', 111) |
| source_issue = fake.MakeTestIssue( |
| 789, 100, 'Source issue', 'New', 222) |
| source_issue.cc_ids.append(111) |
| # Issue without owner |
| source_issue_2 = fake.MakeTestIssue( |
| 789, 101, 'Source issue 2', 'New', 0) |
| |
| self.services.issue.TestAddIssue(target_issue) |
| self.services.issue.TestAddIssue(source_issue) |
| self.services.issue.TestAddIssue(source_issue_2) |
| |
| # We copy this list so that it isn't updated by the test framework |
| initial_issue_comments = ( |
| self.services.issue.GetCommentsForIssue( |
| 'fake cnxn', target_issue.issue_id)[:]) |
| mr = testing_helpers.MakeMonorailRequest(user_info={'user_id': 111}) |
| |
| # Merging source into target should create a comment. |
| self.assertIsNotNone( |
| tracker_helpers.MergeCCsAndAddComment( |
| self.services, mr, source_issue, target_issue)) |
| updated_issue_comments = self.services.issue.GetCommentsForIssue( |
| 'fake cnxn', target_issue.issue_id) |
| for comment in initial_issue_comments: |
| self.assertIn(comment, updated_issue_comments) |
| self.assertEqual( |
| len(initial_issue_comments) + 1, len(updated_issue_comments)) |
| |
| # Merging source into target should add source's owner to target's CCs. |
| updated_target_issue = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 789, 10) |
| self.assertIn(111, updated_target_issue.cc_ids) |
| self.assertIn(222, updated_target_issue.cc_ids) |
| |
| # Merging source 2 into target should make a comment, but not update CCs. |
| self.assertIsNotNone( |
| tracker_helpers.MergeCCsAndAddComment( |
| self.services, mr, source_issue_2, updated_target_issue)) |
| updated_target_issue = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 789, 10) |
| self.assertNotIn(0, updated_target_issue.cc_ids) |
| |
| def testMergeCCsAndAddComment_RestrictedSourceIssue(self): |
| target_issue = fake.MakeTestIssue( |
| 789, 10, 'Target issue', 'New', 222) |
| target_issue_2 = fake.MakeTestIssue( |
| 789, 11, 'Target issue 2', 'New', 222) |
| source_issue = fake.MakeTestIssue( |
| 789, 100, 'Source issue', 'New', 111) |
| source_issue.cc_ids.append(111) |
| source_issue.labels.append('Restrict-View-Commit') |
| target_issue_2.labels.append('Restrict-View-Commit') |
| |
| self.services.issue.TestAddIssue(source_issue) |
| self.services.issue.TestAddIssue(target_issue) |
| self.services.issue.TestAddIssue(target_issue_2) |
| |
| # We copy this list so that it isn't updated by the test framework |
| initial_issue_comments = self.services.issue.GetCommentsForIssue( |
| 'fake cnxn', target_issue.issue_id)[:] |
| mr = testing_helpers.MakeMonorailRequest(user_info={'user_id': 111}) |
| self.assertIsNotNone( |
| tracker_helpers.MergeCCsAndAddComment( |
| self.services, mr, source_issue, target_issue)) |
| |
| # When the source is restricted, we update the target comments... |
| updated_issue_comments = self.services.issue.GetCommentsForIssue( |
| 'fake cnxn', target_issue.issue_id) |
| for comment in initial_issue_comments: |
| self.assertIn(comment, updated_issue_comments) |
| self.assertEqual( |
| len(initial_issue_comments) + 1, len(updated_issue_comments)) |
| # ...but not the target CCs... |
| updated_target_issue = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 789, 10) |
| self.assertNotIn(111, updated_target_issue.cc_ids) |
| # ...unless both issues have the same restrictions. |
| self.assertIsNotNone( |
| tracker_helpers.MergeCCsAndAddComment( |
| self.services, mr, source_issue, target_issue_2)) |
| updated_target_issue_2 = self.services.issue.GetIssueByLocalID( |
| 'fake cnxn', 789, 11) |
| self.assertIn(111, updated_target_issue_2.cc_ids) |
| |
| def testMergeCCsAndAddCommentMultipleIssues(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testGetAttachmentIfAllowed(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testLabelsMaskedByFields(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testLabelsNotMaskedByFields(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testLookupComponentIDs(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| def testParsePostDataUsers(self): |
| pd_users = 'a@example.com, b@example.com' |
| |
| pd_users_ids, pd_users_str = tracker_helpers.ParsePostDataUsers( |
| self.cnxn, pd_users, self.services.user) |
| |
| self.assertEqual([1, 2], sorted(pd_users_ids)) |
| self.assertEqual('a@example.com, b@example.com', pd_users_str) |
| |
| def testParsePostDataUsers_Empty(self): |
| pd_users = '' |
| |
| pd_users_ids, pd_users_str = tracker_helpers.ParsePostDataUsers( |
| self.cnxn, pd_users, self.services.user) |
| |
| self.assertEqual([], sorted(pd_users_ids)) |
| self.assertEqual('', pd_users_str) |
| |
| def testFilterIssueTypes(self): |
| pass # TODO(jrobbins): Write this test. |
| |
| # ParseMergeFields is tested in IssueMergeTest. |
| # AddIssueStarrers is tested in IssueMergeTest.testMergeIssueStars(). |
| # CanEditProjectIssue is tested in IssueMergeTest. |
| |
| def testPairDerivedValuesWithRuleExplanations_Nothing(self): |
| """Test we return nothing for an issue with no derived values.""" |
| proposed_issue = tracker_pb2.Issue() # No derived values. |
| traces = {} |
| derived_users_by_id = {} |
| actual = tracker_helpers.PairDerivedValuesWithRuleExplanations( |
| proposed_issue, traces, derived_users_by_id) |
| (derived_labels_and_why, derived_owner_and_why, |
| derived_cc_and_why, warnings_and_why, errors_and_why) = actual |
| self.assertEqual([], derived_labels_and_why) |
| self.assertEqual([], derived_owner_and_why) |
| self.assertEqual([], derived_cc_and_why) |
| self.assertEqual([], warnings_and_why) |
| self.assertEqual([], errors_and_why) |
| |
| def testPairDerivedValuesWithRuleExplanations_SomeValues(self): |
| """Test we return derived values and explanations for an issue.""" |
| proposed_issue = tracker_pb2.Issue( |
| derived_owner_id=111, derived_cc_ids=[222, 333], |
| derived_labels=['aaa', 'zzz'], |
| derived_warnings=['Watch out'], |
| derived_errors=['Status Assigned requires an owner']) |
| traces = { |
| (tracker_pb2.FieldID.OWNER, 111): 'explain 1', |
| (tracker_pb2.FieldID.CC, 222): 'explain 2', |
| (tracker_pb2.FieldID.CC, 333): 'explain 3', |
| (tracker_pb2.FieldID.LABELS, 'aaa'): 'explain 4', |
| (tracker_pb2.FieldID.WARNING, 'Watch out'): 'explain 6', |
| (tracker_pb2.FieldID.ERROR, |
| 'Status Assigned requires an owner'): 'explain 7', |
| # There can be extra traces that are not used. |
| (tracker_pb2.FieldID.LABELS, 'bbb'): 'explain 5', |
| # If there is no trace for some derived value, why is None. |
| } |
| derived_users_by_id = { |
| 111: testing_helpers.Blank(display_name='one@example.com'), |
| 222: testing_helpers.Blank(display_name='two@example.com'), |
| 333: testing_helpers.Blank(display_name='three@example.com'), |
| } |
| actual = tracker_helpers.PairDerivedValuesWithRuleExplanations( |
| proposed_issue, traces, derived_users_by_id) |
| (derived_labels_and_why, derived_owner_and_why, |
| derived_cc_and_why, warnings_and_why, errors_and_why) = actual |
| self.assertEqual([ |
| {'value': 'aaa', 'why': 'explain 4'}, |
| {'value': 'zzz', 'why': None}, |
| ], derived_labels_and_why) |
| self.assertEqual([ |
| {'value': 'one@example.com', 'why': 'explain 1'}, |
| ], derived_owner_and_why) |
| self.assertEqual([ |
| {'value': 'two@example.com', 'why': 'explain 2'}, |
| {'value': 'three@example.com', 'why': 'explain 3'}, |
| ], derived_cc_and_why) |
| self.assertEqual([ |
| {'value': 'Watch out', 'why': 'explain 6'}, |
| ], warnings_and_why) |
| self.assertEqual([ |
| {'value': 'Status Assigned requires an owner', 'why': 'explain 7'}, |
| ], errors_and_why) |
| |
| |
| class MakeViewsForUsersInIssuesTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.issue1 = _Issue('proj', 1) |
| self.issue1.owner_id = 1001 |
| self.issue1.reporter_id = 1002 |
| |
| self.issue2 = _Issue('proj', 2) |
| self.issue2.owner_id = 2001 |
| self.issue2.reporter_id = 2002 |
| self.issue2.cc_ids.extend([1, 1001, 1002, 1003]) |
| |
| self.issue3 = _Issue('proj', 3) |
| self.issue3.owner_id = 1001 |
| self.issue3.reporter_id = 3002 |
| |
| self.user = fake.UserService() |
| for user_id in [1, 1001, 1002, 1003, 2001, 2002, 3002]: |
| self.user.TestAddUser( |
| 'test%d' % user_id, user_id, add_user=True) |
| |
| def testMakeViewsForUsersInIssues(self): |
| issue_list = [self.issue1, self.issue2, self.issue3] |
| users_by_id = tracker_helpers.MakeViewsForUsersInIssues( |
| 'fake cnxn', issue_list, self.user) |
| six.assertCountEqual( |
| self, [0, 1, 1001, 1002, 1003, 2001, 2002, 3002], |
| list(users_by_id.keys())) |
| for user_id in [1001, 1002, 1003, 2001]: |
| self.assertEqual(users_by_id[user_id].user_id, user_id) |
| |
| def testMakeViewsForUsersInIssuesOmittingSome(self): |
| issue_list = [self.issue1, self.issue2, self.issue3] |
| users_by_id = tracker_helpers.MakeViewsForUsersInIssues( |
| 'fake cnxn', issue_list, self.user, omit_ids=[1001, 1003]) |
| six.assertCountEqual( |
| self, [0, 1, 1002, 2001, 2002, 3002], list(users_by_id.keys())) |
| for user_id in [1002, 2001, 2002, 3002]: |
| self.assertEqual(users_by_id[user_id].user_id, user_id) |
| |
| def testMakeViewsForUsersInIssuesEmpty(self): |
| issue_list = [] |
| users_by_id = tracker_helpers.MakeViewsForUsersInIssues( |
| 'fake cnxn', issue_list, self.user) |
| six.assertCountEqual(self, [], list(users_by_id.keys())) |
| |
| |
| class GetAllIssueProjectsTest(unittest.TestCase): |
| issue_x_1 = tracker_pb2.Issue() |
| issue_x_1.project_id = 789 |
| issue_x_1.local_id = 1 |
| issue_x_1.reporter_id = 1002 |
| |
| issue_x_2 = tracker_pb2.Issue() |
| issue_x_2.project_id = 789 |
| issue_x_2.local_id = 2 |
| issue_x_2.reporter_id = 2002 |
| |
| issue_y_1 = tracker_pb2.Issue() |
| issue_y_1.project_id = 678 |
| issue_y_1.local_id = 1 |
| issue_y_1.reporter_id = 2002 |
| |
| def setUp(self): |
| self.project_service = fake.ProjectService() |
| self.project_service.TestAddProject('proj-x', project_id=789) |
| self.project_service.TestAddProject('proj-y', project_id=678) |
| self.cnxn = 'fake connection' |
| |
| def testGetAllIssueProjects_Empty(self): |
| self.assertEqual( |
| {}, tracker_helpers.GetAllIssueProjects( |
| self.cnxn, [], self.project_service)) |
| |
| def testGetAllIssueProjects_Normal(self): |
| self.assertEqual( |
| {789: self.project_service.GetProjectByName(self.cnxn, 'proj-x')}, |
| tracker_helpers.GetAllIssueProjects( |
| self.cnxn, [self.issue_x_1, self.issue_x_2], self.project_service)) |
| self.assertEqual( |
| {789: self.project_service.GetProjectByName(self.cnxn, 'proj-x'), |
| 678: self.project_service.GetProjectByName(self.cnxn, 'proj-y')}, |
| tracker_helpers.GetAllIssueProjects( |
| self.cnxn, [self.issue_x_1, self.issue_x_2, self.issue_y_1], |
| self.project_service)) |
| |
| |
| class FilterOutNonViewableIssuesTest(unittest.TestCase): |
| owner_id = 111 |
| committer_id = 222 |
| nonmember_1_id = 1002 |
| nonmember_2_id = 2002 |
| nonmember_3_id = 3002 |
| |
| issue1 = tracker_pb2.Issue() |
| issue1.project_name = 'proj' |
| issue1.project_id = 789 |
| issue1.local_id = 1 |
| issue1.reporter_id = nonmember_1_id |
| |
| issue2 = tracker_pb2.Issue() |
| issue2.project_name = 'proj' |
| issue2.project_id = 789 |
| issue2.local_id = 2 |
| issue2.reporter_id = nonmember_2_id |
| issue2.labels.extend(['foo', 'bar']) |
| |
| issue3 = tracker_pb2.Issue() |
| issue3.project_name = 'proj' |
| issue3.project_id = 789 |
| issue3.local_id = 3 |
| issue3.reporter_id = nonmember_3_id |
| issue3.labels.extend(['restrict-view-commit']) |
| |
| issue4 = tracker_pb2.Issue() |
| issue4.project_name = 'proj' |
| issue4.project_id = 789 |
| issue4.local_id = 4 |
| issue4.reporter_id = nonmember_3_id |
| issue4.labels.extend(['Foo', 'Restrict-View-Commit']) |
| |
| def setUp(self): |
| self.user = user_pb2.User() |
| self.project = self.MakeProject(project_pb2.ProjectState.LIVE) |
| self.config = tracker_bizobj.MakeDefaultProjectIssueConfig( |
| self.project.project_id) |
| self.project_dict = {self.project.project_id: self.project} |
| self.config_dict = {self.config.project_id: self.config} |
| |
| def MakeProject(self, state): |
| p = project_pb2.Project( |
| project_id=789, project_name='proj', state=state, |
| owner_ids=[self.owner_id], committer_ids=[self.committer_id]) |
| return p |
| |
| def testFilterOutNonViewableIssues_Member(self): |
| # perms will be permissions.COMMITTER_ACTIVE_PERMISSIONSET |
| filtered_issues = tracker_helpers.FilterOutNonViewableIssues( |
| {self.committer_id}, self.user, self.project_dict, |
| self.config_dict, |
| [self.issue1, self.issue2, self.issue3, self.issue4]) |
| self.assertListEqual([1, 2, 3, 4], |
| [issue.local_id for issue in filtered_issues]) |
| |
| def testFilterOutNonViewableIssues_Owner(self): |
| # perms will be permissions.OWNER_ACTIVE_PERMISSIONSET |
| filtered_issues = tracker_helpers.FilterOutNonViewableIssues( |
| {self.owner_id}, self.user, self.project_dict, self.config_dict, |
| [self.issue1, self.issue2, self.issue3, self.issue4]) |
| self.assertListEqual([1, 2, 3, 4], |
| [issue.local_id for issue in filtered_issues]) |
| |
| def testFilterOutNonViewableIssues_Empty(self): |
| # perms will be permissions.COMMITTER_ACTIVE_PERMISSIONSET |
| filtered_issues = tracker_helpers.FilterOutNonViewableIssues( |
| {self.committer_id}, self.user, self.project_dict, |
| self.config_dict, []) |
| self.assertListEqual([], filtered_issues) |
| |
| def testFilterOutNonViewableIssues_NonMember(self): |
| # perms will be permissions.READ_ONLY_PERMISSIONSET |
| filtered_issues = tracker_helpers.FilterOutNonViewableIssues( |
| {self.nonmember_1_id}, self.user, self.project_dict, |
| self.config_dict, [self.issue1, self.issue2, self.issue3, self.issue4]) |
| self.assertListEqual([1, 2], |
| [issue.local_id for issue in filtered_issues]) |
| |
| def testFilterOutNonViewableIssues_Reporter(self): |
| # perms will be permissions.READ_ONLY_PERMISSIONSET |
| filtered_issues = tracker_helpers.FilterOutNonViewableIssues( |
| {self.nonmember_3_id}, self.user, self.project_dict, |
| self.config_dict, [self.issue1, self.issue2, self.issue3, self.issue4]) |
| self.assertListEqual([1, 2, 3, 4], |
| [issue.local_id for issue in filtered_issues]) |
| |
| |
| class IssueMergeTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services( |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| user=fake.UserService(), |
| project=fake.ProjectService(), |
| issue_star=fake.IssueStarService(), |
| spam=fake.SpamService() |
| ) |
| self.project = self.services.project.TestAddProject('proj', project_id=987) |
| self.config = tracker_bizobj.MakeDefaultProjectIssueConfig( |
| self.project.project_id) |
| self.project_dict = {self.project.project_id: self.project} |
| self.config_dict = {self.config.project_id: self.config} |
| |
| def testParseMergeFields_NotSpecified(self): |
| issue = fake.MakeTestIssue(987, 1, 'summary', 'New', 111) |
| errors = template_helpers.EZTError() |
| post_data = {} |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, None, 'proj', post_data, 'New', self.config, issue, errors) |
| self.assertEqual('', text) |
| self.assertEqual(None, merge_into_issue) |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, None, 'proj', post_data, 'Duplicate', self.config, issue, |
| errors) |
| self.assertEqual('', text) |
| self.assertTrue(errors.merge_into_id) |
| self.assertEqual(None, merge_into_issue) |
| |
| def testParseMergeFields_WrongStatus(self): |
| issue = fake.MakeTestIssue(987, 1, 'summary', 'New', 111) |
| errors = template_helpers.EZTError() |
| post_data = {'merge_into': '12'} |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, None, 'proj', post_data, 'New', self.config, issue, errors) |
| self.assertEqual('', text) |
| self.assertEqual(None, merge_into_issue) |
| |
| def testParseMergeFields_NoSuchIssue(self): |
| issue = fake.MakeTestIssue(987, 1, 'summary', 'New', 111) |
| issue.merged_into = 12 |
| errors = template_helpers.EZTError() |
| post_data = {'merge_into': '12'} |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, self.services, 'proj', post_data, 'Duplicate', |
| self.config, issue, errors) |
| self.assertEqual('12', text) |
| self.assertEqual(None, merge_into_issue) |
| |
| def testParseMergeFields_DontSelfMerge(self): |
| issue = fake.MakeTestIssue(987, 1, 'summary', 'New', 111) |
| errors = template_helpers.EZTError() |
| post_data = {'merge_into': '1'} |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, self.services, 'proj', post_data, 'Duplicate', self.config, |
| issue, errors) |
| self.assertEqual('1', text) |
| self.assertEqual(None, merge_into_issue) |
| self.assertEqual('Cannot merge issue into itself', errors.merge_into_id) |
| |
| def testParseMergeFields_NewIssueToMerge(self): |
| merged_issue = fake.MakeTestIssue( |
| self.project.project_id, |
| 1, |
| 'unused_summary', |
| 'unused_status', |
| 111, |
| reporter_id=111) |
| self.services.issue.TestAddIssue(merged_issue) |
| mergee_issue = fake.MakeTestIssue( |
| self.project.project_id, |
| 2, |
| 'unused_summary', |
| 'unused_status', |
| 111, |
| reporter_id=111) |
| self.services.issue.TestAddIssue(mergee_issue) |
| |
| errors = template_helpers.EZTError() |
| post_data = {'merge_into': str(mergee_issue.local_id)} |
| |
| text, merge_into_issue = tracker_helpers.ParseMergeFields( |
| self.cnxn, self.services, 'proj', post_data, 'Duplicate', self.config, |
| merged_issue, errors) |
| self.assertEqual(str(mergee_issue.local_id), text) |
| self.assertEqual(mergee_issue, merge_into_issue) |
| |
| def testCanEditProjectIssue(self): |
| mr = testing_helpers.MakeMonorailRequest() |
| issue = fake.MakeTestIssue( |
| self.project.project_id, 1, 'summary', 'New', 111) |
| issue.project_name = self.project.project_name |
| |
| non_member_not_allowed = tracker_helpers.CanEditProjectIssue( |
| mr, self.project, issue, None) |
| self.assertEqual(False, non_member_not_allowed) |
| |
| committer_id = 3 |
| self.project.committer_ids.extend([committer_id]) |
| mr.auth.effective_ids.add(committer_id) |
| committer_allowed = tracker_helpers.CanEditProjectIssue( |
| mr, self.project, issue, None) |
| self.assertEqual(True, committer_allowed) |
| |
| self.project.state = project_pb2.ProjectState.ARCHIVED |
| committer_read_only_not_allowed = tracker_helpers.CanEditProjectIssue( |
| mr, self.project, issue, None) |
| self.assertEqual(False, committer_read_only_not_allowed) |
| |
| owner_id = 1 |
| self.project.owner_ids.extend([owner_id]) |
| mr.auth.effective_ids.add(owner_id) |
| owner_read_only_not_allowed = tracker_helpers.CanEditProjectIssue( |
| mr, self.project, issue, None) |
| self.assertEqual(False, owner_read_only_not_allowed) |
| |
| def testMergeIssueStars(self): |
| mr = testing_helpers.MakeMonorailRequest() |
| mr.project_name = self.project.project_name |
| mr.project = self.project |
| |
| config = self.services.config.GetProjectConfig( |
| self.cnxn, self.project.project_id) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 1, 1, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 1, 2, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 1, 3, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 3, 3, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 3, 6, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 2, 3, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 2, 4, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, config, 2, 5, True) |
| |
| new_starrers = tracker_helpers.GetNewIssueStarrers( |
| self.cnxn, self.services, [1, 3], 2) |
| six.assertCountEqual(self, new_starrers, [1, 2, 6]) |
| tracker_helpers.AddIssueStarrers( |
| self.cnxn, self.services, mr, 2, self.project, new_starrers) |
| issue_2_starrers = self.services.issue_star.LookupItemStarrers( |
| self.cnxn, 2) |
| # XXX(jrobbins): these tests incorrectly mix local IDs with IIDs. |
| six.assertCountEqual(self, [1, 2, 3, 4, 5, 6], issue_2_starrers) |
| |
| |
| class MergeLinkedMembersTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services( |
| user=fake.UserService()) |
| self.user1 = self.services.user.TestAddUser('one@example.com', 111) |
| self.user2 = self.services.user.TestAddUser('two@example.com', 222) |
| |
| def testNoLinkedAccounts(self): |
| """When no candidate accounts are linked, they are all returned.""" |
| actual = tracker_helpers._MergeLinkedMembers( |
| self.cnxn, self.services.user, [111, 222]) |
| self.assertEqual([111, 222], actual) |
| |
| def testSomeLinkedButNoMasking(self): |
| """If an account has linked accounts, but they are not here, keep it.""" |
| self.user1.linked_child_ids = [999] |
| self.user2.linked_parent_id = 999 |
| actual = tracker_helpers._MergeLinkedMembers( |
| self.cnxn, self.services.user, [111, 222]) |
| self.assertEqual([111, 222], actual) |
| |
| def testParentMasksChild(self): |
| """When two accounts linked, only the parent is returned.""" |
| self.user2.linked_parent_id = 111 |
| actual = tracker_helpers._MergeLinkedMembers( |
| self.cnxn, self.services.user, [111, 222]) |
| self.assertEqual([111], actual) |
| |
| |
| class FilterMemberDataTest(unittest.TestCase): |
| |
| def setUp(self): |
| services = service_manager.Services( |
| project=fake.ProjectService(), |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| user=fake.UserService()) |
| self.owner_email = 'owner@dom.com' |
| self.committer_email = 'commit@dom.com' |
| self.contributor_email = 'contrib@dom.com' |
| self.indirect_member_email = 'ind@dom.com' |
| self.all_emails = [self.owner_email, self.committer_email, |
| self.contributor_email, self.indirect_member_email] |
| self.project = services.project.TestAddProject('proj') |
| |
| def DoFiltering(self, perms, unsigned_user=False): |
| mr = testing_helpers.MakeMonorailRequest( |
| project=self.project, perms=perms) |
| if not unsigned_user: |
| mr.auth.user_id = 111 |
| mr.auth.user_view = testing_helpers.Blank(domain='jrobbins.org') |
| return tracker_helpers._FilterMemberData( |
| mr, [self.owner_email], [self.committer_email], |
| [self.contributor_email], [self.indirect_member_email], mr.project) |
| |
| def testUnsignedUser_NormalProject(self): |
| visible_members = self.DoFiltering( |
| permissions.READ_ONLY_PERMISSIONSET, unsigned_user=True) |
| six.assertCountEqual( |
| self, [ |
| self.owner_email, self.committer_email, self.contributor_email, |
| self.indirect_member_email |
| ], visible_members) |
| |
| def testUnsignedUser_RestrictedProject(self): |
| self.project.only_owners_see_contributors = True |
| visible_members = self.DoFiltering( |
| permissions.READ_ONLY_PERMISSIONSET, unsigned_user=True) |
| six.assertCountEqual( |
| self, |
| [self.owner_email, self.committer_email, self.indirect_member_email], |
| visible_members) |
| |
| def testOwnersAndAdminsCanSeeAll_NormalProject(self): |
| visible_members = self.DoFiltering( |
| permissions.OWNER_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| visible_members = self.DoFiltering( |
| permissions.ADMIN_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| def testOwnersAndAdminsCanSeeAll_HubAndSpoke(self): |
| self.project.only_owners_see_contributors = True |
| |
| visible_members = self.DoFiltering( |
| permissions.OWNER_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| visible_members = self.DoFiltering( |
| permissions.ADMIN_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| visible_members = self.DoFiltering( |
| permissions.COMMITTER_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| def testNonOwnersCanSeeAll_NormalProject(self): |
| visible_members = self.DoFiltering( |
| permissions.COMMITTER_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| visible_members = self.DoFiltering( |
| permissions.CONTRIBUTOR_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual(self, self.all_emails, visible_members) |
| |
| def testCommittersSeeOnlySameDomain_HubAndSpoke(self): |
| self.project.only_owners_see_contributors = True |
| |
| visible_members = self.DoFiltering( |
| permissions.CONTRIBUTOR_ACTIVE_PERMISSIONSET) |
| six.assertCountEqual( |
| self, |
| [self.owner_email, self.committer_email, self.indirect_member_email], |
| visible_members) |
| |
| |
| class GetLabelOptionsTest(unittest.TestCase): |
| |
| @mock.patch('tracker.tracker_helpers.LabelsNotMaskedByFields') |
| def testGetLabelOptions(self, mockLabelsNotMaskedByFields): |
| mockLabelsNotMaskedByFields.return_value = [] |
| config = tracker_pb2.ProjectIssueConfig() |
| custom_perms = [] |
| actual = tracker_helpers.GetLabelOptions(config, custom_perms) |
| expected = [ |
| {'doc': 'Only users who can edit the issue may access it', |
| 'name': 'Restrict-View-EditIssue'}, |
| {'doc': 'Only users who can edit the issue may add comments', |
| 'name': 'Restrict-AddIssueComment-EditIssue'}, |
| {'doc': 'Custom permission CoreTeam is needed to access', |
| 'name': 'Restrict-View-CoreTeam'} |
| ] |
| self.assertEqual(expected, actual) |
| |
| def testBuildRestrictionChoices(self): |
| choices = tracker_helpers._BuildRestrictionChoices([], [], []) |
| self.assertEqual([], choices) |
| |
| choices = tracker_helpers._BuildRestrictionChoices( |
| [], ['Hop', 'Jump'], []) |
| self.assertEqual([], choices) |
| |
| freq = [('View', 'B', 'You need permission B to do anything'), |
| ('A', 'B', 'You need B to use A')] |
| choices = tracker_helpers._BuildRestrictionChoices(freq, [], []) |
| expected = [dict(name='Restrict-View-B', |
| doc='You need permission B to do anything'), |
| dict(name='Restrict-A-B', |
| doc='You need B to use A')] |
| self.assertListEqual(expected, choices) |
| |
| extra_perms = ['Over18', 'Over21'] |
| choices = tracker_helpers._BuildRestrictionChoices( |
| [], ['Drink', 'Smoke'], extra_perms) |
| expected = [dict(name='Restrict-Drink-Over18', |
| doc='Permission Over18 needed to use Drink'), |
| dict(name='Restrict-Drink-Over21', |
| doc='Permission Over21 needed to use Drink'), |
| dict(name='Restrict-Smoke-Over18', |
| doc='Permission Over18 needed to use Smoke'), |
| dict(name='Restrict-Smoke-Over21', |
| doc='Permission Over21 needed to use Smoke')] |
| self.assertListEqual(expected, choices) |
| |
| |
| class FilterKeptAttachmentsTest(unittest.TestCase): |
| def testFilterKeptAttachments(self): |
| comments = [ |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=1)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[ |
| tracker_pb2.Attachment(attachment_id=2), |
| tracker_pb2.Attachment(attachment_id=3)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| approval_id=24, |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=4)])] |
| |
| filtered = tracker_helpers.FilterKeptAttachments( |
| True, [1, 2, 3, 4], comments, None) |
| self.assertEqual([2, 3], filtered) |
| |
| def testApprovalDescription(self): |
| comments = [ |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=1)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[ |
| tracker_pb2.Attachment(attachment_id=2), |
| tracker_pb2.Attachment(attachment_id=3)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| approval_id=24, |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=4)])] |
| |
| filtered = tracker_helpers.FilterKeptAttachments( |
| True, [1, 2, 3, 4], comments, 24) |
| self.assertEqual([4], filtered) |
| |
| def testNotAnIssueDescription(self): |
| comments = [ |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=1)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| is_description=True, |
| attachments=[ |
| tracker_pb2.Attachment(attachment_id=2), |
| tracker_pb2.Attachment(attachment_id=3)]), |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment( |
| approval_id=24, |
| is_description=True, |
| attachments=[tracker_pb2.Attachment(attachment_id=4)])] |
| |
| filtered = tracker_helpers.FilterKeptAttachments( |
| False, [1, 2, 3, 4], comments, None) |
| self.assertIsNone(filtered) |
| |
| def testNoDescriptionsInComments(self): |
| comments = [ |
| tracker_pb2.IssueComment(), |
| tracker_pb2.IssueComment()] |
| |
| filtered = tracker_helpers.FilterKeptAttachments( |
| True, [1, 2, 3, 4], comments, None) |
| self.assertEqual([], filtered) |
| |
| def testNoComments(self): |
| filtered = tracker_helpers.FilterKeptAttachments( |
| True, [1, 2, 3, 4], [], None) |
| self.assertEqual([], filtered) |
| |
| |
| class EnumFieldHelpersTest(unittest.TestCase): |
| |
| def test_GetEnumFieldValuesAndDocstrings(self): |
| """We can get all choices for an enum field""" |
| fd = tracker_pb2.FieldDef( |
| field_id=123, |
| project_id=1, |
| field_name='yellow', |
| field_type=tracker_pb2.FieldTypes.ENUM_TYPE) |
| ld_1 = tracker_pb2.LabelDef( |
| label='yellow-submarine', label_docstring='ld_1_docstring') |
| ld_2 = tracker_pb2.LabelDef( |
| label='yellow-tisket', label_docstring='ld_2_docstring') |
| ld_3 = tracker_pb2.LabelDef( |
| label='yellow-basket', label_docstring='ld_3_docstring') |
| ld_4 = tracker_pb2.LabelDef( |
| label='yellow', label_docstring='ld_4_docstring') |
| ld_5 = tracker_pb2.LabelDef( |
| label='not-yellow', label_docstring='ld_5_docstring') |
| ld_6 = tracker_pb2.LabelDef( |
| label='yellow-tasket', |
| label_docstring='ld_6_docstring', |
| deprecated=True) |
| config = tracker_pb2.ProjectIssueConfig( |
| default_template_for_developers=1, |
| default_template_for_users=2, |
| well_known_labels=[ld_1, ld_2, ld_3, ld_4, ld_5, ld_6]) |
| actual = tracker_helpers._GetEnumFieldValuesAndDocstrings(fd, config) |
| # Expect to omit labels `yellow` and `not-yellow` due to prefix mismatch |
| # Also expect to omit label `yellow-tasket` because it's deprecated |
| expected = [ |
| ('submarine', 'ld_1_docstring'), ('tisket', 'ld_2_docstring'), |
| ('basket', 'ld_3_docstring') |
| ] |
| self.assertEqual(expected, actual) |
| |
| |
| class CreateIssueHelpersTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.services = service_manager.Services( |
| project=fake.ProjectService(), |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| user=fake.UserService(), |
| usergroup=fake.UserGroupService()) |
| self.cnxn = 'fake cnxn' |
| |
| self.project_member = self.services.user.TestAddUser( |
| 'user_1@example.com', 111) |
| self.project_group_member = self.services.user.TestAddUser( |
| 'group@example.com', 999) |
| self.project = self.services.project.TestAddProject( |
| 'proj', |
| project_id=789, |
| committer_ids=[ |
| self.project_member.user_id, self.project_group_member.user_id |
| ]) |
| self.no_project_user = self.services.user.TestAddUser( |
| 'user_2@example.com', 222) |
| self.config = fake.MakeTestConfig(self.project.project_id, [], []) |
| self.int_fd = tracker_bizobj.MakeFieldDef( |
| 123, 789, 'CPU', tracker_pb2.FieldTypes.INT_TYPE, None, '', False, |
| False, False, None, None, '', False, '', '', |
| tracker_pb2.NotifyTriggers.NEVER, 'no_action', 'doc', False) |
| self.int_fd.max_value = 999 |
| self.config.field_defs = [self.int_fd] |
| self.status_1 = tracker_pb2.StatusDef( |
| status='New', means_open=True, status_docstring='status_1 docstring') |
| self.config.well_known_statuses = [self.status_1] |
| self.component_def_1 = tracker_pb2.ComponentDef( |
| component_id=1, path='compFOO') |
| self.component_def_2 = tracker_pb2.ComponentDef( |
| component_id=2, path='deprecated', deprecated=True) |
| self.config.component_defs = [self.component_def_1, self.component_def_2] |
| self.services.config.StoreConfig('cnxn', self.config) |
| self.services.usergroup.TestAddGroupSettings(999, 'group@example.com') |
| |
| def testAssertValidIssueForCreate_Valid(self): |
| input_issue = tracker_pb2.Issue( |
| summary='sum', |
| status='New', |
| owner_id=111, |
| project_id=789, |
| component_ids=[1], |
| cc_ids=[999]) |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesLabels(self): |
| input_issue = tracker_pb2.Issue( |
| summary='sum', |
| labels=['freeze_new_label'], |
| status='New', |
| owner_id=111, |
| project_id=789) |
| with self.assertRaisesRegex( |
| exceptions.InputException, |
| ("The creation of new labels is blocked for the Chromium project" |
| " in Monorail. To continue with editing your issue, please" |
| " remove: freeze_new_label label\\(s\\)")): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesOwner(self): |
| input_issue = tracker_pb2.Issue( |
| summary='sum', status='New', owner_id=222, project_id=789) |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Issue owner must be a project member'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| input_issue.owner_id = 333 |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Issue owner user ID not found'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| input_issue.owner_id = 999 |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Issue owner cannot be a user group'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesSummary(self): |
| input_issue = tracker_pb2.Issue( |
| summary='', status='New', owner_id=111, project_id=789) |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Summary is required'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| input_issue.summary = ' ' |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesDescription(self): |
| input_issue = tracker_pb2.Issue( |
| summary='sum', status='New', owner_id=111, project_id=789) |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Description is required'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, '') |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, ' ') |
| |
| def testAssertValidIssueForCreate_ValidatesFieldDef(self): |
| fv = tracker_bizobj.MakeFieldValue( |
| self.int_fd.field_id, 1000, None, None, None, None, False) |
| input_issue = tracker_pb2.Issue( |
| summary='sum', |
| status='New', |
| owner_id=111, |
| project_id=789, |
| field_values=[fv]) |
| with self.assertRaises(exceptions.InputException): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesStatus(self): |
| input_issue = tracker_pb2.Issue( |
| summary='sum', status='DNE_status', owner_id=111, project_id=789) |
| |
| def mock_status_lookup(*_args, **_kwargs): |
| return None |
| |
| self.services.config.LookupStatusID = mock_status_lookup |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Undefined status: DNE_status'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesComponents(self): |
| # Tests an undefined component. |
| input_issue = tracker_pb2.Issue( |
| summary='', |
| status='New', |
| owner_id=111, |
| project_id=789, |
| component_ids=[3]) |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Undefined or deprecated component with id: 3'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| # Tests a deprecated component. |
| input_issue = tracker_pb2.Issue( |
| summary='', |
| status='New', |
| owner_id=111, |
| project_id=789, |
| component_ids=[self.component_def_2.component_id]) |
| with self.assertRaisesRegex(exceptions.InputException, |
| 'Undefined or deprecated component with id: 2'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| |
| def testAssertValidIssueForCreate_ValidatesUsers(self): |
| user_fd = tracker_bizobj.MakeFieldDef( |
| 123, 789, 'CPU', tracker_pb2.FieldTypes.INT_TYPE, None, '', False, |
| False, False, None, None, '', False, '', '', |
| tracker_pb2.NotifyTriggers.NEVER, 'no_action', 'doc', False) |
| self.services.config.TestAddFieldDef(user_fd) |
| |
| input_issue = tracker_pb2.Issue( |
| summary='sum', |
| status='New', |
| owner_id=111, |
| project_id=789, |
| cc_ids=[123], |
| field_values=[ |
| tracker_bizobj.MakeFieldValue( |
| user_fd.field_id, None, None, 124, None, None, False) |
| ]) |
| copied_issue = copy.deepcopy(input_issue) |
| with self.assertRaisesRegex(exceptions.InputException, |
| r'users/123: .+\nusers/124: .+'): |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| self.assertEqual(input_issue, copied_issue) |
| |
| self.services.user.TestAddUser('a@test.com', 123) |
| self.services.user.TestAddUser('a@test.com', 124) |
| tracker_helpers.AssertValidIssueForCreate( |
| self.cnxn, self.services, input_issue, 'nonempty description') |
| self.assertEqual(input_issue, copied_issue) |
| |
| |
| class ModifyIssuesHelpersTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.services = service_manager.Services( |
| project=fake.ProjectService(), |
| config=fake.ConfigService(), |
| issue=fake.IssueService(), |
| issue_star=fake.IssueStarService(), |
| user=fake.UserService(), |
| usergroup=fake.UserGroupService()) |
| self.cnxn = 'fake cnxn' |
| |
| self.project_member = self.services.user.TestAddUser( |
| 'user_1@example.com', 111) |
| self.project = self.services.project.TestAddProject( |
| 'proj', project_id=789, committer_ids=[self.project_member.user_id]) |
| self.no_project_user = self.services.user.TestAddUser( |
| 'user_2@example.com', 222) |
| |
| self.config = fake.MakeTestConfig(self.project.project_id, [], []) |
| self.int_fd = tracker_bizobj.MakeFieldDef( |
| 123, 789, 'CPU', tracker_pb2.FieldTypes.INT_TYPE, None, '', False, |
| False, False, None, None, '', False, '', '', |
| tracker_pb2.NotifyTriggers.NEVER, 'no_action', 'doc', False) |
| self.int_fd.max_value = 999 |
| self.config.field_defs = [self.int_fd] |
| self.services.config.StoreConfig('cnxn', self.config) |
| |
| def testApplyAllIssueChanges(self): |
| issue_delta_pairs = [] |
| no_change_iid = 78942 |
| |
| expected_issues_to_update = {} |
| expected_amendments = {} |
| expected_imp_amendments = {} |
| expected_old_owners = {} |
| expected_old_statuses = {} |
| expected_old_components = {} |
| expected_merged_from_add = {} |
| expected_new_starrers = {} |
| |
| issue_main = _Issue('proj', 100) |
| issue_main_ref = ('proj', issue_main.local_id) |
| issue_main.owner_id = 999 |
| issue_main.cc_ids = [111, 222] |
| issue_main.labels = ['dont_touch', 'remove_me'] |
| |
| expected_main = copy.deepcopy(issue_main) |
| expected_main.owner_id = 888 |
| expected_main.cc_ids = [111, 333] |
| expected_main.labels = ['dont_touch', 'add_me'] |
| expected_amendments[issue_main.issue_id] = [ |
| tracker_bizobj.MakeOwnerAmendment(888, 999), |
| tracker_bizobj.MakeCcAmendment([333], [222]), |
| tracker_bizobj.MakeLabelsAmendment(['add_me'], ['remove_me']) |
| ] |
| expected_old_owners[issue_main.issue_id] = 999 |
| |
| # blocked_on issues changes setup. |
| bo_add = _Issue('proj', 1) |
| self.services.issue.TestAddIssue(bo_add) |
| expected_bo_add = copy.deepcopy(bo_add) |
| # All impacted issues should be fetched within ApplyAllIssueChanges |
| # directly from the DB, skipping cache with `use_cache=False` in GetIssue(). |
| # So we expect these issues to have assume_stale=False. |
| expected_bo_add.assume_stale = False |
| expected_bo_add.blocking_iids = [issue_main.issue_id] |
| expected_issues_to_update[expected_bo_add.issue_id] = expected_bo_add |
| expected_imp_amendments[bo_add.issue_id] = [ |
| tracker_bizobj.MakeBlockingAmendment( |
| [issue_main_ref], [], default_project_name='proj') |
| ] |
| |
| bo_remove = _Issue('proj', 2) |
| bo_remove.blocking_iids = [issue_main.issue_id] |
| self.services.issue.TestAddIssue(bo_remove) |
| expected_bo_remove = copy.deepcopy(bo_remove) |
| expected_bo_remove.assume_stale = False |
| expected_bo_remove.blocking_iids = [] |
| expected_issues_to_update[expected_bo_remove.issue_id] = expected_bo_remove |
| expected_imp_amendments[bo_remove.issue_id] = [ |
| tracker_bizobj.MakeBlockingAmendment( |
| [], [issue_main_ref], default_project_name='proj') |
| ] |
| |
| issue_main.blocked_on_iids = [no_change_iid, bo_remove.issue_id] |
| # By default new blocked_on issues that appear in blocked_on_iids |
| # with no prior rank associated with it are un-ranked and assigned rank 0. |
| # See SortBlockedOn in issue_svc.py. |
| issue_main.blocked_on_ranks = [0, 0] |
| expected_main.blocked_on_iids = [no_change_iid, bo_add.issue_id] |
| expected_main.blocked_on_ranks = [0, 0] |
| expected_amendments[issue_main.issue_id].append( |
| tracker_bizobj.MakeBlockedOnAmendment( |
| [('proj', bo_add.local_id)], [('proj', bo_remove.local_id)], |
| default_project_name='proj')) |
| |
| # blocking_issues changes setup. |
| b_add = _Issue('proj', 3) |
| self.services.issue.TestAddIssue(b_add) |
| expected_b_add = copy.deepcopy(b_add) |
| expected_b_add.assume_stale = False |
| expected_b_add.blocked_on_iids = [issue_main.issue_id] |
| expected_b_add.blocked_on_ranks = [0] |
| expected_issues_to_update[expected_b_add.issue_id] = expected_b_add |
| expected_imp_amendments[b_add.issue_id] = [ |
| tracker_bizobj.MakeBlockedOnAmendment( |
| [issue_main_ref], [], default_project_name='proj') |
| ] |
| |
| b_remove = _Issue('proj', 4) |
| b_remove.blocked_on_iids = [issue_main.issue_id] |
| self.services.issue.TestAddIssue(b_remove) |
| expected_b_remove = copy.deepcopy(b_remove) |
| expected_b_remove.assume_stale = False |
| expected_b_remove.blocked_on_iids = [] |
| # Test we can process delta changes and impact changes. |
| delta_b_remove = tracker_pb2.IssueDelta(labels_add=['more_chickens']) |
| expected_b_remove.labels = ['more_chickens'] |
| issue_delta_pairs.append((b_remove, delta_b_remove)) |
| expected_issues_to_update[expected_b_remove.issue_id] = expected_b_remove |
| expected_imp_amendments[b_remove.issue_id] = [ |
| tracker_bizobj.MakeBlockedOnAmendment( |
| [], [issue_main_ref], default_project_name='proj') |
| ] |
| expected_amendments[b_remove.issue_id] = [ |
| tracker_bizobj.MakeLabelsAmendment(['more_chickens'], []) |
| ] |
| |
| issue_main.blocking_iids = [no_change_iid, b_remove.issue_id] |
| expected_main.blocking_iids = [no_change_iid, b_add.issue_id] |
| expected_amendments[issue_main.issue_id].append( |
| tracker_bizobj.MakeBlockingAmendment( |
| [('proj', b_add.local_id)], [('proj', b_remove.local_id)], |
| default_project_name='proj')) |
| |
| # Merged issues changes setup. |
| merge_remove = _Issue('proj', 5) |
| self.services.issue.TestAddIssue(merge_remove) |
| expected_merge_remove = copy.deepcopy(merge_remove) |
| expected_merge_remove.assume_stale = False |
| expected_issues_to_update[ |
| expected_merge_remove.issue_id] = expected_merge_remove |
| expected_imp_amendments[merge_remove.issue_id] = [ |
| tracker_bizobj.MakeMergedIntoAmendment( |
| [], [issue_main_ref], default_project_name='proj') |
| ] |
| |
| merge_add = _Issue('proj', 6) |
| self.services.issue.TestAddIssue(merge_add) |
| expected_merge_add = copy.deepcopy(merge_add) |
| expected_merge_add.assume_stale = False |
| # We are adding 333 and removing 222 in issue_main with delta_main. |
| expected_merge_add.cc_ids = sorted([expected_main.owner_id, 111, 333]) |
| expected_merged_from_add[expected_merge_add.issue_id] = [ |
| issue_main.issue_id |
| ] |
| |
| expected_imp_amendments[merge_add.issue_id] = [ |
| tracker_bizobj.MakeCcAmendment(expected_merge_add.cc_ids, []), |
| tracker_bizobj.MakeMergedIntoAmendment( |
| [issue_main_ref], [], default_project_name='proj') |
| ] |
| # We are merging issue_main into merge_add, so issue_main's starrers |
| # should be merged into merge_add's starrers. |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, issue_main.issue_id, 111, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, issue_main.issue_id, 222, True) |
| expected_merge_add.star_count = 2 |
| expected_new_starrers[merge_add.issue_id] = [222, 111] |
| |
| expected_issues_to_update[expected_merge_add.issue_id] = expected_merge_add |
| |
| |
| issue_main.merged_into = merge_remove.issue_id |
| expected_main.merged_into = merge_add.issue_id |
| expected_amendments[issue_main.issue_id].append( |
| tracker_bizobj.MakeMergedIntoAmendment( |
| [('proj', merge_add.local_id)], [('proj', merge_remove.local_id)], |
| default_project_name='proj')) |
| |
| self.services.issue.TestAddIssue(issue_main) |
| expected_issues_to_update[expected_main.issue_id] = expected_main |
| |
| |
| # Issues we'll put in delta_main.*_remove fields that aren't in issue_main. |
| # These issues should not show up in issues_to_update. |
| missing_1 = _Issue('proj', 404) |
| expected_missing_1 = copy.deepcopy(missing_1) |
| expected_missing_1.assume_stale = False |
| self.services.issue.TestAddIssue(missing_1) |
| missing_2 = _Issue('proj', 405) |
| self.services.issue.TestAddIssue(missing_2) |
| expected_missing_2 = copy.deepcopy(missing_2) |
| expected_missing_2.assume_stale = False |
| |
| delta_main = tracker_pb2.IssueDelta( |
| owner_id=888, |
| cc_ids_remove=[222, 404], cc_ids_add=[333], |
| labels_remove=['remove_me', 'remove_404'], labels_add=['add_me'], |
| merged_into=merge_add.issue_id, |
| blocked_on_add=[bo_add.issue_id], |
| blocked_on_remove=[bo_remove.issue_id, missing_1.issue_id], |
| blocking_add=[b_add.issue_id], |
| blocking_remove=[b_remove.issue_id, missing_2.issue_id]) |
| issue_delta_pairs.append((issue_main, delta_main)) |
| |
| actual_tuple = tracker_helpers.ApplyAllIssueChanges( |
| self.cnxn, issue_delta_pairs, self.services) |
| |
| expected_tuple = tracker_helpers._IssueChangesTuple( |
| expected_issues_to_update, expected_merged_from_add, |
| expected_amendments, expected_imp_amendments, expected_old_owners, |
| expected_old_statuses, expected_old_components, expected_new_starrers) |
| self.assertEqual(actual_tuple, expected_tuple) |
| |
| self.assertEqual(missing_1, expected_missing_1) |
| self.assertEqual(missing_2, expected_missing_2) |
| |
| def testApplyAllIssueChanges_NOOP(self): |
| """Check we can ignore issue-delta pairs that are NOOP.""" |
| noop_issue = _Issue('proj', 1) |
| bo_add_noop = _Issue('proj', 2) |
| bo_remove_noop = _Issue('proj', 3) |
| |
| noop_issue.owner_id = 111 |
| noop_issue.cc_ids = [222] |
| noop_issue.blocked_on_iids = [bo_add_noop.issue_id] |
| bo_add_noop.blocking_iids = [noop_issue.issue_id] |
| |
| self.services.issue.TestAddIssue(noop_issue) |
| self.services.issue.TestAddIssue(bo_add_noop) |
| self.services.issue.TestAddIssue(bo_remove_noop) |
| expected_noop_issue = copy.deepcopy(noop_issue) |
| noop_delta = tracker_pb2.IssueDelta( |
| owner_id=noop_issue.owner_id, |
| cc_ids_add=noop_issue.cc_ids, cc_ids_remove=[333], |
| blocked_on_add=noop_issue.blocked_on_iids, |
| blocked_on_remove=[bo_remove_noop.issue_id]) |
| issue_delta_pairs = [(noop_issue, noop_delta)] |
| |
| actual_tuple = tracker_helpers.ApplyAllIssueChanges( |
| self.cnxn, issue_delta_pairs, self.services) |
| expected_tuple = tracker_helpers._IssueChangesTuple( |
| {}, {}, {}, {}, {}, {}, {}, {}) |
| self.assertEqual(actual_tuple, expected_tuple) |
| |
| self.assertEqual(noop_issue, expected_noop_issue) |
| |
| def testApplyAllIssueChanges_Empty(self): |
| issue_delta_pairs = [] |
| actual_tuple = tracker_helpers.ApplyAllIssueChanges( |
| self.cnxn, issue_delta_pairs, self.services) |
| expected_tuple = tracker_helpers._IssueChangesTuple( |
| {}, {}, {}, {}, {}, {}, {}, {}) |
| self.assertEqual(actual_tuple, expected_tuple) |
| |
| def testUpdateClosedTimestamp(self): |
| config = tracker_pb2.ProjectIssueConfig() |
| config.well_known_statuses.append( |
| tracker_pb2.StatusDef(status='New', means_open=True)) |
| config.well_known_statuses.append( |
| tracker_pb2.StatusDef(status='Accepted', means_open=True)) |
| config.well_known_statuses.append( |
| tracker_pb2.StatusDef(status='Old', means_open=False)) |
| config.well_known_statuses.append( |
| tracker_pb2.StatusDef(status='Closed', means_open=False)) |
| |
| issue = tracker_pb2.Issue() |
| issue.local_id = 1234 |
| issue.status = 'New' |
| |
| # ensure the default value is undef |
| self.assertTrue(not issue.closed_timestamp) |
| |
| # ensure transitioning to the same and other open states |
| # doesn't set the timestamp |
| issue.status = 'New' |
| tracker_helpers.UpdateClosedTimestamp(config, issue, 'New') |
| self.assertTrue(not issue.closed_timestamp) |
| |
| issue.status = 'Accepted' |
| tracker_helpers.UpdateClosedTimestamp(config, issue, 'New') |
| self.assertTrue(not issue.closed_timestamp) |
| |
| # ensure transitioning from open to closed sets the timestamp |
| issue.status = 'Closed' |
| tracker_helpers.UpdateClosedTimestamp(config, issue, 'Accepted') |
| self.assertTrue(issue.closed_timestamp) |
| |
| # ensure that the timestamp is cleared when transitioning from |
| # closed to open |
| issue.status = 'New' |
| tracker_helpers.UpdateClosedTimestamp(config, issue, 'Closed') |
| self.assertTrue(not issue.closed_timestamp) |
| |
| def testGroupUniqueDeltaIssues(self): |
| """We can identify unique IssueDeltas and group Issues by their deltas.""" |
| issue_1 = _Issue('proj', 1) |
| delta_1 = tracker_pb2.IssueDelta(cc_ids_add=[111]) |
| |
| issue_2 = _Issue('proj', 2) |
| delta_2 = tracker_pb2.IssueDelta(cc_ids_add=[111], cc_ids_remove=[222]) |
| |
| issue_3 = _Issue('proj', 3) |
| delta_3 = tracker_pb2.IssueDelta(cc_ids_add=[111]) |
| |
| issue_4 = _Issue('proj', 4) |
| delta_4 = tracker_pb2.IssueDelta() |
| |
| issue_5 = _Issue('proj', 5) |
| delta_5 = tracker_pb2.IssueDelta() |
| |
| issue_delta_pairs = [ |
| (issue_1, delta_1), (issue_2, delta_2), (issue_3, delta_3), |
| (issue_4, delta_4), (issue_5, delta_5) |
| ] |
| unique_deltas, issues_for_deltas = tracker_helpers.GroupUniqueDeltaIssues( |
| issue_delta_pairs) |
| |
| expected_unique_deltas = [delta_1, delta_2, delta_4] |
| self.assertEqual(unique_deltas, expected_unique_deltas) |
| expected_issues_for_deltas = [ |
| [issue_1, issue_3], [issue_2], [issue_4, issue_5] |
| ] |
| self.assertEqual(issues_for_deltas, expected_issues_for_deltas) |
| |
| def testEnforceAttachmentQuotaLimits(self): |
| self.services.project.TestAddProject('Circe', project_id=798) |
| issue_a1 = _Issue('Circe', 1, project_id=798) |
| delta_a1 = tracker_pb2.IssueDelta() |
| |
| issue_a2 = _Issue('Circe', 2, project_id=798) |
| delta_a2 = tracker_pb2.IssueDelta() |
| |
| self.services.project.TestAddProject('Patroclus', project_id=788) |
| issue_b1 = _Issue('Patroclus', 1, project_id=788) |
| delta_b1 = tracker_pb2.IssueDelta() |
| |
| issue_delta_pairs = [ |
| (issue_a1, delta_a1), (issue_a2, delta_a2), (issue_b1, delta_b1) |
| ] |
| |
| upload_1 = framework_helpers.AttachmentUpload( |
| 'dragon', b'OOOOOO\n', 'text/plain') |
| upload_2 = framework_helpers.AttachmentUpload( |
| 'snake', b'ooooo\n', 'text/plain') |
| attachment_uploads = [upload_1, upload_2] |
| |
| actual = tracker_helpers._EnforceAttachmentQuotaLimits( |
| self.cnxn, issue_delta_pairs, self.services, attachment_uploads) |
| |
| expected = { |
| 798: len(upload_1.contents + upload_2.contents) * 2, |
| 788: len(upload_1.contents + upload_2.contents) |
| } |
| self.assertEqual(actual, expected) |
| |
| @mock.patch('tracker.tracker_constants.ISSUE_ATTACHMENTS_QUOTA_HARD', 1) |
| def testEnforceAttachmentQuotaLimits_Exceeded(self): |
| self.services.project.TestAddProject('Circe', project_id=798) |
| issue_a1 = _Issue('Circe', 1, project_id=798) |
| delta_a1 = tracker_pb2.IssueDelta() |
| |
| issue_a2 = _Issue('Circe', 2, project_id=798) |
| delta_a2 = tracker_pb2.IssueDelta() |
| |
| self.services.project.TestAddProject('Patroclus', project_id=788) |
| issue_b1 = _Issue('Patroclus', 1, project_id=788) |
| delta_b1 = tracker_pb2.IssueDelta() |
| |
| issue_delta_pairs = [ |
| (issue_a1, delta_a1), (issue_a2, delta_a2), (issue_b1, delta_b1) |
| ] |
| |
| upload_1 = framework_helpers.AttachmentUpload( |
| 'dragon', b'OOOOOO\n', 'text/plain') |
| upload_2 = framework_helpers.AttachmentUpload( |
| 'snake', b'ooooo\n', 'text/plain') |
| attachment_uploads = [upload_1, upload_2] |
| |
| with self.assertRaisesRegex(exceptions.OverAttachmentQuota, |
| r'.+ project Patroclus\n.+ project Circe'): |
| tracker_helpers._EnforceAttachmentQuotaLimits( |
| self.cnxn, issue_delta_pairs, self.services, attachment_uploads) |
| |
| def testAssertIssueChangesValid_Valid(self): |
| """We can assert when deltas are valid for issues.""" |
| impacted_issue = _Issue('chicken', 101) |
| self.services.issue.TestAddIssue(impacted_issue) |
| |
| issue_1 = _Issue('chicken', 1) |
| self.services.issue.TestAddIssue(issue_1) |
| delta_1 = tracker_pb2.IssueDelta( |
| merged_into=impacted_issue.issue_id, status='Duplicate') |
| exp_d1 = copy.deepcopy(delta_1) |
| |
| issue_2 = _Issue('chicken', 2) |
| self.services.issue.TestAddIssue(issue_2) |
| delta_2 = tracker_pb2.IssueDelta(blocked_on_add=[impacted_issue.issue_id]) |
| exp_d2 = copy.deepcopy(delta_2) |
| |
| issue_3 = _Issue('chicken', 3) |
| self.services.issue.TestAddIssue(issue_3) |
| delta_3 = tracker_pb2.IssueDelta() |
| exp_d3 = copy.deepcopy(delta_3) |
| |
| issue_4 = _Issue('chicken', 4) |
| self.services.issue.TestAddIssue(issue_4) |
| delta_4 = tracker_pb2.IssueDelta(owner_id=self.project_member.user_id) |
| exp_d4 = copy.deepcopy(delta_4) |
| |
| issue_5 = _Issue('chicken', 5) |
| self.services.issue.TestAddIssue(issue_5) |
| fv = tracker_bizobj.MakeFieldValue( |
| self.int_fd.field_id, 998, None, None, None, None, False) |
| delta_5 = tracker_pb2.IssueDelta(field_vals_add=[fv]) |
| exp_d5 = copy.deepcopy(delta_5) |
| |
| issue_6 = _Issue('chicken', 6) |
| self.services.issue.TestAddIssue(issue_6) |
| delta_6 = tracker_pb2.IssueDelta( |
| summary=' ' + 's' * tracker_constants.MAX_SUMMARY_CHARS + ' ') |
| exp_d6 = copy.deepcopy(delta_6) |
| |
| issue_7 = _Issue('chicken', 7) |
| self.services.issue.TestAddIssue(issue_7) |
| issue_8 = _Issue('chicken', 8) |
| self.services.issue.TestAddIssue(issue_8) |
| |
| # We are fine with duplicate/consistent deltas. |
| delta_7 = tracker_pb2.IssueDelta(blocked_on_add=[issue_8.issue_id]) |
| exp_d7 = copy.deepcopy(delta_7) |
| delta_8 = tracker_pb2.IssueDelta(blocking_add=[issue_7.issue_id]) |
| exp_d8 = copy.deepcopy(delta_8) |
| |
| issue_9 = _Issue('chicken', 9) |
| self.services.issue.TestAddIssue(issue_9) |
| issue_10 = _Issue('chicken', 10) |
| self.services.issue.TestAddIssue(issue_10) |
| |
| delta_9 = tracker_pb2.IssueDelta(blocked_on_remove=[issue_10.issue_id]) |
| exp_d9 = copy.deepcopy(delta_9) |
| delta_10 = tracker_pb2.IssueDelta(blocking_remove=[issue_9.issue_id]) |
| exp_d10 = copy.deepcopy(delta_10) |
| |
| issue_11 = _Issue('chicken', 11) |
| user_fd = tracker_bizobj.MakeFieldDef( |
| 123, 789, 'CPU', tracker_pb2.FieldTypes.USER_TYPE, None, '', False, |
| False, False, None, None, '', False, '', '', |
| tracker_pb2.NotifyTriggers.NEVER, 'no_action', 'doc', False) |
| self.services.config.TestAddFieldDef(user_fd) |
| a_user = self.services.user.TestAddUser('a_user@test.com', 123) |
| delta_11 = tracker_pb2.IssueDelta( |
| cc_ids_add=[222], |
| field_vals_add=[ |
| tracker_bizobj.MakeFieldValue( |
| user_fd.field_id, None, None, a_user.user_id, None, None, False) |
| ]) |
| exp_d11 = copy.deepcopy(delta_11) |
| |
| issue_delta_pairs = [ |
| (issue_1, delta_1), (issue_2, delta_2), (issue_3, delta_3), |
| (issue_4, delta_4), (issue_5, delta_5), (issue_6, delta_6), |
| (issue_7, delta_7), (issue_8, delta_8), (issue_9, delta_9), |
| (issue_10, delta_10), (issue_11, delta_11) |
| ] |
| comment = ' ' + 'c' * tracker_constants.MAX_COMMENT_CHARS + ' ' |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services, comment_content=comment) |
| |
| # Check we can handle None `comment_content`. |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services) |
| self.assertEqual( |
| [ |
| exp_d1, exp_d2, exp_d3, exp_d4, exp_d5, exp_d6, exp_d7, exp_d8, |
| exp_d9, exp_d10, exp_d11 |
| ], [ |
| delta_1, delta_2, delta_3, delta_4, delta_5, delta_6, delta_7, |
| delta_8, delta_9, delta_10, delta_11 |
| ]) |
| |
| def testAssertIssueChangesValid_ValidatesLabels(self): |
| """Asserts labels.""" |
| issue_1 = _Issue('chicken', 1) |
| self.services.issue.TestAddIssue(issue_1) |
| delta_1 = tracker_pb2.IssueDelta(labels_add=['freeze_new_label']) |
| issue_delta_pairs = [(issue_1, delta_1)] |
| comment = 'just a plain comment' |
| with self.assertRaisesRegex( |
| exceptions.InputException, |
| ("The creation of new labels is blocked for the Chromium project" |
| " in Monorail. To continue with editing your issue, please" |
| " remove: freeze_new_label label\\(s\\).")): |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services, comment_content=comment) |
| |
| def testAssertIssueChangesValid_RequiredField(self): |
| """Asserts fields and requried fields..""" |
| issue_1 = _Issue('chicken', 1) |
| self.services.issue.TestAddIssue(issue_1) |
| delta_1 = tracker_pb2.IssueDelta() |
| exp_d1 = copy.deepcopy(delta_1) |
| |
| required_fd = tracker_bizobj.MakeFieldDef( |
| 124, 789, 'StrField', tracker_pb2.FieldTypes.STR_TYPE, None, '', True, |
| False, False, None, None, '', False, '', '', |
| tracker_pb2.NotifyTriggers.NEVER, 'no_action', 'doc', False) |
| self.services.config.TestAddFieldDef(required_fd) |
| |
| issue_delta_pairs = [(issue_1, delta_1)] |
| comment = 'just a plain comment' |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services, comment_content=comment) |
| |
| # Check we can handle adding a field value when issue is in invalid state. |
| fv = tracker_bizobj.MakeFieldValue( |
| self.int_fd.field_id, 998, None, None, None, None, False) |
| delta_2 = tracker_pb2.IssueDelta(field_vals_add=[fv]) |
| exp_d2 = copy.deepcopy(delta_2) |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services) |
| self.assertEqual([exp_d1, exp_d2], [delta_1, delta_2]) |
| |
| def testAssertIssueChangesValid_Invalid(self): |
| """We can raise exceptions when deltas are not valid for issues. """ |
| |
| def getRef(issue): |
| return '%s:%d' % (issue.project_name, issue.local_id) |
| |
| issue_delta_pairs = [] |
| expected_err_msgs = [] |
| |
| comment = 'c' * (tracker_constants.MAX_COMMENT_CHARS + 1) |
| expected_err_msgs.append('Comment is too long.') |
| |
| issue_1 = _Issue('chicken', 1) |
| self.services.issue.TestAddIssue(issue_1) |
| issue_1_ref = getRef(issue_1) |
| |
| delta_1 = tracker_pb2.IssueDelta( |
| merged_into=issue_1.issue_id, |
| blocked_on_add=[issue_1.issue_id], |
| summary='', |
| status='', |
| cc_ids_add=[9876]) |
| |
| issue_delta_pairs.append((issue_1, delta_1)) |
| expected_err_msgs.extend( |
| [ |
| ('%s: MERGED type statuses must accompany mergedInto values.') % |
| issue_1_ref, |
| '%s: Cannot merge an issue into itself.' % issue_1_ref, |
| '%s: Cannot block an issue on itself.' % issue_1_ref, |
| 'users/9876: User does not exist.', |
| '%s: Summary required.' % issue_1_ref, |
| '%s: Status is required.' % issue_1_ref |
| ]) |
| |
| issue_2 = _Issue('chicken', 2) |
| self.services.issue.TestAddIssue(issue_2) |
| issue_2_ref = getRef(issue_2) |
| |
| fv = tracker_bizobj.MakeFieldValue( |
| self.int_fd.field_id, 1000, None, None, None, None, False) |
| delta_2 = tracker_pb2.IssueDelta( |
| status='Duplicate', |
| blocking_add=[issue_2.issue_id], |
| summary='s' * (tracker_constants.MAX_SUMMARY_CHARS + 1), |
| owner_id=self.no_project_user.user_id, |
| field_vals_add=[fv]) |
| issue_delta_pairs.append((issue_2, delta_2)) |
| |
| expected_err_msgs.extend( |
| [ |
| ('%s: MERGED type statuses must accompany mergedInto values.') % |
| issue_2_ref, |
| '%s: Cannot block an issue on itself.' % issue_2_ref, |
| '%s: Issue owner must be a project member.' % issue_2_ref, |
| '%s: Summary is too long.' % issue_2_ref, |
| '%s: Error for %r: Value must be <= 999.' % (issue_2_ref, fv) |
| ]) |
| |
| issue_3 = _Issue('chicken', 3) |
| issue_3.status = 'Duplicate' |
| issue_3.merged_into = 78911 |
| self.services.issue.TestAddIssue(issue_3) |
| issue_3_ref = getRef(issue_3) |
| delta_3 = tracker_pb2.IssueDelta( |
| status='Available', merged_into_external='b/123') |
| issue_delta_pairs.append((issue_3, delta_3)) |
| expected_err_msgs.append( |
| '%s: MERGED type statuses must accompany mergedInto values.' % |
| issue_3_ref) |
| |
| with self.assertRaisesRegex(exceptions.InputException, |
| '\n'.join(expected_err_msgs)): |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services, comment_content=comment) |
| |
| def testAssertIssueChangesValid_ConflictingDeltas(self): |
| |
| def getRef(issue): |
| return '%s:%d' % (issue.project_name, issue.local_id) |
| |
| expected_err_msgs = [] |
| issue_3 = _Issue('chicken', 3) |
| self.services.issue.TestAddIssue(issue_3) |
| issue_3_ref = getRef(issue_3) |
| issue_4 = _Issue('chicken', 4) |
| self.services.issue.TestAddIssue(issue_4) |
| issue_4_ref = getRef(issue_4) |
| issue_5 = _Issue('chicken', 5) |
| self.services.issue.TestAddIssue(issue_5) |
| issue_5_ref = getRef(issue_5) |
| issue_6 = _Issue('chicken', 6) |
| self.services.issue.TestAddIssue(issue_6) |
| issue_6_ref = getRef(issue_6) |
| issue_7 = _Issue('chicken', 7) |
| self.services.issue.TestAddIssue(issue_7) |
| issue_7_ref = getRef(issue_7) |
| |
| delta_3 = tracker_pb2.IssueDelta( |
| blocking_add=[issue_4.issue_id], |
| blocked_on_add=[issue_5.issue_id, issue_6.issue_id]) |
| |
| delta_4 = tracker_pb2.IssueDelta( |
| blocked_on_remove=[issue_3.issue_id], blocking_add=[issue_5.issue_id]) |
| expected_err_msgs.append( |
| 'Changes for %s conflict for %s' % (issue_4_ref, issue_3_ref)) |
| |
| delta_5 = tracker_pb2.IssueDelta( |
| blocking_remove=[issue_3.issue_id], |
| blocked_on_remove=[issue_4.issue_id]) |
| expected_err_msgs.append( |
| 'Changes for %s conflict for %s, %s' % |
| (issue_5_ref, issue_3_ref, issue_4_ref)) |
| |
| delta_6 = tracker_pb2.IssueDelta(blocking_remove=[issue_3.issue_id]) |
| expected_err_msgs.append( |
| 'Changes for %s conflict for %s' % (issue_6_ref, issue_3_ref)) |
| |
| impacted_issue = _Issue('chicken', 11) |
| self.services.issue.TestAddIssue(impacted_issue) |
| impacted_issue_ref = getRef(impacted_issue) |
| delta_7 = tracker_pb2.IssueDelta( |
| blocking_remove=[issue_3.issue_id], |
| blocking_add=[issue_3.issue_id], |
| blocked_on_remove=[impacted_issue.issue_id], |
| blocked_on_add=[impacted_issue.issue_id]) |
| expected_err_msgs.append( |
| 'Changes for %s conflict for %s, %s' % |
| (issue_7_ref, issue_3_ref, impacted_issue_ref)) |
| |
| issue_delta_pairs = [ |
| (issue_3, delta_3), |
| (issue_4, delta_4), |
| (issue_5, delta_5), |
| (issue_6, delta_6), |
| (issue_7, delta_7), |
| ] |
| |
| with self.assertRaisesRegex(exceptions.InputException, |
| '\n'.join(expected_err_msgs)): |
| tracker_helpers._AssertIssueChangesValid( |
| self.cnxn, issue_delta_pairs, self.services) |
| |
| def testComputeNewCcsFromIssueMerge(self): |
| """We can compute the new ccs to add to a merge-into issue.""" |
| target_issue = fake.MakeTestIssue(789, 10, 'Target issue', 'New', 111) |
| source_issue_1 = fake.MakeTestIssue( |
| 789, 11, 'Source issue', 'New', 111) # different restrictions |
| source_issue_2 = fake.MakeTestIssue( |
| 789, 12, 'Source issue', 'New', 222) # same restrictions |
| source_issue_3 = fake.MakeTestIssue( |
| 789, 13, 'Source issue', 'New', 222) # no restrictions |
| source_issue_4 = fake.MakeTestIssue( |
| 789, 14, 'Source issue', 'New', 666) # empty ccs |
| source_issue_5 = fake.MakeTestIssue( |
| 788, 15, 'Source issue', 'New', 666) # different project |
| source_issue_1.cc_ids.append(333) |
| source_issue_2.cc_ids.append(444) |
| source_issue_3.cc_ids.append(555) |
| source_issue_5.cc_ids.append(999) |
| |
| target_issue.labels.append('Restrict-View-Chicken') |
| source_issue_1.labels.append('Restrict-View-Cow') |
| source_issue_2.labels.append('Restrict-View-Chicken') |
| |
| self.services.issue.TestAddIssue(target_issue) |
| self.services.issue.TestAddIssue(source_issue_1) |
| self.services.issue.TestAddIssue(source_issue_2) |
| self.services.issue.TestAddIssue(source_issue_3) |
| self.services.issue.TestAddIssue(source_issue_4) |
| self.services.issue.TestAddIssue(source_issue_5) |
| |
| new_cc_ids = tracker_helpers._ComputeNewCcsFromIssueMerge( |
| target_issue, [source_issue_1, source_issue_2, source_issue_3]) |
| six.assertCountEqual(self, new_cc_ids, [444, 555, 222]) |
| |
| def testComputeNewCcsFromIssueMerge_Empty(self): |
| target_issue = fake.MakeTestIssue(789, 10, 'Target issue', 'New', 111) |
| self.services.issue.TestAddIssue(target_issue) |
| new_cc_ids = tracker_helpers._ComputeNewCcsFromIssueMerge(target_issue, []) |
| six.assertCountEqual(self, new_cc_ids, []) |
| |
| def testEnforceNonMergeStatusDeltas(self): |
| # No updates: user is setting to a non-MERGED status with no |
| # existing merged_into values. |
| issue_1 = _Issue('chicken', 1) |
| self.services.issue.TestAddIssue(issue_1) |
| delta_1 = tracker_pb2.IssueDelta(status='Available') |
| exp_delta_1 = copy.deepcopy(delta_1) |
| |
| # No updates: user is setting to a MERGED status. Whether this request |
| # goes through will be handled by _AssertIssueChangesValid(). |
| issue_2 = _Issue('chicken', 2) |
| self.services.issue.TestAddIssue(issue_2) |
| delta_2 = tracker_pb2.IssueDelta(status='Duplicate') |
| exp_delta_2 = copy.deepcopy(delta_2) |
| |
| # No updates: user is setting to a MERGED status. (This test issue starts |
| # out with a merged_into value but a non-MERGED status. We don't expect |
| # real data to ever be in this state) |
| issue_3 = _Issue('chicken', 3) |
| issue_3.merged_into = 7011 |
| self.services.issue.TestAddIssue(issue_3) |
| delta_3 = tracker_pb2.IssueDelta(status='Duplicate') |
| exp_delta_3 = copy.deepcopy(delta_3) |
| |
| # No updates: same situation as above. |
| issue_4 = _Issue('chicken', 4) |
| issue_4.merged_into_external = 'b/123' |
| self.services.issue.TestAddIssue(issue_4) |
| delta_4 = tracker_pb2.IssueDelta(status='Duplicate') |
| exp_delta_4 = copy.deepcopy(delta_4) |
| |
| # Update delta: user is setting status AWAY from a MERGED status, so we |
| # auto-remove any existing merged_into values. |
| issue_5 = _Issue('chicken', 5) |
| issue_5.merged_into = 7011 |
| self.services.issue.TestAddIssue(issue_5) |
| delta_5 = tracker_pb2.IssueDelta(status='Available') |
| exp_delta_5 = copy.deepcopy(delta_5) |
| exp_delta_5.merged_into = 0 |
| |
| # Update delta: user is setting status AWAY from a MERGED status, so we |
| # auto-remove any existing merged_into values. |
| issue_6 = _Issue('chicken', 6) |
| issue_6.merged_into_external = 'b/123' |
| self.services.issue.TestAddIssue(issue_6) |
| delta_6 = tracker_pb2.IssueDelta(status='Available') |
| exp_delta_6 = copy.deepcopy(delta_6) |
| exp_delta_6.merged_into_external = '' |
| |
| # No updates: user is setting to a non-MERGED status while also setting |
| # a merged_into value. This will be rejected down the line by |
| # _AssertIssueChangesValid() |
| issue_7 = _Issue('chicken', 7) |
| issue_7.merged_into = 7011 |
| self.services.issue.TestAddIssue(issue_7) |
| delta_7 = tracker_pb2.IssueDelta( |
| merged_into_external='b/123', status='Available') |
| exp_delta_7 = copy.deepcopy(delta_7) |
| |
| # No updates: user is setting to a non-MERGED status while also setting |
| # a merged_into value. This will be rejected down the line by |
| # _AssertIssueChangesValid() |
| issue_8 = _Issue('chicken', 8) |
| issue_8.merged_into_external = 'b/123' |
| self.services.issue.TestAddIssue(issue_8) |
| delta_8 = tracker_pb2.IssueDelta(merged_into=8011, status='Available') |
| exp_delta_8 = copy.deepcopy(delta_8) |
| |
| pairs = [ |
| (issue_1, delta_1), (issue_2, delta_2), (issue_3, delta_3), |
| (issue_4, delta_4), (issue_5, delta_5), (issue_6, delta_6), |
| (issue_7, delta_7), (issue_8, delta_8) |
| ] |
| |
| tracker_helpers._EnforceNonMergeStatusDeltas( |
| self.cnxn, pairs, self.services) |
| self.assertEqual( |
| [ |
| delta_1, delta_2, delta_3, delta_4, delta_5, delta_6, delta_7, |
| delta_8 |
| ], [ |
| exp_delta_1, exp_delta_2, exp_delta_3, exp_delta_4, exp_delta_5, |
| exp_delta_6, exp_delta_7, exp_delta_8 |
| ]) |
| |
| |
| class IssueChangeImpactedIssuesTest(unittest.TestCase): |
| """Tests for the _IssueChangeImpactedIssues class.""" |
| |
| def setUp(self): |
| self.services = service_manager.Services( |
| issue=fake.IssueService(), issue_star=fake.IssueStarService()) |
| self.cnxn = 'fake connection' |
| |
| def testComputeAllImpactedIDs(self): |
| tracker = tracker_helpers._IssueChangeImpactedIssues() |
| tracker.blocking_add[78901].append(1) |
| tracker.blocking_remove[78902].append(2) |
| tracker.blocked_on_add[78903].append(1) |
| tracker.blocked_on_remove[78904].append(1) |
| tracker.merged_from_add[78905].append(3) |
| tracker.merged_from_remove[78906].append(3) |
| |
| # Repeat a few iids. |
| tracker.blocked_on_remove[78901].append(1) |
| tracker.merged_from_add[78903].append(1) |
| |
| actual = tracker.ComputeAllImpactedIIDs() |
| expected = {78901, 78902, 78903, 78904, 78905, 78906} |
| self.assertEqual(actual, expected) |
| |
| def testComputeAllImpactedIDs_Empty(self): |
| tracker = tracker_helpers._IssueChangeImpactedIssues() |
| actual = tracker.ComputeAllImpactedIIDs() |
| self.assertEqual(actual, set()) |
| |
| def testTrackImpactedIssues(self): |
| issue_delta_pairs = [] |
| |
| issue_1 = _Issue('project', 1) |
| issue_1.merged_into = 78906 |
| delta_1 = tracker_pb2.IssueDelta( |
| merged_into=78905, |
| blocked_on_add=[78901, 78902], |
| blocked_on_remove=[78903, 78904], |
| ) |
| issue_delta_pairs.append((issue_1, delta_1)) |
| |
| issue_2 = _Issue('project', 2) |
| issue_2.merged_into = 78905 |
| delta_2 = tracker_pb2.IssueDelta( |
| merged_into=78905, # This should be ignored. |
| blocking_add=[78901, 78902], |
| blocking_remove=[78903, 78904], |
| ) |
| issue_delta_pairs.append((issue_2, delta_2)) |
| |
| issue_3 = _Issue('project', 3) |
| issue_3.merged_into = 78902 |
| delta_3 = tracker_pb2.IssueDelta(merged_into=78901) |
| issue_delta_pairs.append((issue_3, delta_3)) |
| |
| issue_4 = _Issue('project', 4) |
| issue_4.merged_into = 78901 |
| delta_4 = tracker_pb2.IssueDelta( |
| merged_into=framework_constants.NO_ISSUE_SPECIFIED) |
| issue_delta_pairs.append((issue_4, delta_4)) |
| |
| impacted_issues = tracker_helpers._IssueChangeImpactedIssues() |
| for issue, delta in issue_delta_pairs: |
| impacted_issues.TrackImpactedIssues(issue, delta) |
| |
| self.assertEqual( |
| impacted_issues.blocking_add, { |
| 78901: [issue_1.issue_id], |
| 78902: [issue_1.issue_id] |
| }) |
| self.assertEqual( |
| impacted_issues.blocking_remove, { |
| 78903: [issue_1.issue_id], |
| 78904: [issue_1.issue_id] |
| }) |
| self.assertEqual( |
| impacted_issues.blocked_on_add, { |
| 78901: [issue_2.issue_id], |
| 78902: [issue_2.issue_id] |
| }) |
| self.assertEqual( |
| impacted_issues.blocked_on_remove, { |
| 78903: [issue_2.issue_id], |
| 78904: [issue_2.issue_id] |
| }) |
| self.assertEqual( |
| impacted_issues.merged_from_add, { |
| 78901: [issue_3.issue_id], |
| 78905: [issue_1.issue_id], |
| }) |
| self.assertEqual( |
| impacted_issues.merged_from_remove, { |
| 78901: [issue_4.issue_id], |
| 78902: [issue_3.issue_id], |
| 78906: [issue_1.issue_id], |
| }) |
| |
| def testApplyImpactedIssueChanges(self): |
| impacted_tracker = tracker_helpers._IssueChangeImpactedIssues() |
| impacted_issue = _Issue('proj', 1) |
| self.services.issue.TestAddIssue(impacted_issue) |
| impacted_iid = impacted_issue.issue_id |
| |
| # Setup. |
| bo_add = _Issue('proj', 2) |
| self.services.issue.TestAddIssue(bo_add) |
| impacted_tracker.blocked_on_add[impacted_iid].append(bo_add.issue_id) |
| |
| bo_remove = _Issue('proj', 3) |
| self.services.issue.TestAddIssue(bo_remove) |
| impacted_tracker.blocked_on_remove[impacted_iid].append( |
| bo_remove.issue_id) |
| |
| b_add = _Issue('proj', 4) |
| self.services.issue.TestAddIssue(b_add) |
| impacted_tracker.blocking_add[impacted_iid].append( |
| b_add.issue_id) |
| |
| b_remove = _Issue('proj', 5) |
| self.services.issue.TestAddIssue(b_remove) |
| impacted_tracker.blocking_remove[impacted_iid].append( |
| b_remove.issue_id) |
| |
| m_add = _Issue('proj', 6) |
| m_add.cc_ids = [666, 777] |
| self.services.issue.TestAddIssue(m_add) |
| m_add_no_ccs = _Issue('proj', 7, '', '') |
| self.services.issue.TestAddIssue(m_add_no_ccs) |
| impacted_tracker.merged_from_add[impacted_iid].extend( |
| [m_add.issue_id, m_add_no_ccs.issue_id]) |
| # Set up starrers. |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, impacted_iid, 111, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, impacted_iid, 222, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, m_add.issue_id, 222, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, m_add.issue_id, 333, True) |
| self.services.issue_star.SetStar( |
| self.cnxn, self.services, None, m_add.issue_id, 444, True) |
| |
| m_remove = _Issue('proj', 8) |
| m_remove.cc_ids = [888] |
| self.services.issue.TestAddIssue(m_remove) |
| impacted_tracker.merged_from_remove[impacted_iid].append( |
| m_remove.issue_id) |
| |
| |
| impacted_issue.cc_ids = [666] |
| impacted_issue.blocked_on_iids = [78404, bo_remove.issue_id] |
| impacted_issue.blocking_iids = [78405, b_remove.issue_id] |
| expected_issue = copy.deepcopy(impacted_issue) |
| |
| # Verify. |
| (actual_amendments, |
| actual_new_starrers) = impacted_tracker.ApplyImpactedIssueChanges( |
| self.cnxn, impacted_issue, self.services) |
| expected_amendments = [ |
| tracker_bizobj.MakeBlockedOnAmendment( |
| [('proj', bo_add.local_id)], |
| [('proj', bo_remove.local_id)], default_project_name='proj'), |
| tracker_bizobj.MakeBlockingAmendment( |
| [('proj', b_add.local_id)], |
| [('proj', b_remove.local_id)], default_project_name='proj'), |
| tracker_bizobj.MakeCcAmendment([777], []), |
| tracker_bizobj.MakeMergedIntoAmendment( |
| [('proj', m_add.local_id), ('proj', m_add_no_ccs.local_id)], |
| [('proj', m_remove.local_id)], default_project_name='proj') |
| ] |
| self.assertEqual(actual_amendments, expected_amendments) |
| six.assertCountEqual(self, actual_new_starrers, [333, 444]) |
| |
| expected_issue.cc_ids.append(777) |
| expected_issue.blocked_on_iids = [78404, bo_add.issue_id] |
| # By default new blocked_on issues that appear in blocked_on_iids |
| # with no prior rank associated with it are un-ranked and assigned rank 0. |
| # See SortBlockedOn in issue_svc.py. |
| expected_issue.blocked_on_ranks = [0, 0] |
| expected_issue.blocking_iids = [78405, b_add.issue_id] |
| expected_issue.star_count = 4 |
| self.assertEqual(impacted_issue, expected_issue) |
| |
| def testApplyImpactedIssueChanges_Empty(self): |
| impacted_tracker = tracker_helpers._IssueChangeImpactedIssues() |
| impacted_issue = _Issue('proj', 1) |
| expected_issue = copy.deepcopy(impacted_issue) |
| |
| (actual_amendments, |
| actual_new_starrers) = impacted_tracker.ApplyImpactedIssueChanges( |
| self.cnxn, impacted_issue, self.services) |
| |
| expected_amendments = [] |
| self.assertEqual(actual_amendments, expected_amendments) |
| expected_new_starrers = [] |
| self.assertEqual(actual_new_starrers, expected_new_starrers) |
| self.assertEqual(impacted_issue, expected_issue) |
| |
| def testApplyImpactedIssueChanges_PartiallyEmptyMergedFrom(self): |
| """We can process merged_from changes when one of the lists is empty.""" |
| impacted_tracker = tracker_helpers._IssueChangeImpactedIssues() |
| impacted_issue = _Issue('proj', 1) |
| impacted_iid = impacted_issue.issue_id |
| expected_issue = copy.deepcopy(impacted_issue) |
| |
| m_add = _Issue('proj', 2) |
| self.services.issue.TestAddIssue(m_add) |
| impacted_tracker.merged_from_add[impacted_iid].append( |
| m_add.issue_id) |
| # We're leaving impacted_tracker.merged_from_remove empty. |
| |
| (actual_amendments, |
| actual_new_starrers) = impacted_tracker.ApplyImpactedIssueChanges( |
| self.cnxn, impacted_issue, self.services) |
| |
| expected_amendments = [tracker_bizobj.MakeMergedIntoAmendment( |
| [('proj', m_add.local_id)], [], default_project_name='proj')] |
| self.assertEqual(actual_amendments, expected_amendments) |
| expected_new_starrers = [] |
| self.assertEqual(actual_new_starrers, expected_new_starrers) |
| self.assertEqual(impacted_issue, expected_issue) |
| |
| |
| class AssertUsersExistTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.cnxn = 'fake cnxn' |
| self.services = service_manager.Services(user=fake.UserService()) |
| for user_id in [1, 1001, 1002, 1003, 2001, 2002, 3002]: |
| self.services.user.TestAddUser('test%d' % user_id, user_id, add_user=True) |
| |
| def test_AssertUsersExist_Passes(self): |
| existing = [1, 1001, 1002, 1003, 2001, 2002, 3002] |
| with exceptions.ErrorAggregator(exceptions.InputException) as err_agg: |
| tracker_helpers.AssertUsersExist( |
| self.cnxn, self.services, existing, err_agg) |
| |
| def test_AssertUsersExist_Empty(self): |
| with exceptions.ErrorAggregator(exceptions.InputException) as err_agg: |
| tracker_helpers.AssertUsersExist( |
| self.cnxn, self.services, [], err_agg) |
| |
| def test_AssertUsersExist(self): |
| dne_users = [2, 3] |
| existing = [1, 1001, 1002, 1003, 2001, 2002, 3002] |
| all_users = existing + dne_users |
| with self.assertRaisesRegex( |
| exceptions.InputException, |
| 'users/2: User does not exist.\nusers/3: User does not exist.'): |
| with exceptions.ErrorAggregator(exceptions.InputException) as err_agg: |
| tracker_helpers.AssertUsersExist( |
| self.cnxn, self.services, all_users, err_agg) |
