Changeset - 2181d0058c6a
[Not reviewed]
default
0 2 0
Joseph Rivera - 11 years ago 2015-01-21 15:52:28
rivera.d.joseph@gmail.com
usergroup: more descriptive error message when deleting user group
2 files changed with 4 insertions and 3 deletions:
0 comments (0 inline, 0 general)
kallithea/model/user_group.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# 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 General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.model.users_group
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
user group model for Kallithea
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Oct 1, 2011
 
:author: nvinot, marcink
 
"""
 

	
 

	
 
import logging
 
import traceback
 

	
 
from kallithea.model import BaseModel
 
from kallithea.model.db import UserGroupMember, UserGroup,\
 
    UserGroupRepoToPerm, Permission, UserGroupToPerm, User, UserUserGroupToPerm,\
 
    UserGroupUserGroupToPerm
 
from kallithea.lib.exceptions import UserGroupsAssignedException,\
 
    RepoGroupAssignmentError
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class UserGroupModel(BaseModel):
 

	
 
    cls = UserGroup
 

	
 
    def _get_user_group(self, user_group):
 
        return self._get_instance(UserGroup, user_group,
 
                                  callback=UserGroup.get_by_group_name)
 

	
 
    def _create_default_perms(self, user_group):
 
        # create default permission
 
        default_perm = 'usergroup.read'
 
        def_user = User.get_default_user()
 
        for p in def_user.user_perms:
 
            if p.permission.permission_name.startswith('usergroup.'):
 
                default_perm = p.permission.permission_name
 
                break
 

	
 
        user_group_to_perm = UserUserGroupToPerm()
 
        user_group_to_perm.permission = Permission.get_by_key(default_perm)
 

	
 
        user_group_to_perm.user_group = user_group
 
        user_group_to_perm.user_id = def_user.user_id
 
        return user_group_to_perm
 

	
 
    def _update_permissions(self, user_group, perms_new=None,
 
                            perms_updates=None):
 
        from kallithea.lib.auth import HasUserGroupPermissionAny
 
        if not perms_new:
 
            perms_new = []
 
        if not perms_updates:
 
            perms_updates = []
 

	
 
        # update permissions
 
        for member, perm, member_type in perms_updates:
 
            if member_type == 'user':
 
                # this updates existing one
 
                self.grant_user_permission(
 
                    user_group=user_group, user=member, perm=perm
 
                )
 
            else:
 
                #check if we have permissions to alter this usergroup
 
                if HasUserGroupPermissionAny('usergroup.read', 'usergroup.write',
 
                                             'usergroup.admin')(member):
 
                    self.grant_user_group_permission(
 
                        target_user_group=user_group, user_group=member, perm=perm
 
                    )
 
        # set new permissions
 
        for member, perm, member_type in perms_new:
 
            if member_type == 'user':
 
                self.grant_user_permission(
 
                    user_group=user_group, user=member, perm=perm
 
                )
 
            else:
 
                #check if we have permissions to alter this usergroup
 
                if HasUserGroupPermissionAny('usergroup.read', 'usergroup.write',
 
                                             'usergroup.admin')(member):
 
                    self.grant_user_group_permission(
 
                        target_user_group=user_group, user_group=member, perm=perm
 
                    )
 

	
 
    def get(self, user_group_id, cache=False):
 
        return UserGroup.get(user_group_id)
 

	
 
    def get_group(self, user_group):
 
        return self._get_user_group(user_group)
 

	
 
    def get_by_name(self, name, cache=False, case_insensitive=False):
 
        return UserGroup.get_by_group_name(name, cache, case_insensitive)
 

	
 
    def create(self, name, description, owner, active=True, group_data=None):
 
        try:
 
            new_user_group = UserGroup()
 
            new_user_group.user = self._get_user(owner)
 
            new_user_group.users_group_name = name
 
            new_user_group.user_group_description = description
 
            new_user_group.users_group_active = active
 
            if group_data:
 
                new_user_group.group_data = group_data
 
            self.sa.add(new_user_group)
 
            perm_obj = self._create_default_perms(new_user_group)
 
            self.sa.add(perm_obj)
 

	
 
            self.grant_user_permission(user_group=new_user_group,
 
                                       user=owner, perm='usergroup.admin')
 

	
 
            return new_user_group
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise
 

	
 
    def update(self, user_group, form_data):
 

	
 
        try:
 
            user_group = self._get_user_group(user_group)
 

	
 
            for k, v in form_data.items():
 
                if k == 'users_group_members':
 
                    user_group.members = []
 
                    self.sa.flush()
 
                    members_list = []
 
                    if v:
 
                        v = [v] if isinstance(v, basestring) else v
 
                        for u_id in set(v):
 
                            member = UserGroupMember(user_group.users_group_id, u_id)
 
                            members_list.append(member)
 
                    setattr(user_group, 'members', members_list)
 
                setattr(user_group, k, v)
 

	
 
            self.sa.add(user_group)
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise
 

	
 
    def delete(self, user_group, force=False):
 
        """
 
        Deletes repository group, unless force flag is used
 
        Deletes user group, unless force flag is used
 
        raises exception if there are members in that group, else deletes
 
        group and users
 

	
 
        :param user_group:
 
        :param force:
 
        """
 
        user_group = self._get_user_group(user_group)
 
        try:
 
            # check if this group is not assigned to repo
 
            assigned_groups = UserGroupRepoToPerm.query()\
 
                .filter(UserGroupRepoToPerm.users_group == user_group).all()
 
            assigned_groups = [x.repository.repo_name for x in assigned_groups]
 

	
 
            if assigned_groups and not force:
 
                raise UserGroupsAssignedException(
 
                    'RepoGroup assigned to %s' % assigned_groups)
 
                    'User Group assigned to %s' % ", ".join(assigned_groups))
 
            self.sa.delete(user_group)
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise
 

	
 
    def add_user_to_group(self, user_group, user):
 
        user_group = self._get_user_group(user_group)
 
        user = self._get_user(user)
 

	
 
        for m in user_group.members:
 
            u = m.user
 
            if u.user_id == user.user_id:
 
                # user already in the group, skip
 
                return True
 

	
 
        try:
 
            user_group_member = UserGroupMember()
 
            user_group_member.user = user
 
            user_group_member.users_group = user_group
 

	
 
            user_group.members.append(user_group_member)
 
            user.group_member.append(user_group_member)
 

	
 
            self.sa.add(user_group_member)
 
            return user_group_member
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise
 

	
 
    def remove_user_from_group(self, user_group, user):
 
        user_group = self._get_user_group(user_group)
 
        user = self._get_user(user)
 

	
 
        user_group_member = None
 
        for m in user_group.members:
 
            if m.user.user_id == user.user_id:
 
                # Found this user's membership row
 
                user_group_member = m
 
                break
 

	
 
        if user_group_member:
 
            try:
 
                self.sa.delete(user_group_member)
 
                return True
 
            except Exception:
 
                log.error(traceback.format_exc())
 
                raise
 
        else:
 
            # User isn't in that group
 
            return False
 

	
 
    def has_perm(self, user_group, perm):
 
        user_group = self._get_user_group(user_group)
 
        perm = self._get_perm(perm)
 

	
 
        return UserGroupToPerm.query()\
 
            .filter(UserGroupToPerm.users_group == user_group)\
 
            .filter(UserGroupToPerm.permission == perm).scalar() is not None
 

	
 
    def grant_perm(self, user_group, perm):
 
        user_group = self._get_user_group(user_group)
 
        perm = self._get_perm(perm)
 

	
 
        # if this permission is already granted skip it
 
        _perm = UserGroupToPerm.query()\
 
            .filter(UserGroupToPerm.users_group == user_group)\
 
            .filter(UserGroupToPerm.permission == perm)\
 
            .scalar()
 
        if _perm:
 
            return
 

	
 
        new = UserGroupToPerm()
 
        new.users_group = user_group
 
        new.permission = perm
 
        self.sa.add(new)
 
        return new
 

	
 
    def revokehas_permrevoke_permgrant_perm_perm(self, user_group, perm):
 
        user_group = self._get_user_group(user_group)
 
        perm = self._get_perm(perm)
 

	
 
        obj = UserGroupToPerm.query()\
 
            .filter(UserGroupToPerm.users_group == user_group)\
 
            .filter(UserGroupToPerm.permission == perm).scalar()
 
        if obj:
 
            self.sa.delete(obj)
 

	
 
    def grant_user_permission(self, user_group, user, perm):
 
        """
 
        Grant permission for user on given user group, or update
 
        existing one if found
 

	
 
        :param user_group: Instance of UserGroup, users_group_id,
 
            or users_group_name
 
        :param user: Instance of User, user_id or username
 
        :param perm: Instance of Permission, or permission_name
 
        """
 

	
 
        user_group = self._get_user_group(user_group)
 
        user = self._get_user(user)
 
        permission = self._get_perm(perm)
 

	
 
        # check if we have that permission already
 
        obj = self.sa.query(UserUserGroupToPerm)\
 
            .filter(UserUserGroupToPerm.user == user)\
 
            .filter(UserUserGroupToPerm.user_group == user_group)\
 
            .scalar()
 
        if obj is None:
 
            # create new !
 
            obj = UserUserGroupToPerm()
 
        obj.user_group = user_group
 
        obj.user = user
 
        obj.permission = permission
 
        self.sa.add(obj)
 
        log.debug('Granted perm %s to %s on %s' % (perm, user, user_group))
 
        return obj
 

	
 
    def revoke_user_permission(self, user_group, user):
 
        """
 
        Revoke permission for user on given repository group
 

	
 
        :param user_group: Instance of RepoGroup, repositories_group_id,
 
            or repositories_group name
 
        :param user: Instance of User, user_id or username
 
        """
 

	
 
        user_group = self._get_user_group(user_group)
 
        user = self._get_user(user)
 

	
 
        obj = self.sa.query(UserUserGroupToPerm)\
 
            .filter(UserUserGroupToPerm.user == user)\
 
            .filter(UserUserGroupToPerm.user_group == user_group)\
 
            .scalar()
 
        if obj:
 
            self.sa.delete(obj)
 
            log.debug('Revoked perm on %s on %s' % (user_group, user))
 

	
 
    def grant_user_group_permission(self, target_user_group, user_group, perm):
 
        """
 
        Grant user group permission for given target_user_group
 

	
 
        :param target_user_group:
 
        :param user_group:
 
        :param perm:
 
        """
 
        target_user_group = self._get_user_group(target_user_group)
 
        user_group = self._get_user_group(user_group)
 
        permission = self._get_perm(perm)
 
        # forbid assigning same user group to itself
 
        if target_user_group == user_group:
 
            raise RepoGroupAssignmentError('target repo:%s cannot be '
 
                                           'assigned to itself' % target_user_group)
 

	
 
        # check if we have that permission already
 
        obj = self.sa.query(UserGroupUserGroupToPerm)\
 
            .filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
 
            .filter(UserGroupUserGroupToPerm.user_group == user_group)\
 
            .scalar()
 
        if obj is None:
 
            # create new !
 
            obj = UserGroupUserGroupToPerm()
 
        obj.user_group = user_group
 
        obj.target_user_group = target_user_group
 
        obj.permission = permission
 
        self.sa.add(obj)
 
        log.debug('Granted perm %s to %s on %s' % (perm, target_user_group, user_group))
 
        return obj
 

	
 
    def revoke_user_group_permission(self, target_user_group, user_group):
 
        """
 
        Revoke user group permission for given target_user_group
 

	
 
        :param target_user_group:
 
        :param user_group:
 
        """
 
        target_user_group = self._get_user_group(target_user_group)
 
        user_group = self._get_user_group(user_group)
 

	
 
        obj = self.sa.query(UserGroupUserGroupToPerm)\
 
            .filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
 
            .filter(UserGroupUserGroupToPerm.user_group == user_group)\
 
            .scalar()
 
        if obj:
 
            self.sa.delete(obj)
 
            log.debug('Revoked perm on %s on %s' % (target_user_group, user_group))
 

	
 
    def enforce_groups(self, user, groups, extern_type=None):
 
        user = self._get_user(user)
 
        log.debug('Enforcing groups %s on user %s' % (user, groups))
 
        current_groups = user.group_member
 
        # find the external created groups
 
        externals = [x.users_group for x in current_groups
kallithea/tests/api/api_base.py
Show inline comments
 
@@ -1379,385 +1379,385 @@ class BaseTestApi(object):
 

	
 
    def test_api_get_user_groups(self):
 
        gr_name = 'test_user_group2'
 
        make_user_group(gr_name)
 

	
 
        id_, params = _build_data(self.apikey, 'get_user_groups', )
 
        response = api_call(self, params)
 

	
 
        try:
 
            expected = []
 
            for gr_name in [TEST_USER_GROUP, 'test_user_group2']:
 
                user_group = UserGroupModel().get_group(gr_name)
 
                ret = user_group.get_api_data()
 
                expected.append(ret)
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_create_user_group(self):
 
        group_name = 'some_new_group'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'created new user group `%s`' % group_name,
 
            'user_group': jsonify(UserGroupModel() \
 
                .get_by_name(group_name) \
 
                .get_api_data())
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
        fixture.destroy_user_group(group_name)
 

	
 
    def test_api_get_user_group_that_exist(self):
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=TEST_USER_GROUP)
 
        response = api_call(self, params)
 

	
 
        expected = "user group `%s` already exist" % TEST_USER_GROUP
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'create', crash)
 
    def test_api_get_user_group_exception_occurred(self):
 
        group_name = 'exception_happens'
 
        id_, params = _build_data(self.apikey, 'create_user_group',
 
                                  group_name=group_name)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to create group `%s`' % group_name
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @parameterized.expand([('group_name', {'group_name': 'new_group_name'}),
 
                           ('group_name', {'group_name': 'test_group_for_update'}),
 
                           ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),
 
                           ('active', {'active': False}),
 
                           ('active', {'active': True})])
 
    def test_api_update_user_group(self, changing_attr, updates):
 
        gr_name = 'test_group_for_update'
 
        user_group = fixture.create_user_group(gr_name)
 
        id_, params = _build_data(self.apikey, 'update_user_group',
 
                                  usergroupid=gr_name, **updates)
 
        response = api_call(self, params)
 
        try:
 
            expected = {
 
               'msg': 'updated user group ID:%s %s' % (user_group.users_group_id,
 
                                                     user_group.users_group_name),
 
               'user_group': user_group.get_api_data()
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            if changing_attr == 'group_name':
 
                # switch to updated name for proper cleanup
 
                gr_name = updates['group_name']
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'update', crash)
 
    def test_api_update_user_group_exception_occured(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        id_, params = _build_data(self.apikey, 'update_user_group',
 
                                  usergroupid=gr_name)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'failed to update user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 
        try:
 
            expected = {
 
            'msg': 'added member `%s` to user group `%s`' % (
 
                    TEST_USER_ADMIN_LOGIN, gr_name),
 
            'success': True
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_add_user_to_user_group_that_doesnt_exist(self):
 
        id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                  usergroupid='false-group',
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        expected = 'user group `%s` does not exist' % 'false-group'
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(UserGroupModel, 'add_user_to_group', crash)
 
    def test_api_add_user_to_user_group_exception_occurred(self):
 
        gr_name = 'test_group'
 
        fixture.create_user_group(gr_name)
 
        id_, params = _build_data(self.apikey, 'add_user_to_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        try:
 
            expected = 'failed to add member to user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_remove_user_from_user_group(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 
        id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        try:
 
            expected = {
 
                'msg': 'removed member `%s` from user group `%s`' % (
 
                    TEST_USER_ADMIN_LOGIN, gr_name
 
                ),
 
                'success': True}
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @mock.patch.object(UserGroupModel, 'remove_user_from_group', crash)
 
    def test_api_remove_user_from_user_group_exception_occurred(self):
 
        gr_name = 'test_group_3'
 
        gr = fixture.create_user_group(gr_name)
 
        UserGroupModel().add_user_to_group(gr, user=TEST_USER_ADMIN_LOGIN)
 
        Session().commit()
 
        id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 
        try:
 
            expected = 'failed to remove member from user group `%s`' % gr_name
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        try:
 
            expected = {
 
                'user_group': None,
 
                'msg': 'deleted user group ID:%s %s' % (gr_id, gr_name)
 
            }
 
            self._compare_ok(id_, expected, given=response.body)
 
        finally:
 
            if UserGroupModel().get_by_name(gr_name):
 
                fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group_that_is_assigned(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 

	
 
        ugr_to_perm = RepoModel().grant_user_group_permission(self.REPO, gr_name, 'repository.write')
 
        msg = 'RepoGroup assigned to [%s]' % (ugr_to_perm)
 
        msg = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
 

	
 
        id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 
        response = api_call(self, params)
 

	
 
        try:
 
            expected = msg
 
            self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            if UserGroupModel().get_by_name(gr_name):
 
                fixture.destroy_user_group(gr_name)
 

	
 
    def test_api_delete_user_group_exception_occured(self):
 
        gr_name = 'test_group'
 
        ugroup = fixture.create_user_group(gr_name)
 
        gr_id = ugroup.users_group_id
 
        id_, params = _build_data(self.apikey, 'delete_user_group',
 
                                  usergroupid=gr_name,
 
                                  userid=TEST_USER_ADMIN_LOGIN)
 

	
 
        try:
 
            with mock.patch.object(UserGroupModel, 'delete', crash):
 
                response = api_call(self, params)
 
                expected = 'failed to delete user group ID:%s %s' % (gr_id, gr_name)
 
                self._compare_error(id_, expected, given=response.body)
 
        finally:
 
            fixture.destroy_user_group(gr_name)
 

	
 
    @parameterized.expand([('none', 'repository.none'),
 
                           ('read', 'repository.read'),
 
                           ('write', 'repository.write'),
 
                           ('admin', 'repository.admin')])
 
    def test_api_grant_user_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
 
                perm, TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_permission', crash)
 
    def test_api_grant_user_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=TEST_USER_ADMIN_LOGIN,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_permission(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
 
                TEST_USER_ADMIN_LOGIN, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_permission', crash)
 
    def test_api_revoke_user_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_permission',
 
                                  repoid=self.REPO,
 
                                  userid=TEST_USER_ADMIN_LOGIN, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
 
            TEST_USER_ADMIN_LOGIN, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @parameterized.expand([('none', 'repository.none'),
 
                           ('read', 'repository.read'),
 
                           ('write', 'repository.write'),
 
                           ('admin', 'repository.admin')])
 
    def test_api_grant_user_group_permission(self, name, perm):
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        ret = {
 
            'msg': 'Granted perm: `%s` for user group: `%s` in repo: `%s`' % (
 
                perm, TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        expected = ret
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    def test_api_grant_user_group_permission_wrong_permission(self):
 
        perm = 'haha.no.permission'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'permission `%s` does not exist' % perm
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'grant_user_group_permission', crash)
 
    def test_api_grant_user_group_permission_exception_when_adding(self):
 
        perm = 'repository.read'
 
        id_, params = _build_data(self.apikey,
 
                                  'grant_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP,
 
                                  perm=perm)
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    def test_api_revoke_user_group_permission(self):
 
        RepoModel().grant_user_group_permission(repo=self.REPO,
 
                                                group_name=TEST_USER_GROUP,
 
                                                perm='repository.read')
 
        Session().commit()
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = {
 
            'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
 
                TEST_USER_GROUP, self.REPO
 
            ),
 
            'success': True
 
        }
 
        self._compare_ok(id_, expected, given=response.body)
 

	
 
    @mock.patch.object(RepoModel, 'revoke_user_group_permission', crash)
 
    def test_api_revoke_user_group_permission_exception_when_adding(self):
 
        id_, params = _build_data(self.apikey,
 
                                  'revoke_user_group_permission',
 
                                  repoid=self.REPO,
 
                                  usergroupid=TEST_USER_GROUP, )
 
        response = api_call(self, params)
 

	
 
        expected = 'failed to edit permission for user group: `%s` in repo: `%s`' % (
 
            TEST_USER_GROUP, self.REPO
 
        )
 
        self._compare_error(id_, expected, given=response.body)
 

	
 
    @parameterized.expand([
 
        ('none', 'group.none', 'none'),
 
        ('read', 'group.read', 'none'),
 
        ('write', 'group.write', 'none'),
 
        ('admin', 'group.admin', 'none'),
 

	
 
        ('none', 'group.none', 'all'),
0 comments (0 inline, 0 general)