repo: introduce enable_downloads and enable_statistics when creating repos

These booleans were not shown in the normal repo creation form, so the form
validation applied the "default" values of False. These values were however not
used by the model when creating repos - it just unconditionally used the real
global defaults.

The API already exposed some of this, but it wasn't implemented.

The web form for creating repos lacked these fields, but it was present in the
repo edit form. Just make these fields mandatory. There will thus not be any
defaults to apply in the model for creating repos.

kallithea/controllers/api/api.py

@@ -1,2391 +1,2391 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.api.api
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 API 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: Aug 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 
 import logging
 import traceback
 from datetime import datetime
 
 from tg import request
 
 from kallithea.controllers.api import JSONRPCController, JSONRPCError
 from kallithea.lib.auth import (AuthUser, HasPermissionAny, HasPermissionAnyDecorator, HasRepoGroupPermissionLevel, HasRepoPermissionLevel,
                                 HasUserGroupPermissionLevel)
 from kallithea.lib.exceptions import DefaultUserException, UserGroupsAssignedException
 from kallithea.lib.utils import repo2db_mapper
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import EmptyRepositoryError
 from kallithea.model import db, meta, userlog
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.comment import ChangesetCommentsModel
 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, UserGroupList
 from kallithea.model.user import UserModel
 from kallithea.model.user_group import UserGroupModel
 
 
 log = logging.getLogger(__name__)
 
 
 def store_update(updates, attr, name):
     """
     Stores param in updates dict if it's not None (i.e. if user explicitly set
     a parameter). This allows easy updates of passed in params.
     """
     if attr is not None:
         updates[name] = attr
 
 
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     user = UserModel().get_user(userid)
     if user is None:
         raise JSONRPCError("user `%s` does not exist" % (userid,))
     return user
 
 
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
 
     :param repoid:
     """
     repo = RepoModel().get_repo(repoid)
     if repo is None:
         raise JSONRPCError('repository `%s` does not exist' % (repoid,))
     return repo
 
 
 def get_repo_group_or_error(repogroupid):
     """
     Get repo group by id or name or return JsonRPCError if not found
 
     :param repogroupid:
     """
     repo_group = db.RepoGroup.guess_instance(repogroupid)
     if repo_group is None:
         raise JSONRPCError(
             'repository group `%s` does not exist' % (repogroupid,))
     return repo_group
 
 
 def get_user_group_or_error(usergroupid):
     """
     Get user group by id or name or return JsonRPCError if not found
 
     :param usergroupid:
     """
     user_group = UserGroupModel().get_group(usergroupid)
     if user_group is None:
         raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
     return user_group
 
 
 def get_perm_or_error(permid, prefix=None):
     """
     Get permission by id or name or return JsonRPCError if not found
 
     :param permid:
     """
     perm = db.Permission.get_by_key(permid)
     if perm is None:
         raise JSONRPCError('permission `%s` does not exist' % (permid,))
     if prefix:
         if not perm.permission_name.startswith(prefix):
             raise JSONRPCError('permission `%s` is invalid, '
                                'should start with %s' % (permid, prefix))
     return perm
 
 
 def get_gist_or_error(gistid):
     """
     Get gist by id or gist_access_id or return JsonRPCError if not found
 
     :param gistid:
     """
     gist = GistModel().get_gist(gistid)
     if gist is None:
         raise JSONRPCError('gist `%s` does not exist' % (gistid,))
     return gist
 
 
 class ApiController(JSONRPCController):
     """
     API Controller
 
     The authenticated user can be found as request.authuser.
 
     Example function::
 
         def func(arg1, arg2,...):
             pass
 
     Each function should also **raise** JSONRPCError for any
     errors that happens.
     """
 
     @HasPermissionAnyDecorator('hg.admin')
     def test(self, args):
         return args
 
     @HasPermissionAnyDecorator('hg.admin')
     def pull(self, repoid, clone_uri=None):
         """
         Triggers a pull from remote location on given repo. Can be used to
         automatically keep remote repos up to date. This command can be executed
         only using api_key belonging to user with admin rights
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param clone_uri: repository URI to pull from (optional)
         :type clone_uri: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 "msg" : "Pulled from `<repository name>`",
                 "repository" : "<repository name>"
             }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "Unable to pull changes from `<reponame>`"
             }
         """
         repo = get_repo_or_error(repoid)
 
         try:
             ScmModel().pull_changes(repo.repo_name,
                                     request.authuser.username,
                                     request.ip_addr,
                                     clone_uri=clone_uri)
             return dict(
                 msg='Pulled from `%s`' % repo.repo_name,
                 repository=repo.repo_name
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Unable to pull changes from `%s`' % repo.repo_name
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def rescan_repos(self, remove_obsolete=False):
         """
         Triggers rescan repositories action. If remove_obsolete is set
         than also delete repos that are in database but not in the filesystem.
         aka "clean zombies". This command can be executed only using api_key
         belonging to user with admin rights.
 
         :param remove_obsolete: deletes repositories from
             database that are not found on the filesystem
         :type remove_obsolete: Optional(bool)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 'added': [<added repository name>,...]
                 'removed': [<removed repository name>,...]
             }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 'Error occurred during rescan repositories action'
             }
         """
         try:
             rm_obsolete = remove_obsolete
             added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                             remove_obsolete=rm_obsolete)
             return {'added': added, 'removed': removed}
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during rescan repositories action'
             )
 
     def invalidate_cache(self, repoid):
         """
         Invalidate cache for repository.
         This command can be executed only using api_key belonging to user with admin
         rights or regular user that have write or admin or write access to repository.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 'msg': Cache for repository `<repository name>` was invalidated,
                 'repository': <repository name>
             }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 'Error occurred during cache invalidation action'
             }
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('write')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
         try:
             ScmModel().mark_for_invalidation(repo.repo_name)
             return dict(
                 msg='Cache for repository `%s` was invalidated' % (repoid,),
                 repository=repo.repo_name
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'Error occurred during cache invalidation action'
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_ip(self, userid=None):
         """
         Shows IP address as seen from Kallithea server, together with all
         defined IP addresses for given user. If userid is not passed data is
         returned for user who's calling this function.
         This command can be executed only using api_key belonging to user with
         admin rights.
 
         :param userid: username to show ips for
         :type userid: Optional(str or int)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                          "server_ip_addr" : "<ip_from_client>",
                          "user_ips" : [
                                         {
                                            "ip_addr" : "<ip_with_mask>",
                                            "ip_range" : ["<start_ip>", "<end_ip>"]
                                         },
                                         ...
                                       ]
             }
             error : null
         """
         if userid is None:
             userid = request.authuser.user_id
         user = get_user_or_error(userid)
         ips = db.UserIpMap.query().filter(db.UserIpMap.user == user).all()
         return dict(
             server_ip_addr=request.ip_addr,
             user_ips=ips
         )
 
     # alias for old
     show_ip = get_ip
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_server_info(self):
         """
         return server info, including Kallithea version and installed packages
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 'modules' : [ [<module name>, <module version>], ...]
                 'py_version' : <python version>,
                 'platform' : <platform type>,
                 'kallithea_version' : <kallithea version>,
                 'git_version' : '<git version>',
                 'git_path' : '<git path>'
             }
             error : null
         """
         return db.Setting.get_server_info()
 
     def get_user(self, userid=None):
         """
         Gets a user by username or user_id, Returns empty result if user is
         not found. If userid param is skipped it is set to id of user who is
         calling this method. This command can be executed only using api_key
         belonging to user with admin rights, or regular users that cannot
         specify different userid than theirs
 
         :param userid: user to get data for
         :type userid: Optional(str or int)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : None if user does not exist or
                      {
                         "user_id" :     "<user_id>",
                         "username" :    "<username>",
                         "firstname" :   "<firstname>",
                         "lastname" :    "<lastname>",
                         "email" :       "<email>",
                         "emails" :      "[<list of all emails including additional ones>]",
                         "active" :      "<bool: user active>",
                         "admin" :       "<bool: user is admin>",
                         "permissions" : {
                             "global" : ["hg.create.repository",
                                         "repository.read",
                                         "hg.register.manual_activate"],
                             "repositories" : {"repo1" : "repository.none"},
                             "repositories_groups" : {"Group1" : "group.read"},
                             "user_groups" : { "usrgrp1" : "usergroup.admin" }
                          }
                      }
             error : null
         """
         if not HasPermissionAny('hg.admin')():
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if userid is not None and userid != request.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
                 )
 
         if userid is None:
             userid = request.authuser.user_id
 
         user = get_user_or_error(userid)
         data = user.get_api_data()
         data['permissions'] = AuthUser(user_id=user.user_id).permissions
         return data
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_users(self):
         """
         Lists all existing users. This command can be executed only using api_key
         belonging to user with admin rights.
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : [<user_object>, ...]
             error : null
         """
         return [
             user.get_api_data()
             for user in db.User.query()
                 .order_by(db.User.username)
                 .filter_by(is_default_user=False)
         ]
 
     @HasPermissionAnyDecorator('hg.admin')
     def create_user(self, username, email, password='',
                     firstname='', lastname='',
                     active=True, admin=False,
                     extern_type=db.User.DEFAULT_AUTH_TYPE,
                     extern_name=''):
         """
         Creates new user. Returns new user object. This command can
         be executed only using api_key belonging to user with admin rights.
 
         :param username: new username
         :type username: str or int
         :param email: email
         :type email: str
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: str
         :param lastname: lastname
         :type lastname: str
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name: name of extern
         :type extern_name: Optional(str)
         :param extern_type: extern_type
         :type extern_type: Optional(str)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "created new user `<username>`",
                       "user" : <user_obj>
                      }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "user `<username>` already exist"
                 or
                 "email `<email>` already exist"
                 or
                 "failed to create user `<username>`"
             }
         """
         if db.User.get_by_username(username):
             raise JSONRPCError("user `%s` already exist" % (username,))
 
         if db.User.get_by_email(email):
             raise JSONRPCError("email `%s` already exist" % (email,))
 
         try:
             user = UserModel().create_or_update(
                 username=username,
                 password=password,
                 email=email,
                 firstname=firstname,
                 lastname=lastname,
                 active=active,
                 admin=admin,
                 extern_type=extern_type,
                 extern_name=extern_name
             )
             meta.Session().commit()
             return dict(
                 msg='created new user `%s`' % username,
                 user=user.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create user `%s`' % (username,))
 
     @HasPermissionAnyDecorator('hg.admin')
     def update_user(self, userid, username=None,
                     email=None, password=None,
                     firstname=None, lastname=None,
                     active=None, admin=None,
                     extern_type=None, extern_name=None):
         """
         updates given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
 
         :param userid: userid to update
         :type userid: str or int
         :param username: new username
         :type username: Optional(str or int)
         :param email: email
         :type email: Optional(str)
         :param password: password
         :type password: Optional(str)
         :param firstname: firstname
         :type firstname: Optional(str)
         :param lastname: lastname
         :type lastname: Optional(str)
         :param active: active
         :type active: Optional(bool)
         :param admin: admin
         :type admin: Optional(bool)
         :param extern_name:
         :type extern_name: Optional(str)
         :param extern_type:
         :type extern_type: Optional(str)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "updated user ID:<userid> <username>",
                       "user" : <user_object>
                      }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "failed to update user `<username>`"
             }
         """
         user = get_user_or_error(userid)
 
         # only non optional arguments will be stored in updates
         updates = {}
 
         try:
 
             store_update(updates, username, 'username')
             store_update(updates, password, 'password')
             store_update(updates, email, 'email')
             store_update(updates, firstname, 'name')
             store_update(updates, lastname, 'lastname')
             store_update(updates, active, 'active')
             store_update(updates, admin, 'admin')
             store_update(updates, extern_name, 'extern_name')
             store_update(updates, extern_type, 'extern_type')
 
             user = UserModel().update_user(user, **updates)
             meta.Session().commit()
             return dict(
                 msg='updated user ID:%s %s' % (user.user_id, user.username),
                 user=user.get_api_data()
             )
         except DefaultUserException:
             log.error(traceback.format_exc())
             raise JSONRPCError('editing default user is forbidden')
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user `%s`' % (userid,))
 
     @HasPermissionAnyDecorator('hg.admin')
     def delete_user(self, userid):
         """
         deletes given user if such user exists. This command can
         be executed only using api_key belonging to user with admin rights.
 
         :param userid: user to delete
         :type userid: str or int
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "deleted user ID:<userid> <username>",
                       "user" : null
                      }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "failed to delete user ID:<userid> <username>"
             }
         """
         user = get_user_or_error(userid)
 
         try:
             UserModel().delete(userid)
             meta.Session().commit()
             return dict(
                 msg='deleted user ID:%s %s' % (user.user_id, user.username),
                 user=None
             )
         except Exception:
 
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user ID:%s %s'
                                % (user.user_id, user.username))
 
     # permission check inside
     def get_user_group(self, usergroupid):
         """
         Gets an existing user group. This command can be executed only using api_key
         belonging to user with admin rights or user who has at least
         read access to user group.
 
         :param usergroupid: id of user_group to edit
         :type usergroupid: str or int
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : None if group not exist
                      {
                        "users_group_id" : "<id>",
                        "group_name" :     "<groupname>",
                        "group_description" : "<description>",
                        "active" :         "<bool>",
                        "owner" :          "<username>",
                        "members" :        [<user_obj>,...]
                      }
             error : null
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         data = user_group.get_api_data()
         return data
 
     # permission check inside
     def get_user_groups(self):
         """
         Lists all existing user groups. This command can be executed only using
         api_key belonging to user with admin rights or user who has at least
         read access to user group.
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : [<user_group_obj>,...]
             error : null
         """
         return [
             user_group.get_api_data()
             for user_group in UserGroupList(db.UserGroup.query().all(), perm_level='read')
         ]
 
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create_user_group(self, group_name, description='',
                           owner=None, active=True):
         """
         Creates new user group. This command can be executed only using api_key
         belonging to user with admin rights or an user who has create user group
         permission
 
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: Optional(str)
         :param owner: owner of group. If not passed apiuser is the owner
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "created new user group `<groupname>`",
                       "user_group" : <user_group_object>
                      }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "user group `<group name>` already exist"
                 or
                 "failed to create group `<group name>`"
             }
         """
         if UserGroupModel().get_by_name(group_name):
             raise JSONRPCError("user group `%s` already exist" % (group_name,))
 
         try:
             if owner is None:
                 owner = request.authuser.user_id
 
             owner = get_user_or_error(owner)
             ug = UserGroupModel().create(name=group_name, description=description,
                                          owner=owner, active=active)
             meta.Session().commit()
             return dict(
                 msg='created new user group `%s`' % group_name,
                 user_group=ug.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create group `%s`' % (group_name,))
 
     # permission check inside
     def update_user_group(self, usergroupid, group_name=None,
                           description=None, owner=None,
                           active=None):
         """
         Updates given usergroup.  This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
 
         :param usergroupid: id of user group to update
         :type usergroupid: str or int
         :param group_name: name of new user group
         :type group_name: str
         :param description: group description
         :type description: str
         :param owner: owner of group.
         :type owner: Optional(str or int)
         :param active: group is active
         :type active: Optional(bool)
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             "msg" : 'updated user group ID:<user group id> <user group name>',
             "user_group" : <user_group_object>
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to update user group `<user group name>`"
           }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         if owner is not None:
             owner = get_user_or_error(owner)
 
         updates = {}
         store_update(updates, group_name, 'users_group_name')
         store_update(updates, description, 'user_group_description')
         store_update(updates, owner, 'owner')
         store_update(updates, active, 'users_group_active')
         try:
             UserGroupModel().update(user_group, updates)
             meta.Session().commit()
             return dict(
                 msg='updated user group ID:%s %s' % (user_group.users_group_id,
                                                      user_group.users_group_name),
                 user_group=user_group.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
 
     # permission check inside
     def delete_user_group(self, usergroupid):
         """
         Delete given user group by user group id or name.
         This command can be executed only using api_key
         belonging to user with admin rights or an admin of given user group
 
         :param usergroupid:
         :type usergroupid: str or int
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             "msg" : "deleted user group ID:<user_group_id> <user_group_name>"
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete user group ID:<user_group_id> <user_group_name>"
             or
             "RepoGroup assigned to <repo_groups_list>"
           }
         """
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         try:
             UserGroupModel().delete(user_group)
             meta.Session().commit()
             return dict(
                 msg='deleted user group ID:%s %s' %
                     (user_group.users_group_id, user_group.users_group_name),
                 user_group=None
             )
         except UserGroupsAssignedException as e:
             log.error(traceback.format_exc())
             raise JSONRPCError(str(e))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete user group ID:%s %s' %
                                (user_group.users_group_id,
                                 user_group.users_group_name)
                                )
 
     # permission check inside
     def add_user_to_user_group(self, usergroupid, userid):
         """
         Adds a user to a user group. If user exists in that group success will be
         `false`. This command can be executed only using api_key
         belonging to user with admin rights or an admin of a given user group
 
         :param usergroupid:
         :type usergroupid: str or int
         :param userid:
         :type userid: str or int
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 "success" : True|False # depends on if member is in group
                 "msg" : "added member `<username>` to a user group `<groupname>` |
                          User is already in that group"
             }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "failed to add member to user group `<user_group_name>`"
             }
         """
         user = get_user_or_error(userid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         try:
             ugm = UserGroupModel().add_user_to_group(user_group, user)
             success = True if ugm is not True else False
             msg = 'added member `%s` to user group `%s`' % (
                 user.username, user_group.users_group_name
             )
             msg = msg if success else 'User is already in that group'
             meta.Session().commit()
 
             return dict(
                 success=success,
                 msg=msg
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to add member to user group `%s`' % (
                     user_group.users_group_name,
                 )
             )
 
     # permission check inside
     def remove_user_from_user_group(self, usergroupid, userid):
         """
         Removes a user from a user group. If user is not in given group success will
         be `false`. This command can be executed only
         using api_key belonging to user with admin rights or an admin of given user group
 
         :param usergroupid:
         :param userid:
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "success" : True|False,  # depends on if member is in group
                       "msg" : "removed member <username> from user group <groupname> |
                                User wasn't in group"
                      }
             error : null
         """
         user = get_user_or_error(userid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         try:
             success = UserGroupModel().remove_user_from_group(user_group, user)
             msg = 'removed member `%s` from user group `%s`' % (
                 user.username, user_group.users_group_name
             )
             msg = msg if success else "User wasn't in group"
             meta.Session().commit()
             return dict(success=success, msg=msg)
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to remove member from user group `%s`' % (
                     user_group.users_group_name,
                 )
             )
 
     # permission check inside
     def get_repo(self, repoid,
                  with_revision_names=False,
                  with_pullrequests=False):
         """
         Gets an existing repository by it's name or repository_id. Members will return
         either users_group or user associated to that repository. This command can be
         executed only using api_key belonging to user with admin
         rights or regular user that have at least read access to repository.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                         "repo_id" :          "<repo_id>",
                         "repo_name" :        "<reponame>",
                         "repo_type" :        "<repo_type>",
                         "clone_uri" :        "<clone_uri>",
                         "enable_downloads" : "<bool>",
                         "enable_statistics": "<bool>",
                         "private" :          "<bool>",
                         "created_on" :       "<date_time_created>",
                         "description" :      "<description>",
                         "landing_rev" :      "<landing_rev>",
                         "last_changeset" :   {
                                                  "author" :  "<full_author>",
                                                  "date" :    "<date_time_of_commit>",
                                                  "message" : "<commit_message>",
                                                  "raw_id" :  "<raw_id>",
                                                  "revision": "<numeric_revision>",
                                                  "short_id": "<short_id>"
                                              },
                         "owner" :            "<repo_owner>",
                         "fork_of" :          "<name_of_fork_parent>",
                         "members" :     [
                                             {
                                                 "name" :    "<username>",
                                                 "type" :    "user",
                                                 "permission" : "repository.(read|write|admin)"
                                             },
                                             …
                                             {
                                                 "name" :    "<usergroup name>",
                                                 "type" :    "user_group",
                                                 "permission" : "usergroup.(read|write|admin)"
                                             },
                                             …
                                         ],
                         "followers" :  [<user_obj>, ...],
                         <if with_revision_names == True>
                         "tags" : {
                                     "<tagname>" : "<raw_id>",
                                     ...
                                 },
                         "branches" : {
                                     "<branchname>" : "<raw_id>",
                                     ...
                                 },
                         "bookmarks" : {
                                     "<bookmarkname>" : "<raw_id>",
                                     ...
                                 }
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
 
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('read')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
         members = []
         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)
 
         followers = [
             uf.user.get_api_data()
             for uf in repo.followers
         ]
 
         data = repo.get_api_data(with_revision_names=with_revision_names,
                                  with_pullrequests=with_pullrequests)
         data['members'] = members
         data['followers'] = followers
         return data
 
     # permission check inside
     def get_repos(self):
         """
         Lists all existing repositories. This command can be executed only using
         api_key belonging to user with admin rights or regular user that have
         admin, write or read access to repository.
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : [
                       {
                         "repo_id" :          "<repo_id>",
                         "repo_name" :        "<reponame>",
                         "repo_type" :        "<repo_type>",
                         "clone_uri" :        "<clone_uri>",
                         "private" :          "<bool>",
                         "created_on" :       "<datetimecreated>",
                         "description" :      "<description>",
                         "landing_rev" :      "<landing_rev>",
                         "owner" :            "<repo_owner>",
                         "fork_of" :          "<name_of_fork_parent>",
                         "enable_downloads" : "<bool>",
                         "enable_statistics": "<bool>"
                       },
                       …
                      ]
             error : null
         """
         if not HasPermissionAny('hg.admin')():
             repos = request.authuser.get_all_user_repos()
         else:
             repos = db.Repository.query()
 
         return [
             repo.get_api_data()
             for repo in repos
         ]
 
     # permission check inside
     def get_repo_nodes(self, repoid, revision, root_path,
                        ret_type='all'):
         """
         returns a list of nodes and it's children in a flat list for a given path
         at given revision. It's possible to specify ret_type to show only `files` or
         `dirs`.  This command can be executed only using api_key belonging to
         user with admin rights or regular user that have at least read access to repository.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param revision: revision for which listing should be done
         :type revision: str
         :param root_path: path from which start displaying
         :type root_path: str
         :param ret_type: return type 'all|files|dirs' nodes
         :type ret_type: Optional(str)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : [
                       {
                         "name" :        "<name>",
                         "type" :        "<type>"
                       },
                       …
                      ]
             error : null
         """
         repo = get_repo_or_error(repoid)
 
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('read')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
         _map = {}
         try:
             _d, _f = ScmModel().get_nodes(repo, revision, root_path,
                                           flat=False)
             _map = {
                 'all': _d + _f,
                 'files': _f,
                 'dirs': _d,
             }
             return _map[ret_type]
         except KeyError:
             raise JSONRPCError('ret_type must be one of %s'
                                % (','.join(sorted(_map))))
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to get repo: `%s` nodes' % repo.repo_name
             )
 
     # permission check inside
     def create_repo(self, repo_name, owner=None,
                     repo_type=None, description='',
                     private=False, clone_uri=None,
                     landing_rev='rev:tip',
                     enable_statistics=None,
                     enable_downloads=None,
                     copy_permissions=False):
         """
         Creates a repository. The repository name contains the full path, but the
         parent repository group must exist. For example "foo/bar/baz" require the groups
         "foo" and "bar" (with "foo" as parent), and create "baz" repository with
         "bar" as group. This command can be executed only using api_key
         belonging to user with admin rights or regular user that have create
         repository permission. Regular users cannot specify owner parameter
 
         :param repo_name: repository name
         :type repo_name: str
         :param owner: user_id or username
         :type owner: Optional(str)
         :param repo_type: 'hg' or 'git'
         :type repo_type: Optional(str)
         :param description: repository description
         :type description: Optional(str)
         :param private:
         :type private: bool
         :param clone_uri:
         :type clone_uri: str
         :param landing_rev: <rev_type>:<rev>
         :type landing_rev: str
         :param enable_downloads:
         :type enable_downloads: bool
         :param enable_statistics:
         :type enable_statistics: bool
         :param copy_permissions: Copy permission from group that repository is
             being created.
         :type copy_permissions: bool
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Created new repository `<reponame>`",
                       "success" : true
                      }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 'failed to create repository `<repo_name>`
             }
         """
         group_name = None
         repo_name_parts = repo_name.split('/')
         if len(repo_name_parts) > 1:
             group_name = '/'.join(repo_name_parts[:-1])
             repo_group = db.RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
 
         if not HasPermissionAny('hg.admin')():
             if owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
                 )
         if owner is None:
             owner = request.authuser.user_id
 
         owner = get_user_or_error(owner)
 
         if RepoModel().get_by_repo_name(repo_name):
             raise JSONRPCError("repo `%s` already exist" % repo_name)
 
         defs = db.Setting.get_default_repo_settings(strip_prefix=True)
         if private is None:
             private = defs.get('repo_private') or False
         if repo_type is None:
             repo_type = defs.get('repo_type')
         if enable_statistics is None:
             enable_statistics = defs.get('repo_enable_statistics')
         if enable_downloads is None:
             enable_downloads = defs.get('repo_enable_downloads')
 
         try:
             data = dict(
                 repo_name=repo_name_parts[-1],
                 repo_name_full=repo_name,
                 repo_type=repo_type,
                 repo_description=description,
                 repo_private=private,
                 clone_uri=clone_uri,
                 repo_group=group_name,
                 repo_landing_rev=landing_rev,
-                enable_statistics=enable_statistics,
-                enable_downloads=enable_downloads,
+                repo_enable_statistics=enable_statistics,
+                repo_enable_downloads=enable_downloads,
                 repo_copy_permissions=copy_permissions,
             )
 
             RepoModel().create(form_data=data, cur_user=owner.username)
             # no commit, it's done in RepoModel, or async via celery
             return dict(
                 msg="Created new repository `%s`" % (repo_name,),
                 success=True,  # cannot return the repo data here since fork
                                # can be done async
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to create repository `%s`' % (repo_name,))
 
     # permission check inside
     def update_repo(self, repoid, name=None,
                     owner=None,
                     group=None,
                     description=None, private=None,
                     clone_uri=None, landing_rev=None,
                     enable_statistics=None,
                     enable_downloads=None):
         """
         Updates repo
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param name:
         :param owner:
         :param group:
         :param description:
         :param private:
         :param clone_uri:
         :param landing_rev:
         :param enable_statistics:
         :param enable_downloads:
         """
         repo = get_repo_or_error(repoid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
             if (name != repo.repo_name and repo.group_id is None and
                 not HasPermissionAny('hg.create.repository')()
             ):
                 raise JSONRPCError('no permission to create (or move) top level repositories')
 
             if owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
                 )
 
         updates = {}
         repo_group = group
         if repo_group is not None:
             repo_group = get_repo_group_or_error(repo_group)  # TODO: repos can thus currently not be moved to root
             if repo_group.group_id != repo.group_id:
                 if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(repo_group.group_name)):
                     raise JSONRPCError("no permission to create (or move) repo in %s" % repo_group.group_name)
             repo_group = repo_group.group_id
         try:
             store_update(updates, name, 'repo_name')
             store_update(updates, repo_group, 'repo_group')
             store_update(updates, owner, 'owner')
             store_update(updates, description, 'repo_description')
             store_update(updates, private, 'repo_private')
             store_update(updates, clone_uri, 'clone_uri')
             store_update(updates, landing_rev, 'repo_landing_rev')
             store_update(updates, enable_statistics, 'repo_enable_statistics')
             store_update(updates, enable_downloads, 'repo_enable_downloads')
 
             RepoModel().update(repo, **updates)
             meta.Session().commit()
             return dict(
                 msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
                 repository=repo.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update repo `%s`' % repoid)
 
     # permission check inside
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     def fork_repo(self, repoid, fork_name,
                   owner=None,
                   description='', copy_permissions=False,
                   private=False, landing_rev='rev:tip'):
         """
         Creates a fork of given repo. In case of using celery this will
         immediately return success message, while fork is going to be created
         asynchronous. This command can be executed only using api_key belonging to
         user with admin rights or regular user that have fork permission, and at least
         read access to forking repository. Regular users cannot specify owner parameter.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param fork_name:
         :param owner:
         :param description:
         :param copy_permissions:
         :param private:
         :param landing_rev:
 
         INPUT::
 
             id : <id_for_response>
             api_key : "<api_key>"
             method :  "fork_repo"
             args :    {
                         "repoid" :          "<reponame or repo_id>",
                         "fork_name" :       "<forkname>",
                         "owner" :           "<username or user_id = Optional(=apiuser)>",
                         "description" :     "<description>",
                         "copy_permissions": "<bool>",
                         "private" :         "<bool>",
                         "landing_rev" :     "<landing_rev>"
                       }
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Created fork of `<reponame>` as `<forkname>`",
                       "success" : true
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
         repo_name = repo.repo_name
 
         _repo = RepoModel().get_by_repo_name(fork_name)
         if _repo:
             type_ = 'fork' if _repo.fork else 'repo'
             raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))
 
         group_name = None
         fork_name_parts = fork_name.split('/')
         if len(fork_name_parts) > 1:
             group_name = '/'.join(fork_name_parts[:-1])
             repo_group = db.RepoGroup.get_by_group_name(group_name)
             if repo_group is None:
                 raise JSONRPCError("repo group `%s` not found" % group_name)
             if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):
                 raise JSONRPCError("no permission to create repo in %s" % group_name)
         else:
             if not HasPermissionAny('hg.admin', 'hg.create.repository')():
                 raise JSONRPCError("no permission to create top level repo")
 
         if HasPermissionAny('hg.admin')():
             pass
         elif HasRepoPermissionLevel('read')(repo.repo_name):
             if owner is not None:
                 # forbid setting owner for non-admins
                 raise JSONRPCError(
                     'Only Kallithea admin can specify `owner` param'
                 )
         else:
             raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
         if owner is None:
             owner = request.authuser.user_id
 
         owner = get_user_or_error(owner)
 
         try:
             form_data = dict(
                 repo_name=fork_name_parts[-1],
                 repo_name_full=fork_name,
                 repo_group=group_name,
                 repo_type=repo.repo_type,
                 description=description,
                 private=private,
                 copy_permissions=copy_permissions,
                 landing_rev=landing_rev,
                 update_after_clone=False,
                 fork_parent_id=repo.repo_id,
             )
             RepoModel().create_fork(form_data, cur_user=owner.username)
             # no commit, it's done in RepoModel, or async via celery
             return dict(
                 msg='Created fork of `%s` as `%s`' % (repo.repo_name,
                                                       fork_name),
                 success=True,  # cannot return the repo data here since fork
                                # can be done async
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to fork repository `%s` as `%s`' % (repo_name,
                                                             fork_name)
             )
 
     # permission check inside
     def delete_repo(self, repoid, forks=''):
         """
         Deletes a repository. This command can be executed only using api_key belonging
         to user with admin rights or regular user that have admin access to repository.
         When `forks` param is set it's possible to detach or delete forks of deleting
         repository
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param forks: `detach` or `delete`, what do do with attached forks for repo
         :type forks: Optional(str)
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Deleted repository `<reponame>`",
                       "success" : true
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
 
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
         try:
             handle_forks = forks
             _forks_msg = ''
             _forks = [f for f in repo.forks]
             if handle_forks == 'detach':
                 _forks_msg = ' ' + 'Detached %s forks' % len(_forks)
             elif handle_forks == 'delete':
                 _forks_msg = ' ' + 'Deleted %s forks' % len(_forks)
             elif _forks:
                 raise JSONRPCError(
                     'Cannot delete `%s` it still contains attached forks' %
                     (repo.repo_name,)
                 )
 
             RepoModel().delete(repo, forks=forks)
             meta.Session().commit()
             return dict(
                 msg='Deleted repository `%s`%s' % (repo.repo_name, _forks_msg),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to delete repository `%s`' % (repo.repo_name,)
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def grant_user_permission(self, repoid, userid, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found. This command can be executed only using api_key belonging to user
         with admin rights.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param userid:
         :param perm: (repository.(none|read|write|admin))
         :type perm: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
                       "success" : true
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         perm = get_perm_or_error(perm)
 
         try:
 
             RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)
 
             meta.Session().commit()
             return dict(
                 msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (
                     perm.permission_name, user.username, repo.repo_name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
                 )
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def revoke_user_permission(self, repoid, userid):
         """
         Revoke permission for user on given repository. This command can be executed
         only using api_key belonging to user with admin rights.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param userid:
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
                       "success" : true
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
         user = get_user_or_error(userid)
         try:
             RepoModel().revoke_user_permission(repo=repo, user=user)
             meta.Session().commit()
             return dict(
                 msg='Revoked perm for user: `%s` in repo: `%s`' % (
                     user.username, repo.repo_name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo: `%s`' % (
                     userid, repoid
                 )
             )
 
     # permission check inside
     def grant_user_group_permission(self, repoid, usergroupid, perm):
         """
         Grant permission for user group on given repository, or update
         existing one if found. This command can be executed only using
         api_key belonging to user with admin rights.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param usergroupid: id of usergroup
         :type usergroupid: str or int
         :param perm: (repository.(none|read|write|admin))
         :type perm: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                 "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
                 "success" : true
             }
             error : null
 
         ERROR OUTPUT::
 
             id : <id_given_in_input>
             result : null
             error :  {
                 "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
             }
         """
         repo = get_repo_or_error(repoid)
         perm = get_perm_or_error(perm)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
             if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         try:
             RepoModel().grant_user_group_permission(
                 repo=repo, group_name=user_group, perm=perm)
 
             meta.Session().commit()
             return dict(
                 msg='Granted perm: `%s` for user group: `%s` in '
                     'repo: `%s`' % (
                         perm.permission_name, user_group.users_group_name,
                         repo.repo_name
                     ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user group: `%s` in '
                 'repo: `%s`' % (
                     usergroupid, repo.repo_name
                 )
             )
 
     # permission check inside
     def revoke_user_group_permission(self, repoid, usergroupid):
         """
         Revoke permission for user group on given repository. This command can be
         executed only using api_key belonging to user with admin rights.
 
         :param repoid: repository name or repository id
         :type repoid: str or int
         :param usergroupid:
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
                       "success" : true
                      }
             error : null
         """
         repo = get_repo_or_error(repoid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoPermissionLevel('admin')(repo.repo_name):
                 raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 
             if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
                 raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))
 
         try:
             RepoModel().revoke_user_group_permission(
                 repo=repo, group_name=user_group)
 
             meta.Session().commit()
             return dict(
                 msg='Revoked perm for user group: `%s` in repo: `%s`' % (
                     user_group.users_group_name, repo.repo_name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user group: `%s` in '
                 'repo: `%s`' % (
                     user_group.users_group_name, repo.repo_name
                 )
             )
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_repo_group(self, repogroupid):
         """
         Returns given repo group together with permissions, and repositories
         inside the group
 
         :param repogroupid: id/name of repository group
         :type repogroupid: str or int
         """
         repo_group = get_repo_group_or_error(repogroupid)
 
         members = []
         for user in repo_group.repo_group_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_group.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)
 
         data = repo_group.get_api_data()
         data["members"] = members
         return data
 
     @HasPermissionAnyDecorator('hg.admin')
     def get_repo_groups(self):
         """
         Returns all repository groups
         """
         return [
             repo_group.get_api_data()
             for repo_group in db.RepoGroup.query()
         ]
 
     @HasPermissionAnyDecorator('hg.admin')
     def create_repo_group(self, group_name, description='',
                           owner=None,
                           parent=None,
                           copy_permissions=False):
         """
         Creates a repository group. This command can be executed only using
         api_key belonging to user with admin rights.
 
         :param group_name:
         :type group_name:
         :param description:
         :type description:
         :param owner:
         :type owner:
         :param parent:
         :type parent:
         :param copy_permissions:
         :type copy_permissions:
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
               "msg" : "created new repo group `<repo_group_name>`",
               "repo_group" : <repogroup_object>
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             failed to create repo group `<repogroupid>`
           }
         """
         if db.RepoGroup.get_by_group_name(group_name):
             raise JSONRPCError("repo group `%s` already exist" % (group_name,))
 
         if owner is None:
             owner = request.authuser.user_id
         group_description = description
         parent_group = None
         if parent is not None:
             parent_group = get_repo_group_or_error(parent)
 
         try:
             repo_group = RepoGroupModel().create(
                 group_name=group_name,
                 group_description=group_description,
                 owner=owner,
                 parent=parent_group,
                 copy_permissions=copy_permissions
             )
             meta.Session().commit()
             return dict(
                 msg='created new repo group `%s`' % group_name,
                 repo_group=repo_group.get_api_data()
             )
         except Exception:
 
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create repo group `%s`' % (group_name,))
 
     @HasPermissionAnyDecorator('hg.admin')
     def update_repo_group(self, repogroupid, group_name=None,
                           description=None,
                           owner=None,
                           parent=None):
         repo_group = get_repo_group_or_error(repogroupid)
 
         updates = {}
         try:
             store_update(updates, group_name, 'group_name')
             store_update(updates, description, 'group_description')
             store_update(updates, owner, 'owner')
             store_update(updates, parent, 'parent_group')
             repo_group = RepoGroupModel().update(repo_group, updates)
             meta.Session().commit()
             return dict(
                 msg='updated repository group ID:%s %s' % (repo_group.group_id,
                                                            repo_group.group_name),
                 repo_group=repo_group.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to update repository group `%s`'
                                % (repogroupid,))
 
     @HasPermissionAnyDecorator('hg.admin')
     def delete_repo_group(self, repogroupid):
         """
         :param repogroupid: name or id of repository group
         :type repogroupid: str or int
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             'msg' : 'deleted repo group ID:<repogroupid> <repogroupname>
             'repo_group' : null
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete repo group ID:<repogroupid> <repogroupname>"
           }
         """
         repo_group = get_repo_group_or_error(repogroupid)
 
         try:
             RepoGroupModel().delete(repo_group)
             meta.Session().commit()
             return dict(
                 msg='deleted repo group ID:%s %s' %
                     (repo_group.group_id, repo_group.group_name),
                 repo_group=None
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete repo group ID:%s %s' %
                                (repo_group.group_id, repo_group.group_name)
                                )
 
     # permission check inside
     def grant_user_permission_to_repo_group(self, repogroupid, userid,
                                             perm, apply_to_children='none'):
         """
         Grant permission for user on given repository group, or update existing
         one if found. This command can be executed only using api_key belonging
         to user with admin rights, or user who has admin right to given repository
         group.
 
         :param repogroupid: name or id of repository group
         :type repogroupid: str or int
         :param userid:
         :param perm: (group.(none|read|write|admin))
         :type perm: str
         :param apply_to_children: 'none', 'repos', 'groups', 'all'
         :type apply_to_children: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
                       "success" : true
                      }
             error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
           }
         """
         repo_group = get_repo_group_or_error(repogroupid)
 
         if not HasPermissionAny('hg.admin')():
             if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
                 raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
 
         user = get_user_or_error(userid)
         perm = get_perm_or_error(perm, prefix='group.')
 
         try:
             RepoGroupModel().add_permission(repo_group=repo_group,
                                             obj=user,
                                             obj_type="user",
                                             perm=perm,
                                             recursive=apply_to_children)
             meta.Session().commit()
             return dict(
                 msg='Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (
                     perm.permission_name, apply_to_children, user.username, repo_group.name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo group: `%s`' % (
                     userid, repo_group.name))
 
     # permission check inside
     def revoke_user_permission_from_repo_group(self, repogroupid, userid,
                                                apply_to_children='none'):
         """
         Revoke permission for user on given repository group. This command can
         be executed only using api_key belonging to user with admin rights, or
         user who has admin right to given repository group.
 
         :param repogroupid: name or id of repository group
         :type repogroupid: str or int
         :param userid:
         :type userid:
         :param apply_to_children: 'none', 'repos', 'groups', 'all'
         :type apply_to_children: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Revoked perm (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
                       "success" : true
                      }
             error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
           }
         """
         repo_group = get_repo_group_or_error(repogroupid)
 
         if not HasPermissionAny('hg.admin')():
             if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
                 raise JSONRPCError('repository group `%s` does not exist' % (repogroupid,))
 
         user = get_user_or_error(userid)
 
         try:
             RepoGroupModel().delete_permission(repo_group=repo_group,
                                                obj=user,
                                                obj_type="user",
                                                recursive=apply_to_children)
 
             meta.Session().commit()
             return dict(
                 msg='Revoked perm (recursive:%s) for user: `%s` in repo group: `%s`' % (
                     apply_to_children, user.username, repo_group.name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user: `%s` in repo group: `%s`' % (
                     userid, repo_group.name))
 
     # permission check inside
     def grant_user_group_permission_to_repo_group(
             self, repogroupid, usergroupid, perm,
             apply_to_children='none'):
         """
         Grant permission for user group on given repository group, or update
         existing one if found. This command can be executed only using
         api_key belonging to user with admin rights, or user who has admin
         right to given repository group.
 
         :param repogroupid: name or id of repository group
         :type repogroupid: str or int
         :param usergroupid: id of usergroup
         :type usergroupid: str or int
         :param perm: (group.(none|read|write|admin))
         :type perm: str
         :param apply_to_children: 'none', 'repos', 'groups', 'all'
         :type apply_to_children: str
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
             "success" : true
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
           }
         """
         repo_group = get_repo_group_or_error(repogroupid)
         perm = get_perm_or_error(perm, prefix='group.')
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
                 raise JSONRPCError(
                     'repository group `%s` does not exist' % (repogroupid,))
 
             if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
                 raise JSONRPCError(
                     'user group `%s` does not exist' % (usergroupid,))
 
         try:
             RepoGroupModel().add_permission(repo_group=repo_group,
                                             obj=user_group,
                                             obj_type="user_group",
                                             perm=perm,
                                             recursive=apply_to_children)
             meta.Session().commit()
             return dict(
                 msg='Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                     perm.permission_name, apply_to_children,
                     user_group.users_group_name, repo_group.name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user group: `%s` in '
                 'repo group: `%s`' % (
                     usergroupid, repo_group.name
                 )
             )
 
     # permission check inside
     def revoke_user_group_permission_from_repo_group(
             self, repogroupid, usergroupid,
             apply_to_children='none'):
         """
         Revoke permission for user group on given repository. This command can be
         executed only using api_key belonging to user with admin rights, or
         user who has admin right to given repository group.
 
         :param repogroupid: name or id of repository group
         :type repogroupid: str or int
         :param usergroupid:
         :param apply_to_children: 'none', 'repos', 'groups', 'all'
         :type apply_to_children: str
 
         OUTPUT::
 
             id : <id_given_in_input>
             result : {
                       "msg" : "Revoked perm (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
                       "success" : true
                      }
             error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
           }
         """
         repo_group = get_repo_group_or_error(repogroupid)
         user_group = get_user_group_or_error(usergroupid)
         if not HasPermissionAny('hg.admin')():
             if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):
                 raise JSONRPCError(
                     'repository group `%s` does not exist' % (repogroupid,))
 
             if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):
                 raise JSONRPCError(
                     'user group `%s` does not exist' % (usergroupid,))
 
         try:
             RepoGroupModel().delete_permission(repo_group=repo_group,
                                                obj=user_group,
                                                obj_type="user_group",
                                                recursive=apply_to_children)
             meta.Session().commit()
             return dict(
                 msg='Revoked perm (recursive:%s) for user group: `%s` in repo group: `%s`' % (
                     apply_to_children, user_group.users_group_name, repo_group.name
                 ),
                 success=True
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError(
                 'failed to edit permission for user group: `%s` in repo group: `%s`' % (
                     user_group.users_group_name, repo_group.name
                 )
             )
 
     def get_gist(self, gistid):
         """
         Get given gist by id
 
         :param gistid: id of private or public gist
         :type gistid: str
         """
         gist = get_gist_or_error(gistid)
         if not HasPermissionAny('hg.admin')():
             if gist.owner_id != request.authuser.user_id:
                 raise JSONRPCError('gist `%s` does not exist' % (gistid,))
         return gist.get_api_data()
 
     def get_gists(self, userid=None):
         """
         Get all gists for given user. If userid is empty returned gists
         are for user who called the api
 
         :param userid: user to get gists for
         :type userid: Optional(str or int)
         """
         if not HasPermissionAny('hg.admin')():
             # make sure normal user does not pass someone else userid,
             # he is not allowed to do that
             if userid is not None and userid != request.authuser.user_id:
                 raise JSONRPCError(
                     'userid is not the same as your user'
                 )
 
         if userid is None:
             user_id = request.authuser.user_id
         else:
             user_id = get_user_or_error(userid).user_id
 
         return [
             gist.get_api_data()
             for gist in db.Gist().query()
                 .filter_by(is_expired=False)
                 .filter(db.Gist.owner_id == user_id)
                 .order_by(db.Gist.created_on.desc())
         ]
 
     def create_gist(self, files, owner=None,
                     gist_type=db.Gist.GIST_PUBLIC, lifetime=-1,
                     description=''):
         """
         Creates new Gist
 
         :param files: files to be added to gist
             {'filename': {'content':'...', 'lexer': null},
              'filename2': {'content':'...', 'lexer': null}}
         :type files: dict
         :param owner: gist owner, defaults to api method caller
         :type owner: Optional(str or int)
         :param gist_type: type of gist 'public' or 'private'
         :type gist_type: Optional(str)
         :param lifetime: time in minutes of gist lifetime
         :type lifetime: Optional(int)
         :param description: gist description
         :type description: str
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             "msg" : "created new gist",
             "gist" : <gist_object>
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to create gist"
           }
         """
         try:
             if owner is None:
                 owner = request.authuser.user_id
 
             owner = get_user_or_error(owner)
 
             gist = GistModel().create(description=description,
                                       owner=owner,
                                       ip_addr=request.ip_addr,
                                       gist_mapping=files,
                                       gist_type=gist_type,
                                       lifetime=lifetime)
             meta.Session().commit()
             return dict(
                 msg='created new gist',
                 gist=gist.get_api_data()
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to create gist')
 
     # permission check inside
     def delete_gist(self, gistid):
         """
         Deletes existing gist
 
         :param gistid: id of gist to delete
         :type gistid: str
 
         OUTPUT::
 
           id : <id_given_in_input>
           result : {
             "msg" : "deleted gist ID: <gist_id>",
             "gist" : null
           }
           error : null
 
         ERROR OUTPUT::
 
           id : <id_given_in_input>
           result : null
           error :  {
             "failed to delete gist ID:<gist_id>"
           }
         """
         gist = get_gist_or_error(gistid)
         if not HasPermissionAny('hg.admin')():
             if gist.owner_id != request.authuser.user_id:
                 raise JSONRPCError('gist `%s` does not exist' % (gistid,))
 
         try:
             GistModel().delete(gist)
             meta.Session().commit()
             return dict(
                 msg='deleted gist ID:%s' % (gist.gist_access_id,),
                 gist=None
             )
         except Exception:
             log.error(traceback.format_exc())
             raise JSONRPCError('failed to delete gist ID:%s'
                                % (gist.gist_access_id,))
 
     # permission check inside
     def get_changesets(self, repoid, start=None, end=None, start_date=None,
                        end_date=None, branch_name=None, reverse=False, with_file_list=False, max_revisions=None):
         repo = get_repo_or_error(repoid)
         if not HasRepoPermissionLevel('read')(repo.repo_name):
             raise JSONRPCError('Access denied to repo %s' % repo.repo_name)
 
         format = "%Y-%m-%dT%H:%M:%S"
         try:
             return [e.__json__(with_file_list) for e in
                 repo.scm_instance.get_changesets(start,
                                                  end,
                                                  datetime.strptime(start_date, format) if start_date else None,
                                                  datetime.strptime(end_date, format) if end_date else None,
                                                  branch_name,
                                                  reverse, max_revisions)]
         except EmptyRepositoryError as e:
             raise JSONRPCError('Repository is empty')
 
     # permission check inside
     def get_changeset(self, repoid, raw_id, with_reviews=False):
         repo = get_repo_or_error(repoid)
         if not HasRepoPermissionLevel('read')(repo.repo_name):
             raise JSONRPCError('Access denied to repo %s' % repo.repo_name)
         changeset = repo.get_changeset(raw_id)
         if isinstance(changeset, EmptyChangeset):
             raise JSONRPCError('Changeset %s does not exist' % raw_id)
 
         info = dict(changeset.as_dict())
 
         if with_reviews:
             reviews = ChangesetStatusModel().get_statuses(
                                 repo.repo_name, raw_id)
             info["reviews"] = reviews
 
         return info
 
     # permission check inside
     def get_pullrequest(self, pullrequest_id):
         """
         Get given pull request by id
         """
         pull_request = db.PullRequest.get(pullrequest_id)
         if pull_request is None:
             raise JSONRPCError('pull request `%s` does not exist' % (pullrequest_id,))
         if not HasRepoPermissionLevel('read')(pull_request.org_repo.repo_name):
             raise JSONRPCError('not allowed')
         return pull_request.get_api_data()
 
     # permission check inside
     def comment_pullrequest(self, pull_request_id, comment_msg='', status=None, close_pr=False):
         """
         Add comment, close and change status of pull request.
         """
         apiuser = get_user_or_error(request.authuser.user_id)
         pull_request = db.PullRequest.get(pull_request_id)
         if pull_request is None:
             raise JSONRPCError('pull request `%s` does not exist' % (pull_request_id,))
         if (not HasRepoPermissionLevel('read')(pull_request.org_repo.repo_name)):
             raise JSONRPCError('No permission to add comment. User needs at least reading permissions'
                                ' to the source repository.')
         owner = apiuser.user_id == pull_request.owner_id
         reviewer = apiuser.user_id in [reviewer.user_id for reviewer in pull_request.reviewers]
         if close_pr and not (apiuser.admin or owner):
             raise JSONRPCError('No permission to close pull request. User needs to be admin or owner.')
         if status and not (apiuser.admin or owner or reviewer):
             raise JSONRPCError('No permission to change pull request status. User needs to be admin, owner or reviewer.')
         if pull_request.is_closed():
             raise JSONRPCError('pull request is already closed')
 
         comment = ChangesetCommentsModel().create(
             text=comment_msg,
             repo=pull_request.org_repo.repo_id,
             author=apiuser.user_id,
             pull_request=pull_request.pull_request_id,
             f_path=None,
             line_no=None,
             status_change=db.ChangesetStatus.get_status_lbl(status),
             closing_pr=close_pr
         )
         userlog.action_logger(apiuser,
                       'user_commented_pull_request:%s' % pull_request_id,
                       pull_request.org_repo, request.ip_addr)
         if status:
             ChangesetStatusModel().set_status(
                 pull_request.org_repo_id,
                 status,
                 apiuser.user_id,
                 comment,
                 pull_request=pull_request_id
             )
         if close_pr:
             PullRequestModel().close_pull_request(pull_request_id)
             userlog.action_logger(apiuser,
                           'user_closed_pull_request:%s' % pull_request_id,
                           pull_request.org_repo, request.ip_addr)
         meta.Session().commit()
         return True
 
     # permission check inside
     def edit_reviewers(self, pull_request_id, add=None, remove=None):
         """
         Add and/or remove one or more reviewers to a pull request, by username
         or user ID. Reviewers are specified either as a single-user string or
         as a JSON list of one or more strings.
         """
         if add is None and remove is None:
             raise JSONRPCError('''Invalid request. Neither 'add' nor 'remove' is specified.''')
 
         pull_request = db.PullRequest.get(pull_request_id)
         if pull_request is None:
             raise JSONRPCError('pull request `%s` does not exist' % (pull_request_id,))
 
         apiuser = get_user_or_error(request.authuser.user_id)
         is_owner = apiuser.user_id == pull_request.owner_id
         is_repo_admin = HasRepoPermissionLevel('admin')(pull_request.other_repo.repo_name)
         if not (apiuser.admin or is_repo_admin or is_owner):
             raise JSONRPCError('No permission to edit reviewers of this pull request. User needs to be admin or pull request owner.')
         if pull_request.is_closed():
             raise JSONRPCError('Cannot edit reviewers of a closed pull request.')
 
         if not isinstance(add, list):
             add = [add]
         if not isinstance(remove, list):
             remove = [remove]
 
         # look up actual user objects from given name or id. Bail out if unknown.
         add_objs = set(get_user_or_error(user) for user in add if user is not None)
         remove_objs = set(get_user_or_error(user) for user in remove if user is not None)
 
         new_reviewers = redundant_reviewers = set()
         if add_objs:
             new_reviewers, redundant_reviewers = PullRequestModel().add_reviewers(apiuser, pull_request, add_objs)
         if remove_objs:
             PullRequestModel().remove_reviewers(apiuser, pull_request, remove_objs)
 
         meta.Session().commit()
 
         return {
             'added': [x.username for x in new_reviewers],
             'already_present': [x.username for x in redundant_reviewers],
             # NOTE: no explicit check that removed reviewers were actually present.
             'removed': [x.username for x in remove_objs],
         }

kallithea/model/repo.py

@@ -1,839 +1,836 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
 ~~~~~~~~~~~~~~~~~~~~
 
 Repository 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: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 
 """
 
 import logging
 import os
 import shutil
 import traceback
 from datetime import datetime
 
 import kallithea.lib.utils2
 from kallithea.lib import celerylib, hooks, webutils
 from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.utils import is_valid_repo_uri, make_ui
 from kallithea.lib.utils2 import LazyProperty, get_current_authuser, obfuscate_url_pw, remove_prefix
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.model import db, meta, scm, userlog
 
 
 log = logging.getLogger(__name__)
 
 
 class RepoModel(object):
 
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = db.User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
 
         default_perm = 'repository.none' if private else default
 
         repo_to_perm = db.UserRepoToPerm()
         repo_to_perm.permission = db.Permission.get_by_key(default_perm)
 
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         meta.Session().add(repo_to_perm)
 
         return repo_to_perm
 
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
 
         q = db.Ui.query().filter(db.Ui.ui_key == '/').one()
         return q.ui_value
 
     def get(self, repo_id):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_id == repo_id)
         return repo.scalar()
 
     def get_repo(self, repository):
         return db.Repository.guess_instance(repository)
 
     def get_by_repo_name(self, repo_name):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_name == repo_name)
         return repo.scalar()
 
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         from tg import app_globals, request
         from tg import tmpl_context as c
         from tg.i18n import ugettext as _
 
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
 
         tmpl = template.get_def(tmpl)
         return tmpl.render_unicode(
             *args,
             _=_,
             webutils=webutils,
             c=c,
             request=request,
             **kwargs)
 
     def get_repos_as_dict(self, repos_list, repo_groups_list=None,
                           admin=False,
                           short_name=False):
         """Return repository list for use by DataTable.
         repos_list: list of repositories - but will be filtered for read permission.
         repo_groups_list: added at top of list without permission check.
         admin: return data for action column.
         """
         _render = self._render_datatable
         from tg import request
         from tg import tmpl_context as c
 
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name)
 
         def following(repo_id, is_following):
             return _render('following', repo_id, is_following)
 
         def last_change(last_change):
             return _render("last_change", last_change)
 
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
 
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
 
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
 
         def desc(desc):
             return webutils.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
 
         def state(repo_state):
             return _render("repo_state", repo_state)
 
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name)
 
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
 
         repos_data = []
 
         for gr in repo_groups_list or []:
             repos_data.append(dict(
                 raw_name='\0' + webutils.html_escape(gr.name),  # sort before repositories
                 just_name=webutils.html_escape(gr.name),
                 name=_render('group_name_html', group_name=gr.group_name, name=gr.name),
                 desc=desc(gr.group_description)))
 
         for repo in repos_list:
             if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):
                 continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": webutils.html_escape(repo.repo_name),
                 "just_name": webutils.html_escape(repo.just_name),
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "following": following(
                     repo.repo_id,
                     scm.ScmModel().is_following_repo(repo.repo_name, request.authuser.user_id),
                 ),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": repo.owner.username,
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
             }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id, repo.owner.username)
                 })
             repos_data.append(row)
 
         return {
             "sort": "name",
             "dir": "asc",
             "records": repos_data
         }
 
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
 
         :param repo_name:
         """
 
         repo_info = db.Repository.get_by_repo_name(repo_name)
 
         if repo_info is None:
             return None
 
         defaults = repo_info.get_dict()
         defaults['repo_name'] = repo_info.just_name
         defaults['repo_group'] = repo_info.group_id
 
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
 
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
 
         # fill owner
         if repo_info.owner:
             defaults.update({'owner': repo_info.owner.username})
         else:
             replacement_user = db.User.query().filter(db.User.admin ==
                                                    True).first().username
             defaults.update({'owner': replacement_user})
 
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
 
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
 
         return defaults
 
     def update(self, repo, **kwargs):
         try:
             cur_repo = db.Repository.guess_instance(repo)
             org_repo_name = cur_repo.repo_name
             if 'owner' in kwargs:
                 cur_repo.owner = db.User.get_by_username(kwargs['owner'])
 
             if 'repo_group' in kwargs:
                 assert kwargs['repo_group'] != '-1', kwargs # RepoForm should have converted to None
                 cur_repo.group = db.RepoGroup.get(kwargs['repo_group'])
                 cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 # clone_uri is modified - if given a value, check it is valid
                 if clone_uri != '':
                     # will raise exception on error
                     is_valid_repo_uri(cur_repo.repo_type, clone_uri, make_ui())
                 cur_repo.clone_uri = clone_uri
 
             if 'repo_name' in kwargs:
                 repo_name = kwargs['repo_name']
                 if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                     raise Exception('invalid repo name %s' % repo_name)
                 cur_repo.repo_name = cur_repo.get_new_name(repo_name)
 
             # if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
                 )
                 # handle extra fields
             for field in [k for k in kwargs if k.startswith(db.RepositoryField.PREFIX)]:
                 k = db.RepositoryField.un_prefix_key(field)
                 ex_field = db.RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
 
             if org_repo_name != cur_repo.repo_name:
                 # rename repository
                 self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)
 
             return cur_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def _create_repo(self, repo_name, repo_type, description, owner,
                      private=False, clone_uri=None, repo_group=None,
                      landing_rev='rev:tip', fork_of=None,
                      copy_fork_permissions=False, enable_statistics=False,
                      enable_downloads=False,
                      copy_group_permissions=False, state=db.Repository.STATE_PENDING):
         """
         Create repository inside database with PENDING state. This should only be
         executed by create() repo, with exception of importing existing repos.
 
         """
         owner = db.User.guess_instance(owner)
         fork_of = db.Repository.guess_instance(fork_of)
         repo_group = db.RepoGroup.guess_instance(repo_group)
         try:
             # repo name is just a name of repository
             # while repo_name_full is a full qualified name that is combined
             # with name and path of group
             repo_name_full = repo_name
             repo_name = repo_name.split(kallithea.URL_SEP)[-1]
             if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                 raise Exception('invalid repo name %s' % repo_name)
 
             new_repo = db.Repository()
             new_repo.repo_state = state
             new_repo.enable_statistics = False
             new_repo.repo_name = repo_name_full
             new_repo.repo_type = repo_type
             new_repo.owner = owner
             new_repo.group = repo_group
             new_repo.description = description or repo_name
             new_repo.private = private
             if clone_uri:
                 # will raise exception on error
                 is_valid_repo_uri(repo_type, clone_uri, make_ui())
             new_repo.clone_uri = clone_uri
             new_repo.landing_rev = landing_rev
 
             new_repo.enable_statistics = enable_statistics
             new_repo.enable_downloads = enable_downloads
 
             if fork_of:
                 parent_repo = fork_of
                 new_repo.fork = parent_repo
 
             meta.Session().add(new_repo)
 
             if fork_of and copy_fork_permissions:
                 repo = fork_of
                 user_perms = db.UserRepoToPerm.query() \
                     .filter(db.UserRepoToPerm.repository == repo).all()
                 group_perms = db.UserGroupRepoToPerm.query() \
                     .filter(db.UserGroupRepoToPerm.repository == repo).all()
 
                 for perm in user_perms:
                     db.UserRepoToPerm.create(perm.user, new_repo, perm.permission)
 
                 for perm in group_perms:
                     db.UserGroupRepoToPerm.create(perm.users_group, new_repo,
                                                perm.permission)
 
             elif repo_group and copy_group_permissions:
 
                 user_perms = db.UserRepoGroupToPerm.query() \
                     .filter(db.UserRepoGroupToPerm.group == repo_group).all()
 
                 group_perms = db.UserGroupRepoGroupToPerm.query() \
                     .filter(db.UserGroupRepoGroupToPerm.group == repo_group).all()
 
                 for perm in user_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = db.Permission.get_by_key(perm_name)
                     db.UserRepoToPerm.create(perm.user, new_repo, perm_obj)
 
                 for perm in group_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = db.Permission.get_by_key(perm_name)
                     db.UserGroupRepoToPerm.create(perm.users_group, new_repo, perm_obj)
 
             else:
                 self._create_default_perms(new_repo, private)
 
             # now automatically start following this repository as owner
             scm.ScmModel().toggle_following_repo(new_repo.repo_id, owner.user_id)
             # 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()
             return new_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def create(self, form_data, cur_user):
         """
         Create repository using celery tasks
 
         :param form_data:
         :param cur_user:
         """
         return create_repo(form_data, cur_user)
 
     def _update_permissions(self, repo, perms_new=None, perms_updates=None,
                             check_perms=True):
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
 
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
                 )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
                     )
             # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
                 )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
                     )
 
     def create_fork(self, form_data, cur_user):
         """
         Simple wrapper into executing celery task for fork creation
 
         :param form_data:
         :param cur_user:
         """
         return create_repo_fork(form_data, cur_user)
 
     def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
         """
         Delete given repository, forks parameter defines what do do with
         attached forks. Throws AttachedForksError if deleted repo has attached
         forks
 
         :param repo:
         :param forks: str 'delete' or 'detach'
         :param fs_remove: remove(archive) repo from filesystem
         """
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         repo = db.Repository.guess_instance(repo)
         if repo is not None:
             if forks == 'detach':
                 for r in repo.forks:
                     r.fork = None
             elif forks == 'delete':
                 for r in repo.forks:
                     self.delete(r, forks='delete')
             elif [f for f in repo.forks]:
                 raise AttachedForksError()
 
             old_repo_dict = repo.get_dict()
             try:
                 meta.Session().delete(repo)
                 if fs_remove:
                     self._delete_filesystem_repo(repo)
                 else:
                     log.debug('skipping removal from filesystem')
                 hooks.log_delete_repository(old_repo_dict,
                                       deleted_by=cur_user)
             except Exception:
                 log.error(traceback.format_exc())
                 raise
 
     def grant_user_permission(self, repo, user, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found
 
         :param repo: Instance of Repository, repository_id, or repository name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         user = db.User.guess_instance(user)
         repo = db.Repository.guess_instance(repo)
         permission = db.Permission.guess_instance(perm)
 
         # check if we have that permission already
         obj = db.UserRepoToPerm.query() \
             .filter(db.UserRepoToPerm.user == user) \
             .filter(db.UserRepoToPerm.repository == repo) \
             .scalar()
         if obj is None:
             # create new !
             obj = db.UserRepoToPerm()
             meta.Session().add(obj)
         obj.repository = repo
         obj.user = user
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, user, repo)
         return obj
 
     def revoke_user_permission(self, repo, user):
         """
         Revoke permission for user on given repository
 
         :param repo: Instance of Repository, repository_id, or repository name
         :param user: Instance of User, user_id or username
         """
 
         user = db.User.guess_instance(user)
         repo = db.Repository.guess_instance(repo)
 
         obj = db.UserRepoToPerm.query() \
             .filter(db.UserRepoToPerm.repository == repo) \
             .filter(db.UserRepoToPerm.user == user) \
             .scalar()
         if obj is not None:
             meta.Session().delete(obj)
             log.debug('Revoked perm on %s on %s', repo, user)
 
     def grant_user_group_permission(self, repo, group_name, perm):
         """
         Grant permission for user group on given repository, or update
         existing one if found
 
         :param repo: Instance of Repository, repository_id, or repository name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         :param perm: Instance of Permission, or permission_name
         """
         repo = db.Repository.guess_instance(repo)
         group_name = db.UserGroup.guess_instance(group_name)
         permission = db.Permission.guess_instance(perm)
 
         # check if we have that permission already
         obj = db.UserGroupRepoToPerm.query() \
             .filter(db.UserGroupRepoToPerm.users_group == group_name) \
             .filter(db.UserGroupRepoToPerm.repository == repo) \
             .scalar()
 
         if obj is None:
             # create new
             obj = db.UserGroupRepoToPerm()
             meta.Session().add(obj)
 
         obj.repository = repo
         obj.users_group = group_name
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, group_name, repo)
         return obj
 
     def revoke_user_group_permission(self, repo, group_name):
         """
         Revoke permission for user group on given repository
 
         :param repo: Instance of Repository, repository_id, or repository name
         :param group_name: Instance of UserGroup, users_group_id,
             or user group name
         """
         repo = db.Repository.guess_instance(repo)
         group_name = db.UserGroup.guess_instance(group_name)
 
         obj = db.UserGroupRepoToPerm.query() \
             .filter(db.UserGroupRepoToPerm.repository == repo) \
             .filter(db.UserGroupRepoToPerm.users_group == group_name) \
             .scalar()
         if obj is not None:
             meta.Session().delete(obj)
             log.debug('Revoked perm to %s on %s', repo, group_name)
 
     def delete_stats(self, repo_name):
         """
         removes stats for given repo
 
         :param repo_name:
         """
         repo = db.Repository.guess_instance(repo_name)
         try:
             obj = db.Statistics.query() \
                 .filter(db.Statistics.repository == repo).scalar()
             if obj is not None:
                 meta.Session().delete(obj)
         except Exception:
             log.error(traceback.format_exc())
             raise
 
     def _create_filesystem_repo(self, repo_name, repo_type, repo_group,
                                 clone_uri=None, repo_store_location=None):
         """
         Makes repository on filesystem. Operation is group aware, meaning that it will create
         a repository within a group, and alter the paths accordingly to the group location.
 
         Note: clone_uri is low level and not validated - it might be a file system path used for validated cloning
         """
         from kallithea.lib.utils import is_valid_repo, is_valid_repo_group
 
         if '/' in repo_name:
             raise ValueError('repo_name must not contain groups got `%s`' % repo_name)
 
         if isinstance(repo_group, db.RepoGroup):
             new_parent_path = os.sep.join(repo_group.full_path_splitted)
         else:
             new_parent_path = repo_group or ''
 
         if repo_store_location:
             _paths = [repo_store_location]
         else:
             _paths = [self.repos_path, new_parent_path, repo_name]
         repo_path = os.path.join(*_paths)
 
         # check if this path is not a repository
         if is_valid_repo(repo_path, self.repos_path):
             raise Exception('This path %s is a valid repository' % repo_path)
 
         # check if this path is a group
         if is_valid_repo_group(repo_path, self.repos_path):
             raise Exception('This path %s is a valid group' % repo_path)
 
         log.info('creating repo %s in %s from url: `%s`',
             repo_name, repo_path,
             obfuscate_url_pw(clone_uri))
 
         backend = get_backend(repo_type)
 
         if repo_type == 'hg':
             baseui = make_ui()
             # patch and reset hooks section of UI config to not run any
             # hooks on creating remote repo
             for k, v in baseui.configitems('hooks'):
                 baseui.setconfig('hooks', k, None)
 
             repo = backend(repo_path, create=True, src_url=clone_uri, baseui=baseui)
         elif repo_type == 'git':
             repo = backend(repo_path, create=True, src_url=clone_uri, bare=True)
             # add kallithea hook into this repo
             scm.ScmModel().install_git_hooks(repo)
         else:
             raise Exception('Not supported repo_type %s expected hg/git' % repo_type)
 
         log.debug('Created repo %s with %s backend',
                   repo_name, repo_type)
         return repo
 
     def _rename_filesystem_repo(self, old, new):
         """
         renames repository on filesystem
 
         :param old: old name
         :param new: new name
         """
         log.info('renaming repo from %s to %s', old, new)
 
         old_path = os.path.join(self.repos_path, old)
         new_path = os.path.join(self.repos_path, new)
         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_filesystem_repo(self, repo):
         """
         removes repo from filesystem, the removal is actually done by
         renaming dir to a 'rm__*' prefix which Kallithea will skip.
         It can be undeleted later by reverting the rename.
 
         :param repo: repo object
         """
         rm_path = os.path.join(self.repos_path, repo.repo_name)
         log.info("Removing %s", rm_path)
 
         _now = datetime.now()
         _ms = str(_now.microsecond).rjust(6, '0')
         _d = 'rm__%s__%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                              repo.just_name)
         if repo.group:
             args = repo.group.full_path_splitted + [_d]
             _d = os.path.join(*args)
         if os.path.exists(rm_path):
             shutil.move(rm_path, os.path.join(self.repos_path, _d))
         else:
             log.error("Can't find repo to delete in %r", rm_path)
 
 
 @celerylib.task
 def create_repo(form_data, cur_user):
     cur_user = db.User.guess_instance(cur_user)
 
     owner = cur_user
     repo_name = form_data['repo_name']
     repo_name_full = form_data['repo_name_full']
     repo_type = form_data['repo_type']
     description = form_data['repo_description']
     private = form_data['repo_private']
     clone_uri = form_data.get('clone_uri')
     repo_group = form_data['repo_group']
     landing_rev = form_data['repo_landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
     copy_group_permissions = form_data.get('repo_copy_permissions')
     fork_of = form_data.get('fork_parent_id')
+    enable_statistics = form_data['repo_enable_statistics']
+    enable_downloads = form_data['repo_enable_downloads']
     state = form_data.get('repo_state', db.Repository.STATE_PENDING)
 
-    # repo creation defaults, private and repo_type are filled in form
-    defs = db.Setting.get_default_repo_settings(strip_prefix=True)
-    enable_statistics = defs.get('repo_enable_statistics')
-    enable_downloads = defs.get('repo_enable_downloads')
-
     try:
         db_repo = RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=description,
             owner=owner,
             private=private,
             clone_uri=clone_uri,
             repo_group=repo_group,
             landing_rev=landing_rev,
             fork_of=fork_of,
             copy_fork_permissions=copy_fork_permissions,
             copy_group_permissions=copy_group_permissions,
             enable_statistics=enable_statistics,
             enable_downloads=enable_downloads,
             state=state
         )
 
         userlog.action_logger(cur_user, 'user_created_repo',
                       form_data['repo_name_full'], '')
 
         meta.Session().commit()
         # now create this repo on Filesystem
         RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=db.RepoGroup.guess_instance(repo_group),
             clone_uri=clone_uri,
         )
         db_repo = db.Repository.get_by_repo_name(repo_name_full)
         hooks.log_create_repository(db_repo.get_dict(), created_by=owner.username)
 
         # update repo changeset caches initially
         db_repo.update_changeset_cache()
 
         # set new created state
         db_repo.set_state(db.Repository.STATE_CREATED)
         meta.Session().commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
         db_repo = db.Repository.get_by_repo_name(repo_name_full)
         if db_repo:
             db.Repository.delete(db_repo.repo_id)
             meta.Session().commit()
             RepoModel()._delete_filesystem_repo(db_repo)
         raise
 
 
 @celerylib.task
 def create_repo_fork(form_data, cur_user):
     """
     Creates a fork of repository using interval VCS methods
 
     :param form_data:
     :param cur_user:
     """
     base_path = kallithea.CONFIG['base_path']
     cur_user = db.User.guess_instance(cur_user)
 
     repo_name = form_data['repo_name']  # fork in this case
     repo_name_full = form_data['repo_name_full']
 
     repo_type = form_data['repo_type']
     owner = cur_user
     private = form_data['private']
     clone_uri = form_data.get('clone_uri')
     repo_group = form_data['repo_group']
     landing_rev = form_data['landing_rev']
     copy_fork_permissions = form_data.get('copy_permissions')
 
     try:
         fork_of = db.Repository.guess_instance(form_data.get('fork_parent_id'))
 
         RepoModel()._create_repo(
             repo_name=repo_name_full,
             repo_type=repo_type,
             description=form_data['description'],
             owner=owner,
             private=private,
             clone_uri=clone_uri,
             repo_group=repo_group,
             landing_rev=landing_rev,
             fork_of=fork_of,
             copy_fork_permissions=copy_fork_permissions
         )
         userlog.action_logger(cur_user, 'user_forked_repo:%s' % repo_name_full,
                       fork_of.repo_name, '')
         meta.Session().commit()
 
         source_repo_path = os.path.join(base_path, fork_of.repo_name)
 
         # now create this repo on Filesystem
         RepoModel()._create_filesystem_repo(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_group=db.RepoGroup.guess_instance(repo_group),
             clone_uri=source_repo_path,
         )
         db_repo = db.Repository.get_by_repo_name(repo_name_full)
         hooks.log_create_repository(db_repo.get_dict(), created_by=owner.username)
 
         # update repo changeset caches initially
         db_repo.update_changeset_cache()
 
         # set new created state
         db_repo.set_state(db.Repository.STATE_CREATED)
         meta.Session().commit()
     except Exception as e:
         log.warning('Exception %s occurred when forking repository, '
                     'doing cleanup...' % e)
         # rollback things manually !
         db_repo = db.Repository.get_by_repo_name(repo_name_full)
         if db_repo:
             db.Repository.delete(db_repo.repo_id)
             meta.Session().commit()
             RepoModel()._delete_filesystem_repo(db_repo)
         raise

