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

"""

 (c) 2015-2017 - Copyright Red Hat Inc

 Authors:

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

"""

__requires__ = ['SQLAlchemy >= 0.8']

import pkg_resources

import datetime

import os

import shutil

import sys

import tempfile

import time

import unittest

import pygit2

from mock import patch, MagicMock

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

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

import pagure

import pagure.lib.git

import tests

from pagure.lib.repo import PagureRepo

class PagureLibGittests(tests.Modeltests):

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

    def setUp(self):

        """ Set up the environnment, ran before every tests. """

        super(PagureLibGittests, self).setUp()

        pagure.lib.git.SESSION = self.session

    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.get_authorized_project(self.session, 'test')

        # Add an user to a project

        # The user will be an admin of the project

        msg = pagure.lib.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

"""

        #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

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

"""

        #print data

        self.assertEqual(data, exp)

        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

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 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 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.get_authorized_project(self.session, 'test')

        # Add two deploy keys (one readonly one push)

        msg1 = pagure.lib.add_deploykey_to_project(

            session=self.session,

            project=repo,

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

            pushaccess=False,

            user='pingou'

        )

        msg2 = pagure.lib.add_deploykey_to_project(

            session=self.session,

            project=repo,

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

            pushaccess=True,

            user='pingou'

        )

        self.session.commit()

        self.assertEqual(msg1, 'Deploy key added')

        self.assertEqual(msg2, 'Deploy 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

"""

        #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.get_authorized_project(self.session, 'test')

        # Add an user to a project

        # The user will be an admin of the project

        msg = pagure.lib.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

"""

        #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.get_authorized_project(self.session, 'test')

        # Add an user to a project

        # The user will be an admin of the project

        msg = pagure.lib.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

"""

        #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.get_authorized_project(self.session, 'test')

        # Add a couple of groups

        # They would be admins

        msg = pagure.lib.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.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.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.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.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

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

"""

        #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.get_authorized_project(self.session, 'test')

        # Add a couple of groups

        # They would be ticketers

        msg = pagure.lib.add_group(

            self.session,

            group_name='sysadmin',

            display_name='sysadmin group',

            description=None,

            group_type='user',

            user='pingou',

            is_admin=False,