repo group: introduce editing of owner

The repo group owner concept was only partially implemented. Owners were shown
in the repo group listing, but couldn't be changed. Users owning repo groups
couldn't be deleted, with no other solution than deleting owned repo groups.

This also fixes the existing broken update_repo_group API, which tried to use
unimplemented functionality.

kallithea/controllers/admin/repo_groups.py

@@ -1,402 +1,403 b''
 # -*- 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.controllers.admin.repo_groups
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 Repository groups controller 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: Mar 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import logging
 import traceback
 
 import formencode
 from formencode import htmlfill
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound
 
 from kallithea.controllers import base
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoGroupPermissionLevelDecorator, LoginRequired
 from kallithea.lib.utils2 import safe_int
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import RepoGroupForm, RepoGroupPermsForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import AvailableRepoGroupChoices, RepoGroupList
 
 
 log = logging.getLogger(__name__)
 
 
 class RepoGroupsController(base.BaseController):
 
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(RepoGroupsController, self)._before(*args, **kwargs)
 
     def __load_defaults(self, extras=(), exclude=()):
         """extras is used for keeping current parent ignoring permissions
         exclude is used for not moving group to itself TODO: also exclude descendants
         Note: only admin can create top level groups
         """
         repo_groups = AvailableRepoGroupChoices('admin', extras)
         exclude_group_ids = set(rg.group_id for rg in exclude)
         c.repo_groups = [rg for rg in repo_groups
                          if rg[0] not in exclude_group_ids]
 
     def __load_data(self, group_id):
         """
         Load defaults settings for edit, and update
 
         :param group_id:
         """
         repo_group = db.RepoGroup.get_or_404(group_id)
         data = repo_group.get_dict()
         data['group_name'] = repo_group.name
+        data['owner'] = repo_group.owner.username
 
         # fill repository group users
         for p in repo_group.repo_group_to_perm:
             data.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
 
         # fill repository group groups
         for p in repo_group.users_group_to_perm:
             data.update({'g_perm_%s' % p.users_group.users_group_name:
                              p.permission.permission_name})
 
         return data
 
     def _revoke_perms_on_yourself(self, form_result):
         _up = [u for u in form_result['perms_updates'] if request.authuser.username == u[0]]
         _new = [u for u in form_result['perms_new'] if request.authuser.username == u[0]]
         if _new and _new[0][1] != 'group.admin' or _up and _up[0][1] != 'group.admin':
             return True
         return False
 
     def index(self, format='html'):
         _list = db.RepoGroup.query(sorted=True).all()
         group_iter = RepoGroupList(_list, perm_level='admin')
         repo_groups_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
 
         def repo_group_name(repo_group_name, children_groups):
             return template.get_def("repo_group_name") \
                 .render_unicode(repo_group_name, children_groups, _=_, webutils=webutils, c=c)
 
         def repo_group_actions(repo_group_id, repo_group_name, gr_count):
             return template.get_def("repo_group_actions") \
                 .render_unicode(repo_group_id, repo_group_name, gr_count, _=_, webutils=webutils, c=c,
                         ungettext=ungettext)
 
         for repo_gr in group_iter:
             children_groups = [g.name for g in repo_gr.parents] + [repo_gr.name]
             repo_count = repo_gr.repositories.count()
             repo_groups_data.append({
                 "raw_name": webutils.escape(repo_gr.group_name),
                 "group_name": repo_group_name(repo_gr.group_name, children_groups),
                 "desc": webutils.escape(repo_gr.group_description),
                 "repos": repo_count,
                 "owner": repo_gr.owner.username,
                 "action": repo_group_actions(repo_gr.group_id, repo_gr.group_name,
                                              repo_count)
             })
 
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": repo_groups_data
         }
 
         return base.render('admin/repo_groups/repo_groups.html')
 
     def create(self):
         self.__load_defaults()
 
         # permissions for can create group based on parent_id are checked
         # here in the Form
         repo_group_form = RepoGroupForm(repo_groups=c.repo_groups)
         form_result = None
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             gr = RepoGroupModel().create(
                 group_name=form_result['group_name'],
                 group_description=form_result['group_description'],
                 parent=form_result['parent_group_id'],
-                owner=request.authuser.user_id, # TODO: make editable
+                owner=request.authuser.user_id,
                 copy_permissions=form_result['group_copy_permissions']
             )
             meta.Session().commit()
             # TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             return htmlfill.render(
                 base.render('admin/repo_groups/repo_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             webutils.flash(_('Error occurred during creation of repository group %s')
                     % request.POST.get('group_name'), category='error')
             if form_result is None:
                 raise
             parent_group_id = form_result['parent_group_id']
             # TODO: maybe we should get back to the main view, not the admin one
             raise HTTPFound(location=url('repos_groups', parent_group=parent_group_id))
         webutils.flash(_('Created repository group %s') % gr.group_name,
                 category='success')
         raise HTTPFound(location=url('repos_group_home', group_name=gr.group_name))
 
     def new(self):
         parent_group_id = safe_int(request.GET.get('parent_group') or '-1')
         if HasPermissionAny('hg.admin')('group create'):
             # we're global admin, we're ok and we can create TOP level groups
             pass
         else:
             # we pass in parent group into creation form, thus we know
             # what would be the group, we can check perms here !
             group = db.RepoGroup.get(parent_group_id) if parent_group_id else None
             group_name = group.group_name if group else None
             if HasRepoGroupPermissionLevel('admin')(group_name, 'group create'):
                 pass
             else:
                 raise HTTPForbidden()
 
         self.__load_defaults()
         return htmlfill.render(
             base.render('admin/repo_groups/repo_group_add.html'),
             defaults={'parent_group_id': parent_group_id},
             errors={},
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update(self, group_name):
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
 
         # TODO: kill allow_empty_group - it is only used for redundant form validation!
         if HasPermissionAny('hg.admin')('group edit'):
             # we're global admin, we're ok and we can create TOP level groups
             allow_empty_group = True
         elif not c.repo_group.parent_group:
             allow_empty_group = True
         else:
             allow_empty_group = False
         repo_group_form = RepoGroupForm(
             edit=True,
             old_data=c.repo_group.get_dict(),
             repo_groups=c.repo_groups,
             can_create_in_root=allow_empty_group,
         )()
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
 
             new_gr = RepoGroupModel().update(group_name, form_result)
             meta.Session().commit()
             webutils.flash(_('Updated repository group %s')
                     % form_result['group_name'], category='success')
             # we now have new name !
             group_name = new_gr.group_name
             # TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             c.active = 'settings'
             return htmlfill.render(
                 base.render('admin/repo_groups/repo_group_edit.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             webutils.flash(_('Error occurred during update of repository group %s')
                     % request.POST.get('group_name'), category='error')
 
         raise HTTPFound(location=url('edit_repo_group', group_name=group_name))
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete(self, group_name):
         gr = c.repo_group = db.RepoGroup.guess_instance(group_name)
         parent_group = gr.parent_group
         repos = gr.repositories.all()
         if repos:
             webutils.flash(_('This group contains %s repositories and cannot be '
                       'deleted') % len(repos), category='warning')
             raise HTTPFound(location=url('repos_groups'))
 
         children = gr.children.all()
         if children:
             webutils.flash(_('This group contains %s subgroups and cannot be deleted'
                       % (len(children))), category='warning')
             raise HTTPFound(location=url('repos_groups'))
 
         try:
             RepoGroupModel().delete(group_name)
             meta.Session().commit()
             webutils.flash(_('Removed repository group %s') % group_name,
                     category='success')
             # TODO: in future action_logger(, '', '', '')
         except Exception:
             log.error(traceback.format_exc())
             webutils.flash(_('Error occurred during deletion of repository group %s')
                     % group_name, category='error')
 
         if parent_group:
             raise HTTPFound(location=url('repos_group_home', group_name=parent_group.group_name))
         raise HTTPFound(location=url('repos_groups'))
 
     def show_by_name(self, group_name):
         """
         This is a proxy that does a lookup group_name -> id, and shows
         the group by id view instead
         """
         group_name = group_name.rstrip('/')
         id_ = db.RepoGroup.get_by_group_name(group_name)
         if id_:
             return self.show(group_name)
         raise HTTPNotFound
 
     @HasRepoGroupPermissionLevelDecorator('read')
     def show(self, group_name):
         c.active = 'settings'
 
         c.group = c.repo_group = db.RepoGroup.guess_instance(group_name)
 
         groups = db.RepoGroup.query(sorted=True).filter_by(parent_group=c.group).all()
         repo_groups_list = self.scm_model.get_repo_groups(groups)
 
         repos_list = db.Repository.query(sorted=True).filter_by(group=c.group).all()
         c.data = RepoModel().get_repos_as_dict(repos_list,
                                                repo_groups_list=repo_groups_list,
                                                short_name=True)
 
         return base.render('admin/repo_groups/repo_group_show.html')
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit(self, group_name):
         c.active = 'settings'
 
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
         defaults = self.__load_data(c.repo_group.group_id)
 
         return htmlfill.render(
             base.render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_advanced(self, group_name):
         c.active = 'advanced'
         c.repo_group = db.RepoGroup.guess_instance(group_name)
 
         return base.render('admin/repo_groups/repo_group_edit.html')
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_perms(self, group_name):
         c.active = 'perms'
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults()
         defaults = self.__load_data(c.repo_group.group_id)
 
         return htmlfill.render(
             base.render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
         )
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update_perms(self, group_name):
         """
         Update permissions for given repository group
 
         :param group_name:
         """
 
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         valid_recursive_choices = ['none', 'repos', 'groups', 'all']
         form_result = RepoGroupPermsForm(valid_recursive_choices)().to_python(request.POST)
         if not request.authuser.is_admin:
             if self._revoke_perms_on_yourself(form_result):
                 msg = _('Cannot revoke permission for yourself as admin')
                 webutils.flash(msg, category='warning')
                 raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
         recursive = form_result['recursive']
         # iterate over all members(if in recursive mode) of this groups and
         # set the permissions !
         # this can be potentially heavy operation
         RepoGroupModel()._update_permissions(c.repo_group,
                                              form_result['perms_new'],
                                              form_result['perms_updates'],
                                              recursive)
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         meta.Session().commit()
         webutils.flash(_('Repository group permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
 
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete_perms(self, group_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
 
             if not request.authuser.is_admin:
                 if obj_type == 'user' and request.authuser.user_id == obj_id:
                     msg = _('Cannot revoke permission for yourself as admin')
                     webutils.flash(msg, category='warning')
                     raise Exception('revoke admin permission on self')
             recursive = request.POST.get('recursive', 'none')
             if obj_type == 'user':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id, obj_type='user',
                                                    recursive=recursive)
             elif obj_type == 'user_group':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id,
                                                    obj_type='user_group',
                                                    recursive=recursive)
 
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
             webutils.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()

kallithea/model/forms.py

@@ -1,564 +1,565 b''
 # -*- 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/>.
 """
 these are form validation classes
 http://formencode.org/module-formencode.validators.html
 for list of all available validators
 
 we can create our own validators
 
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 
 
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 
 """
 import logging
 
 import formencode
 from formencode import All
 from tg.i18n import ugettext as _
 
 import kallithea
 from kallithea.model import validators as v
 
 
 log = logging.getLogger(__name__)
 
 
 def LoginForm():
     class _LoginForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = v.UnicodeString(
             strip=True,
             min=1,
             not_empty=True,
             messages={
                'empty': _('Please enter a login'),
                'tooShort': _('Enter a value %(min)i characters long or more')}
         )
 
         password = v.UnicodeString(
             strip=False,
             min=3,
             not_empty=True,
             messages={
                 'empty': _('Please enter a password'),
                 'tooShort': _('Enter %(min)i characters or more')}
         )
 
         remember = v.StringBoolean(if_missing=False)
 
         chained_validators = [v.ValidAuth()]
     return _LoginForm
 
 
 def PasswordChangeForm(username):
     class _PasswordChangeForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         current_password = v.ValidOldPassword(username)(not_empty=True)
         new_password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         new_password_confirmation = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
 
         chained_validators = [v.ValidPasswordsMatch('new_password',
                                                     'new_password_confirmation')]
     return _PasswordChangeForm
 
 
 def UserForm(edit=False, old_data=None):
     old_data = old_data or {}
 
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                        v.ValidUsername(edit, old_data))
         if edit:
             new_password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False),
             )
             admin = v.StringBoolean(if_missing=False)
             chained_validators = [v.ValidPasswordsMatch('new_password',
                                                         'password_confirmation')]
         else:
             password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=True)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             chained_validators = [v.ValidPasswordsMatch('password',
                                                         'password_confirmation')]
 
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
         extern_name = v.UnicodeString(strip=True, if_missing=None)
         extern_type = v.UnicodeString(strip=True, if_missing=None)
     return _UserForm
 
 
 def UserGroupForm(edit=False, old_data=None, available_members=None):
     old_data = old_data or {}
     available_members = available_members or []
 
     class _UserGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         users_group_name = All(
             v.UnicodeString(strip=True, min=1, not_empty=True),
             v.ValidUserGroup(edit, old_data)
         )
         user_group_description = v.UnicodeString(strip=True, min=1,
                                                  not_empty=False)
 
         users_group_active = v.StringBoolean(if_missing=False)
 
         if edit:
             users_group_members = v.OneOf(
                 available_members, hideList=False, testValueList=True,
                 if_missing=None, not_empty=False
             )
 
     return _UserGroupForm
 
 
 def RepoGroupForm(edit=False, old_data=None, repo_groups=None,
                    can_create_in_root=False):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
 
         group_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                          v.SlugifyName(),
                          v.ValidRegex(msg=_('Name must not contain only digits'))(r'(?!^\d+$)^.+$'))
         group_description = v.UnicodeString(strip=True, min=1,
                                             not_empty=False)
         group_copy_permissions = v.StringBoolean(if_missing=False)
 
         if edit:
