|
|
# -*- coding: utf-8 -*-
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
|
# it under the terms of the GNU General Public License as published by
|
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
|
# (at your option) any later version.
|
|
|
#
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
|
# GNU General Public License for more details.
|
|
|
#
|
|
|
# You should have received a copy of the GNU General Public License
|
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
"""
|
|
|
kallithea.model.pull_request
|
|
|
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
|
|
pull request model for Kallithea
|
|
|
|
|
|
This file was forked by the Kallithea project in July 2014.
|
|
|
Original author and date, and relevant copyright and licensing information is below:
|
|
|
:created_on: Jun 6, 2012
|
|
|
:author: marcink
|
|
|
:copyright: (c) 2013 RhodeCode GmbH, and others.
|
|
|
:license: GPLv3, see LICENSE.md for more details.
|
|
|
"""
|
|
|
|
|
|
import datetime
|
|
|
import logging
|
|
|
import re
|
|
|
|
|
|
from tg import request
|
|
|
from tg.i18n import ugettext as _
|
|
|
|
|
|
from kallithea.lib import helpers as h
|
|
|
from kallithea.lib.utils2 import ascii_bytes, extract_mentioned_users
|
|
|
from kallithea.model.db import ChangesetStatus, PullRequest, PullRequestReviewer, User
|
|
|
from kallithea.model.meta import Session
|
|
|
from kallithea.model.notification import NotificationModel
|
|
|
|
|
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
def _assert_valid_reviewers(seq):
|
|
|
"""Sanity check: elements are actual User objects, and not the default user."""
|
|
|
assert not any(user.is_default_user for user in seq)
|
|
|
|
|
|
|
|
|
class PullRequestModel(object):
|
|
|
|
|
|
def add_reviewers(self, user, pr, reviewers, mention_recipients=None):
|
|
|
"""Add reviewer and send notification to them.
|
|
|
"""
|
|
|
reviewers = set(reviewers)
|
|
|
_assert_valid_reviewers(reviewers)
|
|
|
if mention_recipients is not None:
|
|
|
mention_recipients = set(mention_recipients) - reviewers
|
|
|
_assert_valid_reviewers(mention_recipients)
|
|
|
|
|
|
# members
|
|
|
for reviewer in reviewers:
|
|
|
prr = PullRequestReviewer(reviewer, pr)
|
|
|
Session().add(prr)
|
|
|
|
|
|
# notification to reviewers
|
|
|
pr_url = pr.url(canonical=True)
|
|
|
threading = ['%s-pr-%s@%s' % (pr.other_repo.repo_name,
|
|
|
pr.pull_request_id,
|
|
|
h.canonical_hostname())]
|
|
|
subject = h.link_to(
|
|
|
_('%(user)s wants you to review pull request %(pr_nice_id)s: %(pr_title)s') %
|
|
|
{'user': user.username,
|
|
|
'pr_title': pr.title,
|
|
|
'pr_nice_id': pr.nice_id()},
|
|
|
pr_url)
|
|
|
body = pr.description
|
|
|
_org_ref_type, org_ref_name, _org_rev = pr.org_ref.split(':')
|
|
|
_other_ref_type, other_ref_name, _other_rev = pr.other_ref.split(':')
|
|
|
revision_data = [(x.raw_id, x.message)
|
|
|
for x in map(pr.org_repo.get_changeset, pr.revisions)]
|
|
|
email_kwargs = {
|
|
|
'pr_title': pr.title,
|
|
|
'pr_title_short': h.shorter(pr.title, 50),
|
|
|
'pr_user_created': user.full_name_and_username,
|
|
|
'pr_repo_url': h.canonical_url('summary_home', repo_name=pr.other_repo.repo_name),
|
|
|
'pr_url': pr_url,
|
|
|
'pr_revisions': revision_data,
|
|
|
'repo_name': pr.other_repo.repo_name,
|
|
|
'org_repo_name': pr.org_repo.repo_name,
|
|
|
'pr_nice_id': pr.nice_id(),
|
|
|
'pr_target_repo': h.canonical_url('summary_home',
|
|
|
repo_name=pr.other_repo.repo_name),
|
|
|
'pr_target_branch': other_ref_name,
|
|
|
'pr_source_repo': h.canonical_url('summary_home',
|
|
|
repo_name=pr.org_repo.repo_name),
|
|
|
'pr_source_branch': org_ref_name,
|
|
|
'pr_owner': pr.owner,
|
|
|
'pr_owner_username': pr.owner.username,
|
|
|
'pr_username': user.username,
|
|
|
'threading': threading,
|
|
|
'is_mention': False,
|
|
|
}
|
|
|
if reviewers:
|
|
|
NotificationModel().create(created_by=user, subject=subject, body=body,
|
|
|
recipients=reviewers,
|
|
|
type_=NotificationModel.TYPE_PULL_REQUEST,
|
|
|
email_kwargs=email_kwargs)
|
|
|
|
|
|
if mention_recipients:
|
|
|
email_kwargs['is_mention'] = True
|
|
|
subject = _('[Mention]') + ' ' + subject
|
|
|
# FIXME: this subject is wrong and unused!
|
|
|
NotificationModel().create(created_by=user, subject=subject, body=body,
|
|
|
recipients=mention_recipients,
|
|
|
type_=NotificationModel.TYPE_PULL_REQUEST,
|
|
|
email_kwargs=email_kwargs)
|
|
|
|
|
|
def mention_from_description(self, user, pr, old_description=''):
|
|
|
mention_recipients = (extract_mentioned_users(pr.description) -
|
|
|
extract_mentioned_users(old_description))
|
|
|
|
|
|
log.debug("Mentioning %s", mention_recipients)
|
|
|
self.add_reviewers(user, pr, set(), mention_recipients)
|
|
|
|
|
|
def remove_reviewers(self, user, pull_request, reviewers):
|
|
|
"""Remove specified users from being reviewers of the PR."""
|
|
|
if not reviewers:
|
|
|
return # avoid SQLAlchemy warning about empty sequence for IN-predicate
|
|
|
|
|
|
PullRequestReviewer.query() \
|
|
|
.filter_by(pull_request=pull_request) \
|
|
|
.filter(PullRequestReviewer.user_id.in_(r.user_id for r in reviewers)) \
|
|
|
.delete(synchronize_session='fetch') # the default of 'evaluate' is not available
|
|
|
|
|
|
def delete(self, pull_request):
|
|
|
pull_request = PullRequest.guess_instance(pull_request)
|
|
|
Session().delete(pull_request)
|
|
|
if pull_request.org_repo.scm_instance.alias == 'git':
|
|
|
# remove a ref under refs/pull/ so that commits can be garbage-collected
|
|
|
try:
|
|
|
del pull_request.org_repo.scm_instance._repo["refs/pull/%d/head" % pull_request.pull_request_id]
|
|
|
except KeyError:
|
|
|
pass
|
|
|
|
|
|
def close_pull_request(self, pull_request):
|
|
|
pull_request = PullRequest.guess_instance(pull_request)
|
|
|
pull_request.status = PullRequest.STATUS_CLOSED
|
|
|
pull_request.updated_on = datetime.datetime.now()
|
|
|
|
|
|
|
|
|
class CreatePullRequestAction(object):
|
|
|
|
|
|
class ValidationError(Exception):
|
|
|
pass
|
|
|
|
|
|
class Empty(ValidationError):
|
|
|
pass
|
|
|
|
|
|
class AmbiguousAncestor(ValidationError):
|
|
|
pass
|
|
|
|
|
|
class Unauthorized(ValidationError):
|
|
|
pass
|
|
|
|
|
|
@staticmethod
|
|
|
def is_user_authorized(org_repo, other_repo):
|
|
|
"""Performs authorization check with only the minimum amount of
|
|
|
information needed for such a check, rather than a full command
|
|
|
object.
|
|
|
"""
|
|
|
if (h.HasRepoPermissionLevel('read')(org_repo.repo_name) and
|
|
|
h.HasRepoPermissionLevel('read')(other_repo.repo_name)
|
|
|
):
|
|
|
return True
|
|
|
|
|
|
return False
|
|
|
|
|
|
def __init__(self, org_repo, other_repo, org_ref, other_ref, title, description, owner, reviewers):
|
|
|
from kallithea.controllers.compare import CompareController
|
|
|
reviewers = set(reviewers)
|
|
|
_assert_valid_reviewers(reviewers)
|
|
|
|
|
|
(org_ref_type,
|
|
|
org_ref_name,
|
|
|
org_rev) = org_ref.split(':')
|
|
|
org_display = h.short_ref(org_ref_type, org_ref_name)
|
|
|
if org_ref_type == 'rev':
|
|
|
cs = org_repo.scm_instance.get_changeset(org_rev)
|
|
|
org_ref = 'branch:%s:%s' % (cs.branch, cs.raw_id)
|
|
|
|
|
|
(other_ref_type,
|
|
|
other_ref_name,
|
|
|
other_rev) = other_ref.split(':')
|
|
|
if other_ref_type == 'rev':
|
|
|
cs = other_repo.scm_instance.get_changeset(other_rev)
|
|
|
other_ref_name = cs.raw_id[:12]
|
|
|
other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, cs.raw_id)
|
|
|
other_display = h.short_ref(other_ref_type, other_ref_name)
|
|
|
|
|
|
cs_ranges, _cs_ranges_not, ancestor_revs = \
|
|
|
CompareController._get_changesets(org_repo.scm_instance.alias,
|
|
|
other_repo.scm_instance, other_rev, # org and other "swapped"
|
|
|
org_repo.scm_instance, org_rev,
|
|
|
)
|
|
|
if not cs_ranges:
|
|
|
raise self.Empty(_('Cannot create empty pull request'))
|
|
|
|
|
|
if not ancestor_revs:
|
|
|
ancestor_rev = org_repo.scm_instance.EMPTY_CHANGESET
|
|
|
elif len(ancestor_revs) == 1:
|
|
|
ancestor_rev = ancestor_revs[0]
|
|
|
else:
|
|
|
raise self.AmbiguousAncestor(
|
|
|
_('Cannot create pull request - criss cross merge detected, please merge a later %s revision to %s')
|
|
|
% (other_ref_name, org_ref_name))
|
|
|
|
|
|
self.revisions = [cs_.raw_id for cs_ in cs_ranges]
|
|
|
|
|
|
# hack: ancestor_rev is not an other_rev but we want to show the
|
|
|
# requested destination and have the exact ancestor
|
|
|
other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, ancestor_rev)
|
|
|
|
|
|
if not title:
|
|
|
if org_repo == other_repo:
|
|
|
title = '%s to %s' % (org_display, other_display)
|
|
|
else:
|
|
|
title = '%s#%s to %s#%s' % (org_repo.repo_name, org_display,
|
|
|
other_repo.repo_name, other_display)
|
|
|
description = description or _('No description')
|
|
|
|
|
|
self.org_repo = org_repo
|
|
|
self.other_repo = other_repo
|
|
|
self.org_ref = org_ref
|
|
|
self.org_rev = org_rev
|
|
|
self.other_ref = other_ref
|
|
|
self.title = title
|
|
|
self.description = description
|
|
|
self.owner = owner
|
|
|
self.reviewers = reviewers
|
|
|
|
|
|
if not CreatePullRequestAction.is_user_authorized(self.org_repo, self.other_repo):
|
|
|
raise self.Unauthorized(_('You are not authorized to create the pull request'))
|
|
|
|
|
|
def execute(self):
|
|
|
created_by = User.get(request.authuser.user_id)
|
|
|
|
|
|
pr = PullRequest()
|
|
|
pr.org_repo = self.org_repo
|
|
|
pr.org_ref = self.org_ref
|
|
|
pr.other_repo = self.other_repo
|
|
|
pr.other_ref = self.other_ref
|
|
|
pr.revisions = self.revisions
|
|
|
pr.title = self.title
|
|
|
pr.description = self.description
|
|
|
pr.owner = self.owner
|
|
|
Session().add(pr)
|
|
|
Session().flush() # make database assign pull_request_id
|
|
|
|
|
|
if self.org_repo.scm_instance.alias == 'git':
|
|
|
# create a ref under refs/pull/ so that commits don't get garbage-collected
|
|
|
self.org_repo.scm_instance._repo[b"refs/pull/%d/head" % pr.pull_request_id] = ascii_bytes(self.org_rev)
|
|
|
|
|
|
# reset state to under-review
|
|
|
from kallithea.model.changeset_status import ChangesetStatusModel
|
|
|
from kallithea.model.comment import ChangesetCommentsModel
|
|
|
comment = ChangesetCommentsModel().create(
|
|
|
text=u'',
|
|
|
repo=self.org_repo,
|
|
|
author=created_by,
|
|
|
pull_request=pr,
|
|
|
send_email=False,
|
|
|
status_change=ChangesetStatus.STATUS_UNDER_REVIEW,
|
|
|
)
|
|
|
ChangesetStatusModel().set_status(
|
|
|
self.org_repo,
|
|
|
ChangesetStatus.STATUS_UNDER_REVIEW,
|
|
|
created_by,
|
|
|
comment,
|
|
|
pull_request=pr,
|
|
|
)
|
|
|
|
|
|
mention_recipients = extract_mentioned_users(self.description)
|
|
|
PullRequestModel().add_reviewers(created_by, pr, self.reviewers, mention_recipients)
|
|
|
|
|
|
return pr
|
|
|
|
|
|
|
|
|
class CreatePullRequestIterationAction(object):
|
|
|
@staticmethod
|
|
|
def is_user_authorized(old_pull_request):
|
|
|
"""Performs authorization check with only the minimum amount of
|
|
|
information needed for such a check, rather than a full command
|
|
|
object.
|
|
|
"""
|
|
|
if h.HasPermissionAny('hg.admin')():
|
|
|
return True
|
|
|
|
|
|
# Authorized to edit the old PR?
|
|
|
if request.authuser.user_id != old_pull_request.owner_id:
|
|
|
return False
|
|
|
|
|
|
# Authorized to create a new PR?
|
|
|
if not CreatePullRequestAction.is_user_authorized(old_pull_request.org_repo, old_pull_request.other_repo):
|
|
|
return False
|
|
|
|
|
|
return True
|
|
|
|
|
|
def __init__(self, old_pull_request, new_org_rev, new_other_rev, title, description, owner, reviewers):
|
|
|
self.old_pull_request = old_pull_request
|
|
|
|
|
|
org_repo = old_pull_request.org_repo
|
|
|
org_ref_type, org_ref_name, org_rev = old_pull_request.org_ref.split(':')
|
|
|
|
|
|
other_repo = old_pull_request.other_repo
|
|
|
other_ref_type, other_ref_name, other_rev = old_pull_request.other_ref.split(':') # other_rev is ancestor
|
|
|
#assert other_ref_type == 'branch', other_ref_type # TODO: what if not?
|
|
|
|
|
|
new_org_ref = '%s:%s:%s' % (org_ref_type, org_ref_name, new_org_rev)
|
|
|
new_other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, new_other_rev)
|
|
|
|
|
|
self.create_action = CreatePullRequestAction(org_repo, other_repo, new_org_ref, new_other_ref, None, None, owner, reviewers)
|
|
|
|
|
|
# Generate complete title/description
|
|
|
|
|
|
old_revisions = set(old_pull_request.revisions)
|
|
|
revisions = self.create_action.revisions
|
|
|
new_revisions = [r for r in revisions if r not in old_revisions]
|
|
|
lost = old_revisions.difference(revisions)
|
|
|
|
|
|
infos = ['This is a new iteration of %s "%s".' %
|
|
|
(h.canonical_url('pullrequest_show', repo_name=old_pull_request.other_repo.repo_name,
|
|
|
pull_request_id=old_pull_request.pull_request_id),
|
|
|
old_pull_request.title)]
|
|
|
|
|
|
if lost:
|
|
|
infos.append(_('Missing changesets since the previous iteration:'))
|
|
|
for r in old_pull_request.revisions:
|
|
|
if r in lost:
|
|
|
rev_desc = org_repo.get_changeset(r).message.split('\n')[0]
|
|
|
infos.append(' %s %s' % (h.short_id(r), rev_desc))
|
|
|
|
|
|
if new_revisions:
|
|
|
infos.append(_('New changesets on %s %s since the previous iteration:') % (org_ref_type, org_ref_name))
|
|
|
for r in reversed(revisions):
|
|
|
if r in new_revisions:
|
|
|
rev_desc = org_repo.get_changeset(r).message.split('\n')[0]
|
|
|
infos.append(' %s %s' % (h.short_id(r), h.shorter(rev_desc, 80)))
|
|
|
|
|
|
if self.create_action.other_ref == old_pull_request.other_ref:
|
|
|
infos.append(_("Ancestor didn't change - diff since previous iteration:"))
|
|
|
infos.append(h.canonical_url('compare_url',
|
|
|
repo_name=org_repo.repo_name, # other_repo is always same as repo_name
|
|
|
org_ref_type='rev', org_ref_name=h.short_id(org_rev), # use old org_rev as base
|
|
|
other_ref_type='rev', other_ref_name=h.short_id(new_org_rev),
|
|
|
)) # note: linear diff, merge or not doesn't matter
|
|
|
else:
|
|
|
infos.append(_('This iteration is based on another %s revision and there is no simple diff.') % other_ref_name)
|
|
|
else:
|
|
|
infos.append(_('No changes found on %s %s since previous iteration.') % (org_ref_type, org_ref_name))
|
|
|
# TODO: fail?
|
|
|
|
|
|
v = 2
|
|
|
m = re.match(r'(.*)\(v(\d+)\)\s*$', title)
|
|
|
if m is not None:
|
|
|
title = m.group(1)
|
|
|
v = int(m.group(2)) + 1
|
|
|
self.create_action.title = '%s (v%s)' % (title.strip(), v)
|
|
|
|
|
|
# using a mail-like separator, insert new iteration info in description with latest first
|
|
|
descriptions = description.replace('\r\n', '\n').split('\n-- \n', 1)
|
|
|
description = descriptions[0].strip() + '\n\n-- \n' + '\n'.join(infos)
|
|
|
if len(descriptions) > 1:
|
|
|
description += '\n\n' + descriptions[1].strip()
|
|
|
self.create_action.description = description
|
|
|
|
|
|
if not CreatePullRequestIterationAction.is_user_authorized(self.old_pull_request):
|
|
|
raise CreatePullRequestAction.Unauthorized(_('You are not authorized to create the pull request'))
|
|
|
|
|
|
def execute(self):
|
|
|
pull_request = self.create_action.execute()
|
|
|
|
|
|
# Close old iteration
|
|
|
from kallithea.model.comment import ChangesetCommentsModel
|
|
|
ChangesetCommentsModel().create(
|
|
|
text=_('Closed, next iteration: %s .') % pull_request.url(canonical=True),
|
|
|
repo=self.old_pull_request.other_repo_id,
|
|
|
author=request.authuser.user_id,
|
|
|
pull_request=self.old_pull_request.pull_request_id,
|
|
|
closing_pr=True)
|
|
|
PullRequestModel().close_pull_request(self.old_pull_request.pull_request_id)
|
|
|
return pull_request
|
|
|
|
