# -*- 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 os

import shutil

import sys

import tempfile

import time

import unittest

import pygit2

import six

from mock import patch, MagicMock

sys.path.insert(

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

)

import pagure.lib.git

import tests

from pagure.lib.repo import PagureRepo

class PagureLibGittests(tests.Modeltests):

    """ Tests for pagure.lib.git """

    def test_write_gitolite_acls(self):

        """ Test the write_gitolite_acls function of pagure.lib.git.

        when the new uesr is an made an admin """

        tests.create_projects(self.session)

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

        # Add an user to a project

        # The user will be an admin of the project

        msg = pagure.lib.query.add_user_to_project(

            session=self.session, project=repo, new_user="foo", user="pingou"

        )

        self.session.commit()

        self.assertEqual(msg, "User added")

        # Add a forked project

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name="test3",

            description="test project #2",

            is_fork=True,

            parent_id=1,

            hook_token="aaabbbvvv",

        )

        self.session.add(item)

        self.session.commit()

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(

            self.session, configfile=outputconf, project=-1

        )

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """repo test

  R   = @all

  RW+ = pingou

  RW+ = foo

repo docs/test

  R   = @all

  RW+ = pingou

  RW+ = foo

repo tickets/test

  RW+ = pingou

  RW+ = foo

repo requests/test

  RW+ = pingou

  RW+ = foo

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

repo forks/pingou/test3

  R   = @all

  RW+ = pingou

repo docs/forks/pingou/test3

  R   = @all

  RW+ = pingou

repo tickets/forks/pingou/test3

  RW+ = pingou

repo requests/forks/pingou/test3

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_preconf(self):

        """ Test the write_gitolite_acls function of pagure.lib.git with

        a preconf set """

        tests.create_projects(self.session)

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        preconf = os.path.join(self.path, "header_gitolite")

        with open(preconf, "w") as stream:

            stream.write("# this is a header that is manually added")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(

            self.session, outputconf, project=-1, preconf=preconf

        )

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """# this is a header that is manually added

# end of header

repo test

  R   = @all

  RW+ = pingou

repo docs/test

  R   = @all

  RW+ = pingou

repo tickets/test

  RW+ = pingou

repo requests/test

  RW+ = pingou

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data.split("\n"), exp.split("\n"))

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_preconf_postconf(self):

        """ Test the write_gitolite_acls function of pagure.lib.git with

        a postconf set """

        tests.create_projects(self.session)

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        preconf = os.path.join(self.path, "header_gitolite")

        with open(preconf, "w") as stream:

            stream.write("# this is a header that is manually added")

        postconf = os.path.join(self.path, "footer_gitolite")

        with open(postconf, "w") as stream:

            stream.write("# end of generated configuration")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(

            self.session,

            outputconf,

            project=-1,

            preconf=preconf,

            postconf=postconf,

        )

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """# this is a header that is manually added

# end of header

repo test

  R   = @all

  RW+ = pingou

repo docs/test

  R   = @all

  RW+ = pingou

repo tickets/test

  RW+ = pingou

repo requests/test

  RW+ = pingou

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

# end of body

# end of generated configuration

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_postconf(self):

        """ Test the write_gitolite_acls function of pagure.lib.git with

        a preconf and a postconf set """

        tests.create_projects(self.session)

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        postconf = os.path.join(self.path, "footer_gitolite")

        with open(postconf, "w") as stream:

            stream.write("# end of generated configuration")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(

            self.session, outputconf, project=-1, postconf=postconf

        )

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """repo test

  R   = @all

  RW+ = pingou

repo docs/test

  R   = @all

  RW+ = pingou

repo tickets/test

  RW+ = pingou

repo requests/test

  RW+ = pingou

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

# end of body

