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 tg import request

from tg.i18n import ugettext as _

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

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)

            # 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

        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