+            owner = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())
             # FIXME: do a special check that we cannot move a group to one of
             # its children
             pass
 
         parent_group_id = All(v.CanCreateGroup(can_create_in_root),
                               v.OneOf(repo_group_ids, hideList=False,
                                       testValueList=True,
                                       if_missing=None, not_empty=True),
                               v.Int(min=-1, not_empty=True))
         chained_validators = [v.ValidRepoGroup(edit, old_data)]
 
     return _RepoGroupForm
 
 
 def RegisterForm(edit=False, old_data=None):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(
             v.ValidUsername(edit, old_data),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         password = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         password_confirmation = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
 
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirmation')]
 
     return _RegisterForm
 
 
 def PasswordResetRequestForm():
     class _PasswordResetRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = v.Email(not_empty=True)
     return _PasswordResetRequestForm
 
 
 def PasswordResetConfirmationForm():
     class _PasswordResetConfirmationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         email = v.UnicodeString(strip=True, not_empty=True)
         timestamp = v.Number(strip=True, not_empty=True)
         token = v.UnicodeString(strip=True, not_empty=True)
         password = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
         password_confirm = All(v.ValidPassword(), v.UnicodeString(strip=False, min=6))
 
         chained_validators = [v.ValidPasswordsMatch('password',
                                                     'password_confirm')]
     return _PasswordResetConfirmationForm
 
 
 def RepoForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS,
              repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(old_data),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = v.OneOf(supported_backends, required=False,
                             if_missing=old_data.get('repo_type'))
         repo_description = v.UnicodeString(strip=True, min=1, not_empty=False)
         repo_private = v.StringBoolean(if_missing=False)
         repo_landing_rev = v.OneOf(landing_revs, hideList=True)
         repo_copy_permissions = v.StringBoolean(if_missing=False)
         clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))
 
         repo_enable_statistics = v.StringBoolean(if_missing=False)
         repo_enable_downloads = v.StringBoolean(if_missing=False)
 
         if edit:
             owner = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())
             # Not a real field - just for reference for validation:
             # clone_uri_hidden = v.UnicodeString(if_missing='')
 
         chained_validators = [v.ValidCloneUri(),
                               v.ValidRepoName(edit, old_data)]
     return _RepoForm
 
 
 def RepoPermsForm():
     class _RepoPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='repo')]
     return _RepoPermsForm
 
 
 def RepoGroupPermsForm(valid_recursive_choices):
     class _RepoGroupPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         recursive = v.OneOf(valid_recursive_choices)
         chained_validators = [v.ValidPerms(type_='repo_group')]
     return _RepoGroupPermsForm
 
 
 def UserGroupPermsForm():
     class _UserPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms(type_='user_group')]
     return _UserPermsForm
 
 
 def RepoFieldForm():
     class _RepoFieldForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         new_field_key = All(v.FieldKey(),
                             v.UnicodeString(strip=True, min=3, not_empty=True))
         new_field_value = v.UnicodeString(not_empty=False, if_missing='')
         new_field_type = v.OneOf(['str', 'unicode', 'list', 'tuple'],
                                  if_missing='str')
         new_field_label = v.UnicodeString(not_empty=False)
         new_field_desc = v.UnicodeString(not_empty=False)
 
     return _RepoFieldForm
 
 
 def RepoForkForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS,
                  repo_groups=None, landing_revs=None):
     old_data = old_data or {}
     repo_groups = repo_groups or []
     landing_revs = landing_revs or []
     repo_group_ids = [rg[0] for rg in repo_groups]
 
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(),
                          v.OneOf(repo_group_ids, hideList=True),
                          v.Int(min=-1, not_empty=True))
         repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))
         description = v.UnicodeString(strip=True, min=1, not_empty=True)
         private = v.StringBoolean(if_missing=False)
         copy_permissions = v.StringBoolean(if_missing=False)
         update_after_clone = v.StringBoolean(if_missing=False)
         fork_parent_id = v.UnicodeString()
         chained_validators = [v.ValidForkName(edit, old_data)]
         landing_rev = v.OneOf(landing_revs, hideList=True)
 
     return _RepoForkForm
 
 
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         title = v.UnicodeString(strip=True, not_empty=False)
         realm = v.UnicodeString(strip=True, min=1, not_empty=True)
         ga_code = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_public_key = v.UnicodeString(strip=True, min=1, not_empty=False)
         captcha_private_key = v.UnicodeString(strip=True, min=1, not_empty=False)
 
     return _ApplicationSettingsForm
 
 
 def ApplicationVisualisationForm():
     class _ApplicationVisualisationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         show_public_icon = v.StringBoolean(if_missing=False)
         show_private_icon = v.StringBoolean(if_missing=False)
         stylify_metalabels = v.StringBoolean(if_missing=False)
 
         repository_fields = v.StringBoolean(if_missing=False)
         lightweight_journal = v.StringBoolean(if_missing=False)
         dashboard_items = v.Int(min=5, not_empty=True)
         admin_grid_items = v.Int(min=5, not_empty=True)
         show_version = v.StringBoolean(if_missing=False)
         use_gravatar = v.StringBoolean(if_missing=False)
         gravatar_url = v.UnicodeString(min=3)
         clone_uri_tmpl = v.UnicodeString(min=3)
         clone_ssh_tmpl = v.UnicodeString()
 
     return _ApplicationVisualisationForm
 
 
 def ApplicationUiSettingsForm():
     class _ApplicationUiSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         paths_root_path = All(
             v.ValidPath(),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         hooks_changegroup_kallithea_update = v.StringBoolean(if_missing=False)
         hooks_changegroup_kallithea_repo_size = v.StringBoolean(if_missing=False)
 
         extensions_largefiles = v.StringBoolean(if_missing=False)
         extensions_hggit = v.StringBoolean(if_missing=False)
 
     return _ApplicationUiSettingsForm
 
 
 def DefaultPermissionsForm(repo_perms_choices, group_perms_choices,
                            user_group_perms_choices, create_choices,
                            user_group_create_choices, fork_choices,
                            register_choices, extern_activate_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default_repo = v.StringBoolean(if_missing=False)
         overwrite_default_group = v.StringBoolean(if_missing=False)
         overwrite_default_user_group = v.StringBoolean(if_missing=False)
         anonymous = v.StringBoolean(if_missing=False)
         default_repo_perm = v.OneOf(repo_perms_choices)
         default_group_perm = v.OneOf(group_perms_choices)
         default_user_group_perm = v.OneOf(user_group_perms_choices)
 
         default_repo_create = v.OneOf(create_choices)
         default_user_group_create = v.OneOf(user_group_create_choices)
         default_fork = v.OneOf(fork_choices)
 
         default_register = v.OneOf(register_choices)
         default_extern_activate = v.OneOf(extern_activate_choices)
     return _DefaultPermissionsForm
 
 
 def CustomDefaultPermissionsForm():
     class _CustomDefaultPermissionsForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         create_repo_perm = v.StringBoolean(if_missing=False)
         create_user_group_perm = v.StringBoolean(if_missing=False)
         #create_repo_group_perm Impl. later
 
         fork_repo_perm = v.StringBoolean(if_missing=False)
 
     return _CustomDefaultPermissionsForm
 
 
 def DefaultsForm(edit=False, old_data=None, supported_backends=kallithea.BACKENDS):
     class _DefaultsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         default_repo_type = v.OneOf(supported_backends)
         default_repo_private = v.StringBoolean(if_missing=False)
         default_repo_enable_statistics = v.StringBoolean(if_missing=False)
         default_repo_enable_downloads = v.StringBoolean(if_missing=False)
 
     return _DefaultsForm
 
 
 def AuthSettingsForm(current_active_modules):
     class _AuthSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         auth_plugins = All(v.ValidAuthPlugins(),
                            v.UniqueListFromString()(not_empty=True))
 
         def __init__(self, *args, **kwargs):
             # The auth plugins tell us what form validators they use
             if current_active_modules:
                 import kallithea.lib.auth_modules
                 from kallithea.lib.auth_modules import LazyFormencode
                 for module in current_active_modules:
                     plugin = kallithea.lib.auth_modules.loadplugin(module)
                     plugin_name = plugin.name
                     for sv in plugin.plugin_settings():
                         newk = "auth_%s_%s" % (plugin_name, sv["name"])
                         # can be a LazyFormencode object from plugin settings
                         validator = sv["validator"]
                         if isinstance(validator, LazyFormencode):
                             validator = validator()
                         # init all lazy validators from formencode.All
                         if isinstance(validator, All):
                             init_validators = []
                             for validator in validator.validators:
                                 if isinstance(validator, LazyFormencode):
                                     validator = validator()
                                 init_validators.append(validator)
                             validator.validators = init_validators
 
                         self.add_field(newk, validator)
             formencode.Schema.__init__(self, *args, **kwargs)
 
     return _AuthSettingsForm
 
 
 def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,
                      tls_kind_choices):
     class _LdapSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         #pre_validators = [LdapLibValidator]
         ldap_active = v.StringBoolean(if_missing=False)
         ldap_host = v.UnicodeString(strip=True,)
         ldap_port = v.Number(strip=True,)
         ldap_tls_kind = v.OneOf(tls_kind_choices)
         ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)
         ldap_dn_user = v.UnicodeString(strip=True,)
         ldap_dn_pass = v.UnicodeString(strip=True,)
         ldap_base_dn = v.UnicodeString(strip=True,)
         ldap_filter = v.UnicodeString(strip=True,)
         ldap_search_scope = v.OneOf(search_scope_choices)
         ldap_attr_login = v.AttrLoginValidator()(not_empty=True)
         ldap_attr_firstname = v.UnicodeString(strip=True,)
         ldap_attr_lastname = v.UnicodeString(strip=True,)
         ldap_attr_email = v.UnicodeString(strip=True,)
 
     return _LdapSettingsForm
 
 
 def UserExtraEmailForm():
     class _UserExtraEmailForm(formencode.Schema):
         email = All(v.UniqSystemEmail(), v.Email(not_empty=True))
     return _UserExtraEmailForm
 
 
 def UserExtraIpForm():
     class _UserExtraIpForm(formencode.Schema):
         ip = v.ValidIp()(not_empty=True)
     return _UserExtraIpForm
 
 
 def PullRequestForm(repo_id):
     class _PullRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         org_repo = v.UnicodeString(strip=True, required=True)
         org_ref = v.UnicodeString(strip=True, required=True)
         other_repo = v.UnicodeString(strip=True, required=True)
         other_ref = v.UnicodeString(strip=True, required=True)
 
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
 
     return _PullRequestForm
 
 
 def PullRequestPostForm():
     class _PullRequestPostForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         pullrequest_title = v.UnicodeString(strip=True, required=True)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
         org_review_members = v.Set()
         review_members = v.Set()
         updaterev = v.UnicodeString(strip=True, required=False, if_missing=None)
         owner = All(v.UnicodeString(strip=True, required=True),
                     v.ValidRepoUser())
 
     return _PullRequestPostForm
 
 
 def GistForm(lifetime_options):
     class _GistForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         filename = All(v.BasePath()(),
                        v.UnicodeString(strip=True, required=False))
         description = v.UnicodeString(required=False, if_missing='')
         lifetime = v.OneOf(lifetime_options)
         mimetype = v.UnicodeString(required=False, if_missing=None)
         content = v.UnicodeString(required=True, not_empty=True)
         public = v.UnicodeString(required=False, if_missing='')
         private = v.UnicodeString(required=False, if_missing='')
 
     return _GistForm

kallithea/model/repo_group.py