# end of generated configuration

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_deploykeys(self):

        """ Test write_gitolite_acls function to add deploy keys. """

        tests.create_projects(self.session)

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

        # Add two deploy keys (one readonly one push)

        pingou = pagure.lib.query.get_user(self.session, "pingou")

        msg1 = pagure.lib.query.add_sshkey_to_project_or_user(

            session=self.session,

            project=repo,

            ssh_key="ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQDAzBMSIlvPRaEiLOTVInErkRIw9CzQQcnslDekAn1jFnGf+SNa1acvbTiATbCX71AA03giKrPxPH79dxcC7aDXerc6zRcKjJs6MAL9PrCjnbyxCKXRNNZU5U9X/DLaaL1b3caB+WD6OoorhS3LTEtKPX8xyjOzhf3OQSzNjhJp5Q==",

            pushaccess=False,

            creator=pingou,

        )

        msg2 = pagure.lib.query.add_sshkey_to_project_or_user(

            session=self.session,

            project=repo,

            ssh_key="ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC9Xwc2RDzPBhlEDARfHldGjudIVoa04tqT1JVKGQmyllTFz7Rb8CngQL3e7zyNzotnhwYKHdoiLlPkVEiDee4dWMUe48ilqId+FJZQGhyv8fu4BoFdE1AJUVylzmltbLg14VqG5gjTpXgtlrEva9arKwBMHJjRYc8ScaSn3OgyQw==",

            pushaccess=True,

            creator=pingou,

        )

        self.session.commit()

        self.assertEqual(msg1, "SSH key added")

        self.assertEqual(msg2, "SSH key added")

        # Add a forked project

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name="test3",

            description="test project #2",

            is_fork=True,

            parent_id=1,

            hook_token="aaabbbvvv",

        )

        self.session.add(item)

        self.session.commit()

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(self.session, outputconf, project=-1)

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """repo test

  R   = @all

  RW+ = pingou

  R = deploykey_test_1

  RW+ = deploykey_test_2

repo docs/test

  R   = @all

  RW+ = pingou

  R = deploykey_test_1

  RW+ = deploykey_test_2

repo tickets/test

  RW+ = pingou

  R = deploykey_test_1

  RW+ = deploykey_test_2

repo requests/test

  RW+ = pingou

  R = deploykey_test_1

  RW+ = deploykey_test_2

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

repo forks/pingou/test3

  R   = @all

  RW+ = pingou

repo docs/forks/pingou/test3

  R   = @all

  RW+ = pingou

repo tickets/forks/pingou/test3

  RW+ = pingou

repo requests/forks/pingou/test3

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_ticket(self):

        """ Test the write_gitolite_acls function of pagure.lib.git.

        when the new uesr is just a ticketer """

        tests.create_projects(self.session)

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

        # Add an user to a project

        # The user will be an admin of the project

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

        # Add a forked project

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name="test3",

            description="test project #2",

            is_fork=True,

            parent_id=1,

            hook_token="aaabbbvvv",

        )

        self.session.add(item)

        self.session.commit()

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(self.session, outputconf, project=-1)

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """repo test

  R   = @all

  RW+ = pingou

repo docs/test

  R   = @all

  RW+ = pingou

repo tickets/test

  RW+ = pingou

repo requests/test

  RW+ = pingou

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

repo forks/pingou/test3

  R   = @all

  RW+ = pingou

repo docs/forks/pingou/test3

  R   = @all

  RW+ = pingou

repo tickets/forks/pingou/test3

  RW+ = pingou

