# -*- coding: utf-8 -*-

"""

 (c) 2015-2017 - Copyright Red Hat Inc

 Authors:

   Pierre-Yves Chibon <pingou@pingoured.fr>

"""

from __future__ import unicode_literals, absolute_import

import json

import unittest

import shutil

import sys

import tempfile

import time

import os

import re

import pygit2

import six

from mock import patch, MagicMock

from bs4 import BeautifulSoup

from datetime import datetime, timedelta

sys.path.insert(0, os.path.join(os.path.dirname(

    os.path.abspath(__file__)), '..'))

import pagure.lib.query

import pagure.lib.tasks

import tests

from pagure.lib.repo import PagureRepo

def _get_commits(output):

    ''' Returns the commits message in the output. All commits must have

    been made by `Alice Author` or `PY C` to be found.

    '''

    commits = []

    save = False

    cnt = 0

    for row in output.split('\n'):

        if row.strip() in ['Alice Author', 'Alice Äuthòr', 'PY C']:

            save = True

        if save:

            cnt += 1

        if cnt == 7:

            commits.append(row.strip())

            save = False

            cnt = 0

    return commits

def set_up_git_repo(

        session, path, new_project=None, branch_from='feature', mtype='FF',

        prid=1, name_from='test'):

    """ Set up the git repo and create the corresponding PullRequest

    object.

    """

    # Create a git repo to play with

    gitrepo = os.path.join(path, 'repos', '%s.git' % name_from)

    repo = pygit2.init_repository(gitrepo, bare=True)

    newpath = tempfile.mkdtemp(prefix='pagure-fork-test')

    repopath = os.path.join(newpath, 'test')

    clone_repo = pygit2.clone_repository(gitrepo, repopath)

    # Create a file in that git repo

    with open(os.path.join(repopath, 'sources'), 'w') as stream:

        stream.write('foo\n bar')

    clone_repo.index.add('sources')

    clone_repo.index.write()

    try:

        com = repo.revparse_single('HEAD')

        prev_commit = [com.oid.hex]

    except:

        prev_commit = []

    # Commits the files added

    tree = clone_repo.index.write_tree()

    author = pygit2.Signature(

        'Alice Author', 'alice@authors.tld')

    committer = pygit2.Signature(

        'Cecil Committer', 'cecil@committers.tld')

    clone_repo.create_commit(

        'refs/heads/master',  # the name of the reference to update

        author,

        committer,

        'Add sources file for testing',

        # binary string representing the tree object ID

        tree,

        # list of binary strings representing parents of the new commit

        prev_commit

    )

    refname = 'refs/heads/master:refs/heads/master'

    ori_remote = clone_repo.remotes[0]

    PagureRepo.push(ori_remote, refname)

    first_commit = repo.revparse_single('HEAD')

    def compatible_signature(name, email):

        if six.PY2:

            name = name.encode("utf-8")

            email = email.encode("utf-8")

        return pygit2.Signature(name, email)

    if mtype == 'merge':

        with open(os.path.join(repopath, '.gitignore'), 'w') as stream:

            stream.write('*~')

        clone_repo.index.add('.gitignore')

        clone_repo.index.write()

        # Commits the files added

        tree = clone_repo.index.write_tree()

        author = compatible_signature(

            'Alice Äuthòr', 'alice@äuthòrs.tld')

        comitter = compatible_signature(

            'Cecil Cõmmîttër', 'cecil@cõmmîttërs.tld')

        clone_repo.create_commit(

            'refs/heads/master',

            author,

            committer,

            'Add .gitignore file for testing',

            # binary string representing the tree object ID

            tree,

            # list of binary strings representing parents of the new commit

            [first_commit.oid.hex]

        )

        refname = 'refs/heads/master:refs/heads/master'

        ori_remote = clone_repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

    if mtype == 'conflicts':

        with open(os.path.join(repopath, 'sources'), 'w') as stream:

            stream.write('foo\n bar\nbaz')

        clone_repo.index.add('sources')

        clone_repo.index.write()

        # Commits the files added

        tree = clone_repo.index.write_tree()

        author = pygit2.Signature(

            'Alice Author', 'alice@authors.tld')

        committer = pygit2.Signature(

            'Cecil Committer', 'cecil@committers.tld')

        clone_repo.create_commit(

            'refs/heads/master',

            author,

            committer,

            'Add sources conflicting',

            # binary string representing the tree object ID

            tree,

            # list of binary strings representing parents of the new commit

            [first_commit.oid.hex]

        )

        refname = 'refs/heads/master:refs/heads/master'

        ori_remote = clone_repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

    # Set the second repo

    new_gitrepo = repopath

    if new_project:

        # Create a new git repo to play with

        new_gitrepo = os.path.join(newpath, new_project.fullname)

        if not os.path.exists(new_gitrepo):

            os.makedirs(new_gitrepo)

            new_repo = pygit2.clone_repository(gitrepo, new_gitrepo)

    repo = pygit2.Repository(new_gitrepo)

    if mtype != 'nochanges':

        # Edit the sources file again

        with open(os.path.join(new_gitrepo, 'sources'), 'w') as stream:

            stream.write('foo\n bar\nbaz\n boose')

        repo.index.add('sources')

        repo.index.write()

        # Commits the files added

        tree = repo.index.write_tree()

        author = pygit2.Signature(

            'Alice Author', 'alice@authors.tld')

        committer = pygit2.Signature(

            'Cecil Committer', 'cecil@committers.tld')

        repo.create_commit(

            'refs/heads/%s' % branch_from,

            author,

            committer,

            'A commit on branch %s\n\nMore information' % branch_from,

            tree,

            [first_commit.oid.hex]

        )

        refname = 'refs/heads/%s' % (branch_from)

        ori_remote = repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

    # Create a PR for these changes

    project = pagure.lib.query.get_authorized_project(session, 'test')

    req = pagure.lib.query.new_pull_request(

        session=session,

        repo_from=project,

        branch_from=branch_from,

        repo_to=project,

        branch_to='master',

        title='PR from the %s branch' % branch_from,

        user='pingou',

    )

    session.commit()

    assert req.id == prid

    assert req.title == 'PR from the %s branch' % branch_from

    shutil.rmtree(newpath)