kallithea/templates/admin/repos/repo_add_base.html

@@ -1,97 +1,111 b''
 ${h.form(url('repos'))}
     <div class="form">
         <div class="form-group">
             <label class="control-label" for="repo_name">${_('Name')}:</label>
             <div>
                 ${h.text('repo_name',class_='form-control')}
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="repo_type">${_('Type')}:</label>
             <div>
                 ${h.select('repo_type','hg',c.backends,class_='form-control')}
                 <span class="help-block">${_('Type of repository to create.')}</span>
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="clone_uri">${_('Clone remote repository')}:</label>
             <div>
                 ${h.text('clone_uri',class_='form-control', placeholder=_('Repository URL'))}
                 <span class="help-block">
                     ${_('Optional: URL of a remote repository. If set, the repository will be created as a clone from this URL.')}
                 </span>
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="repo_description">${_('Description')}:</label>
             <div>
                 ${h.textarea('repo_description',class_='form-control')}
                 <span class="help-block">${_('Keep it short and to the point. Use a README file for longer descriptions.')}</span>
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="repo_group">${_('Repository group')}:</label>
             <div>
                 ${h.select('repo_group',None,c.repo_groups,class_='form-control')}
                 <span class="help-block">${_('Optionally select a group to put this repository into.')}</span>
             </div>
         </div>
         <div id="copy_perms" class="form-group">
             <label class="control-label" for="repo_copy_permissions">${_('Copy parent group permissions')}:</label>
             <div>
                 ${h.checkbox('repo_copy_permissions',value="True")}
                 <span class="help-block">${_('Copy permission set from parent repository group.')}</span>
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="repo_landing_rev">${_('Landing revision')}:</label>
             <div>
                 ${h.select('repo_landing_rev',None,c.landing_revs,class_='form-control')}
                 <span class="help-block">${_('Default revision for files page, downloads, full text search index and readme generation')}</span>
             </div>
         </div>
         <div class="form-group">
             <label class="control-label" for="repo_private">${_('Private repository')}:</label>
             <div>
                 ${h.checkbox('repo_private',value="True")}
                 <span class="help-block">${_('Private repositories are only visible to people explicitly added as collaborators.')}</span>
             </div>
         </div>
         <div class="form-group">
