rhodecode/tests/models/test_permissions.py

# Copyright (C) 2010-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/

import pytest

from rhodecode.lib.auth import AuthUser
from rhodecode.model.db import (
    RepoGroup, User, UserGroupRepoGroupToPerm, Permission, UserToPerm,
    UserGroupToPerm)
from rhodecode.model.meta import Session
from rhodecode.model.permission import PermissionModel
from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.user import UserModel
from rhodecode.model.user_group import UserGroupModel
from rhodecode.tests.fixtures.rc_fixture import Fixture


fixture = Fixture()


@pytest.fixture()
def repo_name(backend_hg):
    return backend_hg.repo_name


class TestPermissions(object):

    @pytest.fixture(scope='class', autouse=True)
    def default_permissions(self, request, baseapp):
        # recreate default user to get a clean start
        PermissionModel().create_default_user_permissions(
            user=User.DEFAULT_USER, force=True)
        Session().commit()

    @pytest.fixture(autouse=True)
    def prepare_users(self, request):
        # TODO: User creation is a duplicate of test_nofitications, check
        # if that can be unified
        self.u1 = UserModel().create_or_update(
            username='u1', password='qweqwe',
            email='[e-mail]', firstname='u1', lastname='u1'
        )
        self.u2 = UserModel().create_or_update(
            username='u2', password='qweqwe',
            email='[e-mail]', firstname='u2', lastname='u2'
        )
        self.u3 = UserModel().create_or_update(
            username='u3', password='qweqwe',
            email='[e-mail]', firstname='u3', lastname='u3'
        )
        self.anon = User.get_default_user()
        self.a1 = UserModel().create_or_update(
            username='a1', password='qweqwe',
            email='[e-mail]', firstname='a1', lastname='a1',
            admin=True
        )
        Session().commit()

        request.addfinalizer(self.cleanup)

    def cleanup(self):
        if hasattr(self, 'test_repo'):
            RepoModel().delete(repo=self.test_repo)
            Session().commit()

        if hasattr(self, 'g1'):
            RepoGroupModel().delete(self.g1.group_id)
        if hasattr(self, 'g2'):
            RepoGroupModel().delete(self.g2.group_id)
        Session().commit()

        UserModel().delete(self.u1, handle_repos='delete', handle_repo_groups='delete')
        UserModel().delete(self.u2, handle_repos='delete', handle_repo_groups='delete')
        UserModel().delete(self.u3, handle_repos='delete', handle_repo_groups='delete')
        UserModel().delete(self.a1, handle_repos='delete', handle_repo_groups='delete')
        Session().commit()

        if hasattr(self, 'ug1'):
            UserGroupModel().delete(self.ug1, force=True)
            Session().commit()

    def test_default_perms_set(self, repo_name):
        assert repo_perms(self.u1)[repo_name] == 'repository.read'
        new_perm = 'repository.write'
        RepoModel().grant_user_permission(repo=repo_name, user=self.u1,
                                          perm=new_perm)
        Session().commit()
        assert repo_perms(self.u1)[repo_name] == new_perm

    def test_default_admin_perms_set(self, repo_name):
        assert repo_perms(self.a1)[repo_name] == 'repository.admin'
        RepoModel().grant_user_permission(repo=repo_name, user=self.a1,
                                          perm='repository.write')
        Session().commit()
        # cannot really downgrade admins permissions !? they still gets set as
        # admin !
        assert repo_perms(self.a1)[repo_name] == 'repository.admin'

    def test_default_group_perms(self, repo_name):
        self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
        self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)

        assert repo_perms(self.u1)[repo_name] == 'repository.read'
        assert group_perms(self.u1) == {
            'test1': 'group.read', 'test2': 'group.read'}
        assert global_perms(self.u1) == set(
            Permission.DEFAULT_USER_PERMISSIONS)

    def test_default_admin_group_perms(self, repo_name):
        self.g1 = fixture.create_repo_group('test1', skip_if_exists=True)
        self.g2 = fixture.create_repo_group('test2', skip_if_exists=True)

        assert repo_perms(self.a1)[repo_name] == 'repository.admin'
        assert group_perms(self.a1) == {
            'test1': 'group.admin', 'test2': 'group.admin'}

    def test_default_owner_repo_perms(self, backend, user_util, test_repo):
        user = user_util.create_user()
        repo = test_repo('minimal', backend.alias)
        org_owner = repo.user
        assert repo_perms(user)[repo.repo_name] == 'repository.read'

        repo.user = user
        assert repo_perms(user)[repo.repo_name] == 'repository.admin'
        repo.user = org_owner

    def test_default_owner_branch_perms(self, user_util, test_user_group):
        user = user_util.create_user()
        assert branch_perms(user) == {}

    def test_default_owner_repo_group_perms(self, user_util, test_repo_group):
        user = user_util.create_user()
        org_owner = test_repo_group.user

        assert group_perms(user)[test_repo_group.group_name] == 'group.read'

        test_repo_group.user = user
        assert group_perms(user)[test_repo_group.group_name] == 'group.admin'
        test_repo_group.user = org_owner

    def test_default_owner_user_group_perms(self, user_util, test_user_group):
        user = user_util.create_user()
        org_owner = test_user_group.user

        assert user_group_perms(user)[test_user_group.users_group_name] == 'usergroup.read'

        test_user_group.user = user
        assert user_group_perms(user)[test_user_group.users_group_name] == 'usergroup.admin'

        test_user_group.user = org_owner

    def test_propagated_permissions_from_repo_group_to_private_repo(self, repo_name):
        # make group
        self.g1 = fixture.create_repo_group('TOP_LEVEL', skip_if_exists=True)
        # both perms should be read !
        assert group_perms(self.anon) == {
            'TOP_LEVEL': 'group.read'
        }

        # Create repo inside the TOP_LEVEL
        repo_name_in_group = RepoGroup.url_sep().join([self.g1.group_name, 'test_perm_on_private_repo'])
        self.test_repo = fixture.create_repo(name=repo_name_in_group,
                                             repo_type='hg',
                                             repo_group=self.g1,
                                             cur_user=self.u1,)
        assert repo_perms(self.anon)[repo_name_in_group] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_git'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_hg'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_svn'] == 'repository.read'

        # Now change default user permissions
        new_perm = 'repository.write'
        perm_updates = [
            [self.anon.user_id, new_perm, 'user']
        ]
        RepoGroupModel().update_permissions(
            repo_group=self.g1, perm_updates=perm_updates, recursive='all')

        Session().commit()
        assert repo_perms(self.anon)[repo_name_in_group] == new_perm
        assert repo_perms(self.anon)['vcs_test_git'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_hg'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_svn'] == 'repository.read'

        # NOW MARK repo as private
        changes = {
            'repo_private': True
        }
        repo = RepoModel().get_by_repo_name(repo_name_in_group)
        RepoModel().update(repo, **changes)
        Session().commit()

        # Private repo sets 'none' permission for default user
        assert repo_perms(self.anon)[repo_name_in_group] == 'repository.none'
        assert repo_perms(self.anon)['vcs_test_git'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_hg'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_svn'] == 'repository.read'

        # apply same logic of "updated" recursive, but now the anon permissions should be not be impacted
        new_perm = 'repository.write'
        perm_updates = [
            [self.anon.user_id, new_perm, 'user']
        ]
        RepoGroupModel().update_permissions(
            repo_group=self.g1, perm_updates=perm_updates, recursive='all')

        Session().commit()
        assert repo_perms(self.anon)[repo_name_in_group] == 'repository.none'
        assert repo_perms(self.anon)['vcs_test_git'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_hg'] == 'repository.read'
        assert repo_perms(self.anon)['vcs_test_svn'] == 'repository.read'

    def test_propagated_permission_from_users_group_by_explicit_perms_exist(
            self, repo_name):
        # make group
        self.ug1 = fixture.create_user_group('G1')
        UserGroupModel().add_user_to_group(self.ug1, self.u1)

        # set permission to lower
        new_perm = 'repository.none'
        RepoModel().grant_user_permission(
            repo=repo_name, user=self.u1, perm=new_perm)
        Session().commit()
        assert repo_perms(self.u1)[repo_name] == new_perm

        # grant perm for group this should not override permission from user
        # since it has explicitly set
        new_perm_gr = 'repository.write'
        RepoModel().grant_user_group_permission(
            repo=repo_name, group_name=self.ug1, perm=new_perm_gr)
        Session().commit()

        assert repo_perms(self.u1)[repo_name] == new_perm
        assert group_perms(self.u1) == {}

    def test_propagated_permission_from_users_group(self, repo_name):
        # make group
        self.ug1 = fixture.create_user_group('G1')
        UserGroupModel().add_user_to_group(self.ug1, self.u3)

        # grant perm for group
        # this should override default permission from user
        new_perm_gr = 'repository.write'
        RepoModel().grant_user_group_permission(
            repo=repo_name, group_name=self.ug1, perm=new_perm_gr)
        Session().commit()

        assert repo_perms(self.u3)[repo_name] == new_perm_gr
        assert group_perms(self.u3) == {}

    def test_propagated_permission_from_users_group_lower_weight(
            self, repo_name):
        # make group with user
        self.ug1 = fixture.create_user_group('G1')
        UserGroupModel().add_user_to_group(self.ug1, self.u1)

        # set permission to lower
        new_perm_h = 'repository.write'
        RepoModel().grant_user_permission(
            repo=repo_name, user=self.u1, perm=new_perm_h)
        Session().commit()

        assert repo_perms(self.u1)[repo_name] == new_perm_h

        # grant perm for group this should NOT override permission from user
        # since it's lower than granted
        new_perm_l = 'repository.read'
        RepoModel().grant_user_group_permission(
            repo=repo_name, group_name=self.ug1, perm=new_perm_l)
        Session().commit()

        assert repo_perms(self.u1)[repo_name] == new_perm_h
        assert group_perms(self.u1) == {}

    def test_repo_in_group_permissions(self):
        self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
        self.g2 = fixture.create_repo_group('group2', skip_if_exists=True)
        # both perms should be read !
        assert group_perms(self.u1) == \
            {'group1': 'group.read', 'group2': 'group.read'}

        assert group_perms(self.anon) == \
            {'group1': 'group.read', 'group2': 'group.read'}

        # Change perms to none for both groups
        RepoGroupModel().grant_user_permission(
            repo_group=self.g1, user=self.anon, perm='group.none')
        RepoGroupModel().grant_user_permission(
            repo_group=self.g2, user=self.anon, perm='group.none')

        assert group_perms(self.u1) == \
            {'group1': 'group.none', 'group2': 'group.none'}
        assert group_perms(self.anon) == \
            {'group1': 'group.none', 'group2': 'group.none'}

        # add repo to group
        name = RepoGroup.url_sep().join([self.g1.group_name, 'test_perm'])
        self.test_repo = fixture.create_repo(name=name,
                                             repo_type='hg',
                                             repo_group=self.g1,
                                             cur_user=self.u1,)

        assert group_perms(self.u1) == \
            {'group1': 'group.none', 'group2': 'group.none'}
        assert group_perms(self.anon) == \
            {'group1': 'group.none', 'group2': 'group.none'}

        # grant permission for u2 !
        RepoGroupModel().grant_user_permission(
            repo_group=self.g1, user=self.u2, perm='group.read')
        RepoGroupModel().grant_user_permission(
            repo_group=self.g2, user=self.u2, perm='group.read')
        Session().commit()
        assert self.u1 != self.u2

        # u1 and anon should have not change perms while u2 should !
        assert group_perms(self.u1) == \
            {'group1': 'group.none', 'group2': 'group.none'}
        assert group_perms(self.u2) == \
            {'group1': 'group.read', 'group2': 'group.read'}
        assert group_perms(self.anon) == \
            {'group1': 'group.none', 'group2': 'group.none'}

    def test_repo_group_user_as_user_group_member(self):
        # create Group1
        self.g1 = fixture.create_repo_group('group1', skip_if_exists=True)
        assert group_perms(self.anon) == {'group1': 'group.read'}

        # set default permission to none
        RepoGroupModel().grant_user_permission(
            repo_group=self.g1, user=self.anon, perm='group.none')
        Session().commit()

        # make group
        self.ug1 = fixture.create_user_group('G1')
        # add user to group
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
        Session().commit()

        # check if user is in the group
        ug1 = UserGroupModel().get(self.ug1.users_group_id)
        members = [x.user_id for x in ug1.members]
        assert members == [self.u1.user_id]
        # add some user to that group

        # check his permissions
        assert group_perms(self.anon) == {'group1': 'group.none'}
        assert group_perms(self.u1) == {'group1': 'group.none'}

        # grant ug1 read permissions for
        RepoGroupModel().grant_user_group_permission(
            repo_group=self.g1, group_name=self.ug1, perm='group.read')
        Session().commit()

        # check if the
        obj = Session().query(UserGroupRepoGroupToPerm)\
            .filter(UserGroupRepoGroupToPerm.group == self.g1)\
            .filter(UserGroupRepoGroupToPerm.users_group == self.ug1)\
            .scalar()
        assert obj.permission.permission_name == 'group.read'

        assert group_perms(self.anon) == {'group1': 'group.none'}
        assert group_perms(self.u1) == {'group1': 'group.read'}

    def test_inherited_permissions_from_default_on_user_enabled(self):
        # enable fork and create on default user
        _form_result = {
            'default_repo_create': 'hg.create.repository',
            'default_fork_create': 'hg.fork.repository'
        }
        PermissionModel().set_new_user_perms(
            User.get_default_user(), _form_result)
        Session().commit()

        # make sure inherit flag is turned on
        self.u1.inherit_default_permissions = True
        Session().commit()

        # this user will have inherited permissions from default user
        assert global_perms(self.u1) == default_perms()

    def test_inherited_permissions_from_default_on_user_disabled(self):
        # disable fork and create on default user
        _form_result = {
            'default_repo_create': 'hg.create.none',
            'default_fork_create': 'hg.fork.none'
        }
        PermissionModel().set_new_user_perms(
            User.get_default_user(), _form_result)
        Session().commit()

        # make sure inherit flag is turned on
        self.u1.inherit_default_permissions = True
        Session().commit()

        # this user will have inherited permissions from default user
        expected_perms = default_perms(
            added=['hg.create.none', 'hg.fork.none'],
            removed=['hg.create.repository', 'hg.fork.repository'])
        assert global_perms(self.u1) == expected_perms

    def test_non_inherited_permissions_from_default_on_user_enabled(self):
        user_model = UserModel()
        # enable fork and create on default user
        usr = User.DEFAULT_USER
        user_model.revoke_perm(usr, 'hg.create.none')
        user_model.grant_perm(usr, 'hg.create.repository')
        user_model.revoke_perm(usr, 'hg.fork.none')
        user_model.grant_perm(usr, 'hg.fork.repository')

        # disable global perms on specific user
        user_model.revoke_perm(self.u1, 'hg.create.repository')
        user_model.grant_perm(self.u1, 'hg.create.none')
        user_model.revoke_perm(self.u1, 'hg.fork.repository')
        user_model.grant_perm(self.u1, 'hg.fork.none')

        # TODO(marcink): check branch permissions now ?

        # make sure inherit flag is turned off
        self.u1.inherit_default_permissions = False
        Session().commit()

        # this user will have non inherited permissions from he's
        # explicitly set permissions
        assert global_perms(self.u1) == {
            'hg.create.none',
            'hg.fork.none',
            'hg.register.manual_activate',
            'hg.password_reset.enabled',
            'hg.extern_activate.auto',
            'repository.read',
            'group.read',
            'usergroup.read',
            'branch.push_force',
        }

    def test_non_inherited_permissions_from_default_on_user_disabled(self):
        user_model = UserModel()
        # disable fork and create on default user
        usr = User.DEFAULT_USER
        user_model.revoke_perm(usr, 'hg.create.repository')
        user_model.grant_perm(usr, 'hg.create.none')
        user_model.revoke_perm(usr, 'hg.fork.repository')
        user_model.grant_perm(usr, 'hg.fork.none')

        # enable global perms on specific user
        user_model.revoke_perm(self.u1, 'hg.create.none')
        user_model.grant_perm(self.u1, 'hg.create.repository')
        user_model.revoke_perm(self.u1, 'hg.fork.none')
        user_model.grant_perm(self.u1, 'hg.fork.repository')

        # make sure inherit flag is turned off
        self.u1.inherit_default_permissions = False
        Session().commit()

        # TODO(marcink): check branch perms

        # this user will have non inherited permissions from he's
        # explicitly set permissions
        assert global_perms(self.u1) == {
            'hg.create.repository',
            'hg.fork.repository',
            'hg.register.manual_activate',
            'hg.password_reset.enabled',
            'hg.extern_activate.auto',
            'repository.read',
            'group.read',
            'usergroup.read',
            'branch.push_force',
        }

    @pytest.mark.parametrize('perm, expected_perm', [
        ('hg.inherit_default_perms.false', 'repository.none', ),
        ('hg.inherit_default_perms.true', 'repository.read', ),
    ])
    def test_inherited_permissions_on_objects(self, perm, expected_perm):
        _form_result = {
            'default_inherit_default_permissions': perm,
        }
        PermissionModel().set_new_user_perms(
            User.get_default_user(), _form_result)
        Session().commit()

        # make sure inherit flag is turned on
        self.u1.inherit_default_permissions = True
        Session().commit()

        # TODO(marcink): check branch perms

        # this user will have inherited permissions from default user
        assert global_perms(self.u1) == {
            'hg.create.none',
            'hg.fork.none',
            'hg.register.manual_activate',
            'hg.password_reset.enabled',
            'hg.extern_activate.auto',
            'repository.read',
            'group.read',
            'usergroup.read',
            'branch.push_force',
            'hg.create.write_on_repogroup.true',
            'hg.usergroup.create.false',
            'hg.repogroup.create.false',
            perm
        }

        assert set(repo_perms(self.u1).values()) == set([expected_perm])

    def test_repo_owner_permissions_not_overwritten_by_group(self):
        # create repo as USER,
        self.test_repo = fixture.create_repo(name='myownrepo',
                                             repo_type='hg',
                                             cur_user=self.u1)

        # he has permissions of admin as owner
        assert repo_perms(self.u1)['myownrepo'] == 'repository.admin'

        # set his permission as user group, he should still be admin
        self.ug1 = fixture.create_user_group('G1')
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
        RepoModel().grant_user_group_permission(
            self.test_repo,
            group_name=self.ug1,
            perm='repository.none')
        Session().commit()

        assert repo_perms(self.u1)['myownrepo'] == 'repository.admin'

    def test_repo_owner_permissions_not_overwritten_by_others(self):
        # create repo as USER,
        self.test_repo = fixture.create_repo(name='myownrepo',
                                             repo_type='hg',
                                             cur_user=self.u1)

        # he has permissions of admin as owner
        assert repo_perms(self.u1)['myownrepo'] == 'repository.admin'

        # set his permission as user, he should still be admin
        RepoModel().grant_user_permission(
            self.test_repo, user=self.u1, perm='repository.none')
        Session().commit()

        assert repo_perms(self.u1)['myownrepo'] == 'repository.admin'

    def test_repo_group_owner_permissions_not_overwritten_by_group(self):
        # "u1" shall be owner without any special permission assigned
        self.g1 = fixture.create_repo_group('test1')

        # Make user group and grant a permission to user group
        self.ug1 = fixture.create_user_group('G1')
        UserGroupModel().add_user_to_group(self.ug1, self.u1)
        RepoGroupModel().grant_user_group_permission(
            repo_group=self.g1, group_name=self.ug1, perm='group.write')
        Session().commit()

        # Verify that user does not get any special permission if he is not
        # owner
        assert group_perms(self.u1) == {'test1': 'group.write'}

        # Make him owner of the repo group
        self.g1.user = self.u1
        assert group_perms(self.u1) == {'test1': 'group.admin'}

    def test_repo_group_owner_permissions_not_overwritten_by_others(self):
        # "u1" shall be owner without any special permission assigned
        self.g1 = fixture.create_repo_group('test1')
        RepoGroupModel().grant_user_permission(
            repo_group=self.g1, user=self.u1, perm='group.write')
        Session().commit()

        # Verify that user does not get any special permission if he is not
        # owner
        assert group_perms(self.u1) == {'test1': 'group.write'}

        # Make him owner of the repo group
        self.g1.user = self.u1
        assert group_perms(self.u1) == {'test1': 'group.admin'}

    def assert_user_perm_equal(
            self, user, change_factor=0, compare_keys=None):
        perms = UserToPerm.query().filter(UserToPerm.user == user).all()
        assert len(perms) == \
            len(Permission.DEFAULT_USER_PERMISSIONS) + change_factor
        if compare_keys:
            assert set(
                x.permissions.permission_name for x in perms) == compare_keys

    def assert_def_user_group_perm_equal(
            self, user_group, change_factor=0, compare_keys=None):
        perms = UserGroupToPerm.query().filter(
            UserGroupToPerm.users_group == user_group).all()
        assert len(perms) == \
            len(Permission.DEFAULT_USER_PERMISSIONS) + change_factor
        if compare_keys:
            assert set(
                x.permissions.permission_name for x in perms) == compare_keys

    def test_set_default_permissions(self):
        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)

    def test_set_default_permissions_after_one_is_missing(self):
        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)
        # now we delete one, it should be re-created after another call
        perms = UserToPerm.query().filter(UserToPerm.user == self.u1).all()
        Session().delete(perms[0])
        Session().commit()

        self.assert_user_perm_equal(user=self.u1, change_factor=-1)

        # create missing one !
        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)

    @pytest.mark.parametrize("perm, modify_to", [
        ('repository.read', 'repository.none'),
        ('group.read', 'group.none'),
        ('usergroup.read', 'usergroup.none'),
        ('hg.create.repository', 'hg.create.none'),
        ('hg.fork.repository', 'hg.fork.none'),
        ('hg.register.manual_activate', 'hg.register.auto_activate',)
    ])
    def test_set_default_permissions_after_modification(self, perm, modify_to):
        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)

        old = Permission.get_by_key(perm)
        new = Permission.get_by_key(modify_to)
        assert old is not None
        assert new is not None

        # now modify permissions
        p = UserToPerm.query().filter(
            UserToPerm.user == self.u1).filter(
            UserToPerm.permission == old).one()
        p.permission = new
        Session().add(p)
        Session().commit()

        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)

    def test_clear_user_perms(self):
        PermissionModel().create_default_user_permissions(user=self.u1)
        self.assert_user_perm_equal(user=self.u1)

        # now clear permissions
        cleared = PermissionModel()._clear_user_perms(self.u1.user_id)
        self.assert_user_perm_equal(user=self.u1,
                                    change_factor=len(cleared)*-1)

    def test_clear_user_group_perms(self):
        self.ug1 = fixture.create_user_group('G1')
        PermissionModel().create_default_user_group_permissions(
            user_group=self.ug1)
        self.assert_def_user_group_perm_equal(user_group=self.ug1)

        # now clear permissions
        cleared = PermissionModel()._clear_user_group_perms(
            self.ug1.users_group_id)
        self.assert_def_user_group_perm_equal(user_group=self.ug1,
                                              change_factor=len(cleared)*-1)

    @pytest.mark.parametrize("form_result", [
        {},
        {'default_repo_create': 'hg.create.repository'},
        {'default_repo_create': 'hg.create.repository',
         'default_repo_perm': 'repository.read'},
        {'default_repo_create': 'hg.create.none',
         'default_repo_perm': 'repository.write',
         'default_fork_create': 'hg.fork.none'},
    ])
    def test_set_new_user_permissions(self, form_result):
        _form_result = {}
        _form_result.update(form_result)
        PermissionModel().set_new_user_perms(self.u1, _form_result)
        Session().commit()
        change_factor = -1 * (len(Permission.DEFAULT_USER_PERMISSIONS)
                              - len(form_result.keys()))
        self.assert_user_perm_equal(
            self.u1, change_factor=change_factor)

    @pytest.mark.parametrize("form_result", [
        {},
        {'default_repo_create': 'hg.create.repository'},
        {'default_repo_create': 'hg.create.repository',
         'default_repo_perm': 'repository.read'},
        {'default_repo_create': 'hg.create.none',
         'default_repo_perm': 'repository.write',
         'default_fork_create': 'hg.fork.none'},
    ])
    def test_set_new_user_group_permissions(self, form_result):
        _form_result = {}
        _form_result.update(form_result)
        self.ug1 = fixture.create_user_group('G1')
        PermissionModel().set_new_user_group_perms(self.ug1, _form_result)
        Session().commit()
        change_factor = -1 * (len(Permission.DEFAULT_USER_PERMISSIONS)
                              - len(form_result.keys()))
        self.assert_def_user_group_perm_equal(
            self.ug1, change_factor=change_factor)

    @pytest.mark.parametrize("group_active, expected_perm", [
        (True, 'repository.admin'),
        (False, 'repository.read'),
    ])
    def test_get_default_repo_perms_from_user_group_with_active_group(
            self, backend, user_util, group_active, expected_perm):
        repo = backend.create_repo()
        user = user_util.create_user()
        user_group = user_util.create_user_group(
            members=[user], users_group_active=group_active)

        user_util.grant_user_group_permission_to_repo(
            repo, user_group, 'repository.admin')
        permissions = repo_perms(user)
        repo_permission = permissions.get(repo.repo_name)
        assert repo_permission == expected_perm

    @pytest.mark.parametrize("group_active, expected_perm", [
        (True, 'group.admin'),
        (False, 'group.read')
    ])
    def test_get_default_group_perms_from_user_group_with_active_group(
            self, user_util, group_active, expected_perm):
        user = user_util.create_user()
        repo_group = user_util.create_repo_group()
        user_group = user_util.create_user_group(
            members=[user], users_group_active=group_active)

        user_util.grant_user_group_permission_to_repo_group(
            repo_group, user_group, 'group.admin')
        permissions = group_perms(user)
        group_permission = permissions.get(repo_group.name)
        assert group_permission == expected_perm

    @pytest.mark.parametrize("group_active, expected_perm", [
        (True, 'usergroup.admin'),
        (False, 'usergroup.read')
    ])
    def test_get_default_user_group_perms_from_user_group_with_active_group(
            self, user_util, group_active, expected_perm):
        user = user_util.create_user()
        user_group = user_util.create_user_group(
            members=[user], users_group_active=group_active)
        target_user_group = user_util.create_user_group()

        user_util.grant_user_group_permission_to_user_group(
            target_user_group, user_group, 'usergroup.admin')
        permissions = user_group_perms(user)
        group_permission = permissions.get(target_user_group.users_group_name)
        assert group_permission == expected_perm


def repo_perms(user):
    auth_user = AuthUser(user_id=user.user_id)
    return auth_user.permissions['repositories']


def branch_perms(user):
    auth_user = AuthUser(user_id=user.user_id)
    return auth_user.permissions['repository_branches']


def group_perms(user):
    auth_user = AuthUser(user_id=user.user_id)
    return auth_user.permissions['repositories_groups']


def user_group_perms(user):
    auth_user = AuthUser(user_id=user.user_id)
    return auth_user.permissions['user_groups']


def global_perms(user):
    auth_user = AuthUser(user_id=user.user_id)
    return auth_user.permissions['global']


def default_perms(added=None, removed=None):
    expected_perms = set(Permission.DEFAULT_USER_PERMISSIONS)
    if removed:
        expected_perms.difference_update(removed)
    if added:
        expected_perms.update(added)
    return expected_perms