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

"""

 (c) 2015-2018 - Copyright Red Hat Inc

 Authors:

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

"""

from __future__ import unicode_literals, absolute_import

import datetime

import unittest

import shutil

import sys

import tempfile

import os

import six

import json

import pygit2

from mock import patch, MagicMock

sys.path.insert(

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

)

import pagure.lib.query

import tests

class PagureFlaskApptests(tests.Modeltests):

    """ Tests for flask app controller of pagure """

    def test_watch_list(self):

        """ Test for watch list of a user """

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

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

            output = self.app.get("/", follow_redirects=True)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
You have no Projects
',

                output_text,

            )

            tests.create_projects(self.session)

            output = self.app.get("/", follow_redirects=True)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
My Projects
',

                output_text,

            )

    def test_view_users(self):

        """ Test the view_users endpoint. """

        output = self.app.get("/users/?page=abc")

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn(

            '
\n      Users '

            '2',

            output_text,

        )

        self.assertIn(

            '\n                  '

            '
pingou',

            output_text,

        )

        self.assertIn(

            '\n                  '

            '
foo',

            output_text,

        )

    @patch.dict("pagure.config.config", {"ITEM_PER_PAGE": 2})

    def test_view_user_repo_cnt(self):

        """ Test the repo counts on the view_user endpoint. """

        tests.create_projects(self.session)

        self.gitrepos = tests.create_projects_git(

            pagure.config.config["GIT_FOLDER"]

        )

        output = self.app.get("/user/pingou")

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn(

            """

                Projects 

                  3

              """,

            output_text,

        )

        self.assertIn(

            """

                Forks 

                  0

              """,

            output_text,

        )

    def test_view_user(self):

        """ Test the view_user endpoint. """

        output = self.app.get("/user/pingou?repopage=abc&forkpage=def")

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn(

            """

                Projects 

                  0

              """,

            output_text,

        )

        self.assertIn(

            """

                Forks 

                  0

              """,

            output_text,

        )

        tests.create_projects(self.session)

        self.gitrepos = tests.create_projects_git(

            pagure.config.config["GIT_FOLDER"]

        )

        output = self.app.get("/user/pingou?repopage=abc&forkpage=def")

        self.assertEqual(output.status_code, 200)

        output_text = output.get_data(as_text=True)

        self.assertIn(

            """

                Projects 

                  3

              """,

            output_text,

        )

        self.assertIn(

            """

                Forks 

                  0

              """,

            output_text,

        )

        self.assertNotIn(

            'page 1 of 2',

            output_text,

        )

    @patch.dict("pagure.config.config", {"ENABLE_UI_NEW_PROJECTS": False})

    def test_new_project_when_turned_off_in_the_ui(self):

        """ Test the new_project endpoint when new project creation is

        not allowed in the UI of this pagure instance. """

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 404)

            data = {"description": "Project #1", "name": "project-1"}

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 404)

    @patch.dict("pagure.config.config", {"ENABLE_UI_NEW_PROJECTS": False})

    def test_new_project_button_when_turned_off_in_the_ui_no_project(self):

        """ Test the index endpoint when new project creation is

        not allowed in the UI of this pagure instance. """

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

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

            output = self.app.get("/", follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
My Projects
',

                output_text,

            )

            # master template

            self.assertNotIn(

                '

                output_text,

            )

            # index_auth template

            self.assertNotIn(

                'title="Create New Project" aria-hidden="true">', output_text

            )

    @patch.dict("pagure.config.config", {"ENABLE_UI_NEW_PROJECTS": False})

    def test_new_project_button_when_turned_off_in_the_ui_w_project(self):

        """ Test the index endpoint when new project creation is

        not allowed in the UI of this pagure instance. """

        tests.create_projects(self.session)

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

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

            output = self.app.get("/", follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
My Projects
',

                output_text,

            )

            # master template

            self.assertNotIn(

                '

                output_text,

            )

            # index_auth template

            self.assertNotIn(

                'title="Create New Project" aria-hidden="true">', output_text

            )

    def test_new_project_with_dot(self):

        """ Test the new_project endpoint when new project contains a dot.

        """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

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

            csrf_token = self.get_csrf()

            data = {

                "description": "Project #1.",

                "name": "project.1",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                "<title>Overview - project.1 - Pagure</title>", output_text

            )

            self.assertIn(

                'project.1',

                output_text,

            )

        # After

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 1)

    def test_new_project_with_plus(self):

        """ Test the new_project endpoint when new project contains a plus sign.

        """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

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

            csrf_token = self.get_csrf()

            data = {

                "description": "Project #1.",

                "name": "project+1",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                "<title>Overview - project+1 - Pagure</title>", output_text

            )

            self.assertTrue(

                'project+1'

                in output_text

                or 'project+1'

                in output_text

            )

        # After

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 1)

    def test_new_project_when_turned_off(self):

        """ Test the new_project endpoint when new project creation is

        not allowed in the pagure instance. """

        # turn the project creation off

        pagure.config.config["ENABLE_NEW_PROJECTS"] = False

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        self.assertFalse(

            os.path.exists(os.path.join(self.path, "repos", "project-1.git"))

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "tickets", "project-1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "docs", "project-1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "requests", "project-1.git")

            )

        )

        user = tests.FakeUser()

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 404)

            # just get the csrf token

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

            output = self.app.get("/new/")

            pagure.config.config["ENABLE_NEW_PROJECTS"] = False

            csrf_token = (

                output.get_data(as_text=True)

                .split('name="csrf_token" type="hidden" value="')[1]

                .split('">')[0]

            )

            data = {"description": "Project #1", "name": "project-1"}

        user.username = "foo"

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

            data["csrf_token"] = csrf_token

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 404)

        # After

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        self.assertFalse(

            os.path.exists(os.path.join(self.path, "repos", "project-1.git"))

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "tickets", "project-1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "docs", "project-1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "requests", "project-1.git")

            )

        )

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

    def test_new_project_mirrored_invalid_url(self):

        """ Test the new_project with a mirrored repo but an invalid URL. """

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {

                "description": "Project #1",

                "name": "project-1",

                "mirrored_from": "abcd",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("<title>New project - Pagure</title>", output_text)

            self.assertIn("Invalid input. ", output_text)

    def test_new_project_mirrored_invalid_sshurl(self):

        """ Test the new_project with a mirrored repo but an invalid

        SSH-like url.

        """

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            csrf_token = self.get_csrf(output=output)

            data = {

                "description": "Project #1",

                "name": "project-1",

                "mirrored_from": "ssh://git@server.org/foo/bar.git",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("<title>New project - Pagure</title>", output_text)

            self.assertIn("Invalid input. ", output_text)

    def test_new_project_mirrored_valid_url(self):

        """ Test the new_project with a mirrored repo with a valid url. """

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '<label for="mirrored_from">Mirror from URL'</label>

                "",

                output_text,

            )

            csrf_token = self.get_csrf(output=output)

            data = {

                "description": "Project #1",

                "name": "project-1",

                "mirrored_from": "https://example.com/foo/bar.git",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                "<title>Overview - project-1 - Pagure</title>", output_text

            )

            self.assertIn(

                "
This repo is brand new and meant to be mirrored from "

                "https://example.com/foo/bar.git !
",

                output_text,

            )

    @patch.dict("pagure.config.config", {"DISABLE_MIRROR_IN": True})

    def test_new_project_mirrored_mirror_disabled(self):

        """ Test the new_project with a mirrored repo when that feature is

        disabled.

        """

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertNotIn(

                '<label for="mirrored_from">Mirror from URL'</label>

                "",

                output_text,

            )

            csrf_token = self.get_csrf(output=output)

            data = {

                "description": "Project #1",

                "name": "project-1",

                "mirrored_from": "https://example.com/foo/bar.git",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("<title>New project - Pagure</title>", output_text)

            self.assertIn(

                " Mirroring in projects has been disabled in "

                "this instance",

                output_text,

            )

    def test_new_project(self):

        """ Test the new_project endpoint. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        self.assertFalse(

            os.path.exists(os.path.join(self.path, "repos", "project#1.git"))

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "tickets", "project#1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "docs", "project#1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "requests", "project#1.git")

            )

        )

        user = tests.FakeUser()

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            csrf_token = output_text.split(

                'name="csrf_token" type="hidden" value="'

            )[1].split('">')[0]

            data = {"description": "Project #1"}

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertIn(

                "<small>\n            This field is required. \n"</small>

                "          ",

                output_text,

            )

            data["name"] = "project-1"

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertNotIn(

                "<small>\n            This field is required. \n"</small>

                "          ",

                output_text,

            )

            data["csrf_token"] = csrf_token

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertIn("No user " ""username" found", output_text)

        user.username = "foo"

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

            data["csrf_token"] = csrf_token

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1', output_text

            )

            self.assertIn("
This repo is brand new!
", output_text)

            self.assertIn(

                "<title>Overview - project-1 - Pagure</title>", output_text

            )

        # After

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 1)

        self.assertTrue(

            os.path.exists(os.path.join(self.path, "repos", "project-1.git"))

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(self.path, "repos", "tickets", "project-1.git")

            )

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(self.path, "repos", "docs", "project-1.git")

            )

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(self.path, "repos", "requests", "project-1.git")

            )

        )

    @patch.dict(

        "pagure.config.config",

        {

            "PAGURE_ADMIN_USERS": ["pingou"],

            "ALLOW_ADMIN_IGNORE_EXISTING_REPOS": True,

        },

    )

    def test_adopt_repos(self):

        """ Test the new_project endpoint with existing git repo. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

        )

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

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

            data = {

                "csrf_token": self.get_csrf(),

                "name": "test",

                "description": "Project #1",

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("The main repo test.git already exists", output_text)

            data["ignore_existing_repos"] = "y"

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Alice Author", output_text)

    @patch.dict(

        "pagure.config.config",

        {"PAGURE_ADMIN_USERS": [], "USERS_IGNORE_EXISTING_REPOS": ["pingou"]},

    )

    def test_adopt_repos_non_admin(self):

        """ Test the new_project endpoint with existing git repo for non-admins. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

        )

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

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

            data = {

                "csrf_token": self.get_csrf(),

                "name": "test",

                "description": "Project #1",

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("The main repo test.git already exists", output_text)

            data["ignore_existing_repos"] = "y"

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Alice Author", output_text)

    @patch.dict(

        "pagure.config.config",

        {"PAGURE_ADMIN_USERS": [], "USERS_IGNORE_EXISTING_REPOS": []},

    )

    def test_adopt_repos_not_allowed(self):

        """ Test the new_project endpoint with existing git repo for no access. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

        )

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

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

            data = {

                "csrf_token": self.get_csrf(),

                "name": "test",

                "description": "Project #1",

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("The main repo test.git already exists", output_text)

            data["ignore_existing_repos"] = "y"

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("The main repo test.git already exists", output_text)

    @patch.dict("pagure.config.config", {"PROJECT_NAME_REGEX": "^1[a-z]*$"})

    def test_new_project_diff_regex(self):

        """ Test the new_project endpoint with a different regex. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

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

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            csrf_token = self.get_csrf(output=output)

            data = {

                "description": "Project #1",

                "name": "project-1",

                "csrf_token": csrf_token,

            }

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("<title>New project - Pagure</title>", output_text)

            self.assertIn("Create new Project", output_text)

            self.assertIn(

                "<small>\n            Invalid input. \n"</small>

                "          ",

                output_text,

            )

    @patch.dict("pagure.config.config", {"PRIVATE_PROJECTS": True})

    def test_new_project_private(self):

        """ Test the new_project endpoint for a private project. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        self.assertFalse(

            os.path.exists(

                os.path.join(self.path, "repos", "foo", "project#1.git")

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "tickets", "foo", "project#1.git"

                )

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "docs", "foo", "project#1.git"

                )

            )

        )

        self.assertFalse(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "requests", "foo", "project#1.git"

                )

            )

        )

        user = tests.FakeUser()

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

            output = self.app.get("/new/")

            self.assertEqual(output.status_code, 200)

            self.assertIn(

                "Create new Project",

                output.get_data(as_text=True),

            )

            csrf_token = self.get_csrf(output=output)

            data = {"description": "Project #1", "private": True}

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertIn(

                "<small>\n            This field is required. \n"</small>

                "          ",

                output_text,

            )

            data["name"] = "project-1"

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertNotIn(

                "<small>\n            This field is required. \n"</small>

                "          ",

                output_text,

            )

            data["csrf_token"] = csrf_token

            output = self.app.post("/new/", data=data)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn("Create new Project", output_text)

            self.assertIn("No user " ""username" found", output_text)

        user.username = "foo"

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

            data["csrf_token"] = csrf_token

            output = self.app.post("/new/", data=data, follow_redirects=True)

            self.assertEqual(output.status_code, 200)

            output_text = output.get_data(as_text=True)

            self.assertIn(

                '
\nProject #1', output_text

            )

            self.assertIn("
This repo is brand new!
", output_text)

            self.assertIn(

                "<title>Overview - foo/project-1 - Pagure</title>", output_text

            )

        # After

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        projects = pagure.lib.query.search_projects(self.session, private=True)

        self.assertEqual(len(projects), 1)

        self.assertTrue(

            os.path.exists(

                os.path.join(self.path, "repos", "foo", "project-1.git")

            )

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "tickets", "foo", "project-1.git"

                )

            )

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "docs", "foo", "project-1.git"

                )

            )

        )

        self.assertTrue(

            os.path.exists(

                os.path.join(

                    self.path, "repos", "requests", "foo", "project-1.git"

                )

            )

        )

    def test_non_ascii_new_project(self):

        """ Test the new_project endpoint with a non-ascii project. """

        # Before

        projects = pagure.lib.query.search_projects(self.session)

        self.assertEqual(len(projects), 0)

        self.assertFalse(

            os.path.exists(os.path.join(self.path, "repos", "project-1.git"))

        )

        self.assertFalse(

            os.path.exists(