Changeset - 5d60c9a391cd
Parent rev.
Child rev.
[Not reviewed]
default
0 2 0
Søren Løvborg - 9 years ago 2017-03-03 15:34:31
sorenl@unity3d.com
pullrequests: introduce "action objects" for PR creation

Inspired by the command design pattern, this attempts the following:

* move policy and business logic from controllers into the model,
* move validation, authorization and execution logic closer together in
the code,
* establish a reusable pattern for modelling higher-level concepts
(like "create a new PR iteration"),
* make error handling more well-defined, and independent of the
controller layer, and
* provide clear separation between, one one hand, the validation and
authorization of a request, and on the other, the actual execution.
2 files changed with 260 insertions and 190 deletions:
kallithea/controllers/pullrequests.py
21
148
kallithea/model/pull_request.py
239
42
0 comments (0 inline, 0 general)
kallithea/controllers/pullrequests.py
➞
Show inline comments
 
# -*- 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.controllers.pullrequests
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

			




	
	
	
		
 
pull requests controller for Kallithea for initializing pull requests

			




	
	
	
		
 

			




	
	
	
		
 
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: May 7, 2012

			




	
	
	
		
 
:author: marcink

			




	
	
	
		
 
:copyright: (c) 2013 RhodeCode GmbH, and others.

			




	
	
	
		
 
:license: GPLv3, see LICENSE.md for more details.

			




	
	
	
		
 
"""

			




	
	
	
		
 

			




	
	
	
		
 
import logging

			




	
	
	
		
 
import traceback

			




	
	
	
		
 
import formencode

			




	
	
	
		
 
import re

			




	
	
	
		
 

			




	
	
	
		
 
from pylons import request, tmpl_context as c

			




	
	
	
		
 
from pylons.i18n.translation import _

			




	
	
	
		
 
from webob.exc import HTTPFound, HTTPNotFound, HTTPForbidden, HTTPBadRequest

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.config.routing import url

			




	
	
	
		
 
from kallithea.lib import helpers as h

			




	
	
	
		
 
from kallithea.lib import diffs

			




	
	
	
		
 
from kallithea.lib.auth import LoginRequired, HasRepoPermissionLevelDecorator, \

			




	
	
	
		
 
    NotAnonymous

			




	
	
	
		
 
from kallithea.lib.base import BaseRepoController, render, jsonify

			




	
	
	
		
 
from kallithea.lib.diffs import LimitedDiffContainer

			




	
	
	
		
 
from kallithea.lib.page import Page

			




	
	
	
		
 
from kallithea.lib.utils import action_logger

			




	
	
	
		
 
from kallithea.lib.vcs.exceptions import EmptyRepositoryError, ChangesetDoesNotExistError

			




	
	
	
		
 
from kallithea.lib.vcs.utils import safe_str

			




	
	
	
		
 
from kallithea.lib.vcs.utils.hgcompat import unionrepo

			




	
	
	
		
 
from kallithea.model.db import PullRequest, ChangesetStatus, ChangesetComment, \

			




	
	
	
		
 
    PullRequestReviewer, Repository, User

			




	
	
	
		
 
from kallithea.model.pull_request import PullRequestModel

			




	
	
	
		
 
from kallithea.model.pull_request import CreatePullRequestAction, CreatePullRequestIterationAction, PullRequestModel

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.repo import RepoModel

			




	
	
	
		
 
from kallithea.model.comment import ChangesetCommentsModel

			




	
	
	
		
 
from kallithea.model.changeset_status import ChangesetStatusModel

			




	
	
	
		
 
from kallithea.model.forms import PullRequestForm, PullRequestPostForm

			




	
	
	
		
 
from kallithea.lib.utils2 import safe_int

			




	
	
	
		
 
from kallithea.controllers.changeset import _ignorews_url, _context_url, \

			




	
	
	
		
 
    create_comment

			




	
	
	
		
 
from kallithea.controllers.compare import CompareController

			




	
	
	
		
 
from kallithea.lib.graphmod import graph_data

			




	
	
	
		
 

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def _get_reviewer(user_id):

			




	
	
	
		
 
    """Look up user by ID and validate it as a potential reviewer."""

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        user = User.get(int(user_id))

			




	
	
	
		
 
    except ValueError:

			




	
	
	
		
 
        user = None

			




	
	
	
		
 

			




	
	
	
		
 
    if user is None or user.is_default_user:

			




	
	
	
		
 
        h.flash(_('Invalid reviewer "%s" specified') % user_id, category='error')

			




	
	
	
		
 
        raise HTTPBadRequest()

			




	
	
	
		
 

			




	
	
	
		
 
    return user

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class PullrequestsController(BaseRepoController):

			




	
	
	
		
 

			




	
	
	
		
 
    def _get_repo_refs(self, repo, rev=None, branch=None, branch_rev=None):

			




	
	
	
		
 
        """return a structure with repo's interesting changesets, suitable for

			




	
	
	
		
 
        the selectors in pullrequest.html

			




	
	
	
		
 

			




	
	
	
		
 
        rev: a revision that must be in the list somehow and selected by default

			




	
	
	
		
 
        branch: a branch that must be in the list and selected by default - even if closed

			




	
	
	
		
 
        branch_rev: a revision of which peers should be preferred and available."""

			




	
	
	
		
 
        # list named branches that has been merged to this named branch - it should probably merge back

			




	
	
	
		
 
        peers = []

			




	
	
	
		
 

			




	
	
	
		
 
        if rev:

			




	
	
	
		
 
            rev = safe_str(rev)

			




	
	
	
		
 

			




	
	
	
		
 
        if branch:

			




	
	
	
		
 
            branch = safe_str(branch)

			




	
	
	
		
 

			




	
	
	
		
 
        if branch_rev:

			




	
	
	
		
 
            branch_rev = safe_str(branch_rev)

			




	
	
		
 
@@ -286,271 +285,145 @@ class PullrequestsController(BaseRepoCon

			




	
	
	
		
 
            # add parent of this fork also and select it.

			




	
	
	
		
 
            # use the same branch on destination as on source, if available.

			




	
	
	
		
 
            c.a_repos.append((org_repo.parent.repo_name, '%s (parent)' % org_repo.parent.repo_name))

			




	
	
	
		
 
            c.a_repo = org_repo.parent

			




	
	
	
		
 
            c.a_refs, c.default_a_ref = self._get_repo_refs(

			




	
	
	
		
 
                    org_repo.parent.scm_instance, branch=default_cs_branch, rev=other_rev)

			




	
	
	
		
 

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            c.a_repo = org_repo

			




	
	
	
		
 
            c.a_refs, c.default_a_ref = self._get_repo_refs(org_scm_instance, rev=other_rev)

			




	
	
	
		
 

			




	
	
	
		
 
        # gather forks and add to this list ... even though it is rare to

			




	
	
	
		
 
        # request forks to pull from their parent

			




	
	
	
		
 
        for fork in org_repo.forks:

			




	
	
	
		
 
            c.a_repos.append((fork.repo_name, fork.repo_name))

			




	
	
	
		
 

			




	
	
	
		
 
        return render('/pullrequests/pullrequest.html')

			




	
	
	
		
 

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    @HasRepoPermissionLevelDecorator('read')

			




	
	
	
		
 
    @jsonify

			




	
	
	
		
 
    def repo_info(self, repo_name):

			




	
	
	
		
 
        repo = c.db_repo

			




	
	
	
		
 
        refs, selected_ref = self._get_repo_refs(repo.scm_instance)

			




	
	
	
		
 
        return {

			




	
	
	
		
 
            'description': repo.description.split('\n', 1)[0],

			




	
	
	
		
 
            'selected_ref': selected_ref,

			




	
	
	
		
 
            'refs': refs,

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    @HasRepoPermissionLevelDecorator('read')

			




	
	
	
		
 
    def create(self, repo_name):

			




	
	
	
		
 
        repo = c.db_repo

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            _form = PullRequestForm(repo.repo_id)().to_python(request.POST)

			




	
	
	
		
 
        except formencode.Invalid as errors:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            log.error(str(errors))

			




	
	
	
		
 
            msg = _('Error creating pull request: %s') % errors.msg

			




	
	
	
		
 
            h.flash(msg, 'error')

			




	
	
	
		
 
            raise HTTPBadRequest

			




	
	
	
		
 

			




	
	
	
		
 
        # heads up: org and other might seem backward here ...

			




	
	
	
		
 
        org_ref = _form['org_ref'] # will have merge_rev as rev but symbolic name

			




	
	
	
		
 
        org_repo = Repository.guess_instance(_form['org_repo'])

			




	
	
	
		
 
        (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 = _form['other_ref'] # will have symbolic name and head revision

			




	
	
	
		
 
        other_repo = Repository.guess_instance(_form['other_repo'])

			




	
	
	
		
 
        (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,

			




	
	
	
		
 
                                              )

			




	
	
	
		
 
        ancestor_rev = msg = None

			




	
	
	
		
 
        if not cs_ranges:

			




	
	
	
		
 
            msg = _('Cannot create empty pull request')

			




	
	
	
		
 
        elif not ancestor_revs:

			




	
	
	
		
 
            ancestor_rev = org_repo.scm_instance.EMPTY_CHANGESET

			




	
	
	
		
 
        elif len(ancestor_revs) == 1:

			




	
	
	
		
 
            ancestor_rev = ancestor_revs[0]

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            msg = _('Cannot create pull request - criss cross merge detected, please merge a later %s revision to %s'

			




	
	
	
		
 
                    ) % (other_ref_name, org_ref_name)

			




	
	
	
		
 
        if ancestor_rev is None:

			




	
	
	
		
 
            h.flash(msg, category='error', logf=log.error)

			




	
	
	
		
 
            raise HTTPNotFound

			




	
	
	
		
 

			




	
	
	
		
 
        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)

			




	
	
	
		
 

			




	
	
	
		
 
        reviewers = []

			




	
	
	
		
 

			




	
	
	
		
 
        title = _form['pullrequest_title']

			




	
	
	
		
 
        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 = _form['pullrequest_desc'].strip() or _('No description')

			




	
	
	
		
 
        description = _form['pullrequest_desc'].strip()

			




	
	
	
		
 
        owner = User.get(request.authuser.user_id)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            created_by = User.get(request.authuser.user_id)

			




	
	
	
		
 
            pull_request = PullRequestModel().create(

			




	
	
	
		
 
                created_by, org_repo, org_ref, other_repo, other_ref, revisions,

			




	
	
	
		
 
                title, description, reviewers)

			




	
	
	
		
 
            cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, description, owner, reviewers)

			




	
	
	
		
 
        except CreatePullRequestAction.ValidationError as e:

			




	
	
	
		
 
            h.flash(str(e), category='error', logf=log.error)

			




	
	
	
		
 
            raise HTTPNotFound

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            pull_request = cmd.execute()

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            h.flash(_('Successfully opened new pull request'),

			




	
	
	
		
 
                    category='success')

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            h.flash(_('Error occurred while creating pull request'),

			




	
	
	
		
 
                    category='error')

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise HTTPFound(location=url('pullrequest_home', repo_name=repo_name))

			




	
	
	
		
 

			




	
	
	
		
 
        h.flash(_('Successfully opened new pull request'),

			




	
	
	
		
 
                category='success')

			




	
	
	
		
 
        raise HTTPFound(location=pull_request.url())

			




	
	
	
		
 

			




	
	
	
		
 
    def create_new_iteration(self, old_pull_request, new_rev, title, description, reviewers):

			




	
	
	
		
 
        org_repo = old_pull_request.org_repo

			




	
	
	
		
 
        org_ref_type, org_ref_name, org_rev = old_pull_request.org_ref.split(':')

			




	
	
	
		
 
        new_org_rev = self._get_ref_rev(org_repo, 'rev', new_rev)

			




	
	
	
		
 

			




	
	
	
		
 
        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_other_rev = self._get_ref_rev(other_repo, 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, new_other_rev, # org and other "swapped"

			




	
	
	
		
 
            org_repo.scm_instance, new_org_rev)

			




	
	
	
		
 
        ancestor_rev = msg = None

			




	
	
	
		
 
        if not cs_ranges:

			




	
	
	
		
 
            msg = _('Cannot create empty pull request update') # cannot happen!

			




	
	
	
		
 
        elif not ancestor_revs:

			




	
	
	
		
 
            msg = _('Cannot create pull request update - no common ancestor found') # cannot happen

			




	
	
	
		
 
        elif len(ancestor_revs) == 1:

			




	
	
	
		
 
            ancestor_rev = ancestor_revs[0]

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            msg = _('Cannot create pull request update - criss cross merge detected, please merge a later %s revision to %s'

			




	
	
	
		
 
                    ) % (other_ref_name, org_ref_name)

			




	
	
	
		
 
        if ancestor_rev is None:

			




	
	
	
		
 
            h.flash(msg, category='error', logf=log.error)

			




	
	
	
		
 
        owner = User.get(request.authuser.user_id)

			




	
	
	
		
 
        new_org_rev = self._get_ref_rev(old_pull_request.org_repo, 'rev', new_rev)

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            cmd = CreatePullRequestIterationAction(old_pull_request, new_org_rev, title, description, owner, reviewers)

			




	
	
	
		
 
        except CreatePullRequestAction.ValidationError as e:

			




	
	
	
		
 
            h.flash(str(e), category='error', logf=log.error)

			




	
	
	
		
 
            raise HTTPNotFound

			




	
	
	
		
 

			




	
	
	
		
 
        old_revisions = set(old_pull_request.revisions)

			




	
	
	
		
 
        revisions = [cs.raw_id for cs in cs_ranges]

			




	
	
	
		
 
        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 ancestor_rev == other_rev:

			




	
	
	
		
 
                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?

			




	
	
	
		
 

			




	
	
	
		
 
        # hack: ancestor_rev is not an other_ref but we want to show the

			




	
	
	
		
 
        # requested destination and have the exact ancestor

			




	
	
	
		
 
        new_other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, ancestor_rev)

			




	
	
	
		
 
        new_org_ref = '%s:%s:%s' % (org_ref_type, org_ref_name, new_org_rev)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            title, old_v = re.match(r'(.*)\(v(\d+)\)\s*$', title).groups()

			




	
	
	
		
 
            v = int(old_v) + 1

			




	
	
	
		
 
        except (AttributeError, ValueError):

			




	
	
	
		
 
            v = 2

			




	
	
	
		
 
        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()

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            created_by = User.get(request.authuser.user_id)

			




	
	
	
		
 
            pull_request = PullRequestModel().create(

			




	
	
	
		
 
                created_by, org_repo, new_org_ref, other_repo, new_other_ref, revisions,

			




	
	
	
		
 
                title, description, reviewers)

			




	
	
	
		
 
            pull_request = cmd.execute()

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            h.flash(_('Error occurred while creating pull request'),

			




	
	
	
		
 
                    category='error')

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise HTTPFound(location=old_pull_request.url())

			




	
	
	
		
 

			




	
	
	
		
 
        ChangesetCommentsModel().create(

			




	
	
	
		
 
            text=_('Closed, next iteration: %s .') % pull_request.url(canonical=True),

			




	
	
	
		
 
            repo=old_pull_request.other_repo_id,

			




	
	
	
		
 
            author=request.authuser.user_id,

			




	
	
	
		
 
            pull_request=old_pull_request.pull_request_id,

			




	
	
	
		
 
            closing_pr=True)

			




	
	
	
		
 
        PullRequestModel().close_pull_request(old_pull_request.pull_request_id)

			




	
	
	
		
 

			




	
	
	
		
 
        Session().commit()

			




	
	
	
		
 
        h.flash(_('New pull request iteration created'),

			




	
	
	
		
 
                category='success')

			




	
	
	
		
 

			




	
	
	
		
 
        raise HTTPFound(location=pull_request.url())

			




	
	
	
		
 

			




	
	
	
		
 
    # pullrequest_post for PR editing

			




	
	
	
		
 
    @LoginRequired()

			




	
	
	
		
 
    @NotAnonymous()

			




	
	
	
		
 
    @HasRepoPermissionLevelDecorator('read')

			




	
	
	
		
 
    def post(self, repo_name, pull_request_id):

			




	
	
	
		
 
        pull_request = PullRequest.get_or_404(pull_request_id)

			




	
	
	
		
 
        if pull_request.is_closed():

			




	
	
	
		
 
            raise HTTPForbidden()

			




	
	
	
		
 
        assert pull_request.other_repo.repo_name == repo_name

			




	
	
	
		
 
        #only owner or admin can update it

			




	
	
	
		
 
        owner = pull_request.owner_id == request.authuser.user_id

			




	
	
	
		
 
        repo_admin = h.HasRepoPermissionLevel('admin')(c.repo_name)

			




	
	
	
		
 
        if not (h.HasPermissionAny('hg.admin')() or repo_admin or owner):

			




	
	
	
		
 
            raise HTTPForbidden()

			




	
	
	
		
 

			




	
	
	
		
 
        _form = PullRequestPostForm()().to_python(request.POST)

			




	
	
	
		
 

			




	
	
	
		
 
        cur_reviewers = set(pull_request.get_reviewer_users())

			




	
	
	
		
 
        new_reviewers = set(_get_reviewer(s) for s in _form['review_members'])

			




	
	
	
		
 
        old_reviewers = set(_get_reviewer(s) for s in _form['org_review_members'])

			




	
	
	
		
 

			




	
	
	
		
 
        other_added = cur_reviewers - old_reviewers

			




	
	
	
		
 
        other_removed = old_reviewers - cur_reviewers

			




	
	
	
		
 

			




	
	
	
		
 
        if other_added:

			




	
	
	
		
 
            h.flash(_('Meanwhile, the following reviewers have been added: %s') %

			




	
	
	
		
 
                    (', '.join(u.username for u in other_added)),

			




	
	
	
		
 
                    category='warning')

			




	
	
	
		
 
        if other_removed:

			




	
	
	
		
 
            h.flash(_('Meanwhile, the following reviewers have been removed: %s') %

			




	
	
	
		
 
                    (', '.join(u.username for u in other_removed)),

			




	
	
	
		
 
                    category='warning')

			




	
	
	
		
 

			




	
	
	
		
 
        if _form['updaterev']:

			




	
	
	
		
 
            return self.create_new_iteration(pull_request,

			




	
	
	
		
 
                                      _form['updaterev'],

			




	
	
	
		
 
                                      _form['pullrequest_title'],

			




	
	
	
		
 
                                      _form['pullrequest_desc'],

			




	
	
	
		
 
                                      new_reviewers)

			




	
	
	
		
 

			




	
	
	
		
 
        added_reviewers = new_reviewers - old_reviewers - cur_reviewers

			




	
	
	
		
 
        removed_reviewers = (old_reviewers - new_reviewers) & cur_reviewers

			




	
	
	
		
 

			




	
	
	
		
 
        old_description = pull_request.description

			




	
	
	
		
 
        pull_request.title = _form['pullrequest_title']

			




	
	
	
		
 
        pull_request.description = _form['pullrequest_desc'].strip() or _('No description')

			




        

    


    
    

    

        

                

                    kallithea/model/pull_request.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
	
		
 
# -*- 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 logging

			




	
	
	
		
 
import datetime

			




	
	
	
		
 
import re

			




	
	
	
		
 

			




	
	
	
		
 
from pylons import request

			




	
	
	
		
 
from pylons.i18n.translation import _

			




	
	
	
		
 

			




	
	
	
		
 
from sqlalchemy.orm import joinedload

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.lib import helpers as h

			




	
	
	
		
 
from kallithea.model.db import PullRequest, PullRequestReviewer, Notification, \

			




	
	
	
		
 
    ChangesetStatus

			




	
	
	
		
 
    ChangesetStatus, User

			




	
	
	
		
 
from kallithea.model.notification import NotificationModel

			




	
	
	
		
 
from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
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 create(self, created_by, org_repo, org_ref, other_repo, other_ref,

			




	
	
	
		
 
               revisions, title, description, reviewers):

			




	
	
	
		
 
        reviewers = set(reviewers)

			




	
	
	
		
 
        _assert_valid_reviewers(reviewers)

			




	
	
	
		
 

			




	
	
	
		
 
        pr = PullRequest()

			




	
	
	
		
 
        pr.org_repo = org_repo

			




	
	
	
		
 
        pr.org_ref = org_ref

			




	
	
	
		
 
        pr.other_repo = other_repo

			




	
	
	
		
 
        pr.other_ref = other_ref

			




	
	
	
		
 
        pr.revisions = revisions

			




	
	
	
		
 
        pr.title = title

			




	
	
	
		
 
        pr.description = description

			




	
	
	
		
 
        pr.owner = created_by

			




	
	
	
		
 
        Session().add(pr)

			




	
	
	
		
 
        Session().flush() # make database assign pull_request_id

			




	
	
	
		
 

			




	
	
	
		
 
        #reset state to under-review

			




	
	
	
		
 
        from kallithea.model.changeset_status import ChangesetStatusModel

			




	
	
	
		
 
        from kallithea.model.comment import ChangesetCommentsModel

			




	
	
	
		
 
        comment = ChangesetCommentsModel().create(

			




	
	
	
		
 
            text=u'',

			




	
	
	
		
 
            repo=org_repo,

			




	
	
	
		
 
            author=created_by,

			




	
	
	
		
 
            pull_request=pr,

			




	
	
	
		
 
            send_email=False,

			




	
	
	
		
 
            status_change=ChangesetStatus.STATUS_UNDER_REVIEW,

			




	
	
	
		
 
        )

			




	
	
	
		
 
        ChangesetStatusModel().set_status(

			




	
	
	
		
 
            org_repo,

			




	
	
	
		
 
            ChangesetStatus.STATUS_UNDER_REVIEW,

			




	
	
	
		
 
            created_by,

			




	
	
	
		
 
            comment,

			




	
	
	
		
 
            pull_request=pr,

			




	
	
	
		
 
        )

			




	
	
	
		
 

			




	
	
	
		
 
        mention_recipients = extract_mentioned_users(description)

			




	
	
	
		
 
        self.add_reviewers(created_by, pr, reviewers, mention_recipients)

			




	
	
	
		
 

			




	
	
	
		
 
        return pr

			




	
	
	
		
 

			




	
	
	
		
 
    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 = safe_unicode(

			




	
	
	
		
 
            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_=Notification.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_=Notification.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."""

			




	
	
	
		
 

			




	
	
	
		
 
        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)

			




	
	
	
		
 

			




	
	
	
		
 
    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.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

			




	
	
	
		
 

			




	
	
	
		
 
        #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, 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)

			




	
	
	
		
 

			




	
	
	
		
 
        self.create_action = CreatePullRequestAction(org_repo, other_repo, new_org_ref, old_pull_request.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?

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            title, old_v = re.match(r'(.*)\(v(\d+)\)\s*$', title).groups()

			




	
	
	
		
 
            v = int(old_v) + 1

			




	
	
	
		
 
        except (AttributeError, ValueError):

			




	
	
	
		
 
            v = 2

			




	
	
	
		
 
        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

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.