+            <label class="control-label" for="repo_enable_statistics">${_('Enable statistics')}:</label>
+            <div>
+                ${h.checkbox('repo_enable_statistics',value="True")}
+                <span class="help-block">${_('Enable statistics window on summary page.')}</span>
+            </div>
+        </div>
+        <div class="form-group">
+            <label class="control-label" for="repo_enable_downloads">${_('Enable downloads')}:</label>
+            <div>
+                ${h.checkbox('repo_enable_downloads',value="True")}
+                <span class="help-block">${_('Enable download menu on summary page.')}</span>
+            </div>
+        </div>
+        <div class="form-group">
             <div class="buttons">
                 ${h.submit('add',_('Add'),class_="btn btn-default")}
             </div>
         </div>
     </div>
 ${h.end_form()}
 
 <script>
     'use strict';
     $(document).ready(function(){
         $('#repo_type').select2({
             'minimumResultsForSearch': -1
         });
         $('#repo_group').select2({
             'dropdownAutoWidth': true
         });
 
         function setCopyPermsOption(group_val){
             if(group_val != "-1"){
                 $('#copy_perms').show();
             }
             else{
                 $('#copy_perms').hide();
             }
         }
 
         setCopyPermsOption($('#repo_group').val());
         $('#repo_group').on("change", function(e) {
             setCopyPermsOption(e.val);
         });
 
         $('#repo_landing_rev').select2({
             'minimumResultsForSearch': -1
         });
         $('#repo_name').focus();
     });
 </script>

