pagure/tests/test_pagure_flask_ui_fork.py

# -*- 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(
        #'<h3><span class="label label-default">PR#1</span>\n'
        #'  PR from the feature branch\n</h3>',
        # output_text)
        self.assertIn(
            'title="View file as of 2a552b">sources</a>', output_text
        )

        # Test if the `open changed file icon` is displayed.
        self.assertIn(
            'class="open_changed_file_icon_wrap"><span '
            'class="fa fa-file-code-o fa-fw" '
            'alt="Open changed file" title="Open changed file"></span>'
            "</a>",
            output_text,
        )

        self.assertIn(
            '<span class="btn btn-success btn-sm font-weight-bold disabled opacity-100">+3</span>',
            output_text,
        )
        self.assertIn(
            '<span class="btn btn-danger btn-sm font-weight-bold disabled opacity-100">-1</span>',
            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</a>', 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("<p>CI1</p>", 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(
            #'<h3><span class="label label-default">PR#1</span>\n'
            #'  PR from the feature branch\n</h3>',
            # output_text)
            self.assertIn(
                'title="View file as of 2a552b">sources</a>', 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn(
                "Only the assignee can merge this request", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\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(
                '<span class="text-info font-weight-bold">Merged</span> just now\n'
                "            </span>\n            by\n"
                '            <span title="PY C (pingou)">pingou.</span>\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(
                '<span class="text-info font-weight-bold">Merged</span> just now\n'
                "            </span>\n            by\n"
                '            <span title="PY C (pingou)">pingou.</span>\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(
                '<span class="text-info font-weight-bold">Merged</span> just now\n'
                "            </span>\n            by\n"
                '            <span title="PY C (pingou)">pingou.</span>\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(
                '<a class="" href="/test/branch/feature-branch"', output_text
            )

    @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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\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(
            os.path.join(self.path, "requests"), bare=True
        )
        set_up_git_repo(
            self.session,
            self.path,
            new_project=None,
            branch_from="feature-branch",
            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)

            data = {
                "csrf_token": self.get_csrf(output=output),
                "delete_branch": True,
            }

            # 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(
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n'
                '              <span class="text-success font-weight-bold">#1</span>\n'
                '            <span class="font-weight-bold">\n'
                "                  PR from the feature-branch branch\n",
                output_text,
            )
            self.assertIn("Merge conflicts!", output_text)

            # Check the branch still exists
            output = self.app.get("/test/branches")
            self.assertIn("feature-branch", output.get_data(as_text=True))

    @patch("pagure.lib.notify.send_email")
    def test_merge_request_pull_nochange(self, send_email):
        """ Test the merge_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
        )
        set_up_git_repo(
            self.session,
            self.path,
            new_project=None,
            branch_from="master",
            mtype="nochanges",
        )

        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}

            # Nothing to 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(
                "Nothing to do, changes were already merged", output_text
            )

            # Check if the closing notification was added
            output = self.app.get("/test/pull-request/1")
            self.assertIn(
                '<span class="text-info font-weight-bold">Merged</span> just now\n'
                "            </span>\n            by\n"
                '            <span title="PY C (pingou)">pingou.</span>\n',
                output.get_data(as_text=True),
            )

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_close(self, send_email):
        """ Test the request_pull endpoint with a closed PR. """
        send_email.return_value = True

        self.test_merge_request_pull_FF()

        output = self.app.get("/test/pull-request/1")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)

        self.assertIn(
            '<span class="text-info font-weight-bold">Merged</span> '
            "just now\n            </span>\n            by\n",
            output_text,
        )
        self.assertIn(
            'title="View file as of 2a552b">sources</a>', output_text
        )

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_disabled(self, send_email):
        """ Test the request_pull endpoint with PR disabled. """
        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"
        )

        # Project w/o pull-request
        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()

        output = self.app.get("/test/pull-request/1")
        self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    @patch("pagure.lib.git.update_pull_ref")
    def test_request_pull_empty_repo(self, send_email, update_pull_ref):
        """ Test the request_pull endpoint against an empty repo. """
        # Mock update_pull_ref or the repo won't be empty anymore
        # (the PR will have been pushed to refs/pull)
        send_email.return_value = True

        tests.create_projects(self.session)
        item = pagure.lib.model.Project(
            user_id=2,  # foo
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=1,
        )
        self.session.add(item)
        self.session.commit()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "foo"), bare=True
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        self.assertFalse(os.path.exists(gitrepo))
        os.makedirs(gitrepo)
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        # Edit the sources file again
        with open(os.path.join(newpath, "sources"), "w") as stream:
            stream.write("foo\n bar\nbaz\n boose")
        new_repo.index.add("sources")
        new_repo.index.write()

        # Commits the files added
        tree = new_repo.index.write_tree()
        author = pygit2.Signature("Alice Author", "alice@authors.tld")
        committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld")
        new_repo.create_commit(
            "refs/heads/feature",
            author,
            committer,
            "A commit on branch feature",
            tree,
            [],
        )
        refname = "refs/heads/feature:refs/heads/feature"
        ori_remote = new_repo.remotes[0]
        PagureRepo.push(ori_remote, refname)

        # Create a PR for these changes
        project = pagure.lib.query.get_authorized_project(self.session, "test")
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_from=item,
            branch_from="feature",
            repo_to=project,
            branch_to="master",
            title="PR from the feature branch",
            user="pingou",
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "PR from the feature branch")

        output = self.app.get("/test/pull-request/1")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<h4 class="ml-1">\n        <div>\n              '
            '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
            '<span class="text-success '
            'font-weight-bold">#1</span>\n            '
            '<span class="font-weight-bold">\n                  '
            "PR from the feature branch\n",
            output_text,
        )
        self.assertTrue(output_text.count('<span class="commitdate"'), 1)
        self.assertTrue(update_pull_ref.called)

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_empty_fork(self, send_email):
        """ Test the request_pull endpoint from an empty fork. """
        send_email.return_value = True

        tests.create_projects(self.session)
        item = pagure.lib.model.Project(
            user_id=2,  # foo
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=1,
        )
        self.session.add(item)
        self.session.commit()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "foo"), bare=True
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        self.assertFalse(os.path.exists(gitrepo))
        os.makedirs(gitrepo)
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        # Create a PR for these "changes" (there are none, both repos are
        # empty)
        project = pagure.lib.query.get_authorized_project(self.session, "test")
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_from=item,
            branch_from="feature",
            repo_to=project,
            branch_to="master",
            title="PR from the feature branch",
            user="pingou",
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "PR from the feature branch")

        output = self.app.get("/test/pull-request/1", 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(
            "Fork is empty, there are no "
            "commits to create a pull request with",
            output_text,
        )

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_request_pulls_order(self, send_email):
        """Test the request_pulls

        i.e Make sure that the results are displayed
        in the order required by the user"""
        send_email.return_value = True

        # Initially no project
        output = self.app.get("/test/pull-requests")
        self.assertEqual(output.status_code, 404)

        tests.create_projects(self.session)
        tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        item = pagure.lib.model.Project(
            user_id=2,
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=1,
        )
        self.session.add(item)
        self.session.commit()

        # create PR's to play with
        # PR-1
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_to=repo,
            repo_from=item,
            branch_from="feature",
            branch_to="master",
            title="PR from the feature branch",
            user="pingou",
            status="Open",
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "PR from the feature branch")

        # PR-2
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_to=repo,
            branch_to="master",
            branch_from="feature",
            repo_from=item,
            title="test PR",
            user="pingou",
            status="Open",
        )
        self.session.commit()
        self.assertEqual(req.title, "test PR")

        # PR-3
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_to=repo,
            branch_from="feature",
            branch_to="master",
            repo_from=item,
            title="test Invalid PR",
            user="pingou",
            status="Closed",
        )
        self.session.commit()
        self.assertEqual(req.title, "test Invalid PR")

        # PR-4
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_to=repo,
            branch_from="feature",
            title="test PR for sort",
            repo_from=item,
            user="pingou",
            branch_to="master",
            status="Open",
        )
        self.session.commit()
        self.assertEqual(req.title, "test PR for sort")

        # sort by last_updated
        output = self.app.get("/test/pull-requests?order_key=last_updated")
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertEqual(output.status_code, 200)
        # Make sure that issue four is first since it was modified last
        self.assertIn('href="/test/pull-request/4"', tr_elements[0])
        self.assertIn('href="/test/pull-request/2"', tr_elements[1])
        self.assertIn('href="/test/pull-request/1"', tr_elements[2])

        pr_one = pagure.lib.query.search_pull_requests(
            self.session, project_id=1, requestid=1
        )
        pr_one.last_updated = datetime.utcnow() + timedelta(seconds=2)
        self.session.add(pr_one)
        self.session.commit()

        # sort by last_updated
        output = self.app.get("/test/pull-requests?order_key=last_updated")
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertEqual(output.status_code, 200)
        # Make sure that PR four is first since it was modified last
        self.assertIn('href="/test/pull-request/1"', tr_elements[0])
        # Make sure that PR two is second since it was modified second
        self.assertIn('href="/test/pull-request/4"', tr_elements[1])
        # Make sure that PR one is last since it was modified first
        self.assertIn('href="/test/pull-request/2"', tr_elements[2])

        # Now query so that the results are ascending
        output = self.app.get(
            "/test/pull-requests?" "order_key=last_updated&order=asc"
        )
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertIn('href="/test/pull-request/2"', tr_elements[0])
        self.assertIn('href="/test/pull-request/4"', tr_elements[1])
        self.assertIn('href="/test/pull-request/1"', tr_elements[2])

        # check that search_pattern argument works
        output = self.app.get("/test/pull-requests?search_pattern=feature")
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertIn('href="/test/pull-request/1"', tr_elements[0])
        self.assertEqual(len(tr_elements), 1)

        output = self.app.get("/test/pull-requests?search_pattern=PR")
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertIn('href="/test/pull-request/4"', tr_elements[0])
        self.assertIn('href="/test/pull-request/2"', tr_elements[1])
        self.assertIn('href="/test/pull-request/1"', tr_elements[2])
        self.assertEqual(len(tr_elements), 3)

        output = self.app.get("/test/pull-requests?search_pattern=*PR")
        output_text = output.get_data(as_text=True)
        tr_elements = re.findall(
            '<div class="request-row list-group-item list-group-item-action ">(.*?)</div><!--end request-row-->',
            output_text,
            re.M | re.S,
        )
        self.assertEqual(len(tr_elements), 1)
        self.assertIn('href="/test/pull-request/2"', tr_elements[0])

    @patch("pagure.lib.notify.send_email")
    def test_request_pulls(self, send_email):
        """ Test the request_pulls endpoint. """
        send_email.return_value = True

        # No such project
        output = self.app.get("/test/pull-requests")
        self.assertEqual(output.status_code, 404)

        tests.create_projects(self.session)
        tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)

        output = self.app.get("/test/pull-requests")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 0 Open PRs\n',
            output_text,
        )

        set_up_git_repo(
            self.session, self.path, new_project=None, branch_from="feature"
        )

        output = self.app.get("/test/pull-requests")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 1 Open PRs\n',
            output_text,
        )

        output = self.app.get("/test/pull-requests?status=1")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 1 Open PRs\n',
            output_text,
        )

        output = self.app.get("/test/pull-requests?status=true")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 1 Open PRs\n',
            output_text,
        )

        output = self.app.get("/test/pull-requests?status=Merged")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 0 Merged PRs\n',
            output_text,
        )

        output = self.app.get("/test/pull-requests?status=0")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 0 Merged PRs\n',
            output_text,
        )

        output = self.app.get("/test/pull-requests?status=Closed")
        self.assertEqual(output.status_code, 200)
        output_text = output.get_data(as_text=True)
        self.assertIn(
            '<span class="fa fa-fw fa-arrow-circle-down"></span> 0 Cancelled PRs\n',
            output_text,
        )

        # Project w/o pull-request
        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()

        output = self.app.get("/test/pull-requests")
        self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_patch(self, send_email):
        """ Test the request_pull_patch endpoint. """
        send_email.return_value = True

        output = self.app.get("/test/pull-request/1.patch")
        self.assertEqual(output.status_code, 404)

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

        output = self.app.get("/test/pull-request/100.patch")
        self.assertEqual(output.status_code, 404)

        output = self.app.get("/test/pull-request/1.patch")
        self.assertEqual(output.status_code, 200)

        npatch = []
        for row in output.get_data(as_text=True).split("\n"):
            if row.startswith("Date:"):
                continue
            if row.startswith("From "):
                row = row.split(" ", 2)[2]
            npatch.append(row)

        exp = r"""Mon Sep 17 00:00:00 2001
From: Alice Author <alice@authors.tld>
Subject: A commit on branch feature


More information
---

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..e4e5f6c
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+*~
\ No newline at end of file
diff --git a/sources b/sources
index 9f44358..2a552bb 100644
--- a/sources
+++ b/sources
@@ -1,2 +1,4 @@
 foo
- bar
\ No newline at end of file
+ bar
+baz
+ boose
\ No newline at end of file

"""

        patch = "\n".join(npatch)
        # print patch
        self.assertEqual(patch, exp)

        # Project w/o pull-request
        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()

        output = self.app.get("/test/pull-request/1.patch")
        self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_diff(self, send_email):
        """ Test the request_pull_patch endpoint. """
        send_email.return_value = True

        output = self.app.get("/test/pull-request/1.diff")
        self.assertEqual(output.status_code, 404)

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

        output = self.app.get("/test/pull-request/100.diff")
        self.assertEqual(output.status_code, 404)

        output = self.app.get("/test/pull-request/1.diff")
        self.assertEqual(output.status_code, 200)

        exp = r"""diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..e4e5f6c
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+*~
\ No newline at end of file
diff --git a/sources b/sources
index 9f44358..2a552bb 100644
--- a/sources
+++ b/sources
@@ -1,2 +1,4 @@
 foo
- bar
\ No newline at end of file
+ bar
+baz
+ boose
\ No newline at end of file
"""

        self.assertEqual(output.get_data(as_text=True), exp)

        # Project w/o pull-request
        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()

        output = self.app.get("/test/pull-request/1.diff")
        self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_patch_close(self, send_email):
        """ Test the request_pull_patch endpoint with a closed PR. """
        send_email.return_value = True

        self.test_merge_request_pull_FF()

        output = self.app.get("/test/pull-request/1.patch")
        self.assertEqual(output.status_code, 200)

        npatch = []
        for row in output.get_data(as_text=True).split("\n"):
            if row.startswith("Date:"):
                continue
            if row.startswith("From "):
                row = row.split(" ", 2)[2]
            npatch.append(row)

        exp = r"""Mon Sep 17 00:00:00 2001
From: Alice Author <alice@authors.tld>
Subject: A commit on branch feature


More information
---

diff --git a/sources b/sources
index 9f44358..2a552bb 100644
--- a/sources
+++ b/sources
@@ -1,2 +1,4 @@
 foo
- bar
\ No newline at end of file
+ bar
+baz
+ boose
\ No newline at end of file

"""

        patch = "\n".join(npatch)
        # print patch
        self.assertEqual(patch, exp)

    @patch("pagure.lib.notify.send_email")
    @patch("pagure.lib.git.update_pull_ref")
    def test_request_pull_patch_empty_repo(self, send_email, update_pull_ref):
        """ Test the request_pull_patch endpoint against an empty repo. """
        # Mock update_pull_ref or the repo won't be empty anymore
        # (the PR will have been pushed to refs/pull)
        send_email.return_value = True

        tests.create_projects(self.session)
        item = pagure.lib.model.Project(
            user_id=2,  # foo
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=1,
        )
        self.session.add(item)
        self.session.commit()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "foo"), bare=True
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        self.assertFalse(os.path.exists(gitrepo))
        os.makedirs(gitrepo)
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        # Edit the sources file again
        with open(os.path.join(newpath, "sources"), "w") as stream:
            stream.write("foo\n bar\nbaz\n boose")
        new_repo.index.add("sources")
        new_repo.index.write()

        # Commits the files added
        tree = new_repo.index.write_tree()
        author = pygit2.Signature("Alice Author", "alice@authors.tld")
        committer = pygit2.Signature("Cecil Committer", "cecil@committers.tld")
        new_repo.create_commit(
            "refs/heads/feature",
            author,
            committer,
            "A commit on branch feature",
            tree,
            [],
        )
        refname = "refs/heads/feature:refs/heads/feature"
        ori_remote = new_repo.remotes[0]
        PagureRepo.push(ori_remote, refname)

        # Create a PR for these "changes" (there are none, both repos are
        # empty)
        project = pagure.lib.query.get_authorized_project(self.session, "test")
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_from=item,
            branch_from="feature",
            repo_to=project,
            branch_to="master",
            title="PR from the feature branch",
            user="pingou",
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "PR from the feature branch")

        output = self.app.get(
            "/test/pull-request/1.patch", follow_redirects=True
        )
        self.assertEqual(output.status_code, 200)

        npatch = []
        for row in output.get_data(as_text=True).split("\n"):
            if row.startswith("Date:"):
                continue
            if row.startswith("From "):
                row = row.split(" ", 2)[2]
            npatch.append(row)

        exp = r"""Mon Sep 17 00:00:00 2001
From: Alice Author <alice@authors.tld>
Subject: A commit on branch feature


---

diff --git a/sources b/sources
new file mode 100644
index 0000000..2a552bb
--- /dev/null
+++ b/sources
@@ -0,0 +1,4 @@
+foo
+ bar
+baz
+ boose
\ No newline at end of file

"""

        patch = "\n".join(npatch)
        # print patch
        self.assertEqual(patch, exp)

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_patch_empty_fork(self, send_email):
        """ Test the request_pull_patch endpoint from an empty fork. """
        send_email.return_value = True

        tests.create_projects(self.session)
        item = pagure.lib.model.Project(
            user_id=2,  # foo
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=1,
        )
        self.session.add(item)
        self.session.commit()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "foo"), bare=True
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        self.assertFalse(os.path.exists(gitrepo))
        os.makedirs(gitrepo)
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        # Create a PR for these "changes" (there are none, both repos are
        # empty)
        project = pagure.lib.query.get_authorized_project(self.session, "test")
        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_from=item,
            branch_from="feature",
            repo_to=project,
            branch_to="master",
            title="PR from the feature branch",
            user="pingou",
        )
        self.session.commit()
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "PR from the feature branch")

        output = self.app.get(
            "/test/pull-request/1.patch", 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)
        self.assertIn(
            "Fork is empty, there are no "
            "commits to create a pull request with",
            output_text,
        )

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_close_request_pull(self, send_email):
        """ Test the close_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
        )
        set_up_git_repo(
            self.session,
            self.path,
            new_project=None,
            branch_from="feature",
            mtype="merge",
        )

        user = tests.FakeUser()
        with tests.user_set(self.app.application, user):
            output = self.app.post("/test/pull-request/close/1")
            self.assertEqual(output.status_code, 302)

            output = self.app.post(
                "/test/pull-request/close/1", 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
            )
            self.assertIn("Invalid input submitted", output_text)

            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}

            # Invalid project
            output = self.app.post(
                "/foo/pull-request/close/1", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

            # Invalid PR id
            output = self.app.post(
                "/test/pull-request/close/100",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

            # Invalid user for this project
            output = self.app.post(
                "/test/pull-request/close/1", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 403)

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/close/1", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

            # Project w/ pull-request
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            settings = repo.settings
            settings["pull_requests"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            output = self.app.post(
                "/test/pull-request/close/1", 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
            )
            self.assertIn("Pull request closed!", output_text)

    @patch("pagure.lib.notify.send_email")
    def test_reopen_request_pull(self, send_email):
        """ Test the reopen_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
        )
        set_up_git_repo(
            self.session,
            self.path,
            new_project=None,
            branch_from="feature",
            mtype="merge",
        )

        user = tests.FakeUser()
        with tests.user_set(self.app.application, user):
            output = self.app.post("/test/pull-request/1/reopen")
            self.assertEqual(output.status_code, 302)

            output = self.app.post(
                "/test/pull-request/1/reopen", 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(
                #'Pull request reopened!',
                'return window.confirm("Are you sure you want to reopen this requested pull?")',
                output_text,
            )

            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}

            # Invalid project
            output = self.app.post(
                "/foo/pull-request/1/reopen", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

            # Invalid PR id
            output = self.app.post(
                "/test/pull-request/100/reopen",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

            # Invalid user for this project
            output = self.app.post(
                "/test/pull-request/1/reopen", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 403)

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/reopen", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

            # Project w/ pull-request
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            settings = repo.settings
            settings["pull_requests"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            output = self.app.post(
                "/test/pull-request/cancel/1", data=data, follow_redirects=True
            )
            output = self.app.post(
                "/test/pull-request/1/reopen", 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(
                'return window.confirm("Are you sure you want to reopen this requested pull?")',
                output_text,
            )

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
    def test_update_pull_requests_assign(self):
        """ Test the update_pull_requests endpoint when assigning a 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"
        )

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # No such project
            output = self.app.post("/foo/pull-request/1/update")
            self.assertEqual(output.status_code, 404)

            output = self.app.post("/test/pull-request/100/update")
            self.assertEqual(output.status_code, 404)

            # Invalid input
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertNotIn("Request assigned", output_text)

            output = self.app.get("/test/pull-request/1")
            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {"user": "pingou"}

            # No CSRF
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertNotIn("Request assigned", output_text)

            # Invalid assignee
            data = {"csrf_token": csrf_token, "user": "bar"}

            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn("No user &#34;bar&#34; found", output_text)

            # Assign the PR
            data = {"csrf_token": csrf_token, "user": "pingou"}

        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 403)

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn("Request assigned", output_text)

            # Pull-Request closed
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            req = repo.requests[0]
            req.status = "Closed"
            req.closed_by_in = 1
            self.session.add(req)
            self.session.commit()

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
    def test_update_pull_requests_tag(self):
        """ Test the update_pull_requests endpoint when tagging a 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"
        )

        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 = {"tag": "black"}

            # No CSRF
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertNotIn("Request assigned", output_text)

            # Tag the PR
            data = {"csrf_token": csrf_token, "tag": "black"}

            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn("Pull-request tagged with: black", output_text)
            self.assertIn(
                'title="comma separated list of tags"\n              '
                'value="black" />',
                output_text,
            )

        # Try as another user
        user.username = "foo"
        with tests.user_set(self.app.application, user):
            # Tag the PR
            data = {"csrf_token": csrf_token, "tag": "blue, yellow"}

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 403)

            # Make the PR be from foo
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            req = repo.requests[0]
            req.user_id = 2
            self.session.add(req)
            self.session.commit()

            # Re-try to tag the PR
            data = {"csrf_token": csrf_token, "tag": "blue, yellow"}

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            soup = BeautifulSoup(output.get_data(as_text=True), "html.parser")
            self.assertEqual(
                soup.find("title").string,
                "PR#1: PR from the feature branch - test\n - Pagure",
            )

            self.assertIn(
                "Pull-request **un**tagged with: black",
                output.get_data(as_text=True),
            )
            self.assertIn(
                "Pull-request tagged with: blue, yellow",
                output.get_data(as_text=True),
            )

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Pull-Request closed
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            req = repo.requests[0]
            req.status = "Closed"
            req.closed_by_in = 1
            self.session.add(req)
            self.session.commit()

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/update", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_fork_project(self, send_email):
        """ Test the fork_project endpoint. """
        send_email.return_value = True

        tests.create_projects(self.session)
        for folder in ["docs", "tickets", "requests", "repos"]:
            tests.create_projects_git(
                os.path.join(self.path, folder), bare=True
            )

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            output = self.app.post("/do_fork/test")
            self.assertEqual(output.status_code, 400)

            output = self.app.get("/new/")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "<strong>Create new Project</strong>",
                output.get_data(as_text=True),
            )

            csrf_token = self.get_csrf(output=output)

            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/foo", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 404)

        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.post("/do_fork/test")
            self.assertEqual(output.status_code, 400)

            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_branch_space(self, send_email):
        """ Test the new_request_pull endpoint. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        set_up_git_repo(
            self.session,
            self.path,
            new_project=fork,
            branch_from="feature",
            mtype="FF",
        )

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            output = self.app.get("/test/diff/master..foo bar")
            self.assertEqual(output.status_code, 400)
            output_text = output.get_data(as_text=True)
            self.assertIn("<p>Branch foo bar does not exist</p>", output_text)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull(self, send_email):
        """ Test the new_request_pull endpoint. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        set_up_git_repo(
            self.session,
            self.path,
            new_project=fork,
            branch_from="feature",
            mtype="FF",
        )

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.get("/foo/diff/master..feature")
            self.assertEqual(output.status_code, 404)

            output = self.app.get("/test/diff/master..foo")
            self.assertEqual(output.status_code, 400)

            output = self.app.get("/test/diff/foo..master")
            self.assertEqual(output.status_code, 400)

            output = self.app.get("/test/diff/feature..master")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Diff from master to feature - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<p class="error"> No commits found </p>', output_text
            )

            output = self.app.get("/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Diff from feature to master - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertNotIn(
                '<input type="submit" class="submit positive button" '
                'value="Create">',
                output_text,
            )

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            output = self.app.get("/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )
            # Check that we prefilled the input fields as expected:
            self.assertIn(
                '<input class="form-control" id="title" name="title" '
                'placeholder="Pull Request Title" required="required" '
                'type="text" value="A commit on branch feature">',
                output_text,
            )
            self.assertIn(
                """<textarea class="form-control" rows=8 id="initial_comment" name="initial_comment"
            placeholder="Describe your changes" tabindex=1>
More information</textarea>
          <div id="preview" class="p-1">""",
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="master"><span class="fa fa-random">'
                "</span> master</a>",
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Case 1 - Add an initial comment
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/test/diff/master..feature", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#2: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)
            self.assertEqual(output_text.count('title="PY C (pingou)"'), 2)

            # Test if the `open changed file icon` is displayed.
            self.assertIn(
                'class="open_changed_file_icon_wrap"><span '
                'class="fa fa-file-code-o fa-fw" '
                'alt="Open changed file" title="Open changed file"></span>'
                "</a>",
                output_text,
            )

            # Case 2 - Add an empty initial comment
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "",
            }

            output = self.app.post(
                "/test/diff/master..feature", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#3: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertNotIn('<div id="comment-', output_text)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_filename_unicode(self, send_email):
        """ Test the new_request_pull endpoint. """
        send_email.return_value = True

        # Create the main project in the DB
        item = pagure.lib.model.Project(
            user_id=1,  # pingou
            name="test",
            description="test project #1",
            hook_token="aaabbbccc",
        )
        item.close_status = [
            "Invalid",
            "Insufficient data",
            "Fixed",
            "Duplicate",
        ]
        self.session.add(item)
        self.session.commit()

        # Create the fork
        item = pagure.lib.model.Project(
            user_id=1,  # pingou
            name="test",
            description="test project #1",
            hook_token="aaabbbcccdd",
            parent_id=1,
            is_fork=True,
        )
        item.close_status = [
            "Invalid",
            "Insufficient data",
            "Fixed",
            "Duplicate",
        ]
        self.session.add(item)
        self.session.commit()

        # Create two git repos, one has 6 commits, the other 4 of which only
        # 1 isn't present in the first repo
        gitrepo = os.path.join(self.path, "repos", "test.git")
        pygit2.init_repository(gitrepo, bare=True)

        gitrepo2 = os.path.join(
            self.path, "repos", "forks", "pingou", "test.git"
        )
        pygit2.init_repository(gitrepo2, bare=True)

        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        repopath = os.path.join(newpath, "test")
        clone_repo = pygit2.clone_repository(gitrepo, repopath)

        # Do 2 commits to the main repo
        for i in range(2):
            with open(os.path.join(repopath, "sources"), "w") as stream:
                stream.write("foo%s\n bar%s\n" % (i, i))
            clone_repo.index.add("sources")
            clone_repo.index.write()

            parents = []
            try:
                last_commit = clone_repo.revparse_single("HEAD")
                parents = [last_commit.oid.hex]
            except KeyError:
                pass

            # 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,
                "Editing the file sources for testing #%s" % i,
                # binary string representing the tree object ID
                tree,
                # list of binary strings representing parents of the new commit
                parents,
            )

        # Push to the main repo
        refname = "refs/heads/master:refs/heads/master"
        ori_remote = clone_repo.remotes[0]
        PagureRepo.push(ori_remote, refname)

        # Push to the fork repo
        remote = clone_repo.create_remote("pingou_fork", gitrepo2)
        PagureRepo.push(remote, refname)

        # Add 1 commits to the fork repo
        repopath = os.path.join(newpath, "pingou_test")
        clone_repo = pygit2.clone_repository(gitrepo2, repopath)

        with open(os.path.join(repopath, "soürces"), "w") as stream:
            stream.write("foo\n bar\n")
        clone_repo.index.add("soürces")
        clone_repo.index.write()
        with open(os.path.join(repopath, "fóß"), "w") as stream:
            stream.write("foo\n bar\n")
        clone_repo.index.add("fóß")
        clone_repo.index.write()

        last_commit = clone_repo.revparse_single("HEAD")

        # 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")
        last_commit = clone_repo.create_commit(
            "refs/heads/feature_foo",  # the name of the reference to update
            author,
            committer,
            "New edition on side branch of the file sources for testing",
            # binary string representing the tree object ID
            tree,
            # list of binary strings representing parents of the new commit
            [last_commit.oid.hex],
        )

        # Push to the fork repo
        ori_remote = clone_repo.remotes[0]
        refname = "refs/heads/feature_foo:refs/heads/feature_foo"
        PagureRepo.push(ori_remote, refname)

        shutil.rmtree(newpath)

        # Create the PR between the two repos
        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        forked_repo = pagure.lib.query.get_authorized_project(
            self.session, "test", user="pingou"
        )

        req = pagure.lib.query.new_pull_request(
            session=self.session,
            repo_from=forked_repo,
            branch_from="feature_foo",
            repo_to=repo,
            branch_to="master",
            title="test pull-request",
            user="pingou",
        )
        self.assertEqual(req.id, 1)
        self.assertEqual(req.title, "test pull-request")

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            output = self.app.get("/fork/pingou/test/diff/master..feature_foo")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - fork/pingou/test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )
            # Check that we prefilled the input fields as expected:
            self.assertIn(
                '<input class="form-control" id="title" name="title" '
                'placeholder="Pull Request Title" required="required" '
                'type="text" value="New edition on side branch of the file '
                'sources for testing">',
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="master"><span class="fa fa-random">'
                "</span> master</a>",
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Case 1 - Add an initial comment
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/fork/pingou/test/diff/master..feature_foo",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#2: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)
            self.assertEqual(output_text.count('title="PY C (pingou)"'), 2)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_req_sign_off_view(self, send_email):
        """ Test the new_request_pull endpoint. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        # Enforce Signed-of-by in the repo
        settings = repo.settings
        settings["Enforce_signed-off_commits_in_pull-request"] = True
        repo.settings = settings
        self.session.add(repo)
        self.session.commit()

        set_up_git_repo(
            self.session,
            self.path,
            new_project=fork,
            branch_from="feature",
            mtype="FF",
        )

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):

            output = self.app.get("/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Diff from feature to master - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                "This project enforces the "
                "Signed-off-by statement on all commits",
                output_text,
            )
            self.assertNotIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )
            self.assertNotIn(
                "This repo enforces that "
                "all commits are signed off by their author.",
                output_text,
            )

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_req_sign_off_submit(self, send_email):
        """ Test the new_request_pull endpoint. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        # Enforce Signed-of-by in the repo
        settings = repo.settings
        settings["Enforce_signed-off_commits_in_pull-request"] = True
        repo.settings = settings
        self.session.add(repo)
        self.session.commit()

        set_up_git_repo(
            self.session,
            self.path,
            new_project=fork,
            branch_from="feature",
            mtype="FF",
        )

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):

            output = self.app.get("/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                "This project enforces the "
                "Signed-off-by statement on all commits",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Try to create the PR
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/test/diff/master..feature", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - test\n - "
                "Pagure</title>",
                output_text,
            )
            # Flashed information message
            self.assertIn(
                "This project enforces the "
                "Signed-off-by statement on all commits",
                output_text,
            )
            # Flashed error message
            self.assertIn(
                "This repo enforces that "
                "all commits are signed off by their author.",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )

    @patch("pagure.lib.notify.send_email")
    def test_request_pull_commit_start_stop(self, send_email):
        """ Test the the commit start and stop of brand new PR. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        set_up_git_repo(
            self.session,
            self.path,
            new_project=fork,
            branch_from="feature",
            mtype="FF",
        )

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            output = self.app.get("/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Case 1 - Add an initial comment
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/test/diff/master..feature", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#2: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)

        # Check if commit start and stop have been set for PR#2
        request = pagure.lib.query.search_pull_requests(
            self.session, project_id=1, requestid=2
        )
        self.assertIsNotNone(request.commit_start)
        self.assertIsNotNone(request.commit_stop)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_from_fork_branch(self, send_email):
        """ Test creating a fork to fork PR. """
        send_email.return_value = True

        # Create main repo with some content
        tests.create_projects(self.session)
        tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)
        tests.add_content_git_repo(
            os.path.join(self.path, "repos", "test.git")
        )

        # Create fork repo with more content
        tests.create_projects(
            self.session, is_fork=True, hook_token_suffix="fork"
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "pingou"), bare=True
        )
        tests.add_content_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git")
        )
        tests.add_readme_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git"),
            branch="feature",
        )
        tests.add_readme_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git"),
            branch="random_branch",
        )

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            data = {"csrf_token": self.get_csrf()}

            output = self.app.post(
                "/do_fork/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that Ralph's fork do exist
            output = self.app.get("/fork/pingou/test")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
            )

            # Try opening a pull-request
            output = self.app.get("/fork/pingou/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - "
                "fork/pingou/test\n - Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="master"><span class="fa fa-random">'
                "</span> master</a>",
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="random_branch"><span class="fa fa-random">'
                "</span> random_branch</a>",
                output_text,
            )

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
    def test_new_request_pull_from_fork_fixing_ticket(self):
        """ Test creating a fork to fork PR fixing a ticket. """
        # Create main repo with some content
        tests.create_projects(self.session)
        tests.create_projects_git(os.path.join(self.path, "repos"), bare=True)
        tests.add_content_git_repo(
            os.path.join(self.path, "repos", "test.git")
        )

        # Create fork repo with more content
        tests.create_projects(
            self.session, is_fork=True, hook_token_suffix="fork"
        )
        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "pingou"), bare=True
        )
        tests.add_content_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git")
        )
        tests.add_readme_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git"),
            branch="feature",
        )
        tests.add_readme_git_repo(
            os.path.join(self.path, "repos", "forks", "pingou", "test.git"),
            branch="random_branch",
        )

        # Check relations before we create the PR
        project = pagure.lib.query.get_authorized_project(self.session, "test")
        self.assertEqual(len(project.requests), 0)
        self.assertEqual(len(project.issues), 0)

        # Create issues to link to
        msg = pagure.lib.query.new_issue(
            session=self.session,
            repo=project,
            title="tést íssüé",
            content="We should work on this",
            user="pingou",
        )
        self.session.commit()
        self.assertEqual(msg.title, "tést íssüé")

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            csrf_token = self.get_csrf()
            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that pingou's fork do exist
            output = self.app.get("/fork/pingou/test")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
                prid=2,
            )

            # Try opening a pull-request
            output = self.app.get("/fork/pingou/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - "
                "fork/pingou/test\n - Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="master"><span class="fa fa-random">'
                "</span> master</a>",
                output_text,
            )
            self.assertIn(
                '<a class="dropdown-item branch_from_item pointer" '
                'data-value="random_branch"><span class="fa fa-random">'
                "</span> random_branch</a>",
                output_text,
            )

            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment\n\nFixes #1",
            }

            output = self.app.post(
                "/fork/pingou/test/diff/master..feature",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#3: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertIn(
                "<p>Test Initial Comment</p>\n<p>Fixes <a href", output_text
            )

        project = pagure.lib.query.get_authorized_project(self.session, "test")
        self.assertEqual(len(project.requests), 2)
        self.assertEqual(len(project.requests[0].related_issues), 0)
        self.assertEqual(len(project.requests[1].related_issues), 1)
        self.assertEqual(len(project.issues), 1)
        self.assertEqual(len(project.issues[0].related_prs), 1)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_fork_to_fork_pr_disabled(self, send_email):
        """ Test creating a fork to fork PR. """
        send_email.return_value = True

        self.test_fork_project()

        # Create a 3rd user
        item = pagure.lib.model.User(
            user="ralph",
            fullname="Ralph bar",
            password="ralph_foo",
            default_email="ralph@bar.com",
        )
        self.session.add(item)
        item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com")
        self.session.add(item)
        self.session.commit()

        user = tests.FakeUser()
        user.username = "ralph"
        with tests.user_set(self.app.application, user):
            # Have Ralph fork, foo's fork of test
            output = self.app.get("/fork/foo/test")
            self.assertEqual(output.status_code, 200)

            output = self.app.post("/do_fork/fork/foo/test")
            self.assertEqual(output.status_code, 400)

            csrf_token = self.get_csrf()
            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/fork/foo/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that Ralph's fork do exist
            output = self.app.get("/fork/ralph/test")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
            )

            # Try opening a pull-request
            output = self.app.get("/fork/ralph/test/diff/master..feature")
            self.assertEqual(output.status_code, 404)
            self.assertIn(
                "<p>No pull-request allowed on this project</p>",
                output.get_data(as_text=True),
            )

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_fork_to_fork(self, send_email):
        """ Test creating a fork to fork PR. """
        send_email.return_value = True

        self.test_fork_project()

        # Create a 3rd user
        item = pagure.lib.model.User(
            user="ralph",
            fullname="Ralph bar",
            password="ralph_foo",
            default_email="ralph@bar.com",
        )
        self.session.add(item)
        item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com")
        self.session.add(item)
        self.session.commit()

        user = tests.FakeUser()
        user.username = "ralph"
        with tests.user_set(self.app.application, user):
            # Have Ralph fork, foo's fork of test
            output = self.app.get("/fork/foo/test")
            self.assertEqual(output.status_code, 200)

            output = self.app.post("/do_fork/fork/foo/test")
            self.assertEqual(output.status_code, 400)

            csrf_token = self.get_csrf()
            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/fork/foo/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that Ralph's fork do exist
            output = self.app.get("/fork/ralph/test")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            # Turn on pull-request on the fork
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test", user="foo"
            )
            settings = repo.settings
            settings["pull_requests"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            # Add some content to the parent
            set_up_git_repo(
                self.session,
                self.path,
                new_project=repo,
                branch_from="master",
                mtype="FF",
                name_from=repo.fullname,
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
                prid=2,
                name_from=fork.fullname,
            )

            # Try opening a pull-request
            output = self.app.get("/fork/ralph/test/diff/master..feature")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - fork/ralph/test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Case 1 - Add an initial comment
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/fork/ralph/test/diff/master..feature",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#1: foo bar PR - fork/foo/test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_fork_to_other_fork(self, send_email):
        """ Test creating a PR from fork to a fork of the same family. """
        send_email.return_value = True

        self.test_fork_project()

        # Create a 3rd user
        item = pagure.lib.model.User(
            user="ralph",
            fullname="Ralph bar",
            password="ralph_foo",
            default_email="ralph@bar.com",
        )
        self.session.add(item)
        item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com")
        self.session.add(item)
        self.session.commit()

        user = tests.FakeUser()
        user.username = "ralph"
        with tests.user_set(self.app.application, user):
            csrf_token = self.get_csrf()
            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/test", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that Ralph's fork do exist
            output = self.app.get("/fork/ralph/test")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            # Turn on pull-request on the fork
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test", user="foo"
            )
            settings = repo.settings
            settings["pull_requests"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            # Add some content to the parents
            set_up_git_repo(
                self.session,
                self.path,
                new_project=repo,
                branch_from="master",
                mtype="FF",
            )
            set_up_git_repo(
                self.session,
                self.path,
                new_project=repo,
                branch_from="master",
                mtype="FF",
                name_from=repo.fullname,
                prid=2,
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
                prid=3,
                name_from=fork.fullname,
            )

            # Try opening a pull-request
            output = self.app.get(
                "/fork/ralph/test/diff/master..feature?project_to=fork/foo/test"
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Create new Pull Request for master - fork/ralph/test\n - "
                "Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<input type="submit" class="btn btn-primary" value="Create Pull Request">\n',
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            # Case 1 - Opening PR to fork/foo/test
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/fork/ralph/test/diff/master..feature?project_to=fork/foo/test",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#1: foo bar PR - fork/foo/test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)

            # Case 1 - Opening PR to parent repo, shows project_to works
            output = self.app.post(
                "/fork/ralph/test/diff/master..feature",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>PR#4: foo bar PR - test\n - Pagure</title>",
                output_text,
            )
            self.assertIn("<p>Test Initial Comment</p>", output_text)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_fork_to_other_unrelated_fork(self, send_email):
        """ Test creating a PR from  fork to fork that isn't from the same
        family.
        """
        send_email.return_value = True

        self.test_fork_project()

        # Create a 3rd user
        item = pagure.lib.model.User(
            user="ralph",
            fullname="Ralph bar",
            password="ralph_foo",
            default_email="ralph@bar.com",
        )
        self.session.add(item)
        item = pagure.lib.model.UserEmail(user_id=3, email="ralph@bar.com")
        self.session.add(item)
        self.session.commit()

        user = tests.FakeUser()
        user.username = "ralph"
        with tests.user_set(self.app.application, user):
            csrf_token = self.get_csrf()
            data = {"csrf_token": csrf_token}

            output = self.app.post(
                "/do_fork/test2", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)

            # Check that Ralph's fork do exist
            output = self.app.get("/fork/ralph/test2")
            self.assertEqual(output.status_code, 200)

            tests.create_projects_git(
                os.path.join(self.path, "requests"), bare=True
            )

            # Turn on pull-request on the fork
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test", user="foo"
            )
            settings = repo.settings
            settings["pull_requests"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            # Add some content to the parent
            set_up_git_repo(
                self.session,
                self.path,
                new_project=repo,
                branch_from="master",
                mtype="FF",
                name_from=repo.fullname,
            )

            fork = pagure.lib.query.get_authorized_project(
                self.session, "test2", user="ralph"
            )

            set_up_git_repo(
                self.session,
                self.path,
                new_project=fork,
                branch_from="feature",
                mtype="FF",
                prid=2,
                name_from=fork.fullname,
            )

            # Case 1 - Opening PR to fork/foo/test
            data = {
                "csrf_token": csrf_token,
                "title": "foo bar PR",
                "initial_comment": "Test Initial Comment",
            }

            output = self.app.post(
                "/fork/ralph/test2/diff/master..feature?project_to=fork/foo/test",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 400)
            self.assertIn(
                "<p>fork/foo/test is not part of fork/ralph/test2's "
                "family</p>",
                output.get_data(as_text=True),
            )

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_empty_repo(self, send_email):
        """ Test the new_request_pull endpoint against an empty repo. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.get(
                "/fork/foo/test/diff/master..feature", follow_redirects=True
            )
            self.assertEqual(output.status_code, 400)
            self.assertIn(
                "<p>Fork is empty, there are no commits to create a pull "
                "request with</p>",
                output.get_data(as_text=True),
            )

            output = self.app.get("/test/new_issue")
            csrf_token = self.get_csrf(output=output)

            data = {"csrf_token": csrf_token, "title": "foo bar PR"}

            output = self.app.post(
                "/test/diff/master..feature", data=data, follow_redirects=True
            )
            self.assertEqual(output.status_code, 400)
            self.assertIn(
                "<p>Fork is empty, there are no commits to create a pull "
                "request with</p>",
                output.get_data(as_text=True),
            )

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_new_request_pull_empty_fork(self, send_email):
        """ Test the new_request_pull endpoint against an empty repo. """
        send_email.return_value = True

        self.test_fork_project()

        tests.create_projects_git(
            os.path.join(self.path, "requests"), bare=True
        )

        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        fork = pagure.lib.query.get_authorized_project(
            self.session, "test", user="foo"
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.get(
                "/fork/foo/test/diff/master..master", follow_redirects=True
            )
            self.assertEqual(output.status_code, 400)
            self.assertIn(
                "<p>Fork is empty, there are no commits to create a pull "
                "request with</p>",
                output.get_data(as_text=True),
            )

        shutil.rmtree(newpath)

    @patch("pagure.lib.notify.send_email")
    def test_pull_request_add_comment(self, send_email):
        """ Test the pull_request_add_comment endpoint. """
        send_email.return_value = True

        self.test_request_pull()

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            output = self.app.post("/foo/pull-request/1/comment")
            self.assertEqual(output.status_code, 404)

            output = self.app.post("/test/pull-request/100/comment")
            self.assertEqual(output.status_code, 404)

            output = self.app.post("/test/pull-request/1/comment")
            self.assertEqual(output.status_code, 200)
            self.assertTrue(
                output.get_data(as_text=True).startswith(
                    '\n<section class="add_comment">'
                )
            )

            csrf_token = self.get_csrf(output=output)

            data = {
                "csrf_token": csrf_token,
                "comment": "This look alright but we can do better",
            }
            output = self.app.post(
                "/test/pull-request/1/comment",
                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("Comment added", output_text)
            self.assertEqual(output_text.count('title="PY C (pingou)"'), 2)

            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/comment",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_pull_request_drop_comment(self, send_email):
        """ Test the pull_request_drop_comment endpoint. """
        send_email.return_value = True

        self.test_pull_request_add_comment()
        # Project w/ pull-request
        repo = pagure.lib.query.get_authorized_project(self.session, "test")
        settings = repo.settings
        settings["pull_requests"] = True
        repo.settings = settings
        self.session.add(repo)
        self.session.commit()

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):
            output = self.app.post("/foo/pull-request/1/comment/drop")
            self.assertEqual(output.status_code, 404)

            output = self.app.post("/test/pull-request/100/comment/drop")
            self.assertEqual(output.status_code, 404)

            output = self.app.post(
                "/test/pull-request/1/comment/drop", follow_redirects=True
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            # self.assertIn('href="#comment-1">¶</a>', output_text)
            self.assertIn(
                "<p>This look alright but we can do better</p>", output_text
            )

            csrf_token = self.get_csrf(output=output)

            # Invalid comment id
            data = {"csrf_token": csrf_token, "drop_comment": "10"}
            output = self.app.post(
                "/test/pull-request/1/comment/drop",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

            data["drop_comment"] = "1"
            output = self.app.post(
                "/test/pull-request/1/comment/drop",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 403)

        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Drop comment
            output = self.app.post(
                "/test/pull-request/1/comment/drop",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn("Comment removed", output_text)

            # Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/comment/drop",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_pull_request_edit_comment(self, send_email):
        """ Test the pull request edit comment endpoint """
        send_email.return_value = True

        self.test_request_pull()

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Repo 'foo' does not exist so it is verifying that condition
            output = self.app.post("/foo/pull-request/1/comment/1/edit")
            self.assertEqual(output.status_code, 404)

            # Here no comment is present in the PR so its verifying that condition
            output = self.app.post("/test/pull-request/100/comment/100/edit")
            self.assertEqual(output.status_code, 404)

            output = self.app.post("/test/pull-request/1/comment")
            self.assertEqual(output.status_code, 200)
            # Creating comment to play with
            self.assertTrue(
                output.get_data(as_text=True).startswith(
                    '\n<section class="add_comment">'
                )
            )

            csrf_token = self.get_csrf(output=output)

            data = {
                "csrf_token": csrf_token,
                "comment": "This look alright but we can do better",
            }
            output = self.app.post(
                "/test/pull-request/1/comment",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)
            self.assertIn(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn("Comment added", output_text)
            # Check if the comment is there
            self.assertIn(
                "<p>This look alright but we can do better</p>", output_text
            )
            output = self.app.get("/test/pull-request/1/comment/1/edit")
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)

            self.assertIn('<section class="edit_comment">', output_text)
            # Checking if the comment is there in the update page
            self.assertIn(
                "This look alright but we can do better</textarea>",
                output_text,
            )

            csrf_token = self.get_csrf(output=output)

            data = {
                "csrf_token": csrf_token,
                "update_comment": "This look alright but we can do better than this.",
            }
            output = self.app.post(
                "/test/pull-request/1/comment/1/edit",
                data=data,
                follow_redirects=True,
            )
            output_text = output.get_data(as_text=True)
            # Checking if the comment is updated in the main page
            self.assertIn(
                "<p>This look alright but we can do better than this.</p>",
                output_text,
            )
            self.assertIn(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            # Checking if Edited by User is there or not
            self.assertTrue(
                "<small>Edited just now by pingou </small>" in output_text
                or "<small>Edited seconds ago by pingou </small>"
                in output_text
            )
            self.assertIn("Comment updated", output_text)

            #  Project w/o pull-request
            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()

            output = self.app.post(
                "/test/pull-request/1/comment/edit/1",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

    @patch("pagure.lib.notify.send_email")
    def test_merge_request_pull_FF_w_merge_commit(self, send_email):
        """ Test the merge_request_pull endpoint with a FF PR but with a
        merge commit.
        """
        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(
                '<h4 class="ml-1">\n        <div>\n              '
                '<span class="fa fa-fw text-success fa-arrow-circle-down pt-1"></span>\n              '
                '<span class="text-success '
                'font-weight-bold">#1</span>\n            '
                '<span class="font-weight-bold">\n                  '
                "PR from the feature branch\n",
                output_text,
            )
            self.assertIn(
                'title="View file as of 2a552b">sources</a>', 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 requiring a merge commit
            repo = pagure.lib.query.get_authorized_project(
                self.session, "test"
            )
            settings = repo.settings
            settings["always_merge"] = True
            repo.settings = settings
            self.session.add(repo)
            self.session.commit()

            # Merge
            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("Merge #1 `PR from the feature branch`", output_text)
            self.assertIn("A commit on branch feature", output_text)

            # Check if the closing notification was added
            output = self.app.get("/test/pull-request/1")
            self.assertIn(
                '<span class="text-info font-weight-bold">Merged</span> just now\n'
                "            </span>\n            by\n"
                '            <span title="PY C (pingou)">pingou.</span>\n',
                output.get_data(as_text=True),
            )

    @patch("pagure.lib.notify.send_email")
    def test_internal_endpoint_main_ahead(self, send_email):
        """ Test the new_request_pull endpoint when the main repo is ahead
        of the fork.
        """
        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"
        )

        gitrepo = os.path.join(self.path, "repos", "test.git")
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Make the main repo be ahead of the fork

        # First commit
        newpath = tempfile.mkdtemp(prefix="pagure-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, "testfile"), "w") as stream:
            stream.write("foo\n bar")
        clone_repo.index.add("testfile")
        clone_repo.index.write()

        # Commits the files added
        last_commit = clone_repo.revparse_single("HEAD")
        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 testfile file for testing",
            # binary string representing the tree object ID
            tree,
            # list of binary strings representing parents of the new commit
            [last_commit.oid.hex],
        )

        # Second commit
        with open(os.path.join(repopath, "testfile"), "a") as stream:
            stream.write("\nfoo2\n bar2")
        clone_repo.index.add("testfile")
        clone_repo.index.write()

        # Commits the files added
        last_commit = clone_repo.revparse_single("HEAD")
        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 a second commit to testfile for testing",
            # binary string representing the tree object ID
            tree,
            # list of binary strings representing parents of the new commit
            [last_commit.oid.hex],
        )

        # Third commit
        with open(os.path.join(repopath, "testfile"), "a") as stream:
            stream.write("\nfoo3\n bar3")
        clone_repo.index.add("testfile")
        clone_repo.index.write()

        # Commits the files added
        last_commit = clone_repo.revparse_single("HEAD")
        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 a third commit to testfile for testing",
            # binary string representing the tree object ID
            tree,
            # list of binary strings representing parents of the new commit
            [last_commit.oid.hex],
        )

        refname = "refs/heads/master:refs/heads/master"
        ori_remote = clone_repo.remotes[0]
        PagureRepo.push(ori_remote, refname)

        shutil.rmtree(newpath)

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):

            csrf_token = self.get_csrf()

            output = self.app.post(
                "/pv/pull-request/ready",
                data={"repo": "test", "csrf_token": csrf_token},
            )
            self.assertEqual(output.status_code, 200)
            data = json.loads(output.get_data(as_text=True))
            self.assertEqual(sorted(data.keys()), ["code", "task"])
            self.assertEqual(data["code"], "OK")

    @patch("pagure.lib.notify.send_email")
    def test_fork_edit_file(self, send_email):
        """ Test the fork_edit file endpoint. """

        send_email.return_value = True

        # Git repo not found
        output = self.app.post("fork_edit/test/edit/master/f/sources")
        self.assertEqual(output.status_code, 404)

        tests.create_projects(self.session)
        for folder in ["docs", "tickets", "requests", "repos"]:
            tests.create_projects_git(
                os.path.join(self.path, folder), bare=True
            )

        # User not logged in
        output = self.app.post("fork_edit/test/edit/master/f/sources")
        self.assertEqual(output.status_code, 302)

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Invalid request
            output = self.app.post("fork_edit/test/edit/master/f/source")
            self.assertEqual(output.status_code, 400)

            output = self.app.get("/new/")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "<strong>Create new Project</strong>",
                output.get_data(as_text=True),
            )

            csrf_token = self.get_csrf(output=output)

            data = {"csrf_token": csrf_token}

            # No files can be found since they are not added
            output = self.app.post(
                "fork_edit/test/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):

            data = {"csrf_token": csrf_token}

            # Invalid request
            output = self.app.post(
                "fork_edit/test/edit/master/f/sources", follow_redirects=True
            )
            self.assertEqual(output.status_code, 400)

            # Add content to the repo
            tests.add_content_git_repo(
                os.path.join(pagure.config.config["GIT_FOLDER"], "test.git")
            )

            tests.add_readme_git_repo(
                os.path.join(pagure.config.config["GIT_FOLDER"], "test.git")
            )

            tests.add_binary_git_repo(
                os.path.join(pagure.config.config["GIT_FOLDER"], "test.git"),
                "test.jpg",
            )

            # Check if button exists
            output = self.app.get("/test/blob/master/f/sources")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "Fork and Edit\n                    </button>\n",
                output.get_data(as_text=True),
            )

            # Check fork-edit doesn't show for binary files
            output = self.app.get("/test/blob/master/f/test.jpg")
            self.assertEqual(output.status_code, 200)
            self.assertNotIn(
                "Fork and Edit\n                    </button>\n",
                output.get_data(as_text=True),
            )

            # Check for edit panel
            output = self.app.post(
                "fork_edit/test/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                '<li><a href="/fork/foo/test/tree/master">'
                '<span class="fa fa-random"></span>&nbsp; master</a>'
                '</li><li class="active"><span class="fa fa-file">'
                "</span>&nbsp; sources</li>",
                output_text,
            )
            self.assertIn(
                '<textarea id="textareaCode" name="content">foo\n bar</textarea>',
                output_text,
            )

            # Check for edit panel- Fork already done
            output = self.app.post(
                "fork_edit/test/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn("<title>Edit - test - Pagure</title>", output_text)
            self.assertIn(
                "You had already forked " "this project", output_text
            )
            self.assertIn(
                '<i class="fa fa-code-fork fa-fw"></i> View Upstream',
                output_text,
            )
            self.assertIn(
                '<li><a href="/fork/foo/test/tree/master">'
                '<span class="fa fa-random"></span>&nbsp; master</a>'
                '</li><li class="active"><span class="fa fa-file">'
                "</span>&nbsp; sources</li>",
                output_text,
            )
            self.assertIn(
                '<textarea id="textareaCode" name="content">foo\n bar</textarea>',
                output_text,
            )

            # View what's supposed to be an image
            output = self.app.post(
                "fork_edit/test/edit/master/f/test.jpg",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 400)
            self.assertIn(
                "<p>Cannot edit binary files</p>",
                output.get_data(as_text=True),
            )

        # Check fork-edit shows when user is not logged in
        output = self.app.get("/test/blob/master/f/sources")
        self.assertEqual(output.status_code, 200)
        self.assertIn(
            "Fork and Edit\n                    </button>\n",
            output.get_data(as_text=True),
        )

        # Check if fork-edit shows for different user
        user.username = "pingou"
        with tests.user_set(self.app.application, user):

            # Check if button exists
            output = self.app.get("/test/blob/master/f/sources")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "Edit in your fork\n                    </button>\n",
                output.get_data(as_text=True),
            )

            # Check fork-edit doesn't show for binary
            output = self.app.get("/test/blob/master/f/test.jpg")
            self.assertEqual(output.status_code, 200)
            self.assertNotIn(
                "Edit in your fork\n                    </button>\n",
                output.get_data(as_text=True),
            )

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))
    def test_fork_edit_file_namespace(self):
        """ Test the fork_edit file endpoint on a namespaced project. """

        tests.create_projects(self.session)
        for folder in ["docs", "tickets", "requests", "repos"]:
            tests.create_projects_git(
                os.path.join(self.path, folder), bare=True
            )

        # User not logged in
        output = self.app.post(
            "fork_edit/somenamespace/test3/edit/master/f/sources"
        )
        self.assertEqual(output.status_code, 302)

        user = tests.FakeUser()
        user.username = "pingou"
        with tests.user_set(self.app.application, user):
            # Invalid request
            output = self.app.post(
                "fork_edit/somenamespace/test3/edit/master/f/sources"
            )
            self.assertEqual(output.status_code, 400)

            output = self.app.get("/new/")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "<strong>Create new Project</strong>",
                output.get_data(as_text=True),
            )

            csrf_token = self.get_csrf(output=output)

            data = {"csrf_token": csrf_token}

            # No files can be found since they are not added
            output = self.app.post(
                "fork_edit/somenamespace/test3/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 404)

        user = tests.FakeUser()
        user.username = "foo"
        with tests.user_set(self.app.application, user):

            data = {"csrf_token": csrf_token}

            # Invalid request
            output = self.app.post(
                "fork_edit/somenamespace/test3/edit/master/f/sources",
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 400)

            # Add content to the repo
            tests.add_content_git_repo(
                os.path.join(
                    pagure.config.config["GIT_FOLDER"],
                    "somenamespace",
                    "test3.git",
                )
            )

            tests.add_readme_git_repo(
                os.path.join(
                    pagure.config.config["GIT_FOLDER"],
                    "somenamespace",
                    "test3.git",
                )
            )

            tests.add_binary_git_repo(
                os.path.join(
                    pagure.config.config["GIT_FOLDER"],
                    "somenamespace",
                    "test3.git",
                ),
                "test.jpg",
            )

            # Check if button exists
            output = self.app.get("/somenamespace/test3/blob/master/f/sources")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "Fork and Edit\n                    </button>\n",
                output.get_data(as_text=True),
            )

            # Check fork-edit doesn't show for binary files
            output = self.app.get(
                "/somenamespace/test3/blob/master/f/test.jpg"
            )
            self.assertEqual(output.status_code, 200)
            self.assertNotIn(
                "Fork and Edit\n                    </button>\n",
                output.get_data(as_text=True),
            )

            # Check for edit panel
            output = self.app.post(
                "fork_edit/somenamespace/test3/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Edit - somenamespace/test3 - Pagure</title>",
                output_text,
            )
            self.assertIn(
                '<i class="fa fa-code-fork fa-fw"></i> View Upstream',
                output_text,
            )
            self.assertIn(
                '<li><a href="/fork/foo/somenamespace/test3/tree/master">'
                '<span class="fa fa-random"></span>&nbsp; master</a>'
                '</li><li class="active"><span class="fa fa-file">'
                "</span>&nbsp; sources</li>",
                output_text,
            )
            self.assertIn(
                '<textarea id="textareaCode" name="content">foo\n bar</textarea>',
                output_text,
            )

            # Check for edit panel - while the project was already forked
            output = self.app.post(
                "fork_edit/somenamespace/test3/edit/master/f/sources",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)
            output_text = output.get_data(as_text=True)
            self.assertIn(
                "<title>Edit - somenamespace/test3 - Pagure</title>",
                output_text,
            )
            self.assertIn(
                "You had already forked " "this project", output_text
            )
            self.assertIn(
                '<i class="fa fa-code-fork fa-fw"></i> View Upstream',
                output_text,
            )
            self.assertIn(
                '<li><a href="/fork/foo/somenamespace/test3/tree/master">'
                '<span class="fa fa-random"></span>&nbsp; master</a>'
                '</li><li class="active"><span class="fa fa-file">'
                "</span>&nbsp; sources</li>",
                output_text,
            )
            self.assertIn(
                '<textarea id="textareaCode" name="content">foo\n bar</textarea>',
                output_text,
            )

    @patch("pagure.lib.notify.send_email")
    def test_fork_without_main_repo(self, send_email):
        """ Test the fork without the main repo. """
        send_email.return_value = True

        tests.create_projects(self.session)

        # Create a fork with no parent i.e parent_id = None
        item = pagure.lib.model.Project(
            user_id=2,  # foo
            name="test",
            description="test project #1",
            hook_token="aaabbb",
            is_fork=True,
            parent_id=None,
        )
        self.session.add(item)
        self.session.commit()

        # Get fork project
        project = pagure.lib.query._get_project(self.session, "test", "foo")

        # Pull-requests and issue-trackers are off for forks
        # lib function is not used here so mannually turning them off
        project_settings = project.settings
        project_settings["pull_requests"] = False
        project_settings["issue_tracker"] = False
        project.settings = project_settings
        self.session.add(project)
        self.session.commit()

        tests.create_projects_git(
            os.path.join(self.path, "repos", "forks", "foo"), bare=True
        )

        # Create a git repo to play with
        gitrepo = os.path.join(self.path, "repos", "test.git")
        self.assertFalse(os.path.exists(gitrepo))
        os.makedirs(gitrepo)
        repo = pygit2.init_repository(gitrepo, bare=True)

        # Create a fork of this repo
        newpath = tempfile.mkdtemp(prefix="pagure-fork-test")
        gitrepo = os.path.join(self.path, "repos", "forks", "foo", "test.git")
        new_repo = pygit2.clone_repository(gitrepo, newpath)
        tests.add_content_git_repo(gitrepo)

        # UI test for deleted main
        output = self.app.get("/fork/foo/test")
        self.assertEqual(output.status_code, 200)
        self.assertIn(
            "Forked from a deleted repository", output.get_data(as_text=True)
        )

        # Testing commit endpoint
        output = self.app.get("/fork/foo/test/commits/master")
        self.assertEqual(output.status_code, 200)
        self.assertIn(
            'Commits <span class="badge badge-secondary"> 2</span>\n',
            output.get_data(as_text=True),
        )

        # Test pull-request endpoint
        output = self.app.get("/fork/foo/test/pull-requests")
        self.assertEqual(output.status_code, 404)

        # Test issue-tracker endpoint
        output = self.app.get("/fork/foo/test/issues")
        self.assertEqual(output.status_code, 404)

        shutil.rmtree(newpath)

    def _set_up_for_reaction_test(self):
        self.session.add(
            pagure.lib.model.User(
                user="jdoe",
                fullname="John Doe",
                password=b"password",
                default_email="jdoe@example.com",
            )
        )
        self.session.commit()
        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"
        )
        pagure.lib.query.get_authorized_project(self.session, "test")
        request = pagure.lib.query.search_pull_requests(
            self.session, requestid=1, project_id=1
        )
        pagure.lib.query.add_pull_request_comment(
            self.session,
            request=request,
            commit=None,
            tree_id=None,
            filename=None,
            row=None,
            comment="Hello",
            user="jdoe",
        )
        self.session.commit()

    @patch("pagure.lib.notify.send_email")
    def test_add_reaction(self, send_email):
        """ Test the request_pull endpoint. """
        send_email.return_value = True

        self._set_up_for_reaction_test()

        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),
                "reaction": "Thumbs up",
            }

            output = self.app.post(
                "/test/pull-request/1/comment/1/react",
                data=data,
                follow_redirects=True,
            )
            self.assertEqual(output.status_code, 200)

            # Load the page and check reaction is added.
            output = self.app.get("/test/pull-request/1")
            self.assertEqual(output.status_code, 200)
            self.assertIn(
                "Thumbs up sent by pingou", output.get_data(as_text=True)
            )

    @patch("pagure.lib.notify.send_email")
    def test_add_reaction_unauthenticated(self, send_email):
        """ Test the request_pull endpoint. """
        send_email.return_value = True

        self._set_up_for_reaction_test()

        output = self.app.get("/test/pull-request/1")
        self.assertEqual(output.status_code, 200)

        data = {
            "csrf_token": self.get_csrf(output=output),
            "reaction": "Thumbs down",
        }

        output = self.app.post(
            "/test/pull-request/1/comment/1/react",
            data=data,
            follow_redirects=False,
        )
        # Redirect to login page
        self.assertEqual(output.status_code, 302)
        self.assertIn("/login/", output.headers["Location"])


class TestTicketAccessEditPRMetadata(tests.Modeltests):
    """ Tests that people with ticket access on a project can edit the
    meta-data of a PR """

    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(TestTicketAccessEditPRMetadata, self).setUp()
        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"
        )

        # Add user "foo" to the project "test"
        repo = pagure.lib.query._get_project(self.session, "test")
        msg = pagure.lib.query.add_user_to_project(
            session=self.session,
            project=repo,
            new_user="foo",
            user="pingou",
            access="ticket",
        )
        self.session.commit()
        self.assertEqual(msg, "User added")

    def test_unauth_cannot_view_edit_metadata_ui(self):
        """ Test that unauthenticated users cannot view the edit the
        metadata fields in the UI. """

        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(
            '<i class="fa fa-fw fa-pencil"></i></a>',
            '<a class="btn btn-outline-primary border-0 btn-sm '
            "issue-metadata-display editmetadatatoggle pointer inline-block"
            '"><i class="fa fa-fw fa-pencil"></i></a>',
            output_text,
        )
        self.assertNotIn(
            '<form method="POST" action="/test/pull-request/1/update">',
            output_text,
        )

    def test_admin_can_view_edit_metadata_ui(self):
        """ Test that admin users can view the edit the metadata fields in
        the UI. """

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            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(
                '<i class="fa fa-fw fa-pencil"></i></a>',
                '<a class="btn btn-outline-primary border-0 btn-sm '
                "issue-metadata-display editmetadatatoggle pointer inline-block"
                '"><i class="fa fa-fw fa-pencil"></i></a>',
                output_text,
            )
            self.assertIn(
                '<form method="POST" action="/test/pull-request/1/update">',
                output_text,
            )

    def test_admin_can_edit_metadata_ui(self):
        """ Test that admin users can edit the metadata in the UI. """

        user = tests.FakeUser(username="pingou")
        with tests.user_set(self.app.application, user):
            data = {"csrf_token": self.get_csrf(), "user": "foo"}
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<i class="fa fa-fw fa-pencil"></i></a>',
                '<a class="btn btn-outline-primary border-0 btn-sm '
                "issue-metadata-display editmetadatatoggle pointer inline-block"
                '"><i class="fa fa-fw fa-pencil"></i></a>',
                output_text,
            )
            self.assertIn(
                '<form method="POST" action="/test/pull-request/1/update">',
                output_text,
            )
            self.assertIn(
                '<input value="foo"\n                    name="user" '
                'id="assignee" placeholder="username" >',
                output_text,
            )

    def test_ticket_can_view_edit_metadata_ui(self):
        """ Test that users with ticket access can view the edit the
        metadata fields in the UI. """

        user = tests.FakeUser(username="foo")
        with tests.user_set(self.app.application, user):
            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(
                '<i class="fa fa-fw fa-pencil"></i></a>',
                '<a class="btn btn-outline-primary border-0 btn-sm '
                "issue-metadata-display editmetadatatoggle pointer inline-block"
                '"><i class="fa fa-fw fa-pencil"></i></a>',
                output_text,
            )

            self.assertIn(
                '<form method="POST" action="/test/pull-request/1/update">',
                output_text,
            )

    def test_ticket_can_edit_metadata_ui(self):
        """ Test that users with ticket access can edit the metadata in the
        UI. """

        user = tests.FakeUser(username="foo")
        with tests.user_set(self.app.application, user):
            data = {"csrf_token": self.get_csrf(), "user": "pingou"}
            output = self.app.post(
                "/test/pull-request/1/update", 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(
                '<i class="fa fa-fw fa-pencil"></i></a>',
                '<a class="btn btn-outline-primary border-0 btn-sm '
                "issue-metadata-display editmetadatatoggle pointer inline-block"
                '"><i class="fa fa-fw fa-pencil"></i></a>',
                output_text,
            )
            self.assertIn(
                '<form method="POST" action="/test/pull-request/1/update">',
                output_text,
            )
            self.assertIn(
                '<input value="pingou"\n                    name="user" '
                'id="assignee" placeholder="username" >',
                output_text,
            )


if __name__ == "__main__":
    unittest.main(verbosity=2)