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

"""

 (c) 2018 - Copyright Red Hat Inc

 Authors:

   Pierre-Yves Chibon <pingou@pingoured.fr></pingou@pingoured.fr>

"""

from __future__ import unicode_literals, absolute_import

import json

import os

import re

import shutil

import sys

import tempfile

import time

import unittest

import pygit2

import wtforms

from mock import patch, MagicMock

from bs4 import BeautifulSoup

sys.path.insert(

    0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")

)

import pagure.lib.query

import tests

from pagure.lib.repo import PagureRepo

from pagure.lib.git import _make_signature

class PagureRemotePRtests(tests.Modeltests):

    """ Tests for remote PRs in pagure """

    def setUp(self):

        """ Set up the environment. """

        super(PagureRemotePRtests, self).setUp()

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

        self.old_value = pagure.config.config["REMOTE_GIT_FOLDER"]

        pagure.config.config["REMOTE_GIT_FOLDER"] = os.path.join(

            self.path, "remotes"

        )

    def tearDown(self):

        """ Clear things up. """

        super(PagureRemotePRtests, self).tearDown()

        pagure.config.config["REMOTE_GIT_FOLDER"] = self.old_value

        shutil.rmtree(self.newpath)

    def set_up_git_repo(self, new_project=None, branch_from="feature"):

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

        object.

        """

        # Create a git repo to play with

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

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

        repopath = os.path.join(self.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 = _make_signature("Alice Author", "alice@authors.tld")

        committer = _make_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,

        )

        # time.sleep(1)

        refname = "refs/heads/master:refs/heads/master"

        ori_remote = clone_repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

        first_commit = repo.revparse_single("HEAD")

        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 = _make_signature("Alice Äuthòr", "alice@äuthòrs.tld")

        committer = _make_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)

        # Set the second repo

        new_gitrepo = repopath

        if new_project:

            # Create a new git repo to play with

            new_gitrepo = os.path.join(self.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)

        # 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 = _make_signature("Alice Author", "alice@authors.tld")

        committer = _make_signature("Cecil Committer", "cecil@committers.tld")

        repo.create_commit(

            "refs/heads/%s" % branch_from,

            author,

            committer,

            "A commit on branch %s" % branch_from,

            tree,

            [first_commit.oid.hex],

        )

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

        ori_remote = repo.remotes[0]

        PagureRepo.push(ori_remote, refname)

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    def test_new_remote_pr_unauth(self):

        """ Test creating a new remote PR un-authenticated. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        self.set_up_git_repo()

        # Before

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

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

        # Try creating a remote PR

        output = self.app.get("/test/diff/remote")

        self.assertEqual(output.status_code, 302)

        self.assertIn(

            "You should be redirected automatically to target URL: "

            '

            output.get_data(as_text=True),

        )

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    def test_new_remote_pr_auth(self):

        """ Test creating a new remote PR authenticated. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        self.set_up_git_repo()

        # Before

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

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

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

        # Try creating a remote PR

        user = tests.FakeUser(username="foo")

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

            output = self.app.get("/test/diff/remote")

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                "
New remote pull-request
",

                output.get_data(as_text=True),

            )

            csrf_token = self.get_csrf(output=output)

            with patch(

                "pagure.forms.RemoteRequestPullForm.git_repo.args",

                MagicMock(

                    return_value=(

                        "Git Repo address",

                        [wtforms.validators.DataRequired()],

                    )

                ),

            ):

                data = {

                    "csrf_token": csrf_token,

                    "title": "Remote PR title",

                    "branch_from": "feature",

                    "branch_to": "master",

                    "git_repo": os.path.join(self.newpath, "test"),

                }

                output = self.app.post("/test/diff/remote", data=data)

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn("Create Pull Request\n    \n", output_text)

                self.assertIn('
\n', output_text)

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

                # Not saved yet

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

                project = pagure.lib.query.get_authorized_project(

                    self.session, "test"

                )

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

                data = {

                    "csrf_token": csrf_token,

                    "title": "Remote PR title",

                    "branch_from": "feature",

                    "branch_to": "master",

                    "git_repo": os.path.join(self.newpath, "test"),

                    "confirm": 1,

                }

                self.old_value = pagure.config.config["DISABLE_REMOTE_PR"]

                pagure.config.config["DISABLE_REMOTE_PR"] = True

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 404)

                pagure.config.config["DISABLE_REMOTE_PR"] = self.old_value

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn(

                    '#1',

                    output_text,

                )

                self.assertIn('
\n', output_text)

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

                # Show the filename in the Changes summary

                self.assertIn(

                    '

                )

                self.assertIn(

                    '
'

                    "\n          <small>.gitignore</small>",

                    output_text,

                )

                self.assertIn(

                    '

                )

                self.assertIn(

                    '
'

                    "\n          <small>sources</small>",

                    output_text,

                )

        # Remote PR Created

        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)

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    def test_new_remote_no_title(self):

        """ Test creating a new remote PR authenticated when no title is

        specified. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        self.set_up_git_repo()

        # Before

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

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

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

        # Try creating a remote PR

        user = tests.FakeUser(username="foo")

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

            output = self.app.get("/test/diff/remote")

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                "
New remote pull-request
",

                output.get_data(as_text=True),

            )

            csrf_token = self.get_csrf(output=output)

            with patch(

                "pagure.forms.RemoteRequestPullForm.git_repo.args",

                MagicMock(

                    return_value=(

                        "Git Repo address",

                        [wtforms.validators.DataRequired()],

                    )

                ),

            ):

                data = {

                    "csrf_token": csrf_token,

                    "branch_from": "master",

                    "branch_to": "feature",

                    "git_repo": os.path.join(self.newpath, "test"),

                }

                output = self.app.post("/test/diff/remote", data=data)

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn("
New remote pull-request
", output_text)

                self.assertIn("<option selected="">feature</option>", output_text)

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    def test_new_remote_pr_empty_target(self):

        """ Test creating a new remote PR authenticated against an empty

        git repo. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        # Create empty target git repo

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

        pygit2.init_repository(gitrepo, bare=True)

        # Create git repo we'll pull from

        gitrepo = os.path.join(self.path, "repos", "test_origin.git")

        repo = pygit2.init_repository(gitrepo)

        # Create a file in that git repo

        with open(os.path.join(gitrepo, "sources"), "w") as stream:

            stream.write("foo\n bar")

        repo.index.add("sources")

        repo.index.write()

        prev_commit = []

        # Commits the files added

        tree = repo.index.write_tree()

        author = _make_signature("Alice Author", "alice@authors.tld")

        committer = _make_signature("Cecil Committer", "cecil@committers.tld")

        repo.create_commit(

            "refs/heads/feature",  # 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,

        )

        # Before

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

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

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

        # Try creating a remote PR

        user = tests.FakeUser(username="foo")

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

            output = self.app.get("/test/diff/remote")

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                "
New remote pull-request
",

                output.get_data(as_text=True),

            )

            csrf_token = self.get_csrf(output=output)

            with patch(

                "pagure.forms.RemoteRequestPullForm.git_repo.args",

                MagicMock(

                    return_value=(

                        "Git Repo address",

                        [wtforms.validators.DataRequired()],

                    )

                ),

            ):

                data = {

                    "csrf_token": csrf_token,

                    "title": "Remote PR title",

                    "branch_from": "feature",

                    "branch_to": "master",

                    "git_repo": gitrepo,

                }

                output = self.app.post("/test/diff/remote", data=data)

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn("Create Pull Request\n    \n", output_text)

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

                # Not saved yet

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

                project = pagure.lib.query.get_authorized_project(

                    self.session, "test"

                )

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

                data = {

                    "csrf_token": csrf_token,

                    "title": "Remote PR title",

                    "branch_from": "feature",

                    "branch_to": "master",

                    "git_repo": gitrepo,

                    "confirm": 1,

                }

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn(

                    "<title>PR#1: Remote PR title - test\n - Pagure</title>",

                    output_text,

                )

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

                # Show the filename in the Changes summary

                self.assertIn(

                    '

                )

                self.assertIn(

                    '
'

                    "\n          <small>sources</small>",

                    output_text,

                )

        # Remote PR Created

        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)

        # Check the merge state of the PR

        data = {"csrf_token": csrf_token, "requestid": project.requests[0].uid}

        output = self.app.post("/pv/pull-request/merge", data=data)

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        data = json.loads(output_text)

        self.assertEqual(

            data,

            {

                "code": "FFORWARD",

                "message": "The pull-request can be merged and fast-forwarded",

                "short_code": "Ok",

            },

        )

        user = tests.FakeUser(username="pingou")

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

            # Merge the PR

            data = {"csrf_token": csrf_token}

            output = self.app.post(

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

            )

            output_text = output.get_data(as_text=True)

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                "<title>PR#1: Remote PR title - test\n - Pagure</title>",

                output_text,

            )

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    @patch("pagure.lib.tasks_services.trigger_ci_build")

    def test_new_remote_pr_ci_off(self, trigger_ci):

        """ Test creating a new remote PR when CI is not configured. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        self.set_up_git_repo()

        # Before

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

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

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

        # Create a remote PR

        user = tests.FakeUser(username="foo")

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

            csrf_token = self.get_csrf()

            data = {

                "csrf_token": csrf_token,

                "title": "Remote PR title",

                "branch_from": "feature",

                "branch_to": "master",

                "git_repo": os.path.join(self.newpath, "test"),

            }

            with patch(

                "pagure.forms.RemoteRequestPullForm.git_repo.args",

                MagicMock(

                    return_value=(

                        "Git Repo address",

                        [wtforms.validators.DataRequired()],

                    )

                ),

            ):

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                data["confirm"] = 1

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn(

                    '#1',

                    output_text,

                )

                self.assertIn('
\n', output_text)

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

        # Remote PR Created

        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)

        trigger_ci.assert_not_called()

    @patch("pagure.lib.notify.send_email", MagicMock(return_value=True))

    @patch("pagure.lib.tasks_services.trigger_ci_build")

    def test_new_remote_pr_ci_on(self, trigger_ci):

        """ Test creating a new remote PR when CI is configured. """

        tests.create_projects(self.session)

        tests.create_projects_git(

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

        )

        self.set_up_git_repo()

        # Before

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

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

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

        # Create a remote PR

        user = tests.FakeUser(username="pingou")

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

            csrf_token = self.get_csrf()

            # Activate CI hook

            data = {

                "active_pr": "y",

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

                "ci_job": "test/job",

                "ci_type": "jenkins",

                "csrf_token": csrf_token,

            }

            output = self.app.post(

                "/test/settings/Pagure CI", data=data, follow_redirects=True

            )

            self.assertEqual(output.status_code, 200)

        user = tests.FakeUser(username="foo")

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

            data = {

                "csrf_token": csrf_token,

                "title": "Remote PR title",

                "branch_from": "feature",

                "branch_to": "master",

                "git_repo": os.path.join(self.newpath, "test"),

            }

            # Disables checking the URL pattern for git_repo

            with patch(

                "pagure.forms.RemoteRequestPullForm.git_repo.args",

                MagicMock(

                    return_value=(

                        "Git Repo address",

                        [wtforms.validators.DataRequired()],

                    )

                ),

            ):

                # Do the preview, triggers the cache & all

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                # Confirm the PR creation

                data["confirm"] = 1

                output = self.app.post(

                    "/test/diff/remote", data=data, follow_redirects=True

                )

                self.assertEqual(output.status_code, 200)

                output_text = output.get_data(as_text=True)

                self.assertIn(

                    '#1',

                    output_text,

                )

                self.assertIn('
\n', output_text)

                self.assertIn('
\n', output_text)

                self.assertNotIn(

                    '
\n', output_text

                )

        # Remote PR Created

        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)

        trigger_ci.assert_not_called()

if __name__ == "__main__":

    unittest.main(verbosity=2)