kallithea/tests/api/api_base.py

@@ -1,2880 +1,2880 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}),  # currently broken
-        #('enable_downloads', {'enable_downloads': True}),  # currently broken
+        ('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('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/fixture.py

@@ -1,434 +1,436 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/>.
 
 """
 Helpers for fixture generation
 """
 
 import logging
 import os
 import shutil
 import tarfile
 from os.path import dirname
 
 from tg import request
 from tg.util.webtest import test_context
 
 from kallithea.lib.auth import AuthUser
 from kallithea.lib.db_manage import DbManage
 from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
 from kallithea.lib.pidlock import DaemonLock
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.model import db, meta
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.comment import ChangesetCommentsModel
 from kallithea.model.gist import GistModel
 from kallithea.model.pull_request import CreatePullRequestAction  # , CreatePullRequestIterationAction, 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.base import (GIT_REPO, HG_REPO, IP_ADDR, TEST_USER_ADMIN_EMAIL, TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_PASS, TEST_USER_REGULAR2_EMAIL,
                                   TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR_EMAIL, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS,
                                   TESTS_TMP_PATH, invalidate_all_caches)
 
 
 log = logging.getLogger(__name__)
 
 FIXTURES = os.path.join(dirname(dirname(os.path.abspath(__file__))), 'tests', 'fixtures')
 
 
 def raise_exception(*args, **kwargs):
     raise Exception('raise_exception raised exception')
 
 
 class Fixture(object):
 
     def __init__(self):
         pass
 
     def anon_access(self, status):
         """
         Context manager for controlling anonymous access.
         Anon access will be set and committed, but restored again when exiting the block.
 
         Usage:
 
         fixture = Fixture()
         with fixture.anon_access(False):
             stuff
         """
 
         class context(object):
             def __enter__(self):
                 anon = db.User.get_default_user()
                 self._before = anon.active
                 anon.active = status
                 meta.Session().commit()
                 invalidate_all_caches()
 
             def __exit__(self, exc_type, exc_val, exc_tb):
                 anon = db.User.get_default_user()
                 anon.active = self._before
                 meta.Session().commit()
 
         return context()
 
     def _get_repo_create_params(self, **custom):
         """Return form values to be validated through RepoForm"""
         defs = dict(
             repo_name=None,
             repo_type='hg',
             clone_uri='',
             repo_group='-1',
             repo_description='DESC',
             repo_private=False,
+            repo_enable_statistics=False,
+            repo_enable_downloads=False,
             repo_landing_rev='rev:tip',
             repo_copy_permissions=False,
             repo_state=db.Repository.STATE_CREATED,
         )
         defs.update(custom)
         if 'repo_name_full' not in custom:
             defs.update({'repo_name_full': defs['repo_name']})
 
         # fix the repo name if passed as repo_name_full
         if defs['repo_name']:
             defs['repo_name'] = defs['repo_name'].split('/')[-1]
 
         return defs
 
     def _get_repo_group_create_params(self, **custom):
         """Return form values to be validated through RepoGroupForm"""
         defs = dict(
             group_name=None,
             group_description='DESC',
             parent_group_id='-1',
             perms_updates=[],
             perms_new=[],
             recursive=False
         )
         defs.update(custom)
 
         return defs
 
     def _get_user_create_params(self, name, **custom):
         defs = dict(
             username=name,
             password='qweqwe',
             email='[e-mail]' % name,
             firstname='TestUser',
             lastname='Test',
             active=True,
             admin=False,
             extern_type='internal',
             extern_name=None
         )
         defs.update(custom)
 
         return defs
 
     def _get_user_group_create_params(self, name, **custom):
         defs = dict(
             users_group_name=name,
             user_group_description='DESC',
             users_group_active=True,
             user_group_data={},
         )
         defs.update(custom)
 
         return defs
 
     def create_repo(self, name, repo_group=None, cur_user=TEST_USER_ADMIN_LOGIN, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             r = db.Repository.get_by_repo_name(name)
             if r:
                 return r
 
         if isinstance(repo_group, db.RepoGroup):
             repo_group = repo_group.group_id
 
         form_data = self._get_repo_create_params(repo_name=name, **kwargs)
         form_data['repo_group'] = repo_group # patch form dict so it can be used directly by model
         RepoModel().create(form_data, cur_user=cur_user)
         meta.Session().commit()
         ScmModel().mark_for_invalidation(name)
         return db.Repository.get_by_repo_name(name)
 
     def create_fork(self, repo_to_fork, fork_name, cur_user=TEST_USER_ADMIN_LOGIN, **kwargs):
         repo_to_fork = db.Repository.get_by_repo_name(repo_to_fork)
 
         form_data = self._get_repo_create_params(repo_name=fork_name,
                                             fork_parent_id=repo_to_fork.repo_id,
                                             repo_type=repo_to_fork.repo_type,
                                             **kwargs)
         # patch form dict so it can be used directly by model
         form_data['description'] = form_data['repo_description']
         form_data['private'] = form_data['repo_private']
         form_data['landing_rev'] = form_data['repo_landing_rev']
 
         RepoModel().create_fork(form_data, cur_user=cur_user)
         meta.Session().commit()
         ScmModel().mark_for_invalidation(fork_name)
         r = db.Repository.get_by_repo_name(fork_name)
         assert r
         return r
 
     def destroy_repo(self, repo_name, **kwargs):
         RepoModel().delete(repo_name, **kwargs)
         meta.Session().commit()
 
     def create_repo_group(self, name, parent_group_id=None, cur_user=TEST_USER_ADMIN_LOGIN, **kwargs):
         assert '/' not in name, (name, kwargs) # use group_parent_id to make nested groups
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = db.RepoGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_repo_group_create_params(group_name=name, **kwargs)
         gr = RepoGroupModel().create(
             group_name=form_data['group_name'],
             group_description=form_data['group_name'],
             parent=parent_group_id,
             owner=cur_user,
             )
         meta.Session().commit()
         gr = db.RepoGroup.get_by_group_name(gr.group_name)
         return gr
 
     def destroy_repo_group(self, repogroupid):
         RepoGroupModel().delete(repogroupid)
         meta.Session().commit()
 
     def create_user(self, name, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             user = db.User.get_by_username(name)
             if user:
                 return user
         form_data = self._get_user_create_params(name, **kwargs)
         user = UserModel().create(form_data)
         meta.Session().commit()
         user = db.User.get_by_username(user.username)
         return user
 
     def destroy_user(self, userid):
         UserModel().delete(userid)
         meta.Session().commit()
 
     def create_user_group(self, name, cur_user=TEST_USER_ADMIN_LOGIN, **kwargs):
         if 'skip_if_exists' in kwargs:
             del kwargs['skip_if_exists']
             gr = db.UserGroup.get_by_group_name(group_name=name)
             if gr:
                 return gr
         form_data = self._get_user_group_create_params(name, **kwargs)
         user_group = UserGroupModel().create(
             name=form_data['users_group_name'],
             description=form_data['user_group_description'],
             owner=cur_user,
             active=form_data['users_group_active'],
             group_data=form_data['user_group_data'])
         meta.Session().commit()
         user_group = db.UserGroup.get_by_group_name(user_group.users_group_name)
         return user_group
 
     def destroy_user_group(self, usergroupid):
         UserGroupModel().delete(user_group=usergroupid, force=True)
         meta.Session().commit()
 
     def create_gist(self, **kwargs):
         form_data = {
             'description': 'new-gist',
             'owner': TEST_USER_ADMIN_LOGIN,
             'gist_type': db.Gist.GIST_PUBLIC,
             'lifetime': -1,
             'gist_mapping': {'filename1.txt': {'content': 'hello world'}}
         }
         form_data.update(kwargs)
         gist = GistModel().create(
             description=form_data['description'], owner=form_data['owner'], ip_addr=IP_ADDR,
             gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
             lifetime=form_data['lifetime']
         )
         meta.Session().commit()
 
         return gist
 
     def destroy_gists(self, gistid=None):
         for g in db.Gist.query():
             if gistid:
                 if gistid == g.gist_access_id:
                     GistModel().delete(g)
             else:
                 GistModel().delete(g)
         meta.Session().commit()
 
     def load_resource(self, resource_name, strip=True):
         with open(os.path.join(FIXTURES, resource_name), 'rb') as f:
             source = f.read()
             if strip:
                 source = source.strip()
 
         return source
 
     def commit_change(self, repo, filename, content, message, vcs_type,
                       parent=None, newfile=False, author=None):
         repo = db.Repository.get_by_repo_name(repo)
         _cs = parent
         if parent is None:
             _cs = EmptyChangeset(alias=vcs_type)
         if author is None:
             author = '%s <%s>' % (TEST_USER_ADMIN_LOGIN, TEST_USER_ADMIN_EMAIL)
 
         if newfile:
             nodes = {
                 filename: {
                     'content': content
                 }
             }
             cs = ScmModel().create_nodes(
                 user=TEST_USER_ADMIN_LOGIN,
                 ip_addr=IP_ADDR,
                 repo=repo,
                 message=message,
                 nodes=nodes,
                 parent_cs=_cs,
                 author=author,
             )
         else:
             cs = ScmModel().commit_change(
                 repo=repo.scm_instance, repo_name=repo.repo_name,
                 cs=parent,
                 user=TEST_USER_ADMIN_LOGIN,
                 ip_addr=IP_ADDR,
                 author=author,
                 message=message,
                 content=content,
                 f_path=filename
             )
         return cs
 
     def review_changeset(self, repo, revision, status, author=TEST_USER_ADMIN_LOGIN):
         comment = ChangesetCommentsModel().create("review comment", repo, author, revision=revision, send_email=False)
         csm = ChangesetStatusModel().set_status(repo, db.ChangesetStatus.STATUS_APPROVED, author, comment, revision=revision)
         meta.Session().commit()
         return csm
 
     def create_pullrequest(self, testcontroller, repo_name, pr_src_rev, pr_dst_rev, title='title'):
         org_ref = 'branch:stable:%s' % pr_src_rev
         other_ref = 'branch:default:%s' % pr_dst_rev
         with test_context(testcontroller.app): # needed to be able to mock request user and routes.url
             org_repo = other_repo = db.Repository.get_by_repo_name(repo_name)
             owner_user = db.User.get_by_username(TEST_USER_ADMIN_LOGIN)
             reviewers = [db.User.get_by_username(TEST_USER_REGULAR_LOGIN)]
             request.authuser = AuthUser(dbuser=owner_user)
             # creating a PR sends a message with an absolute URL - without routing that requires mocking
             request.environ['routes.url'] = lambda arg, qualified=False, **kwargs: ('https://localhost' if qualified else '') + '/fake/' + arg
             cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, 'No description', owner_user, reviewers)
             pull_request = cmd.execute()
             meta.Session().commit()
         return pull_request.pull_request_id
 
 
 #==============================================================================
 # Global test environment setup
 #==============================================================================
 
 def create_test_env(repos_test_path, config, reuse_database):
     """
     Makes a fresh database and
     install test repository into tmp dir
     """
 
     # PART ONE create db
     dbconf = config['sqlalchemy.url']
     log.debug('making test db %s', dbconf)
 
     # create test dir if it doesn't exist
     if not os.path.isdir(repos_test_path):
         log.debug('Creating testdir %s', repos_test_path)
         os.makedirs(repos_test_path)
 
     dbmanage = DbManage(dbconf=dbconf, root=config['here'],
                         cli_args={
                             'force_ask': True,
                             'username': TEST_USER_ADMIN_LOGIN,
                             'password': TEST_USER_ADMIN_PASS,
                             'email': TEST_USER_ADMIN_EMAIL,
                         })
     dbmanage.create_tables(reuse_database=reuse_database)
     # for tests dynamically set new root paths based on generated content
     dbmanage.create_settings(dbmanage.prompt_repo_root_path(repos_test_path))
     dbmanage.create_default_user()
     dbmanage.create_admin_user()
     dbmanage.create_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, TEST_USER_REGULAR_EMAIL, False)
     dbmanage.create_user(TEST_USER_REGULAR2_LOGIN, TEST_USER_REGULAR2_PASS, TEST_USER_REGULAR2_EMAIL, False)
     dbmanage.create_permissions()
     dbmanage.populate_default_permissions()
     meta.Session().commit()
     # PART TWO make test repo
     log.debug('making test vcs repositories')
 
     idx_path = config['index_dir']
     data_path = config['cache_dir']
 
     # clean index and data
     if idx_path and os.path.exists(idx_path):
         log.debug('remove %s', idx_path)
         shutil.rmtree(idx_path)
 
     if data_path and os.path.exists(data_path):
         log.debug('remove %s', data_path)
         shutil.rmtree(data_path)
 
     # CREATE DEFAULT TEST REPOS
     tar = tarfile.open(os.path.join(FIXTURES, 'vcs_test_hg.tar.gz'))
     tar.extractall(os.path.join(TESTS_TMP_PATH, HG_REPO))
     tar.close()
 
     tar = tarfile.open(os.path.join(FIXTURES, 'vcs_test_git.tar.gz'))
     tar.extractall(os.path.join(TESTS_TMP_PATH, GIT_REPO))
     tar.close()
 
 
 def create_test_index(repo_location, config, full_index):
     """
     Makes default test index
     """
 
     index_location = os.path.join(config['index_dir'])
     if not os.path.exists(index_location):
         os.makedirs(index_location)
 
     l = DaemonLock(os.path.join(index_location, 'make_index.lock'))
     WhooshIndexingDaemon(index_location=index_location,
                          repo_location=repo_location) \
         .run(full_index=full_index)
     l.release()
 
 
 def failing_test_hook(ui, repo, **kwargs):
     ui.write(b"failing_test_hook failed\n")
     return 1
 
 
 def exception_test_hook(ui, repo, **kwargs):
     raise Exception("exception_test_hook threw an exception")
 
 
 def passing_test_hook(ui, repo, **kwargs):
     ui.write(b"passing_test_hook succeeded\n")
     return 0