kallithea Changeset - 56451a7ca82f

Changeset - 56451a7ca82f

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Thomas De Schampheleire - 5 years ago 2020-10-09 15:34:41
thomas.de_schampheleire@nokia.com

api: new method: edit_reviewers

Allow adding and removing reviewers of a pull request with the API call
'edit_reviewers', taking a pull request ID and a list of usernames or
userids to add and remove. For single-user operation, a string can be used
instead of a list.

Note that the 'kallithea-api' tool only accepts strings, so can only perform
single-user operations. Use python 'requests', 'curl' or similar instead if
you need to operate on multiple users at a time.

2 files changed with 362 insertions and 1 deletions:

kallithea/controllers/api/api.py

kallithea/tests/api/api_base.py

316

0 comments (0 inline, 0 general)

kallithea/controllers/api/api.py

➞

Show inline comments

@@ @@ -2421,3 +2421,49 @@ class ApiController(JSONRPCController): @@
                           pull_request.org_repo, request.ip_addr)
         Session().commit()
         return True
     # permission check inside
     def edit_reviewers(self, pull_request_id, add=None, remove=None):
         """
         Add and/or remove one or more reviewers to a pull request, by username
         or user ID. Reviewers are specified either as a single-user string or
         as a JSON list of one or more strings.
         """
         if add is None and remove is None:
             raise JSONRPCError('''Invalid request. Neither 'add' nor 'remove' is specified.''')
         pull_request = PullRequest.get(pull_request_id)
         if pull_request is None:
             raise JSONRPCError('pull request `%s` does not exist' % (pull_request_id,))
         apiuser = get_user_or_error(request.authuser.user_id)
         is_owner = apiuser.user_id == pull_request.owner_id
         is_repo_admin = HasRepoPermissionLevel('admin')(pull_request.other_repo.repo_name)
         if not (apiuser.admin or is_repo_admin or is_owner):
             raise JSONRPCError('No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.')
         if pull_request.is_closed():
             raise JSONRPCError('Cannot edit reviewers of a closed pull request.')
         if not isinstance(add, list):
             add = [add]
         if not isinstance(remove, list):
             remove = [remove]
         # look up actual user objects from given name or id. Bail out if unknown.
         add_objs = set(get_user_or_error(user) for user in add if user is not None)
         remove_objs = set(get_user_or_error(user) for user in remove if user is not None)
         new_reviewers = redundant_reviewers = set()
         if add_objs:
             new_reviewers, redundant_reviewers = PullRequestModel().add_reviewers(apiuser, pull_request, add_objs)
         if remove_objs:
             PullRequestModel().remove_reviewers(apiuser, pull_request, remove_objs)
         Session().commit()
         return {
             'added': [x.username for x in new_reviewers],
             'already_present': [x.username for x in redundant_reviewers],
             # NOTE: no explicit check that removed reviewers were actually present.
             'removed': [x.username for x in remove_objs],
+        }

kallithea/tests/api/api_base.py

➞

Show inline comments

@@ @@ -27,9 +27,10 @@ from kallithea.lib import ext_json @@
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.utils2 import ascii_bytes
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.db import ChangesetStatus, PullRequest, RepoGroup, Repository, Setting, Ui, User
+from kallithea.model.db import ChangesetStatus, PullRequest, PullRequestReviewer, RepoGroup, Repository, Setting, Ui, User
 from kallithea.model.gist import GistModel
 from kallithea.model.meta import Session
 from kallithea.model.pull_request import PullRequestModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import ScmModel
@@ @@ -2528,3 +2529,317 @@ class _BaseTestApi(object): @@
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == 'Looks good to me'
     def test_api_edit_reviewers_add_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN], 'already_present': [], 'removed': [] }
         self._compare_ok(random_id, expected, given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": 999},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
     def test_api_edit_reviewers_add_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ self.TEST_USER_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN, self.TEST_USER_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(self.TEST_USER_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_already_present(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [base.TEST_USER_REGULAR_LOGIN],
                      'removed': [],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         PullRequestModel().close_pull_request(pull_request_id)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
     def test_api_edit_reviewers_add_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         pullrequest.owner = User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
     def test_api_edit_reviewers_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": 999},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_nonpresent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         # NOTE: no explicit indication that removed user was not even a reviewer
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR2_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         prr = PullRequestReviewer(User.get_by_username(base.TEST_USER_REGULAR2_LOGIN), pullrequest)
         Session().add(prr)
         Session().commit()
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ] },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set()
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN])
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         PullRequestModel().close_pull_request(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_remove_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": base.TEST_USER_REGULAR2_LOGIN,
                      "remove": base.TEST_USER_REGULAR_LOGIN
                     },
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
+                   }
         self._compare_ok(random_id, expected, given=response.body)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_add_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         prr = PullRequestReviewer(User.get_by_username(base.TEST_USER_ADMIN_LOGIN), pullrequest)
         Session().add(prr)
         Session().commit()
         assert User.get_by_username(base.TEST_USER_ADMIN_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": [ base.TEST_USER_REGULAR2_LOGIN ],
                      "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN ],
                     },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN])
         assert User.get_by_username(base.TEST_USER_ADMIN_LOGIN) not in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
     def test_api_edit_reviewers_invalid_params(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = PullRequest().get(pull_request_id)
         assert User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         pullrequest.owner = User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Invalid request. Neither 'add' nor 'remove' is specified.", given=response.body)
         assert ext_json.loads(response.body)['result'] is None

0 comments (0 inline, 0 general)