class PagureFlaskForktests(tests.Modeltests):

    """ Tests for flask fork controller of pagure """

    def test_request_pull_reference(self):

        """ Test if there is a reference created for a new PR. """

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None, branch_from='feature')

        project = pagure.lib.query.get_authorized_project(self.session, 'test')

        self.assertEqual(len(project.requests), 1)

        # View the pull-request

        output = self.app.get('/test/pull-request/1')

        self.assertEqual(output.status_code, 200)

        gitrepo = os.path.join(self.path, 'repos', 'test.git')

        repo = pygit2.Repository(gitrepo)

        self.assertEqual(

            list(repo.listall_references()),

            ['refs/heads/feature', 'refs/heads/master', 'refs/pull/1/head']

        )

    @patch('pagure.lib.notify.send_email')

    def test_request_pull(self, send_email):

        """ Test the request_pull endpoint. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        # Non-existant project

        output = self.app.get('/foobar/pull-request/1')

        self.assertEqual(output.status_code, 404)

        # Project has no PR

        output = self.app.get('/test/pull-request/1')

        self.assertEqual(output.status_code, 404)

        set_up_git_repo(

            self.session, self.path, new_project=None, branch_from='feature')

        project = pagure.lib.query.get_authorized_project(self.session, 'test')

        self.assertEqual(len(project.requests), 1)

        # View the pull-request

        output = self.app.get('/test/pull-request/1')

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        #self.assertIn(

            #'
PR#1\n'

            #'  PR from the feature branch\n',

            #output_text)

        self.assertIn(

            'title="View file as of 2a552b">sources', output_text)

        # Test if the `open changed file icon` is displayed.

        self.assertIn(

            'class="open_changed_file_icon_wrap">

            'class="fa fa-file-code-o fa-fw" '

            'alt="Open changed file" title="Open changed file">'

            '', output_text)

        self.assertIn(

            '+3', output_text)

        self.assertIn(

            '-1',

            output_text)

    @patch('pagure.lib.notify.send_email')

    def test_task_update_request_pull(self, send_email):

        """ Test the task update_pull_request endpoint. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None, branch_from='feature')

        self.session = pagure.lib.query.create_session(self.dbpath)

        project = pagure.lib.query.get_authorized_project(self.session, 'test')

        self.assertEqual(len(project.requests), 1)

        request = project.requests[0]

        self.assertEqual(len(request.comments), 0)

        start_commit = request.commit_start

        stop_commit = request.commit_stop

        # View the pull-request

        output = self.app.get('/test/pull-request/1')

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn(

            '<title>PR#1: PR from the feature branch - test\n - Pagure</title>',

            output_text)

        self.assertIn(

            'title="View file as of 2a552b">sources', output_text)

        # Add a new commit on the repo from

        newpath = tempfile.mkdtemp(prefix='pagure-fork-test')

        gitrepo = os.path.join(self.path, 'repos', 'test.git')

        repopath = os.path.join(newpath, 'test')

        clone_repo = pygit2.clone_repository(

            gitrepo, repopath, checkout_branch='feature')

        def compatible_signature(name, email):

            if six.PY2:

                name = name.encode("utf-8")

                email = email.encode("utf-8")

            return pygit2.Signature(name, email)

        with open(os.path.join(repopath, '.gitignore'), 'w') as stream:

                stream.write('*~')

        clone_repo.index.add('.gitignore')

        clone_repo.index.write()

        com = clone_repo.revparse_single('HEAD')

        prev_commit = [com.oid.hex]

        # Commits the files added

        tree = clone_repo.index.write_tree()

        author = compatible_signature(

            'Alice Äuthòr', 'alice@äuthòrs.tld')

        comitter = compatible_signature(

            'Cecil Cõmmîttër', 'cecil@cõmmîttërs.tld')

        clone_repo.create_commit(

            'refs/heads/feature',

            author,

            comitter,

            'Add .gitignore file for testing',

            # binary string representing the tree object ID

            tree,

            # list of binary strings representing parents of the new commit

            prev_commit

        )

        refname = 'refs/heads/feature:refs/heads/feature'

        ori_remote = clone_repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

        shutil.rmtree(newpath)

        pagure.lib.tasks.update_pull_request(request.uid)

        self.session = pagure.lib.query.create_session(self.dbpath)

        project = pagure.lib.query.get_authorized_project(self.session, 'test')

        self.assertEqual(len(project.requests), 1)

        request = project.requests[0]

        self.assertEqual(len(request.comments), 1)

        self.assertIsNotNone(request.commit_start)

        self.assertIsNotNone(request.commit_stop)

        self.assertNotEqual(start_commit, request.commit_start)

        self.assertNotEqual(stop_commit, request.commit_stop)

    @patch('pagure.lib.notify.send_email')

    def test_request_pull_ci_dropdown(self, send_email):

        """ Test presence of the "Rerun CI" dropdown with various settings. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None, branch_from='feature')

        user = tests.FakeUser()

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            # old-style TRIGGER_CI list - test backwards compatibility

            with patch.dict('pagure.config.config',

                            {'TRIGGER_CI': ['old-style-trigger-ci']}):

                output = self.app.get('/test/pull-request/1')

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertNotIn('Rerun CI', output_text)

            # new-style TRIGGER_CI, but no button to show

            with patch.dict('pagure.config.config',

                            {'TRIGGER_CI': {'no-button': None}}):

                output = self.app.get('/test/pull-request/1')

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertNotIn('Rerun CI', output_text)

            trigger_ci = {

                'foobar-ci': {

                    'name': 'foobar-ci-name',

                    'description': 'barfoo',

                },

                'spam-ci': {

                    'name': 'spam-ci-name',

                    'description': 'with beans and eggs',

                },

                'no-button-for-me-ci': None,

            }

            # new-style TRIGGER_CI, several buttons to show

            with patch.dict('pagure.config.config',

                            {'TRIGGER_CI': trigger_ci}):

                output = self.app.get('/test/pull-request/1')

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn('Rerun CI', output_text)

                self.assertIn('foobar-ci-name', output_text)

                self.assertIn('spam-ci-name', output_text)

                self.assertNotIn('no-button-for-me-ci', output_text)

            trigger_ci = {

                'foobar-ci': {

                    'name': 'foobar-ci-name',

                    'description': 'barfoo',

                    'requires_project_hook_attr': ('ci_hook', 'active_pr', True),

                },

            }

            # new-style TRIGGER_CI with requires_project_hook_attr that is

            # not fulfilled by the project

            with patch.dict('pagure.config.config',

                            {'TRIGGER_CI': trigger_ci}):

                output = self.app.get('/test/pull-request/1')

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertNotIn('Rerun CI', output_text)

            # now activate the hook and try again

            data = {

                'active_pr': 'y',

                'ci_url': 'https://jenkins.fedoraproject.org',

                'ci_job': 'ci_job',

                'ci_type': 'jenkins',

                'csrf_token': self.get_csrf()

            }

            output = self.app.post('/test/settings/Pagure CI', data=data,

                                   follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            with patch.dict('pagure.config.config',

                            {'TRIGGER_CI': trigger_ci}):

                output = self.app.get('/test/pull-request/1')

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn('Rerun CI', output_text)

                self.assertIn('foobar-ci-name', output_text)

        # shouldn't show up if user is not logged in

        with patch.dict('pagure.config.config',

                        {'TRIGGER_CI': trigger_ci}):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertNotIn('Rerun CI', output_text)

    @patch('pagure.lib.notify.send_email')

    @patch.dict('pagure.config.config',

                {'TRIGGER_CI': {'CI1': {'name': 'CI1', 'description': 'CI1!'}}})

    def test_request_pull_ci_rerun(self, send_email):

        """ Test rerunning CI using button from the "Rerun CI" dropdown. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None, branch_from='feature')

        user = tests.FakeUser()

        user.username = 'pingou'

        project = pagure.lib.query.get_authorized_project(self.session, 'test')

        request = project.requests[0]

        with tests.user_set(self.app.application, user):

            # no csrf token

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            output = self.app.post(

                '/test/pull-request/1/trigger-ci', follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn('Invalid input', output.get_data(as_text=True))

            # no such PR

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            output = self.app.post(

                '/test/pull-request/2/trigger-ci', follow_redirects=True)

            self.assertEqual(output.status_code, 404)

            # wrong comment

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {'csrf_token': csrf_token, 'comment': 'this doesnt exist'}

            output = self.app.post(

                '/test/pull-request/1/trigger-ci', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn('Invalid input', output.get_data(as_text=True))

            # everything ok

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {'csrf_token': csrf_token, 'comment': 'CI1'}

            output = self.app.post(

                '/test/pull-request/1/trigger-ci', data=data, follow_redirects=True)

            output_text = output.get_data(as_text=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn('
CI1
', output_text)

            comment = request.comments[0]

            self.assertTrue(comment.notification)

            self.assertEqual(comment.comment, 'CI1')

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_FF(self, send_email):

        """ Test the merge_request_pull endpoint with a FF PR. """

        send_email.return_value = True

        self.test_request_pull()

        user = tests.FakeUser()

        with tests.user_set(self.app.application, user):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            # No CSRF

            output = self.app.post(

                '/test/pull-request/1/merge', data={}, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<title>PR#1: PR from the feature branch - test\n - '

                'Pagure</title>', output_text)

            #self.assertIn(

                #'
PR#1\n'

                #'  PR from the feature branch\n',

                #output_text)

            self.assertIn(

                'title="View file as of 2a552b">sources', output_text)

            # Wrong project

            data = {

                'csrf_token': csrf_token,

            }

            output = self.app.post(

                '/foobar/pull-request/100/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 404)

            # Wrong project

            data = {

                'csrf_token': csrf_token,

            }

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 403)

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            # Wrong request id

            data = {

                'csrf_token': csrf_token,

            }

            output = self.app.post(

                '/test/pull-request/100/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 404)

            # Project w/o pull-request

            self.session.commit()

            repo = pagure.lib.query.get_authorized_project(self.session, 'test')

            settings = repo.settings

            settings['pull_requests'] = False

            repo.settings = settings

            self.session.add(repo)

            self.session.commit()

            # Pull-request disabled

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 404)

            # Project w pull-request but only assignee can merge

            self.session.commit()

            repo = pagure.lib.query.get_authorized_project(self.session, 'test')

            settings['pull_requests'] = True

            settings['Only_assignee_can_merge_pull-request'] = True

            repo.settings = settings

            self.session.add(repo)

            self.session.commit()

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<title>PR#1: PR from the feature branch - test\n - '

                'Pagure</title>', output_text)

            self.assertIn(

                '
\n        
\n              '

                '\n              '

                '

                'font-weight-bold">#1\n            '

                '\n                  '

                'PR from the feature branch\n',

                output_text)

            self.assertIn(

                'This request must be '

                'assigned to be merged', output_text)

            # PR assigned but not to this user

            self.session.commit()

            repo = pagure.lib.query.get_authorized_project(self.session, 'test')

            req = repo.requests[0]

            req.assignee_id = 2

            self.session.add(req)

            self.session.commit()

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\n        
\n              '

                '\n              '

                '

                'font-weight-bold">#1\n            '

                '\n                  '

                'PR from the feature branch\n',

                output_text)

            self.assertIn(

                'Only the assignee can '

                'merge this review', output_text)

            # Project w/ minimal PR score

            self.session.commit()

            repo = pagure.lib.query.get_authorized_project(self.session, 'test')

            settings['Only_assignee_can_merge_pull-request'] = False

            settings['Minimum_score_to_merge_pull-request'] = 2

            repo.settings = settings

            self.session.add(repo)

            self.session.commit()

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\n        
\n              '

                '\n              '

                '

                'font-weight-bold">#1\n            '

                '\n                  '

                'PR from the feature branch\n',

                output_text)

            self.assertIn(

                'This request does not '

                'have the minimum review score necessary to be merged',

                output_text)

            # Merge

            self.session.commit()

            repo = pagure.lib.query.get_authorized_project(self.session, 'test')

            settings['Minimum_score_to_merge_pull-request'] = -1

            repo.settings = settings

            self.session.add(repo)

            self.session.commit()

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output = self.app.get('/test/commits')

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<title>Commits - test - Pagure</title>', output_text)

            self.assertIn(

                'A commit on branch feature', output_text)

            self.assertNotIn(

                'Merge #1 `PR from the feature branch`', output_text)

            # Check if the closing notification was added

            output = self.app.get('/test/pull-request/1')

            self.assertIn(

                'Merged just now\n'

                '            \n            by\n'

                '            pingou.\n',

                output.get_data(as_text=True))

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_merge(self, send_email):

        """ Test the merge_request_pull endpoint with a merge PR. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None,

            branch_from='feature', mtype='merge')

        user = tests.FakeUser()

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {

                'csrf_token': csrf_token,

            }

            # Merge

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                '<title>Overview - test - Pagure</title>', output.get_data(as_text=True))

            # Check if the closing notification was added

            output = self.app.get('/test/pull-request/1')

            self.assertIn(

                'Merged just now\n'

                '            \n            by\n'

                '            pingou.\n',

                output.get_data(as_text=True))

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_merge_with_comment(self, send_email):

        """ Test the merge_request_pull endpoint with a merge PR. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None,

            branch_from='feature', mtype='merge')

        self.session = pagure.lib.query.create_session(self.dbpath)

        request = pagure.lib.query.search_pull_requests(

            self.session, project_id=1, requestid=1)

        self.assertEqual(len(request.comments), 0)

        user = tests.FakeUser()

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {

                'csrf_token': csrf_token,

                'comment': 'Thanks for the review and the suggestions!'

            }

            # Merge

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                '<title>Overview - test - Pagure</title>', output.get_data(as_text=True))

            # Check if the closing notification was added

            output = self.app.get('/test/pull-request/1')

            output_text = output.get_data(as_text=True)

            self.assertIn(

                'Merged just now\n'

                '            \n            by\n'

                '            pingou.\n',

                output_text)

            self.assertIn(

                'Thanks for the review and the suggestions!', output_text)

            self.session = pagure.lib.query.create_session(self.dbpath)

            request = pagure.lib.query.search_pull_requests(

                self.session, project_id=1, requestid=1)

            self.assertEqual(len(request.comments), 2)

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_merge_with_delete_branch(self, send_email):

        """ Test the merge_request_pull endpoint with a merge PR and delete source branch. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None,

            branch_from='feature-branch', mtype='merge')

        user = tests.FakeUser()

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            data = {

                'csrf_token': self.get_csrf(output=output),

                'delete_branch': True,

            }

            # Merge

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<title>Overview - test - Pagure</title>', output_text)

            # Check the branch is not mentioned

            self.assertNotIn(

                '

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_conflicts(self, send_email):

        """ Test the merge_request_pull endpoint with a conflicting PR. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(

            os.path.join(self.path, 'requests'), bare=True)

        set_up_git_repo(

            self.session, self.path, new_project=None,

            branch_from='feature', mtype='conflicts')

        user = tests.FakeUser()

        user.username = 'pingou'

        with tests.user_set(self.app.application, user):

            output = self.app.get('/test/pull-request/1')

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {

                'csrf_token': csrf_token,

            }

            # Merge conflicts

            output = self.app.post(

                '/test/pull-request/1/merge', data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\n        
\n              '

                '\n              '

                '

                'font-weight-bold">#1\n            '

                '\n                  '

                'PR from the feature branch\n',

                output_text)

            self.assertIn('Merge conflicts!', output_text)

    @patch('pagure.lib.notify.send_email')

    def test_merge_request_pull_conflicts_with_delete_branch(self, send_email):

        """ Test the merge_request_pull endpoint with a conflicting PR and request deletion of branch. """

        send_email.return_value = True

        tests.create_projects(self.session)

        tests.create_projects_git(