pagure/tests/test_pagure_lib_gitolite_config.py

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

"""
 (c) 2017 - Copyright Red Hat Inc

 Authors:
   Pierre-Yves Chibon <pingou@pingoured.fr>

"""

from __future__ import unicode_literals, absolute_import

import datetime
import os
import shutil
import sys
import tempfile
import time
import unittest
from io import open

import pygit2
from mock import patch, MagicMock

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

import pagure.lib.git
import pagure.lib.query
import tests
from pagure.lib.repo import PagureRepo


CORE_CONFIG = """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"""


class PagureLibGitoliteConfigtests(tests.Modeltests):
    """ Tests for pagure.lib.git """

    maxDiff = None

    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(PagureLibGitoliteConfigtests, self).setUp()

        tests.create_projects(self.session)

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

        self.preconf = os.path.join(self.path, "header_gitolite")
        with open(self.preconf, "w") as stream:
            stream.write("# this is a header that is manually added\n")
            stream.write("\n")
            stream.write("@group1 = foo bar baz\n")
            stream.write("@group2 = threebean puiterwijk kevin pingou\n")

        self.postconf = os.path.join(self.path, "footer_gitolite")
        with open(self.postconf, "w", encoding="utf-8") as stream:
            stream.write("# end of generated configuration\n")
            stream.write(r"# \ó/")
            stream.write("\n# end of footer\n")

    def tearDown(self):
        """ Tearn down the environnment, ran before every tests. """
        super(PagureLibGitoliteConfigtests, self).tearDown()

        if os.path.exists(self.outputconf):
            os.unlink(self.outputconf)
        self.assertFalse(os.path.exists(self.outputconf))

    def test_write_gitolite_pre_post_projectNone(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set """

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=None,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf, "r") as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
%s

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
            % CORE_CONFIG
        )
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_pre_post_projectNone_to_existing_file(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set with existing output file """

        with open(self.outputconf, "w") as stream:
            pass

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=None,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()
        self.assertEqual(data, "")

    def test_write_gitolite_pre_post_project_1(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set """

        with open(self.outputconf, "w") as stream:
            pass

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=-1,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
%s

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
            % CORE_CONFIG
        )

        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_pre_post_project_test(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set """

        with open(self.outputconf, "w") as stream:
            pass

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

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=project,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# 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

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_pre_post_project_test_full_file(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set """

        # Re-generate the gitolite config for all the projects
        self.test_write_gitolite_pre_post_project_1()
        self.assertTrue(os.path.exists(self.outputconf))

        project = pagure.lib.query._get_project(self.session, "test")
        project.user_id = 2
        self.session.add(project)
        self.session.commit()

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

        project = pagure.lib.query._get_project(self.session, "test")
        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=project,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
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 test
  R   = @all
  RW+ = foo
  RW+ = pingou

repo docs/test
  R   = @all
  RW+ = foo
  RW+ = pingou

repo tickets/test
  RW+ = foo
  RW+ = pingou

repo requests/test
  RW+ = foo
  RW+ = pingou

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
        # print data
        self.assertEqual(data, exp)

    @patch.dict(
        "pagure.config.config", {"ENABLE_DOCS": False, "ENABLE_TICKETS": False}
    )
    def test_write_gitolite_disabled_docs_tickets(self):
        """ Test the write_gitolite_acls function when docs and tickets
        are disabled """

        # Re-generate the gitolite config for all the projects
        project = pagure.lib.query._get_project(self.session, "test")
        project.user_id = 2
        self.session.add(project)
        self.session.commit()

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

        project = pagure.lib.query._get_project(self.session, "test")
        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session, self.outputconf, project=project
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

        exp = """repo test
  R   = @all
  RW+ = foo
  RW+ = pingou

repo requests/test
  RW+ = foo
  RW+ = pingou

repo test2
  R   = @all
  RW+ = pingou

repo requests/test2
  RW+ = pingou

repo somenamespace/test3
  R   = @all
  RW+ = pingou

repo requests/somenamespace/test3
  RW+ = pingou

# end of body
"""
        self.assertEqual(data, exp)


class PagureLibGitoliteGroupConfigtests(tests.Modeltests):
    """ Tests for generating the gitolite configuration file for a group
    change

    """

    maxDiff = None

    def setUp(self):
        """ Set up the environnment, ran before every tests. """
        super(PagureLibGitoliteGroupConfigtests, self).setUp()

        tests.create_projects(self.session)

        pagure.lib.query.add_group(
            self.session,
            group_name="grp",
            display_name="grp group",
            description=None,
            group_type="user",
            user="pingou",
            is_admin=False,
            blacklist=[],
        )
        pagure.lib.query.add_group(
            self.session,
            group_name="grp2",
            display_name="grp2 group",
            description=None,
            group_type="user",
            user="foo",
            is_admin=False,
            blacklist=[],
        )
        self.session.commit()

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

        self.preconf = os.path.join(self.path, "header_gitolite")
        with open(self.preconf, "w") as stream:
            stream.write("# this is a header that is manually added\n")
            stream.write("\n")
            stream.write("@group1 = foo bar baz\n")
            stream.write("@group2 = threebean puiterwijk kevin pingou\n")

        self.postconf = os.path.join(self.path, "footer_gitolite")
        with open(self.postconf, "w") as stream:
            stream.write("# end of generated configuration\n")
            stream.write(r"# \ó/")
            stream.write("\n# end of footer\n")

    def tearDown(self):
        """ Tearn down the environnment, ran before every tests. """
        super(PagureLibGitoliteGroupConfigtests, self).tearDown()

        if os.path.exists(self.outputconf):
            os.unlink(self.outputconf)
        self.assertFalse(os.path.exists(self.outputconf))

    def test_write_gitolite_project_test_group(self):
        """ Test the write_gitolite_acls when updating a single group. """

        with open(self.outputconf, "w") as stream:
            pass

        project = pagure.lib.query._get_project(self.session, "test")
        group = pagure.lib.query.search_groups(self.session, group_name="grp")

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=project,
            preconf=self.preconf,
            postconf=self.postconf,
            group=group,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
@grp  = pingou

repo test
  R   = @all
  RW+ = pingou

repo docs/test
  R   = @all
  RW+ = pingou

repo tickets/test
  RW+ = pingou

repo requests/test
  RW+ = pingou

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_project_test_all_groups(self):
        """ Test the write_gitolite_acls when updating all groups. """

        with open(self.outputconf, "w") as stream:
            pass

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

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=project,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
@grp  = pingou
@grp2  = foo
# end of groups

repo test
  R   = @all
  RW+ = pingou

repo docs/test
  R   = @all
  RW+ = pingou

repo tickets/test
  RW+ = pingou

repo requests/test
  RW+ = pingou

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_project_all_projects_groups(self):
        """ Test the generating the entire gitolite config. """

        with open(self.outputconf, "w") as stream:
            pass

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=-1,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
@grp  = pingou
@grp2  = foo
# end of groups

%s

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
            % CORE_CONFIG
        )
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_project_all_projects_one_group(self):
        """ Test the generating the entire gitolite config. """

        # Generate the full gitolite config that we will update
        self.test_write_gitolite_project_all_projects_groups()

        project = pagure.lib.query._get_project(self.session, "test")
        group = pagure.lib.query.search_groups(self.session, group_name="grp")

        # Let's add `foo` to `grp` so something changes
        msg = pagure.lib.query.add_user_to_group(
            self.session,
            username="foo",
            group=group,
            user="pingou",
            is_admin=False,
        )
        self.session.commit()
        self.assertEqual(msg, "User `foo` added to the group `grp`.")

        # Let's add `foo` to `test` so the project changes as well
        msg = pagure.lib.query.add_user_to_project(
            self.session,
            project=project,
            new_user="foo",
            user="pingou",
            access="commit",
        )
        self.assertEqual(msg, "User added")
        self.session.commit()

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=project,
            group=group,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
@grp  = foo pingou
@grp2  = foo
# end of groups

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

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
        # print data
        self.assertEqual(data, exp)

    def test_write_gitolite_delete_group(self):
        """ Test the updating the gitolite config after having
        deleted a group.
        """

        # Generate the full gitolite config that we will update
        self.test_write_gitolite_project_all_projects_groups()

        # Delete the group `grp`
        self.assertEqual(len(pagure.lib.query.search_groups(self.session)), 2)
        group = pagure.lib.query.search_groups(self.session, group_name="grp")
        self.session.delete(group)
        self.session.commit()
        self.assertEqual(len(pagure.lib.query.search_groups(self.session)), 1)

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(
            self.session,
            self.outputconf,
            project=None,
            preconf=self.preconf,
            postconf=self.postconf,
        )
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

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

@group1 = foo bar baz
@group2 = threebean puiterwijk kevin pingou

# end of header
@grp2  = foo
# end of groups

%s

# end of body
# end of generated configuration
# \ó/
# end of footer

"""
            % CORE_CONFIG
        )
        # print data
        self.assertEqual(data, exp)

    @patch("pagure.lib.git_auth.get_git_auth_helper")
    def test_task_generate_gitolite_acls_one_group(self, get_helper):
        """ Test the generate_gitolite_acls task to ensure if group is None
        then None is passed to the helper. """
        helper = MagicMock()
        get_helper.return_value = helper
        pagure.lib.query.SESSIONMAKER = self.session.session_factory

        pagure.lib.tasks.generate_gitolite_acls(
            namespace=None, name="test", user=None, group=None
        )

        get_helper.assert_called_with()
        args = helper.generate_acls.call_args
        self.assertIsNone(args[1].get("group"))
        self.assertIsNotNone(args[1].get("project"))

    def test_write_gitolite_project_test_private(self):
        """ Test the write_gitolite_acls function of pagure.lib.git with
        a postconf set """

        # Make the test project private
        project = pagure.lib.query._get_project(self.session, "test")
        project.private = True
        self.session.add(project)
        self.session.commit()

        # Re-generate the gitolite config just for this project
        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(self.session, self.outputconf, project=None)
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

        exp = """@grp  = pingou
@grp2  = foo
# end of groups

repo test
  RW+ = pingou

repo docs/test
  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, exp)

    def test_remove_acls(self):
        """ Test the remove_acls function of pagure.lib.git when deleting
        a project """
        pagure.config.config["GITOLITE_CONFIG"] = self.outputconf

        with open(self.outputconf, "w") as stream:
            pass

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(self.session, self.outputconf, project=-1)
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

        exp = (
            """@grp  = pingou
@grp2  = foo
# end of groups

%s

# end of body
"""
            % CORE_CONFIG
        )

        # print data
        self.assertEqual(data, exp)

        # Test removing a project from the existing config
        project = pagure.lib.query.get_authorized_project(
            self.session, project_name="test"
        )

        helper.remove_acls(self.session, project=project)

        with open(self.outputconf) as stream:
            data = stream.read()

        exp = """@grp  = pingou
@grp2  = foo
# end of groups

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

    def test_remove_acls_no_project(self):
        """ Test the remove_acls function of pagure.lib.git when no project
        is specified """
        pagure.config.config["GITOLITE_CONFIG"] = self.outputconf

        with open(self.outputconf, "w") as stream:
            pass

        helper = pagure.lib.git_auth.get_git_auth_helper("gitolite3")
        helper.write_gitolite_acls(self.session, self.outputconf, project=-1)
        self.assertTrue(os.path.exists(self.outputconf))

        with open(self.outputconf) as stream:
            data = stream.read()

        exp = (
            """@grp  = pingou
@grp2  = foo
# end of groups

%s

# end of body
"""
            % CORE_CONFIG
        )

        # print data
        self.assertEqual(data, exp)

        # Test nothing changes if no project is specified

        self.assertRaises(
            RuntimeError, helper.remove_acls, self.session, project=None
        )

        with open(self.outputconf) as stream:
            data = stream.read()

        self.assertEqual(data, exp)


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