# -*- 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 mercurial.unionrepo

from tg import request

from tg import tmpl_context as c

from tg.i18n import ugettext as _

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

from kallithea.config.routing import url

from kallithea.controllers.changeset import _context_url, _ignorews_url, create_cs_pr_comment, delete_cs_pr_comment

from kallithea.lib import diffs

from kallithea.lib import helpers as h

from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired

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

from kallithea.lib.graphmod import graph_data

from kallithea.lib.page import Page

from kallithea.lib.utils2 import ascii_bytes, safe_bytes, safe_int

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

from kallithea.model.changeset_status import ChangesetStatusModel

from kallithea.model.comment import ChangesetCommentsModel

from kallithea.model.db import ChangesetStatus, PullRequest, PullRequestReviewer, Repository, User

from kallithea.model.forms import PullRequestForm, PullRequestPostForm

from kallithea.model.meta import Session

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

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 branch_rev:

            # a revset not restricting to merge() would be better

            # (especially because it would get the branch point)

            # ... but is currently too expensive

            # including branches of children could be nice too

            peerbranches = set()

            for i in repo._repo.revs(

                b"sort(parents(branch(id(%s)) and merge()) - branch(id(%s)), -rev)",

                ascii_bytes(branch_rev), ascii_bytes(branch_rev),

            ):

                for abranch in repo.get_changeset(i).branches:

                    if abranch not in peerbranches:

                        n = 'branch:%s:%s' % (abranch, repo.get_changeset(abranch).raw_id)

                        peers.append((n, abranch))

                        peerbranches.add(abranch)

        selected = None

        tiprev = repo.tags.get('tip')

        tipbranch = None

        branches = []

        for abranch, branchrev in repo.branches.items():

            n = 'branch:%s:%s' % (abranch, branchrev)

            desc = abranch

            if branchrev == tiprev:

                tipbranch = abranch

                desc = '%s (current tip)' % desc

            branches.append((n, desc))

            if rev == branchrev:

                selected = n

            if branch == abranch:

                if not rev:

                    selected = n

                branch = None

        if branch:  # branch not in list - it is probably closed

            branchrev = repo.closed_branches.get(branch)

            if branchrev:

                n = 'branch:%s:%s' % (branch, branchrev)

                branches.append((n, _('%s (closed)') % branch))

                selected = n

                branch = None

            if branch:

                log.debug('branch %r not found in %s', branch, repo)

        bookmarks = []

        for bookmark, bookmarkrev in repo.bookmarks.items():

            n = 'book:%s:%s' % (bookmark, bookmarkrev)

            bookmarks.append((n, bookmark))

            if rev == bookmarkrev:

                selected = n

        tags = []

        for tag, tagrev in repo.tags.items():

            if tag == 'tip':

                continue

            n = 'tag:%s:%s' % (tag, tagrev)

            tags.append((n, tag))

            # note: even if rev == tagrev, don't select the static tag - it must be chosen explicitly

        # prio 1: rev was selected as existing entry above

        # prio 2: create special entry for rev; rev _must_ be used

        specials = []

        if rev and selected is None:

            selected = 'rev:%s:%s' % (rev, rev)

            specials = [(selected, '%s: %s' % (_("Changeset"), rev[:12]))]

        # prio 3: most recent peer branch

        if peers and not selected:

            selected = peers[0][0]

        # prio 4: tip revision

        if not selected:

            if h.is_hg(repo):

                if tipbranch:

                    selected = 'branch:%s:%s' % (tipbranch, tiprev)

                else:

                    selected = 'tag:null:' + repo.EMPTY_CHANGESET

                    tags.append((selected, 'null'))

                    selected = 'branch:master:%s' % repo.branches['master']

                else:

                    k, v = list(repo.branches.items())[0]

                    selected = 'branch:%s:%s' % (k, v)

        groups = [(specials, _("Special")),

                  (peers, _("Peer branches")),

                  (bookmarks, _("Bookmarks")),

                  (branches, _("Branches")),

                  (tags, _("Tags")),

                  ]

        return [g for g in groups if g[0]], selected

    def _is_allowed_to_change_status(self, pull_request):

        if pull_request.is_closed():

            return False

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

        reviewer = PullRequestReviewer.query() \

            .filter(PullRequestReviewer.pull_request == pull_request) \

            .filter(PullRequestReviewer.user_id == request.authuser.user_id) \

            .count() != 0

        return request.authuser.admin or owner or reviewer

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def show_all(self, repo_name):

        c.from_ = request.GET.get('from_') or ''

        c.closed = request.GET.get('closed') or ''

        url_params = {}

        if c.from_:

            url_params['from_'] = 1

        if c.closed:

            url_params['closed'] = 1

        p = safe_int(request.GET.get('page'), 1)

        q = PullRequest.query(include_closed=c.closed, sorted=True)

        if c.from_:

            q = q.filter_by(org_repo=c.db_repo)

        else:

            q = q.filter_by(other_repo=c.db_repo)

        c.pull_requests = q.all()

        c.pullrequests_pager = Page(c.pull_requests, page=p, items_per_page=100, **url_params)

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

    @LoginRequired()

    def show_my(self):

        c.closed = request.GET.get('closed') or ''

        c.my_pull_requests = PullRequest.query(

            include_closed=c.closed,

            sorted=True,

        ).filter_by(owner_id=request.authuser.user_id).all()

        c.participate_in_pull_requests = []

        c.participate_in_pull_requests_todo = []

        done_status = set([ChangesetStatus.STATUS_APPROVED, ChangesetStatus.STATUS_REJECTED])

        for pr in PullRequest.query(

            include_closed=c.closed,

            reviewer_id=request.authuser.user_id,

            sorted=True,

        ):

            status = pr.user_review_status(request.authuser.user_id) # very inefficient!!!

            if status in done_status:

                c.participate_in_pull_requests.append(pr)

            else:

                c.participate_in_pull_requests_todo.append(pr)

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

    @LoginRequired()

    @HasRepoPermissionLevelDecorator('read')

    def index(self):

        org_repo = c.db_repo

        org_scm_instance = org_repo.scm_instance

        try:

            org_scm_instance.get_changeset()

        except EmptyRepositoryError as e:

            h.flash(_('There are no changesets yet'),

                    category='warning')

            raise HTTPFound(location=url('summary_home', repo_name=org_repo.repo_name))

        org_rev = request.GET.get('rev_end')

        # rev_start is not directly useful - its parent could however be used

        # as default for other and thus give a simple compare view

        rev_start = request.GET.get('rev_start')

        other_rev = None

        if rev_start:

            starters = org_repo.get_changeset(rev_start).parents

            if starters:

                other_rev = starters[0].raw_id

            else:

                other_rev = org_repo.scm_instance.EMPTY_CHANGESET

        branch = request.GET.get('branch')

        c.cs_repos = [(org_repo.repo_name, org_repo.repo_name)]

        c.default_cs_repo = org_repo.repo_name

        c.cs_refs, c.default_cs_ref = self._get_repo_refs(org_scm_instance, rev=org_rev, branch=branch)

        default_cs_ref_type, default_cs_branch, default_cs_rev = c.default_cs_ref.split(':')

        if default_cs_ref_type != 'branch':

            default_cs_branch = org_repo.get_changeset(default_cs_rev).branch

        # add org repo to other so we can open pull request against peer branches on itself

        c.a_repos = [(org_repo.repo_name, '%s (self)' % org_repo.repo_name)]

        if org_repo.parent:

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

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

    @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'])

        other_ref = _form['other_ref'] # will have symbolic name and head revision

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

        reviewers = []

        title = _form['pullrequest_title']

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

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

        try:

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

        except CreatePullRequestAction.ValidationError as e:

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

            raise HTTPNotFound

        try:

            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=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):

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

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

        new_other_rev = self._get_ref_rev(old_pull_request.other_repo, old_pull_request.other_ref_parts[0], old_pull_request.other_ref_parts[1])

        try:

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

        except CreatePullRequestAction.ValidationError as e:

import re

import pytest

from kallithea.controllers.pullrequests import PullrequestsController

from kallithea.model.meta import Session

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

fixture = Fixture()

class TestPullrequestsController(base.TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(base.url(controller='pullrequests', action='index',

    def test_create_trivial(self):

        self.log_user()

        response = self.app.post(base.url(controller='pullrequests', action='create',

                                  'pullrequest_title': 'title',

                                  'pullrequest_desc': 'description',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                 status=302)

        response = response.follow()

        assert response.status == '200 OK'

        response.mustcontain('Successfully opened new pull request')

        invalid_user_id = 99999

        self.log_user()

        # create a valid pull request

        response = self.app.post(base.url(controller='pullrequests', action='create',

                                 {

                                  'pullrequest_title': 'title',

                                  'pullrequest_desc': 'description',

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                 },

                                status=302)

        # location is of the form:

        m = re.search(r'/pull-request/(\d+)/', response.location)

        assert m is not None

        pull_request_id = m.group(1)

        # edit it

        response = self.app.post(base.url(controller='pullrequests', action='post',

                                 {

                                  'pullrequest_title': 'title',

                                  'pullrequest_desc': 'description',

                                  'owner': base.TEST_USER_ADMIN_LOGIN,

                                  '_session_csrf_secret_token': self.session_csrf_secret_token(),

                                  'review_members': [str(invalid_user_id)],

                                 },

                                 status=400)

        response.mustcontain('Invalid reviewer "%s" specified' % invalid_user_id)

@pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods

class TestPullrequestsGetRepoRefs(base.TestController):

    def setup_method(self, method):

        self.repo_name = 'main'

        self.repo_scm_instance = repo.scm_instance

        Session().commit()

        self.c = PullrequestsController()

    def teardown_method(self, method):

        fixture.destroy_repo('main')

        Session().commit()

        Session.remove()

    def test_repo_refs_empty_repo(self):

        # empty repo with no commits, no branches, no bookmarks, just one tag

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

    def test_repo_refs_one_commit_no_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

    def test_repo_refs_one_commit_rev_hint(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)

        assert default == expected

    def test_repo_refs_two_commits_no_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        cs1 = fixture.commit_change(self.repo_name, filename='file2',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

        assert default == expected

    def test_repo_refs_two_commits_rev_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        cs1 = fixture.commit_change(self.repo_name, filename='file2',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)

        expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)

        assert default == expected

        assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)

        assert default == expected

    def test_repo_refs_two_commits_branch_hint(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        cs1 = fixture.commit_change(self.repo_name, filename='file2',

                parent=None, newfile=True)

        assert default == expected

    def test_repo_refs_one_branch_no_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                parent=None, newfile=True)

        # TODO