@@ -1,532 +1,534 b''
 # -*- 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.repo_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 repo 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: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 
 import datetime
 import logging
 import os
 import shutil
 import traceback
 
 import kallithea.lib.utils2
 from kallithea.lib.utils2 import LazyProperty
 from kallithea.model import db, meta, repo
 
 
 log = logging.getLogger(__name__)
 
 
 class RepoGroupModel(object):
 
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
 
         q = db.Ui.get_by_key('paths', '/')
         return q.ui_value
 
     def _create_default_perms(self, new_group):
         # create default permission
         default_perm = 'group.read'
         def_user = db.User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('group.'):
                 default_perm = p.permission.permission_name
                 break
 
         repo_group_to_perm = db.UserRepoGroupToPerm()
         repo_group_to_perm.permission = db.Permission.get_by_key(default_perm)
 
         repo_group_to_perm.group = new_group
         repo_group_to_perm.user_id = def_user.user_id
         meta.Session().add(repo_group_to_perm)
         return repo_group_to_perm
 
     def _create_group(self, group_name):
         """
         makes repository group on filesystem
 
         :param repo_name:
         :param parent_id:
         """
 
         create_path = os.path.join(self.repos_path, group_name)
         log.debug('creating new group in %s', create_path)
 
         if os.path.isdir(create_path):
             raise Exception('That directory already exists !')
 
         os.makedirs(create_path)
         log.debug('Created group in %s', create_path)
 
     def _rename_group(self, old, new):
         """
         Renames a group on filesystem
 
         :param group_name:
         """
 
         if old == new:
             log.debug('skipping group rename')
             return
 
         log.debug('renaming repository group from %s to %s', old, new)
 
         old_path = os.path.join(self.repos_path, old)
         new_path = os.path.join(self.repos_path, new)
 
         log.debug('renaming repos paths from %s to %s', old_path, new_path)
 
         if os.path.isdir(new_path):
             raise Exception('Was trying to rename to already '
                             'existing dir %s' % new_path)
         shutil.move(old_path, new_path)
 
     def _delete_group(self, group, force_delete=False):
         """
         Deletes a group from a filesystem
 
         :param group: instance of group from database
         :param force_delete: use shutil rmtree to remove all objects
         """
         paths = group.full_path.split(kallithea.URL_SEP)
         paths = os.sep.join(paths)
 
         rm_path = os.path.join(self.repos_path, paths)
         log.info("Removing group %s", rm_path)
         # delete only if that path really exists
         if os.path.isdir(rm_path):
             if force_delete:
                 shutil.rmtree(rm_path)
             else:
                 # archive that group
                 _now = datetime.datetime.now()
                 _ms = str(_now.microsecond).rjust(6, '0')
                 _d = 'rm__%s_GROUP_%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                                           group.name)
                 shutil.move(rm_path, os.path.join(self.repos_path, _d))
 
     def create(self, group_name, group_description, owner, parent=None,
                just_db=False, copy_permissions=False):
         try:
             if kallithea.lib.utils2.repo_name_slug(group_name) != group_name:
                 raise Exception('invalid repo group name %s' % group_name)
 
             owner = db.User.guess_instance(owner)
             parent_group = db.RepoGroup.guess_instance(parent)
             new_repo_group = db.RepoGroup()
             new_repo_group.owner = owner
             new_repo_group.group_description = group_description or group_name
             new_repo_group.parent_group = parent_group
             new_repo_group.group_name = new_repo_group.get_new_name(group_name)
 
             meta.Session().add(new_repo_group)
 
             # create an ADMIN permission for owner except if we're super admin,
             # later owner should go into the owner field of groups
             if not owner.is_admin:
                 self.grant_user_permission(repo_group=new_repo_group,
                                            user=owner, perm='group.admin')
 
             if parent_group and copy_permissions:
                 # copy permissions from parent
                 user_perms = db.UserRepoGroupToPerm.query() \
                     .filter(db.UserRepoGroupToPerm.group == parent_group).all()
 
                 group_perms = db.UserGroupRepoGroupToPerm.query() \
                     .filter(db.UserGroupRepoGroupToPerm.group == parent_group).all()
 
                 for perm in user_perms:
                     # don't copy over the permission for user who is creating
                     # this group, if he is not super admin he get's admin
                     # permission set above
                     if perm.user != owner or owner.is_admin:
                         db.UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)
 
                 for perm in group_perms:
                     db.UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)
             else:
                 self._create_default_perms(new_repo_group)
 
             if not just_db:
                 # we need to flush here, in order to check if database won't
                 # throw any exceptions, create filesystem dirs at the very end
                 meta.Session().flush()
                 self._create_group(new_repo_group.group_name)
 
             return new_repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def _update_permissions(self, repo_group, perms_new=None,
                             perms_updates=None, recursive=None,
                             check_perms=True):
         from kallithea.lib.auth import HasUserGroupPermissionLevel
 
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
 
         def _set_perm_user(obj, user, perm):
             if isinstance(obj, db.RepoGroup):
                 self.grant_user_permission(repo_group=obj, user=user, perm=perm)
             elif isinstance(obj, db.Repository):
                 user = db.User.guess_instance(user)
 
                 # private repos will not allow to change the default permissions
                 # using recursive mode
                 if obj.private and user.is_default_user:
                     return
 
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 repo.RepoModel().grant_user_permission(
                     repo=obj, user=user, perm=perm
                 )
 
         def _set_perm_group(obj, users_group, perm):
             if isinstance(obj, db.RepoGroup):
                 self.grant_user_group_permission(repo_group=obj,
                                                   group_name=users_group,
                                                   perm=perm)
             elif isinstance(obj, db.Repository):
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 repo.RepoModel().grant_user_group_permission(
                     repo=obj, group_name=users_group, perm=perm
                 )
 
         # start updates
         updates = []
         log.debug('Now updating permissions for %s in recursive mode:%s',
                   repo_group, recursive)
 
         for obj in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(obj, db.RepoGroup) and not obj == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(obj, db.Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 obj = repo_group
                 # also we do a break at the end of this loop.
 
             # update permissions
             for member, perm, member_type in perms_updates:
                 ## set for user
                 if member_type == 'user':
                     # this updates also current one if found
                     _set_perm_user(obj, user=member, perm=perm)
                 ## set for user group
                 else:
                     # check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             # set new permissions
             for member, perm, member_type in perms_new:
                 if member_type == 'user':
                     _set_perm_user(obj, user=member, perm=perm)
                 else:
                     # check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             updates.append(obj)
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
 
         return updates
 
     def update(self, repo_group, repo_group_args):
         try:
             repo_group = db.RepoGroup.guess_instance(repo_group)
             old_path = repo_group.full_path
 
             # change properties
+            if 'owner' in repo_group_args:
+                repo_group.owner = db.User.get_by_username(repo_group_args['owner'])
             if 'group_description' in repo_group_args:
                 repo_group.group_description = repo_group_args['group_description']
             if 'parent_group_id' in repo_group_args:
                 assert repo_group_args['parent_group_id'] != '-1', repo_group_args  # RepoGroupForm should have converted to None
                 repo_group.parent_group = db.RepoGroup.get(repo_group_args['parent_group_id'])
                 repo_group.group_name = repo_group.get_new_name(repo_group.name)
             if 'group_name' in repo_group_args:
                 group_name = repo_group_args['group_name']
                 if kallithea.lib.utils2.repo_name_slug(group_name) != group_name:
                     raise Exception('invalid repo group name %s' % group_name)
                 repo_group.group_name = repo_group.get_new_name(group_name)
             new_path = repo_group.full_path
             meta.Session().add(repo_group)
 
             # iterate over all members of this groups and do fixes
             # if obj is a repoGroup also fix the name of the group according
             # to the parent
             # if obj is a Repo fix it's name
             # this can be potentially heavy operation
             for obj in repo_group.recursive_groups_and_repos():
                 # set the value from it's parent
                 if isinstance(obj, db.RepoGroup):
                     new_name = obj.get_new_name(obj.name)
                     log.debug('Fixing group %s to new name %s'
                                 % (obj.group_name, new_name))
                     obj.group_name = new_name
                 elif isinstance(obj, db.Repository):
                     # we need to get all repositories from this new group and
                     # rename them accordingly to new group path
                     new_name = obj.get_new_name(obj.just_name)
                     log.debug('Fixing repo %s to new name %s'
                                 % (obj.repo_name, new_name))
                     obj.repo_name = new_name
 
             self._rename_group(old_path, new_path)
 
             return repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def delete(self, repo_group, force_delete=False):
         repo_group = db.RepoGroup.guess_instance(repo_group)
         try:
             meta.Session().delete(repo_group)
             self._delete_group(repo_group, force_delete)
         except Exception:
             log.error('Error removing repo_group %s', repo_group)
             raise
 
     def add_permission(self, repo_group, obj, obj_type, perm, recursive):
         repo_group = db.RepoGroup.guess_instance(repo_group)
         perm = db.Permission.guess_instance(perm)
 
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, db.RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, db.Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
 
             if isinstance(el, db.RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().grant_user_permission(el, user=obj, perm=perm)
                 elif obj_type == 'user_group':
                     RepoGroupModel().grant_user_group_permission(el, group_name=obj, perm=perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, db.Repository):
                 # for repos we need to hotfix the name of permission
                 _perm = perm.permission_name.replace('group.', 'repository.')
                 if obj_type == 'user':
                     repo.RepoModel().grant_user_permission(el, user=obj, perm=_perm)
                 elif obj_type == 'user_group':
                     repo.RepoModel().grant_user_group_permission(el, group_name=obj, perm=_perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
 
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
 
     def delete_permission(self, repo_group, obj, obj_type, recursive):
         """
         Revokes permission for repo_group for given obj(user or users_group),
         obj_type can be user or user group
 
         :param repo_group:
         :param obj: user or user group id
         :param obj_type: user or user group type
         :param recursive: recurse to all children of group
         """
         repo_group = db.RepoGroup.guess_instance(repo_group)
 
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, db.RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, db.Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
 
             if isinstance(el, db.RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     RepoGroupModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, db.Repository):
                 if obj_type == 'user':
                     repo.RepoModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     repo.RepoModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
 
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
 
     def grant_user_permission(self, repo_group, user, perm):
         """
         Grant permission for user on given repository group, or update
         existing one if found
 
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
 
         repo_group = db.RepoGroup.guess_instance(repo_group)
         user = db.User.guess_instance(user)
         permission = db.Permission.guess_instance(perm)
 
         # check if we have that permission already
         obj = db.UserRepoGroupToPerm.query() \
             .filter(db.UserRepoGroupToPerm.user == user) \
             .filter(db.UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is None:
             # create new !
             obj = db.UserRepoGroupToPerm()
             meta.Session().add(obj)
         obj.group = repo_group
         obj.user = user
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, user, repo_group)
         return obj
 
     def revoke_user_permission(self, repo_group, user):
         """
         Revoke permission for user on given repository group
 
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         """
 
         repo_group = db.RepoGroup.guess_instance(repo_group)
         user = db.User.guess_instance(user)
 
         obj = db.UserRepoGroupToPerm.query() \
             .filter(db.UserRepoGroupToPerm.user == user) \
             .filter(db.UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is not None:
             meta.Session().delete(obj)
             log.debug('Revoked perm on %s on %s', repo_group, user)
 
     def grant_user_group_permission(self, repo_group, group_name, perm):
         """
         Grant permission for user group on given repository group, or update
         existing one if found
 
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         :param perm: Instance of Permission, or permission_name
         """
         repo_group = db.RepoGroup.guess_instance(repo_group)
         group_name = db.UserGroup.guess_instance(group_name)
         permission = db.Permission.guess_instance(perm)
 
         # check if we have that permission already
         obj = db.UserGroupRepoGroupToPerm.query() \
             .filter(db.UserGroupRepoGroupToPerm.group == repo_group) \
             .filter(db.UserGroupRepoGroupToPerm.users_group == group_name) \
             .scalar()
 
         if obj is None:
             # create new
             obj = db.UserGroupRepoGroupToPerm()
             meta.Session().add(obj)
 
         obj.group = repo_group
         obj.users_group = group_name
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, group_name, repo_group)
         return obj
 
     def revoke_user_group_permission(self, repo_group, group_name):
         """
         Revoke permission for user group on given repository group
 
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         """
         repo_group = db.RepoGroup.guess_instance(repo_group)
         group_name = db.UserGroup.guess_instance(group_name)
 
         obj = db.UserGroupRepoGroupToPerm.query() \
             .filter(db.UserGroupRepoGroupToPerm.group == repo_group) \
             .filter(db.UserGroupRepoGroupToPerm.users_group == group_name) \
             .scalar()
         if obj is not None:
             meta.Session().delete(obj)
             log.debug('Revoked perm to %s on %s', repo_group, group_name)

kallithea/templates/admin/repo_groups/repo_group_edit_settings.html

@@ -1,51 +1,59 b''
 ## -*- coding: utf-8 -*-
 ${h.form(url('update_repos_group',group_name=c.repo_group.group_name))}
 <div class="form">
         <div class="form-group">
             <label class="control-label" for="group_name">${_('Group name')}:</label>
             <div>
                 ${h.text('group_name',class_='form-control')}
             </div>
         </div>
 
         <div class="form-group">
+            <label class="control-label" for="owner">${_('Owner')}:</label>
+            <div>
+               ${h.text('owner',class_='form-control', placeholder=_('Type name of user'))}
+            </div>
+        </div>
+
+        <div class="form-group">
             <label class="control-label" for="group_description">${_('Description')}:</label>
             <div>
                 ${h.textarea('group_description',cols=23,rows=5,class_='form-control')}
             </div>
         </div>
 
         <div class="form-group">
             <label class="control-label" for="parent_group_id">${_('Group parent')}:</label>
             <div>
                 ${h.select('parent_group_id','',c.repo_groups,class_='form-control')}
             </div>
         </div>
 
         <div class="form-group">
             <div class="buttons">
                 ${h.submit('save',_('Save'),class_="btn btn-default")}
                 ${h.reset('reset',_('Reset'),class_="btn btn-default")}
             </div>
         </div>
 </div>
 ${h.end_form()}
 
 ${h.form(url('delete_repo_group', group_name=c.repo_group.group_name))}
 <div class="form">
         <div class="form-group">
             <div class="buttons">
                 ${h.submit('remove_%s' % c.repo_group.group_name,_('Remove this group'),class_="btn btn-danger",onclick="return confirm('"+_('Confirm to delete this group')+"');")}
             </div>
         </div>
 </div>
 ${h.end_form()}
 
 <script>
     'use strict';
     $(document).ready(function(){
         $("#parent_group_id").select2({
             'dropdownAutoWidth': true
         });
+        SimpleUserAutoComplete($('#owner'));
     });
 </script>

kallithea/tests/api/api_base.py

@@ -1,2921 +1,2921 b''
 # -*- 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/>.
 
 """
 Tests for the JSON-RPC web api.
 """
 
 import datetime
 import os
 import random
 import re
 import string
 from typing import Sized
 
 import mock
 import pytest
 from webtest import TestApp
 
 from kallithea.lib import ext_json
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.utils2 import ascii_bytes
 from kallithea.model import db, meta
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.gist import GistModel
 from kallithea.model.pull_request import PullRequestModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import ScmModel
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture, raise_exception
 
 
 API_URL = '/_admin/api'
 TEST_USER_GROUP = 'test_user_group'
 TEST_REPO_GROUP = 'test_repo_group'
 
 fixture = Fixture()
 
 
 def _build_data(apikey, method, **kw):
     """
     Builds API data with given random ID
     For convenience, the json is returned as str
     """
     random_id = random.randrange(1, 9999)
     return random_id, ext_json.dumps({
         "id": random_id,
         "api_key": apikey,
         "method": method,
         "args": kw
     })
 
 
 jsonify = lambda obj: ext_json.loads(ext_json.dumps(obj))
 
 
 def api_call(test_obj, params):
     response = test_obj.app.post(API_URL, content_type='application/json',
                                  params=params)
     return response
 
 
 ## helpers
 def make_user_group(name=TEST_USER_GROUP):
     gr = fixture.create_user_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
     UserGroupModel().add_user_to_group(user_group=gr,
                                        user=base.TEST_USER_ADMIN_LOGIN)
     meta.Session().commit()
     return gr
 
 
 def make_repo_group(name=TEST_REPO_GROUP):
     gr = fixture.create_repo_group(name, cur_user=base.TEST_USER_ADMIN_LOGIN)
     meta.Session().commit()
     return gr
 
 
 class _BaseTestApi(object):
     app: TestApp  # assigned by app_fixture in subclass TestController mixin
     # assigned in subclass:
     REPO: str
     REPO_TYPE: str
     TEST_REVISION: str
     TEST_PR_SRC: str
     TEST_PR_DST: str
     TEST_PR_REVISIONS: Sized
 
     @classmethod
     def setup_class(cls):
         cls.usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         cls.apikey = cls.usr.api_key
         cls.test_user = UserModel().create_or_update(
             username='test-api',
             password='test',
             email='[e-mail]',
             firstname='first',
             lastname='last'
         )
         meta.Session().commit()
         cls.TEST_USER_LOGIN = cls.test_user.username
         cls.apikey_regular = cls.test_user.api_key
 
     @classmethod
     def teardown_class(cls):
         pass
 
     def setup_method(self, method):
         make_user_group()
         make_repo_group()
 
     def teardown_method(self, method):
         fixture.destroy_user_group(TEST_USER_GROUP)
         fixture.destroy_gists()
         fixture.destroy_repo_group(TEST_REPO_GROUP)
 
     def _compare_ok(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': None,
             'result': expected
         })
         given = ext_json.loads(given)
         assert expected == given, (expected, given)
 
     def _compare_error(self, id_, expected, given):
         expected = jsonify({
             'id': id_,
             'error': expected,
             'result': None
         })
         given = ext_json.loads(given)
         assert expected == given, (expected, given)
 
     def test_api_wrong_key(self):
         id_, params = _build_data('trololo', 'get_user')
         response = api_call(self, params)
 
         expected = 'Invalid API key'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_missing_non_optional_param(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         response = api_call(self, params)
 
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_missing_non_optional_param_args_null(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
 
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_missing_non_optional_param_args_bad(self):
         id_, params = _build_data(self.apikey, 'get_repo')
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
 
         expected = 'Missing non optional `repoid` arg in JSON DATA'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_args_is_null(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": null')
         response = api_call(self, params)
         assert response.status == '200 OK'
 
     def test_api_args_is_bad(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         params = params.replace('"args": {}', '"args": 1')
         response = api_call(self, params)
         assert response.status == '200 OK'
 
     def test_api_args_different_args(self):
         expected = {
             'ascii_letters': string.ascii_letters,
             'ws': string.whitespace,
             'printables': string.printable
         }
         id_, params = _build_data(self.apikey, 'test', args=expected)
         response = api_call(self, params)
         assert response.status == '200 OK'
         self._compare_ok(id_, expected, response.body)
 
     def test_api_get_users(self):
         id_, params = _build_data(self.apikey, 'get_users', )
         response = api_call(self, params)
         ret_all = []
         _users = db.User.query().filter_by(is_default_user=False) \
             .order_by(db.User.username).all()
         for usr in _users:
             ret = usr.get_api_data()
             ret_all.append(jsonify(ret))
         expected = ret_all
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_user(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
 
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
 
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_user_that_does_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_user',
                                   userid='trololo')
         response = api_call(self, params)
 
         expected = "user `%s` does not exist" % 'trololo'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_user_without_giving_userid(self):
         id_, params = _build_data(self.apikey, 'get_user')
         response = api_call(self, params)
 
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
 
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_user_without_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user')
         response = api_call(self, params)
 
         usr = db.User.get_by_username(self.TEST_USER_LOGIN)
         ret = usr.get_api_data()
         ret['permissions'] = AuthUser(dbuser=usr).permissions
 
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_user_with_giving_userid_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_user',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
 
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_pull_remote(self):
         # Note: pulling from local repos is a misfeature - it will bypass access control
         # ... but ok, if the path already has been set in the database
         repo_name = 'test_pull'
         r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         # hack around that clone_uri can't be set to to a local path
         # (as shown by test_api_create_repo_clone_uri_local)
         r.clone_uri = os.path.join(db.Ui.get_by_key('paths', '/').ui_value, self.REPO)
         meta.Session().commit()
 
         pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
 
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,)
         response = api_call(self, params)
 
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
 
         post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in db.Repository.query().filter(db.Repository.repo_name == repo_name)]
 
         fixture.destroy_repo(repo_name)
 
         assert pre_cached_tip != post_cached_tip
 
     def test_api_pull_fork(self):
         fork_name = 'fork'
         fixture.create_fork(self.REPO, fork_name)
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=fork_name,)
         response = api_call(self, params)
 
         expected = {'msg': 'Pulled from `%s`' % fork_name,
                     'repository': fork_name}
         self._compare_ok(id_, expected, given=response.body)
 
         fixture.destroy_repo(fork_name)
 
     def test_api_pull_error_no_remote_no_fork(self):
         # should fail because no clone_uri is set
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=self.REPO, )
         response = api_call(self, params)
 
         expected = 'Unable to pull changes from `%s`' % self.REPO
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_pull_custom_remote(self):
         repo_name = 'test_pull_custom_remote'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
         custom_remote_path = os.path.join(db.Ui.get_by_key('paths', '/').ui_value, self.REPO)
 
         id_, params = _build_data(self.apikey, 'pull',
                                   repoid=repo_name,
                                   clone_uri=custom_remote_path)
         response = api_call(self, params)
 
         expected = {'msg': 'Pulled from `%s`' % repo_name,
                     'repository': repo_name}
         self._compare_ok(id_, expected, given=response.body)
 
         fixture.destroy_repo(repo_name)
 
     def test_api_rescan_repos(self):
         id_, params = _build_data(self.apikey, 'rescan_repos')
         response = api_call(self, params)
 
         expected = {'added': [], 'removed': []}
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(ScmModel, 'repo_scan', raise_exception)
     def test_api_rescann_error(self):
         id_, params = _build_data(self.apikey, 'rescan_repos', )
         response = api_call(self, params)
 
         expected = 'Error occurred during rescan repositories action'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_existing_user(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=base.TEST_USER_ADMIN_LOGIN,
                                   email='[e-mail]',
                                   password='trololo')
         response = api_call(self, params)
 
         expected = "user `%s` already exist" % base.TEST_USER_ADMIN_LOGIN
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_user_with_existing_email(self):
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=base.TEST_USER_ADMIN_LOGIN + 'new',
                                   email=base.TEST_USER_REGULAR_EMAIL,
                                   password='trololo')
         response = api_call(self, params)
 
         expected = "email `%s` already exist" % base.TEST_USER_REGULAR_EMAIL
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_user(self):
         username = 'test_new_api_user'
         email = username + "@example.com"
 
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
 
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
         )
 
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_without_password(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
 
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email)
         response = api_call(self, params)
 
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
         )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     def test_api_create_user_with_extern_name(self):
         username = 'test_new_api_user_passwordless'
         email = username + "@example.com"
 
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email, extern_name='internal')
         response = api_call(self, params)
 
         usr = db.User.get_by_username(username)
         ret = dict(
             msg='created new user `%s`' % username,
             user=jsonify(usr.get_api_data())
         )
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_user(usr.user_id)
 
     @mock.patch.object(UserModel, 'create_or_update', raise_exception)
     def test_api_create_user_when_exception_happened(self):
 
         username = 'test_new_api_user'
         email = username + "@example.com"
 
         id_, params = _build_data(self.apikey, 'create_user',
                                   username=username,
                                   email=email,
                                   password='trololo')
         response = api_call(self, params)
         expected = 'failed to create user `%s`' % username
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_delete_user(self):
         usr = UserModel().create_or_update(username='test_user',
                                            password='qweqwe',
                                            email='[e-mail]',
                                            firstname='u1', lastname='u1')
         meta.Session().commit()
         username = usr.username
         email = usr.email
         usr_id = usr.user_id
         ## DELETE THIS USER NOW
 
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
 
         ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),
                'user': None}
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(UserModel, 'delete', raise_exception)
     def test_api_delete_user_when_exception_happened(self):
         usr = UserModel().create_or_update(username='test_user',
                                            password='qweqwe',
                                            email='[e-mail]',
                                            firstname='u1', lastname='u1')
         meta.Session().commit()
         username = usr.username
 
         id_, params = _build_data(self.apikey, 'delete_user',
                                   userid=username, )
         response = api_call(self, params)
         ret = 'failed to delete user ID:%s %s' % (usr.user_id,
                                                   usr.username)
         expected = ret
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,expected', [
         ('firstname', 'new_username'),
         ('lastname', 'new_username'),
         ('email', 'new_username'),
         ('admin', True),
         ('admin', False),
         ('extern_type', 'ldap'),
         ('extern_type', None),
         ('extern_name', 'test'),
         ('extern_name', None),
         ('active', False),
         ('active', True),
         ('password', 'newpass'),
     ])
     def test_api_update_user(self, name, expected):
         usr = db.User.get_by_username(self.TEST_USER_LOGIN)
         kw = {name: expected,
               'userid': usr.user_id}
         id_, params = _build_data(self.apikey, 'update_user', **kw)
         response = api_call(self, params)
 
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, self.TEST_USER_LOGIN),
             'user': jsonify(db.User \
                 .get_by_username(self.TEST_USER_LOGIN) \
                 .get_api_data())
         }
 
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_update_user_no_changed_params(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
 
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, base.TEST_USER_ADMIN_LOGIN),
             'user': ret
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_update_user_by_user_id(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
 
         response = api_call(self, params)
         ret = {
             'msg': 'updated user ID:%s %s' % (
                 usr.user_id, base.TEST_USER_ADMIN_LOGIN),
             'user': ret
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_update_user_default_user(self):
         usr = db.User.get_default_user()
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
 
         response = api_call(self, params)
         expected = 'editing default user is forbidden'
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(UserModel, 'update_user', raise_exception)
     def test_api_update_user_when_exception_happens(self):
         usr = db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN)
         ret = jsonify(usr.get_api_data())
         id_, params = _build_data(self.apikey, 'update_user',
                                   userid=usr.user_id)
 
         response = api_call(self, params)
         ret = 'failed to update user `%s`' % usr.user_id
 
         expected = ret
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_repo(self):
         new_group = 'some_new_group'
         make_user_group(new_group)
         RepoModel().grant_user_group_permission(repo=self.REPO,
                                                 group_name=new_group,
                                                 perm='repository.read')
         meta.Session().commit()
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
         assert "tags" not in response.json['result']
         assert 'pull_requests' not in response.json['result']
 
         repo = RepoModel().get_by_repo_name(self.REPO)
         ret = repo.get_api_data()
 
         members = []
         followers = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user = user.user
             user_data = {'name': user.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
 
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group = user_group.users_group
             user_group_data = {'name': user_group.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
 
         for user in repo.followers:
             followers.append(user.user.get_api_data())
 
         ret['members'] = members
         ret['followers'] = followers
 
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_user_group(new_group)
 
         id_, params = _build_data(self.apikey, 'get_repo', repoid=self.REPO,
                                   with_revision_names=True,
                                   with_pullrequests=True)
         response = api_call(self, params)
         assert "v0.2.0" in response.json['result']['tags']
         assert 'pull_requests' in response.json['result']
 
     @base.parametrize('grant_perm', [
         ('repository.admin'),
         ('repository.write'),
         ('repository.read'),
     ])
     def test_api_get_repo_by_non_admin(self, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
         id_, params = _build_data(self.apikey_regular, 'get_repo',
                                   repoid=self.REPO)
         response = api_call(self, params)
 
         repo = RepoModel().get_by_repo_name(self.REPO)
         assert len(repo.repo_to_perm) >= 2  # make sure we actually are testing something - probably the default 2 permissions, possibly more
 
         expected = repo.get_api_data()
 
         members = []
         for user in repo.repo_to_perm:
             perm = user.permission.permission_name
             user_obj = user.user
             user_data = {'name': user_obj.username, 'type': "user",
                          'permission': perm}
             members.append(user_data)
         for user_group in repo.users_group_to_perm:
             perm = user_group.permission.permission_name
             user_group_obj = user_group.users_group
             user_group_data = {'name': user_group_obj.users_group_name,
                                'type': "user_group", 'permission': perm}
             members.append(user_group_data)
         expected['members'] = members
 
         followers = []
 
         for user in repo.followers:
             followers.append(user.user.get_api_data())
 
         expected['followers'] = followers
 
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
 
     def test_api_get_repo_by_non_admin_no_permission_to_repo(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=db.User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         try:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=self.TEST_USER_LOGIN,
                                               perm='repository.none')
 
             id_, params = _build_data(self.apikey_regular, 'get_repo',
                                       repoid=self.REPO)
             response = api_call(self, params)
 
             expected = 'repository `%s` does not exist' % (self.REPO)
             self._compare_error(id_, expected, given=response.body)
         finally:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=db.User.DEFAULT_USER_NAME,
                                               perm='repository.read')
 
     def test_api_get_repo_that_doesn_not_exist(self):
         id_, params = _build_data(self.apikey, 'get_repo',
                                   repoid='no-such-repo')
         response = api_call(self, params)
 
         ret = 'repository `%s` does not exist' % 'no-such-repo'
         expected = ret
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_repos(self):
         id_, params = _build_data(self.apikey, 'get_repos')
         response = api_call(self, params)
 
         expected = jsonify([
             repo.get_api_data()
             for repo in db.Repository.query()
         ])
 
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_repos_non_admin(self):
         id_, params = _build_data(self.apikey_regular, 'get_repos')
         response = api_call(self, params)
 
         expected = jsonify([
             repo.get_api_data()
             for repo in AuthUser(dbuser=db.User.get_by_username(self.TEST_USER_LOGIN)).get_all_user_repos()
         ])
 
         self._compare_ok(id_, expected, given=response.body)
 
     @base.parametrize('name,ret_type', [
         ('all', 'all'),
         ('dirs', 'dirs'),
         ('files', 'files'),
     ])
     def test_api_get_repo_nodes(self, name, ret_type):
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
 
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_bad_revisions(self):
         rev = 'i-dont-exist'
         path = '/'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
 
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_bad_path(self):
         rev = 'tip'
         path = '/idontexits'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path, )
         response = api_call(self, params)
 
         expected = 'failed to get repo: `%s` nodes' % self.REPO
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_repo_nodes_bad_ret_type(self):
         rev = 'tip'
         path = '/'
         ret_type = 'error'
         id_, params = _build_data(self.apikey, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
 
         expected = ('ret_type must be one of %s'
                     % (','.join(sorted(['files', 'dirs', 'all']))))
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,ret_type,grant_perm', [
         ('all', 'all', 'repository.write'),
         ('dirs', 'dirs', 'repository.admin'),
         ('files', 'files', 'repository.read'),
     ])
     def test_api_get_repo_nodes_by_regular_user(self, name, ret_type, grant_perm):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=grant_perm)
         meta.Session().commit()
 
         rev = 'tip'
         path = '/'
         id_, params = _build_data(self.apikey_regular, 'get_repo_nodes',
                                   repoid=self.REPO, revision=rev,
                                   root_path=path,
                                   ret_type=ret_type)
         response = api_call(self, params)
 
         # we don't the actual return types here since it's tested somewhere
         # else
         expected = response.json['result']
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)
 
     @base.parametrize('changing_attr,updates', [
         ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
         ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a misfeature - it would bypass access control
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('private', {'private': True}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('repo_group', {'group': 'test_group_for_update'}),
     ])
     def test_api_create_repo(self, changing_attr, updates):
         repo_name = repo_name_full = 'new_repo'
 
         if changing_attr == 'repo_group':
             group_name = updates['group']
             fixture.create_repo_group(group_name)
             repo_name_full = '/'.join([group_name, repo_name])
             updates = {}
 
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_type=self.REPO_TYPE, repo_name=repo_name_full, **updates)
         response = api_call(self, params)
 
         try:
             expected = {
                 'msg': 'Created new repository `%s`' % repo_name_full,
                 'success': True}
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to create repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
                 return
             else:
                 self._compare_ok(id_, expected, given=response.body)
 
             repo = db.Repository.get_by_repo_name(repo_name_full)
             assert repo is not None
 
             expected_data = {
                     'clone_uri': None,
                     'created_on': repo.created_on,
                     'description': repo_name,
                     'enable_downloads': False,
                     'enable_statistics': False,
                     'fork_of': None,
                     'landing_rev': ['rev', 'tip'],
                     'last_changeset': {'author': '',
                                        'date': datetime.datetime(1970, 1, 1, 0, 0),
                                        'message': '',
                                        'raw_id': '0000000000000000000000000000000000000000',
                                        'revision': -1,
                                        'short_id': '000000000000'},
                     'owner': 'test_admin',
                     'private': False,
                     'repo_id': repo.repo_id,
                     'repo_name': repo_name_full,
                     'repo_type': self.REPO_TYPE,
             }
             expected_data.update(updates)
             if changing_attr == 'landing_rev':
                 expected_data['landing_rev'] = expected_data['landing_rev'].split(':', 1)
             assert repo.get_api_data() == expected_data
         finally:
             fixture.destroy_repo(repo_name_full)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(group_name)
 
     @base.parametrize('repo_name', [
         '',
         '.',
         '..',
         ':',
         '/',
         '<test>',
     ])
     def test_api_create_repo_bad_names(self, repo_name):
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,
         )
         response = api_call(self, params)
         if repo_name == '/':
             expected = "repo group `` not found"
             self._compare_error(id_, expected, given=response.body)
         else:
             expected = "failed to create repository `%s`" % repo_name
             self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_clone_uri_local(self):
         # cloning from local repos was a misfeature - it would bypass access control
         # TODO: introduce other test coverage of actual remote cloning
         clone_uri = os.path.join(base.TESTS_TMP_PATH, self.REPO)
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,
                                   clone_uri=clone_uri,
         )
         response = api_call(self, params)
         expected = "failed to create repository `%s`" % repo_name
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_and_repo_group(self):
         repo_group_name = 'my_gr'
         repo_name = '%s/api-repo' % repo_group_name
 
         # repo creation can no longer also create repo group
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = 'repo group `%s` not found' % repo_group_name
         self._compare_error(id_, expected, given=response.body)
         assert RepoModel().get_by_repo_name(repo_name) is None
 
         # create group before creating repo
         rg = fixture.create_repo_group(repo_group_name)
         meta.Session().commit()
 
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
         }
         self._compare_ok(id_, expected, given=response.body)
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
 
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_in_repo_group_without_permission(self):
         repo_group_basename = 'api-repo-repo'
         repo_group_name = '%s/%s' % (TEST_REPO_GROUP, repo_group_basename)
         repo_name = '%s/api-repo' % repo_group_name
 
         top_group = db.RepoGroup.get_by_group_name(TEST_REPO_GROUP)
         assert top_group
         rg = fixture.create_repo_group(repo_group_basename, parent_group_id=top_group)
         meta.Session().commit()
         RepoGroupModel().grant_user_permission(repo_group_name,
                                                self.TEST_USER_LOGIN,
                                                'group.none')
         meta.Session().commit()
 
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
         )
         response = api_call(self, params)
 
         # API access control match Web access control:
         expected = 'no permission to create repo in test_repo_group/api-repo-repo'
         self._compare_error(id_, expected, given=response.body)
 
         fixture.destroy_repo(repo_name)
         fixture.destroy_repo_group(repo_group_name)
 
     def test_api_create_repo_unknown_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=owner,
                                   repo_type=self.REPO_TYPE,
         )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_repo_dont_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
         )
         response = api_call(self, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
         )
         response = api_call(self, params)
 
         repo = RepoModel().get_by_repo_name(repo_name)
         assert repo is not None
         ret = {
             'msg': 'Created new repository `%s`' % repo_name,
             'success': True,
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_by_non_admin_specify_owner(self):
         repo_name = 'api-repo'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey_regular, 'create_repo',
                                   repo_name=repo_name,
                                   repo_type=self.REPO_TYPE,
                                   owner=owner)
         response = api_call(self, params)
 
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     def test_api_create_repo_exists(self):
         repo_name = self.REPO
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = "repo `%s` already exist" % repo_name
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_repo_dot_dot(self):
         # it is only possible to create repositories in existing repo groups - and '..' can't be used
         group_name = '%s/..' % TEST_REPO_GROUP
         repo_name = '%s/%s' % (group_name, 'could-be-outside')
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = 'repo group `%s` not found' % group_name
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(repo_name)
 
     @mock.patch.object(RepoModel, 'create', raise_exception)
     def test_api_create_repo_exception_occurred(self):
         repo_name = 'api-repo'
         id_, params = _build_data(self.apikey, 'create_repo',
                                   repo_name=repo_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
                                   repo_type=self.REPO_TYPE,)
         response = api_call(self, params)
         expected = 'failed to create repository `%s`' % repo_name
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('changing_attr,updates', [
         ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
         ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a misfeature - it would bypass access control
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('private', {'private': True}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('name', {'name': 'new_repo_name'}),
         ('repo_group', {'group': 'test_group_for_update'}),
     ])
     def test_api_update_repo(self, changing_attr, updates):
         repo_name = 'api_update_me'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
 
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
 
         if changing_attr == 'name':
             repo_name = updates['name']
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name])
         expected = {
             'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
             'repository': repo.get_api_data()
         }
         expected['repository'].update(updates)
         if changing_attr == 'clone_uri' and updates['clone_uri'] is None:
             expected['repository']['clone_uri'] = ''
         if changing_attr == 'landing_rev':
             expected['repository']['landing_rev'] = expected['repository']['landing_rev'].split(':', 1)
         if changing_attr == 'name':
             expected['repository']['repo_name'] = expected['repository'].pop('name')
         if changing_attr == 'repo_group':
             expected['repository']['repo_name'] = expected['repository'].pop('group') + '/' + repo.repo_name
 
         response = api_call(self, params)
 
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to update repo `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
             else:
                 self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
 
     @base.parametrize('changing_attr,updates', [
         ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('clone_uri', {'clone_uri': 'http://example.com/repo'}), # will fail - pulling from non-existing repo should fail
         ('clone_uri', {'clone_uri': '/repo'}), # will fail - pulling from local repo was a misfeature - it would bypass access control
         ('clone_uri', {'clone_uri': None}),
         ('landing_rev', {'landing_rev': 'branch:master'}),
         ('enable_statistics', {'enable_statistics': True}),
         ('enable_downloads', {'enable_downloads': True}),
         ('name', {'name': 'new_repo_name'}),
         ('repo_group', {'group': 'test_group_for_update'}),
     ])
     def test_api_update_group_repo(self, changing_attr, updates):
         group_name = 'lololo'
         fixture.create_repo_group(group_name)
         repo_name = '%s/api_update_me' % group_name
         repo = fixture.create_repo(repo_name, repo_group=group_name, repo_type=self.REPO_TYPE)
         if changing_attr == 'repo_group':
             fixture.create_repo_group(updates['group'])
 
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         if changing_attr == 'name':
             repo_name = '%s/%s' % (group_name, updates['name'])
         if changing_attr == 'repo_group':
             repo_name = '/'.join([updates['group'], repo_name.rsplit('/', 1)[-1]])
         try:
             if changing_attr == 'clone_uri' and updates['clone_uri']:
                 expected = 'failed to update repo `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
             else:
                 expected = {
                     'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo_name),
                     'repository': repo.get_api_data()
                 }
                 self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             if changing_attr == 'repo_group':
                 fixture.destroy_repo_group(updates['group'])
         fixture.destroy_repo_group(group_name)
 
     def test_api_update_repo_repo_group_does_not_exist(self):
         repo_name = 'admin_owned'
         fixture.create_repo(repo_name)
         updates = {'group': 'test_group_for_update'}
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'repository group `%s` does not exist' % updates['group']
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_update_repo_regular_user_not_allowed(self):
         repo_name = 'admin_owned'
         fixture.create_repo(repo_name)
         updates = {'description': 'something else'}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'repository `%s` does not exist' % repo_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     @mock.patch.object(RepoModel, 'update', raise_exception)
     def test_api_update_repo_exception_occurred(self):
         repo_name = 'api_update_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         id_, params = _build_data(self.apikey, 'update_repo',
                                   repoid=repo_name, owner=base.TEST_USER_ADMIN_LOGIN,)
         response = api_call(self, params)
         try:
             expected = 'failed to update repo `%s`' % repo_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_update_repo_regular_user_change_top_level_repo_name(self):
         repo_name = 'admin_owned'
         new_repo_name = 'new_repo_name'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'no permission to create (or move) top level repositories'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
 
     def test_api_update_repo_regular_user_change_repo_name_allowed(self):
         repo_name = 'admin_owned'
         new_repo_name = 'new_repo_name'
         repo = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         UserModel().revoke_perm('default', 'hg.create.none')
         UserModel().grant_perm('default', 'hg.create.repository')
         updates = {'name': new_repo_name}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = {
                 'msg': 'updated repo ID:%s %s' % (repo.repo_id, new_repo_name),
                 'repository': repo.get_api_data()
             }
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
             fixture.destroy_repo(new_repo_name)
 
     def test_api_update_repo_regular_user_change_owner(self):
         repo_name = 'admin_owned'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         RepoModel().grant_user_permission(repo=repo_name,
                                           user=self.TEST_USER_LOGIN,
                                           perm='repository.admin')
         updates = {'owner': base.TEST_USER_ADMIN_LOGIN}
         id_, params = _build_data(self.apikey_regular, 'update_repo',
                                   repoid=repo_name, **updates)
         response = api_call(self, params)
         try:
             expected = 'Only Kallithea admin can specify `owner` param'
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_delete_repo(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
 
         id_, params = _build_data(self.apikey, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
 
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
         }
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_delete_repo_by_non_admin(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,
                             cur_user=self.TEST_USER_LOGIN)
         id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                   repoid=repo_name, )
         response = api_call(self, params)
 
         ret = {
             'msg': 'Deleted repository `%s`' % repo_name,
             'success': True
         }
         try:
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_delete_repo_by_non_admin_no_permission(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             id_, params = _build_data(self.apikey_regular, 'delete_repo',
                                       repoid=repo_name, )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (repo_name)
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_delete_repo_exception_occurred(self):
         repo_name = 'api_delete_me'
         fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)
         try:
             with mock.patch.object(RepoModel, 'delete', raise_exception):
                 id_, params = _build_data(self.apikey, 'delete_repo',
                                           repoid=repo_name, )
                 response = api_call(self, params)
 
                 expected = 'failed to delete repository `%s`' % repo_name
                 self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(repo_name)
 
     def test_api_fork_repo(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
         )
         response = api_call(self, params)
 
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
 
     @base.parametrize('fork_name', [
         'api-repo-fork',
         '%s/api-repo-fork' % TEST_REPO_GROUP,
     ])
     def test_api_fork_repo_non_admin(self, fork_name):
         RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                self.TEST_USER_LOGIN,
                                                'group.write')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
         )
         response = api_call(self, params)
 
         ret = {
             'msg': 'Created fork of `%s` as `%s`' % (self.REPO,
                                                      fork_name),
             'success': True,
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin_specify_owner(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
         )
         response = api_call(self, params)
         expected = 'Only Kallithea admin can specify `owner` param'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_non_admin_no_permission_to_fork(self):
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=db.User.DEFAULT_USER_NAME,
                                           perm='repository.none')
         fork_name = 'api-repo-fork'
         try:
             id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
             )
             response = api_call(self, params)
             expected = 'repository `%s` does not exist' % (self.REPO)
             self._compare_error(id_, expected, given=response.body)
         finally:
             RepoModel().grant_user_permission(repo=self.REPO,
                                               user=db.User.DEFAULT_USER_NAME,
                                               perm='repository.read')
             fixture.destroy_repo(fork_name)
 
     @base.parametrize('name,perm', [
         ('read', 'repository.read'),
         ('write', 'repository.write'),
         ('admin', 'repository.admin'),
     ])
     def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):
         fork_name = 'api-repo-fork'
         # regardless of base repository permission, forking is disallowed
         # when repository creation is disabled
         RepoModel().grant_user_permission(repo=self.REPO,
                                           user=self.TEST_USER_LOGIN,
                                           perm=perm)
         UserModel().revoke_perm('default', 'hg.create.repository')
         UserModel().grant_perm('default', 'hg.create.none')
         id_, params = _build_data(self.apikey_regular, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
         )
         response = api_call(self, params)
         expected = 'no permission to create top level repo'
         self._compare_error(id_, expected, given=response.body)
         fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_unknown_owner(self):
         fork_name = 'api-repo-fork'
         owner = 'i-dont-exist'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=owner,
         )
         response = api_call(self, params)
         expected = 'user `%s` does not exist' % owner
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_fork_repo_fork_exists(self):
         fork_name = 'api-repo-fork'
         fixture.create_fork(self.REPO, fork_name)
 
         try:
             fork_name = 'api-repo-fork'
 
             id_, params = _build_data(self.apikey, 'fork_repo',
                                       repoid=self.REPO,
                                       fork_name=fork_name,
                                       owner=base.TEST_USER_ADMIN_LOGIN,
             )
             response = api_call(self, params)
 
             expected = "fork `%s` already exist" % fork_name
             self._compare_error(id_, expected, given=response.body)
         finally:
             fixture.destroy_repo(fork_name)
 
     def test_api_fork_repo_repo_exists(self):
         fork_name = self.REPO
 
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
         )
         response = api_call(self, params)
 
         expected = "repo `%s` already exist" % fork_name
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'create_fork', raise_exception)
     def test_api_fork_repo_exception_occurred(self):
         fork_name = 'api-repo-fork'
         id_, params = _build_data(self.apikey, 'fork_repo',
                                   repoid=self.REPO,
                                   fork_name=fork_name,
                                   owner=base.TEST_USER_ADMIN_LOGIN,
         )
         response = api_call(self, params)
 
         expected = 'failed to fork repository `%s` as `%s`' % (self.REPO,
                                                                fork_name)
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_user_group(self):
         id_, params = _build_data(self.apikey, 'get_user_group',
                                   usergroupid=TEST_USER_GROUP)
         response = api_call(self, params)
 
         user_group = UserGroupModel().get_group(TEST_USER_GROUP)
         members = []
         for user in user_group.members:
             user = user.user
             members.append(user.get_api_data())
 
         ret = user_group.get_api_data()
         ret['members'] = members
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_user_groups(self):
         gr_name = 'test_user_group2'
         make_user_group(gr_name)
 
         try:
             id_, params = _build_data(self.apikey, 'get_user_groups', )
             response = api_call(self, params)
 
             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', raise_exception)
     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)
 
     @base.parametrize('changing_attr,updates', [
         ('group_name', {'group_name': 'new_group_name'}),
         ('group_name', {'group_name': 'test_group_for_update'}),
         ('owner', {'owner': base.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)
         try:
             id_, params = _build_data(self.apikey, 'update_user_group',
                                       usergroupid=gr_name, **updates)
             response = api_call(self, params)
             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', raise_exception)
     def test_api_update_user_group_exception_occurred(self):
         gr_name = 'test_group'
         fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'update_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             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)
         try:
             id_, params = _build_data(self.apikey, 'add_user_to_user_group',
                                       usergroupid=gr_name,
                                       userid=base.TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = {
             'msg': 'added member `%s` to user group `%s`' % (
                     base.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=base.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', raise_exception)
     def test_api_add_user_to_user_group_exception_occurred(self):
         gr_name = 'test_group'
         fixture.create_user_group(gr_name)
         try:
             id_, params = _build_data(self.apikey, 'add_user_to_user_group',
                                       usergroupid=gr_name,
                                       userid=base.TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             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=base.TEST_USER_ADMIN_LOGIN)
         meta.Session().commit()
         try:
             id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
                                       usergroupid=gr_name,
                                       userid=base.TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             expected = {
                 'msg': 'removed member `%s` from user group `%s`' % (
                     base.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', raise_exception)
     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=base.TEST_USER_ADMIN_LOGIN)
         meta.Session().commit()
         try:
             id_, params = _build_data(self.apikey, 'remove_user_from_user_group',
                                       usergroupid=gr_name,
                                       userid=base.TEST_USER_ADMIN_LOGIN)
             response = api_call(self, params)
             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
         try:
             id_, params = _build_data(self.apikey, 'delete_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             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 = 'User Group assigned to %s' % ugr_to_perm.repository.repo_name
 
         try:
             id_, params = _build_data(self.apikey, 'delete_user_group',
                                       usergroupid=gr_name)
             response = api_call(self, params)
             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_occurred(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)
 
         try:
             with mock.patch.object(UserGroupModel, 'delete', raise_exception):
                 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)
 
     @base.parametrize('name,perm', [
         ('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=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
 
         ret = {
             'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                 perm, base.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=base.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', raise_exception)
     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=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
             base.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=base.TEST_USER_ADMIN_LOGIN, )
         response = api_call(self, params)
 
         expected = {
             'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
                 base.TEST_USER_ADMIN_LOGIN, self.REPO
             ),
             'success': True
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(RepoModel, 'revoke_user_permission', raise_exception)
     def test_api_revoke_user_permission_exception_when_adding(self):
         id_, params = _build_data(self.apikey,
                                   'revoke_user_permission',
                                   repoid=self.REPO,
                                   userid=base.TEST_USER_ADMIN_LOGIN, )
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user: `%s` in repo: `%s`' % (
             base.TEST_USER_ADMIN_LOGIN, self.REPO
         )
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,perm', [
         ('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', raise_exception)
     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')
         meta.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', raise_exception)
     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)
 
     @base.parametrize('changing_attr,updates', [
-        #('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),  # currently broken
+        ('owner', {'owner': base.TEST_USER_REGULAR_LOGIN}),
         ('description', {'description': 'new description'}),
         ('group_name', {'group_name': 'new_repo_name'}),
         ('parent', {'parent': 'test_group_for_update'}),
     ])
     def test_api_update_repo_group(self, changing_attr, updates):
         group_name = 'lololo'
         repo_group = fixture.create_repo_group(group_name)
 
         new_group_name = group_name
         if changing_attr == 'group_name':
             assert repo_group.parent_group_id is None  # lazy assumption for this test
             new_group_name = updates['group_name']
         if changing_attr == 'parent':
             new_group_name = '/'.join([updates['parent'], group_name.rsplit('/', 1)[-1]])
 
         expected = {
             'msg': 'updated repository group ID:%s %s' % (repo_group.group_id, new_group_name),
             'repo_group': repo_group.get_api_data()
         }
         expected['repo_group'].update(updates)
         if 'description' in updates:
             expected['repo_group']['group_description'] = expected['repo_group'].pop('description')
 
         if changing_attr == 'parent':
             new_parent = fixture.create_repo_group(updates['parent'])
             expected['repo_group']['parent_group'] = expected['repo_group'].pop('parent')
             expected['repo_group']['group_name'] = new_group_name
 
         id_, params = _build_data(self.apikey, 'update_repo_group',
                                   repogroupid=group_name, **updates)
         response = api_call(self, params)
 
         try:
             self._compare_ok(id_, expected, given=response.body)
         finally:
             if changing_attr == 'parent':
                 fixture.destroy_repo_group(new_parent.group_id)
             fixture.destroy_repo_group(new_group_name)
 
     @base.parametrize('name,perm,apply_to_children', [
         ('none', 'group.none', 'none'),
         ('read', 'group.read', 'none'),
         ('write', 'group.write', 'none'),
         ('admin', 'group.admin', 'none'),
 
         ('none', 'group.none', 'all'),
         ('read', 'group.read', 'all'),
         ('write', 'group.write', 'all'),
         ('admin', 'group.admin', 'all'),
 
         ('none', 'group.none', 'repos'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
 
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_permission_to_repo_group(self, name, perm, apply_to_children):
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm, apply_to_children=apply_to_children)
         response = api_call(self, params)
 
         ret = {
             'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
                 perm, apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
             ),
             'success': True
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
         ('none_fails', 'group.none', 'none', False, False),
         ('read_fails', 'group.read', 'none', False, False),
         ('write_fails', 'group.write', 'none', False, False),
         ('admin_fails', 'group.admin', 'none', False, False),
 
         # with granted perms
         ('none_ok', 'group.none', 'none', True, True),
         ('read_ok', 'group.read', 'none', True, True),
         ('write_ok', 'group.write', 'none', True, True),
         ('admin_ok', 'group.admin', 'none', True, True),
     ])
     def test_api_grant_user_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok):
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
 
         id_, params = _build_data(self.apikey_regular,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm, apply_to_children=apply_to_children)
         response = api_call(self, params)
         if access_ok:
             ret = {
                 'msg': 'Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
                     perm, apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
                 ),
                 'success': True
             }
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         else:
             expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
             self._compare_error(id_, expected, given=response.body)
 
     def test_api_grant_user_permission_to_repo_group_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.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(RepoGroupModel, 'grant_user_permission', raise_exception)
     def test_api_grant_user_permission_to_repo_group_exception_when_adding(self):
         perm = 'group.read'
         id_, params = _build_data(self.apikey,
                                   'grant_user_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   perm=perm)
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
             base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
         )
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,apply_to_children', [
         ('none', 'none'),
         ('all', 'all'),
         ('repos', 'repos'),
         ('groups', 'groups'),
     ])
     def test_api_revoke_user_permission_from_repo_group(self, name, apply_to_children):
         RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
                                                user=base.TEST_USER_ADMIN_LOGIN,
                                                perm='group.read',)
         meta.Session().commit()
 
         id_, params = _build_data(self.apikey,
                                   'revoke_user_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
 
         expected = {
             'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
                 apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
             ),
             'success': True
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @base.parametrize('name,apply_to_children,grant_admin,access_ok', [
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
 
         # after granting admin rights
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
     ])
     def test_api_revoke_user_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok):
         RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
                                                user=base.TEST_USER_ADMIN_LOGIN,
                                                perm='group.read',)
         meta.Session().commit()
 
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
 
         id_, params = _build_data(self.apikey_regular,
                                   'revoke_user_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
         if access_ok:
             expected = {
                 'msg': 'Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
                     apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
                 ),
                 'success': True
             }
             self._compare_ok(id_, expected, given=response.body)
         else:
             expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
             self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'revoke_user_permission', raise_exception)
     def test_api_revoke_user_permission_from_repo_group_exception_when_adding(self):
         id_, params = _build_data(self.apikey,
                                   'revoke_user_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   userid=base.TEST_USER_ADMIN_LOGIN, )
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user: `%s` in repo group: `%s`' % (
             base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
         )
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,perm,apply_to_children', [
         ('none', 'group.none', 'none'),
         ('read', 'group.read', 'none'),
         ('write', 'group.write', 'none'),
         ('admin', 'group.admin', 'none'),
 
         ('none', 'group.none', 'all'),
         ('read', 'group.read', 'all'),
         ('write', 'group.write', 'all'),
         ('admin', 'group.admin', 'all'),
 
         ('none', 'group.none', 'repos'),
         ('read', 'group.read', 'repos'),
         ('write', 'group.write', 'repos'),
         ('admin', 'group.admin', 'repos'),
 
         ('none', 'group.none', 'groups'),
         ('read', 'group.read', 'groups'),
         ('write', 'group.write', 'groups'),
         ('admin', 'group.admin', 'groups'),
     ])
     def test_api_grant_user_group_permission_to_repo_group(self, name, perm, apply_to_children):
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
 
         ret = {
             'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                 perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
             ),
             'success': True
         }
         expected = ret
         self._compare_ok(id_, expected, given=response.body)
 
     @base.parametrize('name,perm,apply_to_children,grant_admin,access_ok', [
         ('none_fails', 'group.none', 'none', False, False),
         ('read_fails', 'group.read', 'none', False, False),
         ('write_fails', 'group.write', 'none', False, False),
         ('admin_fails', 'group.admin', 'none', False, False),
 
         # with granted perms
         ('none_ok', 'group.none', 'none', True, True),
         ('read_ok', 'group.read', 'none', True, True),
         ('write_ok', 'group.write', 'none', True, True),
         ('admin_ok', 'group.admin', 'none', True, True),
     ])
     def test_api_grant_user_group_permission_to_repo_group_by_regular_user(
             self, name, perm, apply_to_children, grant_admin, access_ok):
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
 
         id_, params = _build_data(self.apikey_regular,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
         if access_ok:
             ret = {
                 'msg': 'Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                     perm, apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
                 ),
                 'success': True
             }
             expected = ret
             self._compare_ok(id_, expected, given=response.body)
         else:
             expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
             self._compare_error(id_, expected, given=response.body)
 
     def test_api_grant_user_group_permission_to_repo_group_wrong_permission(self):
         perm = 'haha.no.permission'
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   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(RepoGroupModel, 'grant_user_group_permission', raise_exception)
     def test_api_grant_user_group_permission_exception_when_adding_to_repo_group(self):
         perm = 'group.read'
         id_, params = _build_data(self.apikey,
                                   'grant_user_group_permission_to_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   perm=perm)
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
             TEST_USER_GROUP, TEST_REPO_GROUP
         )
         self._compare_error(id_, expected, given=response.body)
 
     @base.parametrize('name,apply_to_children', [
         ('none', 'none'),
         ('all', 'all'),
         ('repos', 'repos'),
         ('groups', 'groups'),
     ])
     def test_api_revoke_user_group_permission_from_repo_group(self, name, apply_to_children):
         RepoGroupModel().grant_user_group_permission(repo_group=TEST_REPO_GROUP,
                                                      group_name=TEST_USER_GROUP,
                                                      perm='group.read',)
         meta.Session().commit()
         id_, params = _build_data(self.apikey,
                                   'revoke_user_group_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
 
         expected = {
             'msg': 'Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                 apply_to_children, TEST_USER_GROUP, TEST_REPO_GROUP
             ),
             'success': True
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @base.parametrize('name,apply_to_children,grant_admin,access_ok', [
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
 
         # after granting admin rights
         ('none', 'none', False, False),
         ('all', 'all', False, False),
         ('repos', 'repos', False, False),
         ('groups', 'groups', False, False),
     ])
     def test_api_revoke_user_group_permission_from_repo_group_by_regular_user(
             self, name, apply_to_children, grant_admin, access_ok):
         RepoGroupModel().grant_user_permission(repo_group=TEST_REPO_GROUP,
                                                user=base.TEST_USER_ADMIN_LOGIN,
                                                perm='group.read',)
         meta.Session().commit()
 
         if grant_admin:
             RepoGroupModel().grant_user_permission(TEST_REPO_GROUP,
                                                    self.TEST_USER_LOGIN,
                                                    'group.admin')
             meta.Session().commit()
 
         id_, params = _build_data(self.apikey_regular,
                                   'revoke_user_group_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,
                                   apply_to_children=apply_to_children,)
         response = api_call(self, params)
         if access_ok:
             expected = {
                 'msg': 'Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                     apply_to_children, base.TEST_USER_ADMIN_LOGIN, TEST_REPO_GROUP
                 ),
                 'success': True
             }
             self._compare_ok(id_, expected, given=response.body)
         else:
             expected = 'repository group `%s` does not exist' % TEST_REPO_GROUP
             self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(RepoGroupModel, 'revoke_user_group_permission', raise_exception)
     def test_api_revoke_user_group_permission_from_repo_group_exception_when_adding(self):
         id_, params = _build_data(self.apikey, 'revoke_user_group_permission_from_repo_group',
                                   repogroupid=TEST_REPO_GROUP,
                                   usergroupid=TEST_USER_GROUP,)
         response = api_call(self, params)
 
         expected = 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
             TEST_USER_GROUP, TEST_REPO_GROUP
         )
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_gist(self):
         gist = fixture.create_gist()
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         id_, params = _build_data(self.apikey, 'get_gist',
                                   gistid=gist_id, )
         response = api_call(self, params)
 
         expected = {
             'access_id': gist_id,
             'created_on': gist_created_on,
             'description': 'new-gist',
             'expires': -1.0,
             'gist_id': int(gist_id),
             'type': 'public',
             'url': 'http://localhost:80/_admin/gists/%s' % gist_id
         }
 
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_gist_that_does_not_exist(self):
         id_, params = _build_data(self.apikey_regular, 'get_gist',
                                   gistid='12345', )
         response = api_call(self, params)
         expected = 'gist `%s` does not exist' % ('12345',)
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_gist_private_gist_without_permission(self):
         gist = fixture.create_gist()
         gist_id = gist.gist_access_id
         gist_created_on = gist.created_on
         id_, params = _build_data(self.apikey_regular, 'get_gist',
                                   gistid=gist_id, )
         response = api_call(self, params)
 
         expected = 'gist `%s` does not exist' % gist_id
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_gists(self):
         fixture.create_gist()
         fixture.create_gist()
 
         id_, params = _build_data(self.apikey, 'get_gists')
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 2
         #self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_gists_regular_user(self):
         # by admin
         fixture.create_gist()
         fixture.create_gist()
 
         # by reg user
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
 
         id_, params = _build_data(self.apikey_regular, 'get_gists')
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 3
         #self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_gists_only_for_regular_user(self):
         # by admin
         fixture.create_gist()
         fixture.create_gist()
 
         # by reg user
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
         fixture.create_gist(owner=self.TEST_USER_LOGIN)
 
         id_, params = _build_data(self.apikey, 'get_gists',
                                   userid=self.TEST_USER_LOGIN)
         response = api_call(self, params)
         expected = response.json
         assert len(response.json['result']) == 3
         #self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_gists_regular_user_with_different_userid(self):
         id_, params = _build_data(self.apikey_regular, 'get_gists',
                                   userid=base.TEST_USER_ADMIN_LOGIN)
         response = api_call(self, params)
         expected = 'userid is not the same as your user'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_create_gist(self):
         id_, params = _build_data(self.apikey_regular, 'create_gist',
                                   lifetime=10,
                                   description='foobar-gist',
                                   gist_type='public',
                                   files={'foobar': {'content': 'foo'}})
         response = api_call(self, params)
         expected = {
             'gist': {
                 'access_id': response.json['result']['gist']['access_id'],
                 'created_on': response.json['result']['gist']['created_on'],
                 'description': 'foobar-gist',
                 'expires': response.json['result']['gist']['expires'],
                 'gist_id': response.json['result']['gist']['gist_id'],
                 'type': 'public',
                 'url': response.json['result']['gist']['url']
             },
             'msg': 'created new gist'
         }
         self._compare_ok(id_, expected, given=response.body)
 
     @mock.patch.object(GistModel, 'create', raise_exception)
     def test_api_create_gist_exception_occurred(self):
         id_, params = _build_data(self.apikey_regular, 'create_gist',
                                   files={})
         response = api_call(self, params)
         expected = 'failed to create gist'
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_delete_gist(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_delete_gist_regular_user(self):
         gist_id = fixture.create_gist(owner=self.TEST_USER_LOGIN).gist_access_id
         id_, params = _build_data(self.apikey_regular, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = {'gist': None, 'msg': 'deleted gist ID:%s' % gist_id}
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_delete_gist_regular_user_no_permission(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey_regular, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = 'gist `%s` does not exist' % (gist_id,)
         self._compare_error(id_, expected, given=response.body)
 
     @mock.patch.object(GistModel, 'delete', raise_exception)
     def test_api_delete_gist_exception_occurred(self):
         gist_id = fixture.create_gist().gist_access_id
         id_, params = _build_data(self.apikey, 'delete_gist',
                                   gistid=gist_id)
         response = api_call(self, params)
         expected = 'failed to delete gist ID:%s' % (gist_id,)
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_ip(self):
         id_, params = _build_data(self.apikey, 'get_ip')
         response = api_call(self, params)
         expected = {
             'server_ip_addr': '0.0.0.0',
             'user_ips': []
         }
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_server_info(self):
         id_, params = _build_data(self.apikey, 'get_server_info')
         response = api_call(self, params)
         expected = db.Setting.get_server_info()
         self._compare_ok(id_, expected, given=response.body)
 
     def test_api_get_changesets(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start=0, end=2)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' not in result[0]
 
     def test_api_get_changesets_with_max_revisions(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2011-02-24T00:00:00", max_revisions=10)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 10
         assert 'message' in result[0]
         assert 'added' not in result[0]
 
     def test_api_get_changesets_with_branch(self):
         if self.REPO == 'vcs_test_hg':
             branch = 'stable'
         else:
             pytest.skip("skipping due to missing branches in git test repo")
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, branch_name=branch, start_date="2011-02-24T00:00:00")
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 5
         assert 'message' in result[0]
         assert 'added' not in result[0]
 
     def test_api_get_changesets_with_file_list(self):
         id_, params = _build_data(self.apikey, 'get_changesets',
                                   repoid=self.REPO, start_date="2010-04-07T23:30:30", end_date="2010-04-08T00:31:14", with_file_list=True)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert len(result) == 3
         assert 'message' in result[0]
         assert 'added' in result[0]
 
     def test_api_get_changeset(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" not in result
 
     def test_api_get_changeset_with_reviews(self):
         reviewobjs = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION,
                                   with_reviews=True)
         response = api_call(self, params)
         result = ext_json.loads(response.body)["result"]
         assert result["raw_id"] == self.TEST_REVISION
         assert "reviews" in result
         assert len(result["reviews"]) == 1
         review = result["reviews"][0]
         expected = {
             'status': 'approved',
             'modified_at': reviewobjs[0].modified_at.replace(microsecond=0).isoformat(),
             'reviewer': 'test_admin',
         }
         assert review == expected
 
     def test_api_get_changeset_that_does_not_exist(self):
         """ Fetch changeset status for non-existant changeset.
         revision id is the above git hash used in the test above with the
         last 3 nibbles replaced with 0xf.  Should not exist for git _or_ hg.
         """
         id_, params = _build_data(self.apikey, 'get_changeset',
                                   repoid=self.REPO, raw_id = '7ab37bc680b4aa72c34d07b230c866c28e9fcfff')
         response = api_call(self, params)
         expected = 'Changeset %s does not exist' % ('7ab37bc680b4aa72c34d07b230c866c28e9fcfff',)
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_changeset_without_permission(self):
         review = fixture.review_changeset(self.REPO, self.TEST_REVISION, "approved")
         RepoModel().revoke_user_permission(repo=self.REPO, user=self.TEST_USER_LOGIN)
         RepoModel().revoke_user_permission(repo=self.REPO, user="default")
         id_, params = _build_data(self.apikey_regular, 'get_changeset',
                                   repoid=self.REPO, raw_id=self.TEST_REVISION)
         response = api_call(self, params)
         expected = 'Access denied to repo %s' % self.REPO
         self._compare_error(id_, expected, given=response.body)
 
     def test_api_get_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'get test')
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": 'get_pullrequest',
             "args": {"pullrequest_id": pull_request_id},
         }))
         response = api_call(self, params)
         pullrequest = db.PullRequest().get(pull_request_id)
         expected = {
             "status": "new",
             "pull_request_id": pull_request_id,
             "description": "No description",
             "url": "/%s/pull-request/%s/_/%s" % (self.REPO, pull_request_id, "stable"),
             "reviewers": [{"username": "test_regular"}],
             "org_repo_url": "http://localhost:80/%s" % self.REPO,
             "org_ref_parts": ["branch", "stable", self.TEST_PR_SRC],
             "other_ref_parts": ["branch", "default", self.TEST_PR_DST],
             "comments": [{"username": base.TEST_USER_ADMIN_LOGIN, "text": "",
                          "comment_id": pullrequest.comments[0].comment_id}],
             "owner": base.TEST_USER_ADMIN_LOGIN,
             "statuses": [{"status": "under_review", "reviewer": base.TEST_USER_ADMIN_LOGIN, "modified_at": "2000-01-01T00:00:00"} for i in range(0, len(self.TEST_PR_REVISIONS))],
             "title": "get test",
             "revisions": self.TEST_PR_REVISIONS,
             "created_on": "2000-01-01T00:00:00",
             "updated_on": "2000-01-01T00:00:00",
         }
         self._compare_ok(random_id, expected,
                          given=re.sub(br"\d\d\d\d\-\d\d\-\d\dT\d\d\:\d\d\:\d\d",
                                       b"2000-01-01T00:00:00", response.body))
 
     def test_api_close_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'close test')
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "close_pr": True},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == ''
         assert pullrequest.status == db.PullRequest.STATUS_CLOSED
         assert pullrequest.is_closed() == True
 
     def test_api_status_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, "status test")
 
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": db.ChangesetStatus.STATUS_APPROVED},
         }))
         response = api_call(self, params)
         pullrequest = db.PullRequest().get(pull_request_id)
         self._compare_error(random_id, "No permission to change pull request status. User needs to be admin, owner or reviewer.", given=response.body)
         assert db.ChangesetStatus.STATUS_UNDER_REVIEW == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN).api_key,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "status": db.ChangesetStatus.STATUS_APPROVED},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.ChangesetStatus.STATUS_APPROVED == ChangesetStatusModel().calculate_pull_request_result(pullrequest)[2]
 
     def test_api_comment_pullrequest(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, "comment test")
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey,
             "method": "comment_pullrequest",
             "args": {"pull_request_id": pull_request_id, "comment_msg": "Looks good to me"},
         }))
         response = api_call(self, params)
         self._compare_ok(random_id, True, given=response.body)
         pullrequest = db.PullRequest().get(pull_request_id)
         assert pullrequest.comments[-1].text == 'Looks good to me'
 
     def test_api_edit_reviewers_add_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN], 'already_present': [], 'removed': [] }
 
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_add_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": 999},
         }))
         response = api_call(self, params)
 
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
 
     def test_api_edit_reviewers_add_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ self.TEST_USER_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN, self.TEST_USER_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set()
 
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(self.TEST_USER_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_add_already_present(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {
                 "pull_request_id": pull_request_id,
                 "add": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ]
             },
         }))
         response = api_call(self, params)
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [base.TEST_USER_REGULAR_LOGIN],
                      'removed': [],
                    }
 
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_add_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = self.test_user
         PullRequestModel().close_pull_request(pull_request_id)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
 
     def test_api_edit_reviewers_add_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "add": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
 
 
     def test_api_edit_reviewers_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
 
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
                    }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_remove_nonexistent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": 999},
         }))
         response = api_call(self, params)
 
         self._compare_error(random_id, "user `999` does not exist", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_remove_nonpresent(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR2_LOGIN},
         }))
         response = api_call(self, params)
 
         # NOTE: no explicit indication that removed user was not even a reviewer
         expected = { 'added': [],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR2_LOGIN],
                    }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         prr = db.PullRequestReviewer(db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN), pullrequest)
         meta.Session().add(prr)
         meta.Session().commit()
 
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN ] },
         }))
         response = api_call(self, params)
 
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set()
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR2_LOGIN])
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_remove_closed(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         PullRequestModel().close_pull_request(pull_request_id)
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
 
         self._compare_error(random_id, "Cannot edit reviewers of a closed pull request.", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_remove_not_owner(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id, "remove": base.TEST_USER_REGULAR_LOGIN},
         }))
         response = api_call(self, params)
 
         self._compare_error(random_id, "No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.", given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_add_remove_single(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": base.TEST_USER_REGULAR2_LOGIN,
                      "remove": base.TEST_USER_REGULAR_LOGIN
                     },
         }))
         response = api_call(self, params)
 
         expected = { 'added': [base.TEST_USER_REGULAR2_LOGIN],
                      'already_present': [],
                      'removed': [base.TEST_USER_REGULAR_LOGIN],
                    }
         self._compare_ok(random_id, expected, given=response.body)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_add_remove_multiple(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         prr = db.PullRequestReviewer(db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN), pullrequest)
         meta.Session().add(prr)
         meta.Session().commit()
         assert db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) not in pullrequest.get_reviewer_users()
 
         pullrequest.owner = self.test_user
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id,
                      "add": [ base.TEST_USER_REGULAR2_LOGIN ],
                      "remove": [ base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN ],
                     },
         }))
         response = api_call(self, params)
 
         # list order depends on python sorting hash, which is randomized
         assert set(ext_json.loads(response.body)['result']['added']) == set([base.TEST_USER_REGULAR2_LOGIN])
         assert set(ext_json.loads(response.body)['result']['already_present']) == set()
         assert set(ext_json.loads(response.body)['result']['removed']) == set([base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_ADMIN_LOGIN])
         assert db.User.get_by_username(base.TEST_USER_ADMIN_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) not in pullrequest.get_reviewer_users()
         assert db.User.get_by_username(base.TEST_USER_REGULAR2_LOGIN) in pullrequest.get_reviewer_users()
 
     def test_api_edit_reviewers_invalid_params(self):
         pull_request_id = fixture.create_pullrequest(self, self.REPO, self.TEST_PR_SRC, self.TEST_PR_DST, 'edit reviewer test')
         pullrequest = db.PullRequest().get(pull_request_id)
         assert db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN) in pullrequest.get_reviewer_users()
 
         pullrequest.owner = db.User.get_by_username(base.TEST_USER_REGULAR_LOGIN)
         random_id = random.randrange(1, 9999)
         params = ascii_bytes(ext_json.dumps({
             "id": random_id,
             "api_key": self.apikey_regular,
             "method": "edit_reviewers",
             "args": {"pull_request_id": pull_request_id},
         }))
         response = api_call(self, params)
 
         self._compare_error(random_id, "Invalid request. Neither 'add' nor 'remove' is specified.", given=response.body)
         assert ext_json.loads(response.body)['result'] is None

kallithea/tests/functional/test_admin_repo_groups.py

@@ -1,150 +1,151 b''
 from kallithea.model import db, meta
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.tests import base
 from kallithea.tests.fixture import Fixture
 
 
 fixture = Fixture()
 
 
 class TestRepoGroupsController(base.TestController):
 
     def test_index(self):
         self.log_user()
         response = self.app.get(base.url('repos_groups'))
         response.mustcontain('"records": []')
 
     def test_new(self):
         self.log_user()
         response = self.app.get(base.url('new_repos_group'))
 
     def test_create(self):
         self.log_user()
 
         group_name = 'foo'
 
         # creation with form error
         response = self.app.post(base.url('repos_groups'),
                                          {'group_name': group_name,
                                           '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain('<!-- for: group_description -->')
 
         # creation
         response = self.app.post(base.url('repos_groups'),
                                          {'group_name': group_name,
                                          'group_description': 'lala',
                                          'parent_group_id': '-1',
                                          'group_copy_permissions': 'True',
                                           '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Created repository group %s' % group_name)
 
         # edit form
         response = self.app.get(base.url('edit_repo_group', group_name=group_name))
         response.mustcontain('>lala<')
 
         # edit with form error
         response = self.app.post(base.url('update_repos_group', group_name=group_name),
                                          {'group_name': group_name,
                                           '_session_csrf_secret_token': self.session_csrf_secret_token()})
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain('<!-- for: group_description -->')
 
         # edit
         response = self.app.post(base.url('update_repos_group', group_name=group_name),
                                          {'group_name': group_name,
+                                         'owner': base.TEST_USER_REGULAR2_LOGIN,
                                          'group_description': 'lolo',
                                           '_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Updated repository group %s' % group_name)
         response = response.follow()
         response.mustcontain('name="group_name" type="text" value="%s"' % group_name)
         response.mustcontain(no='<!-- for: group_description -->')
         response.mustcontain('>lolo<')
 
         # listing
         response = self.app.get(base.url('repos_groups'))
         response.mustcontain('raw_name": "%s"' % group_name)
 
         # show
         response = self.app.get(base.url('repos_group', group_name=group_name))
         response.mustcontain('href="/_admin/repo_groups/%s/edit"' % group_name)
 
         # show ignores extra trailing slashes in the URL
         response = self.app.get(base.url('repos_group', group_name='%s//' % group_name))
         response.mustcontain('href="/_admin/repo_groups/%s/edit"' % group_name)
 
         # delete
         response = self.app.post(base.url('delete_repo_group', group_name=group_name),
                                  {'_session_csrf_secret_token': self.session_csrf_secret_token()})
         self.checkSessionFlash(response, 'Removed repository group %s' % group_name)
 
     def test_new_by_regular_user(self):
         self.log_user(base.TEST_USER_REGULAR_LOGIN, base.TEST_USER_REGULAR_PASS)
         response = self.app.get(base.url('new_repos_group'), status=403)
 
     def test_case_insensitivity(self):
         self.log_user()
         group_name = 'newgroup'
         response = self.app.post(base.url('repos_groups'),
                                  fixture._get_repo_group_create_params(group_name=group_name,
                                                                  _session_csrf_secret_token=self.session_csrf_secret_token()))
         # try to create repo group with swapped case
         swapped_group_name = group_name.swapcase()
         response = self.app.post(base.url('repos_groups'),
                                  fixture._get_repo_group_create_params(group_name=swapped_group_name,
                                                                  _session_csrf_secret_token=self.session_csrf_secret_token()))
         response.mustcontain('already exists')
 
         RepoGroupModel().delete(group_name)
         meta.Session().commit()
 
     def test_subgroup_deletion(self):
         self.log_user()
         parent = None
         parent_name = 'parent'
         sub = None
         sub_name = 'sub'
         sub_path = 'parent/sub'
 
         try:
             # create parent group
             assert db.RepoGroup.guess_instance(parent_name) is None
             response = self.app.post(
                 base.url('repos_groups'),
                 fixture._get_repo_group_create_params(
                     group_name=parent_name,
                     _session_csrf_secret_token=self.session_csrf_secret_token()
                 )
             )
             parent = db.RepoGroup.guess_instance(parent_name)
             assert parent is not None
 
             # create sub group
             assert db.RepoGroup.guess_instance(sub_path) is None
             response = self.app.post(
                 base.url('repos_groups'),
                 fixture._get_repo_group_create_params(
                     group_name=sub_name,
                     parent_group_id=parent.group_id,
                     _session_csrf_secret_token=self.session_csrf_secret_token()
                 )
             )
             sub = db.RepoGroup.guess_instance(sub_path)
             assert sub is not None
 
             # delete sub group
             response = self.app.post(
                 base.url('delete_repo_group', group_name=sub_path),
                 params={
                     '_session_csrf_secret_token': self.session_csrf_secret_token()
                 },
             )
             sub = db.RepoGroup.guess_instance(sub_path)
             assert sub is None
 
         finally:
             if sub:
                 RepoGroupModel().delete(sub)
             if parent:
                 RepoGroupModel().delete(parent)
             meta.Session().commit()