repo requests/forks/pingou/test3

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_commit(self):

        """ Test the write_gitolite_acls function of pagure.lib.git.

        when the new uesr is just a committer """

        tests.create_projects(self.session)

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

        # Add an user to a project

        # The user will be an admin of the project

        msg = pagure.lib.query.add_user_to_project(

            session=self.session,

            project=repo,

            new_user="foo",

            user="pingou",

            access="commit",

        )

        self.session.commit()

        self.assertEqual(msg, "User added")

        # Add a forked project

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name="test3",

            description="test project #2",

            is_fork=True,

            parent_id=1,

            hook_token="aaabbbvvv",

        )

        self.session.add(item)

        self.session.commit()

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(self.session, outputconf, project=-1)

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """repo test

  R   = @all

  RW+ = pingou

  RW+ = foo

repo docs/test

  R   = @all

  RW+ = pingou

  RW+ = foo

repo tickets/test

  RW+ = pingou

  RW+ = foo

repo requests/test

  RW+ = pingou

  RW+ = foo

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

repo forks/pingou/test3

  R   = @all

  RW+ = pingou

repo docs/forks/pingou/test3

  R   = @all

  RW+ = pingou

repo tickets/forks/pingou/test3

  RW+ = pingou

repo requests/forks/pingou/test3

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data, exp)

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_groups(self):

        """ Test the write_gitolite_acls function of pagure.lib.git with

        groups as admin

        """

        tests.create_projects(self.session)

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

        # Add a couple of groups

        # They would be admins

        msg = pagure.lib.query.add_group(

            self.session,

            group_name="sysadmin",

            display_name="sysadmin group",

            description=None,

            group_type="user",

            user="pingou",

            is_admin=False,

            blacklist=[],

        )

        self.session.commit()

        self.assertEqual(msg, "User `pingou` added to the group `sysadmin`.")

        msg = pagure.lib.query.add_group(

            self.session,

            group_name="devs",

            display_name="devs group",

            description=None,

            group_type="user",

            user="pingou",

            is_admin=False,

            blacklist=[],

        )

        self.session.commit()

        self.assertEqual(msg, "User `pingou` added to the group `devs`.")

        # Associate these groups to a project

        msg = pagure.lib.query.add_group_to_project(

            session=self.session,

            project=repo,

            new_group="sysadmin",

            user="pingou",

        )

        self.session.commit()

        self.assertEqual(msg, "Group added")

        msg = pagure.lib.query.add_group_to_project(

            session=self.session, project=repo, new_group="devs", user="pingou"

        )

        self.session.commit()

        self.assertEqual(msg, "Group added")

        # Add an user to a project

        msg = pagure.lib.query.add_user_to_project(

            session=self.session, project=repo, new_user="foo", user="pingou"

        )

        self.session.commit()

        self.assertEqual(msg, "User added")

        # Add a forked project

        item = pagure.lib.model.Project(

            user_id=1,  # pingou

            name="test2",

            description="test project #2",

            is_fork=True,

            parent_id=1,

            hook_token="aaabbbvvv",

        )

        self.session.add(item)

        self.session.commit()

        outputconf = os.path.join(self.path, "test_gitolite.conf")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")

        helper.write_gitolite_acls(self.session, outputconf, project=-1)

        self.assertTrue(os.path.exists(outputconf))

        with open(outputconf) as stream:

            data = stream.read()

        exp = """@devs  = pingou

@sysadmin  = pingou

# end of groups

repo test

  R   = @all

  RW+ = @devs @sysadmin

  RW+ = pingou

  RW+ = foo

repo docs/test

  R   = @all

  RW+ = @devs @sysadmin

  RW+ = pingou

  RW+ = foo

repo tickets/test

  RW+ = @devs @sysadmin

  RW+ = pingou

  RW+ = foo

repo requests/test

  RW+ = @devs @sysadmin

  RW+ = pingou

  RW+ = foo

repo test2

  R   = @all

  RW+ = pingou

repo docs/test2

  R   = @all

  RW+ = pingou

repo tickets/test2

  RW+ = pingou

repo requests/test2

  RW+ = pingou

repo somenamespace/test3

  R   = @all

  RW+ = pingou

repo docs/somenamespace/test3

  R   = @all

  RW+ = pingou

repo tickets/somenamespace/test3

  RW+ = pingou

repo requests/somenamespace/test3

  RW+ = pingou

repo forks/pingou/test2

  R   = @all

  RW+ = pingou

repo docs/forks/pingou/test2

  R   = @all

  RW+ = pingou

repo tickets/forks/pingou/test2

  RW+ = pingou

repo requests/forks/pingou/test2

  RW+ = pingou

# end of body

"""

        # print data

        self.assertEqual(data.split("\n"), exp.split("\n"))

        os.unlink(outputconf)

        self.assertFalse(os.path.exists(outputconf))

    def test_write_gitolite_acls_groups_ticket(self):

        """ Test the write_gitolite_acls function of pagure.lib.git with

        groups as ticketers

        """

        tests.create_projects(self.session)

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

        # Add a couple of groups

        # They would be ticketers

        msg = pagure.lib.query.add_group(

            self.session,

            group_name="sysadmin",

            display_name="sysadmin group",

            description=None,

            group_type="user",

            user="pingou",

            is_admin=False,

            blacklist=[],

        )

        self.session.commit()

        self.assertEqual(msg, "User `pingou` added to the group `sysadmin`.")

        msg = pagure.lib.query.add_group(

            self.session,

            group_name="devs",

            display_name="devs group",

            description=None,

            group_type="user",

            user="pingou",

            is_admin=False,

            blacklist=[],

        )

        self.session.commit()

        self.assertEqual(msg, "User `pingou` added to the group `devs`.")