allow_repo_location_change = True

## allows to setup custom hooks in settings page

allow_custom_hooks_settings = True

## extra extensions for indexing, space separated and without the leading '.'.

#index.extensions =

#    gemfile

#    lock

## extra filenames for indexing, space separated

#index.filenames =

#    .dockerignore

#    .editorconfig

#    INSTALL

#    CHANGELOG

####################################

##            SSH CONFIG          ##

####################################

## SSH is disabled by default, until an Administrator decides to enable it.

ssh_enabled = false

## File where users' SSH keys will be stored *if* ssh_enabled is true.

#ssh_authorized_keys = /home/kallithea/.ssh/authorized_keys

## Path to be used in ssh_authorized_keys file to invoke kallithea-cli with ssh-serve.

#kallithea_cli_path = /srv/kallithea/venv/bin/kallithea-cli

## Locale to be used in the ssh-serve command.

## This is needed because an SSH client may try to use its own locale

## settings, which may not be available on the server.

## See `locale -a` for valid values on this system.

#ssh_locale = C.UTF-8

####################################

##         CELERY CONFIG          ##

####################################

## Note: Celery doesn't support Windows.

use_celery = false

## Celery config settings from https://docs.celeryproject.org/en/4.4.0/userguide/configuration.html prefixed with 'celery.'.

## Example: use the message queue on the local virtual host 'kallitheavhost' as the RabbitMQ user 'kallithea':

celery.broker_url = amqp://kallithea:thepassword@localhost:5672/kallitheavhost

celery.worker_concurrency = 2

celery.worker_max_tasks_per_child = 100

####################################

##          BEAKER CACHE          ##

####################################

beaker.cache.data_dir = %(here)s/data/cache/data

beaker.cache.lock_dir = %(here)s/data/cache/lock

beaker.cache.regions = long_term,long_term_file

beaker.cache.long_term.type = memory

beaker.cache.long_term.expire = 36000

beaker.cache.long_term.key_length = 256

beaker.cache.long_term_file.type = file

beaker.cache.long_term_file.expire = 604800

beaker.cache.long_term_file.key_length = 256

####################################

##        BEAKER SESSION          ##

####################################

## Name of session cookie. Should be unique for a given host and path, even when running

## on different ports. Otherwise, cookie sessions will be shared and messed up.

session.key = kallithea

## Sessions should always only be accessible by the browser, not directly by JavaScript.

session.httponly = true

## Session lifetime. 2592000 seconds is 30 days.

session.timeout = 2592000

## Server secret used with HMAC to ensure integrity of cookies.

#session.secret = VERY-SECRET

session.secret = development-not-secret

## Further, encrypt the data with AES.

#session.encrypt_key = <key_for_encryption>

#session.validate_key = <validation_key>

## Type of storage used for the session, current types are

## dbm, file, memcached, database, and memory.

## File system storage of session data. (default)

#session.type = file

## Cookie only, store all session data inside the cookie. Requires secure secrets.

#session.type = cookie

        c.repo_groups = AvailableRepoGroupChoices('write', extras)

        c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo)

    def __load_data(self):

        """

        Load defaults settings for edit, and update

        """

        c.repo_info = self._load_repo()

        self.__load_defaults(c.repo_info)

        defaults = RepoModel()._get_defaults(c.repo_name)

        defaults['clone_uri'] = c.repo_info.clone_uri_hidden # don't show password

        defaults['permanent_url'] = c.repo_info.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True)

        return defaults

    def index(self, format='html'):

        repos_list = RepoList(db.Repository.query(sorted=True).all(), perm_level='admin')

        # the repo list will be filtered to only show repos where the user has read permissions

        repos_data = RepoModel().get_repos_as_dict(repos_list, admin=True)

        # data used to render the grid

        c.data = repos_data

        return base.render('admin/repos/repos.html')

    @NotAnonymous()

    def create(self):

        self.__load_defaults()

        try:

            # CanWriteGroup validators checks permissions of this POST

            form_result = RepoForm(repo_groups=c.repo_groups,

                                   landing_revs=c.landing_revs_choices)() \

                            .to_python(dict(request.POST))

        except formencode.Invalid as errors:

            log.info(errors)

            return htmlfill.render(

                base.render('admin/repos/repo_add.html'),

                defaults=errors.value,

                errors=errors.error_dict or {},

                prefix_error=False,

                force_defaults=False,

                encoding="UTF-8")

        try:

            # create is done sometimes async on celery, db transaction

            # management is handled there.

            task = RepoModel().create(form_result, request.authuser.user_id)

        except Exception:

            log.error(traceback.format_exc())

            msg = (_('Error creating repository %s')

                   % form_result.get('repo_name'))

            webutils.flash(msg, category='error')

            raise HTTPFound(location=url('home'))

        raise HTTPFound(location=webutils.url('repo_creating_home',

                              repo_name=form_result['repo_name_full'],

    @NotAnonymous()

    def create_repository(self):

        self.__load_defaults()

        if not c.repo_groups:

            raise HTTPForbidden

        parent_group = request.GET.get('parent_group')

        ## apply the defaults from defaults page

        defaults = db.Setting.get_default_repo_settings(strip_prefix=True)

        if parent_group:

            prg = db.RepoGroup.get(parent_group)

            if prg is None or not any(rgc[0] == prg.group_id

                                      for rgc in c.repo_groups):

                raise HTTPForbidden

        else:

            parent_group = '-1'

        defaults.update({'repo_group': parent_group})

        return htmlfill.render(

            base.render('admin/repos/repo_add.html'),

            defaults=defaults,

            errors={},

            prefix_error=False,

            encoding="UTF-8",

            force_defaults=False)

    @LoginRequired()

    def repo_creating(self, repo_name):

        c.repo = repo_name

        if not c.repo:

            raise HTTPNotFound()

        return base.render('admin/repos/repo_creating.html')

    @LoginRequired()

    @base.jsonify

    def repo_check(self, repo_name):

        c.repo = repo_name

        repo = db.Repository.get_by_repo_name(repo_name)

        if repo and repo.repo_state == db.Repository.STATE_CREATED:

            if repo.clone_uri:

                webutils.flash(_('Created repository %s from %s')

                        % (repo.repo_name, repo.clone_uri_hidden), category='success')

            else:

                repo_url = webutils.link_to(repo.repo_name,

                                     webutils.url('summary_home',

                                           repo_name=repo.repo_name))

                fork = repo.fork

                if fork is not None:

                    fork_name = fork.repo_name

                    webutils.flash(webutils.HTML(_('Forked repository %s as %s'))

                            % (fork_name, repo_url), category='success')

                else:

                    webutils.flash(webutils.HTML(_('Created repository %s')) % repo_url,

                            category='success')

            return {'result': True}

        return {'result': False}

    @HasRepoPermissionLevelDecorator('admin')

    def update(self, repo_name):

        c.repo_info = self._load_repo()

        self.__load_defaults(c.repo_info)

        c.active = 'settings'

        c.repo_fields = db.RepositoryField.query() \

            .filter(db.RepositoryField.repository == c.repo_info).all()

        repo_model = RepoModel()

        changed_name = repo_name

        repo = db.Repository.get_by_repo_name(repo_name)

        old_data = {

            'repo_name': repo_name,

            'repo_group': repo.group.get_dict() if repo.group else {},

            'repo_type': repo.repo_type,

        }

        _form = RepoForm(edit=True, old_data=old_data,

                         repo_groups=c.repo_groups,

                         landing_revs=c.landing_revs_choices)()

        try:

            form_result = _form.to_python(dict(request.POST))

            repo = repo_model.update(repo_name, **form_result)

            ScmModel().mark_for_invalidation(repo_name)

            webutils.flash(_('Repository %s updated successfully') % repo_name,

                    category='success')

            changed_name = repo.repo_name

            userlog.action_logger(request.authuser, 'admin_updated_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:

            repo_name_parts = repo_name.split('/')

            repo_group = None

            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)

            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=repo_group,

                repo_landing_rev=landing_rev,

                enable_statistics=enable_statistics,

                enable_downloads=enable_downloads,

                repo_copy_permissions=copy_permissions,

            )

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

                not HasPermissionAny('hg.create.repository')()

            ):

                raise JSONRPCError('no permission to create (or move) 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

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

        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'

                )

            if not HasPermissionAny('hg.create.repository')():

                raise JSONRPCError('no permission to create repositories')

        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:

            fork_name_parts = fork_name.split('/')

            repo_group = None

            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)

            form_data = dict(

                repo_name=fork_name_parts[-1],

                repo_name_full=fork_name,

                repo_group=repo_group,

                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,

            )

            task = 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' %

        # add suffix to fork

        defaults['repo_name'] = '%s-fork' % defaults['repo_name']

        return defaults

    @LoginRequired(allow_default_user=True)

    @HasRepoPermissionLevelDecorator('read')

    def forks(self, repo_name):

        p = safe_int(request.GET.get('page'), 1)

        repo_id = c.db_repo.repo_id

        d = []

        for r in db.Repository.get_repo_forks(repo_id):

            if not HasRepoPermissionLevel('read')(r.repo_name, 'get forks check'):

                continue

            d.append(r)

        c.forks_pager = Page(d, page=p, items_per_page=20)

        if request.environ.get('HTTP_X_PARTIAL_XHR'):

            return base.render('/forks/forks_data.html')

        return base.render('/forks/forks.html')

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')

    @HasRepoPermissionLevelDecorator('read')

    def fork(self, repo_name):

        c.repo_info = db.Repository.get_by_repo_name(repo_name)

        if not c.repo_info:

            raise HTTPNotFound()

        defaults = self.__load_data()

        return htmlfill.render(

            base.render('forks/fork.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    @LoginRequired()

    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')

    @HasRepoPermissionLevelDecorator('read')

    def fork_create(self, repo_name):

        self.__load_defaults()

        c.repo_info = db.Repository.get_by_repo_name(repo_name)

        _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},

                             repo_groups=c.repo_groups,

                             landing_revs=c.landing_revs_choices)()

        form_result = {}

        try:

            form_result = _form.to_python(dict(request.POST))

            # an approximation that is better than nothing

            if not db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE).ui_active:

                form_result['update_after_clone'] = False

            # create fork is done sometimes async on celery, db transaction

            # management is handled there.

            task = RepoModel().create_fork(form_result, request.authuser.user_id)

        except formencode.Invalid as errors:

            return htmlfill.render(

                base.render('forks/fork.html'),

                defaults=errors.value,

                errors=errors.error_dict or {},

                prefix_error=False,

                encoding="UTF-8",

                force_defaults=False)

        except Exception:

            log.error(traceback.format_exc())

            webutils.flash(_('An error occurred during repository forking %s') %

                    repo_name, category='error')

        raise HTTPFound(location=webutils.url('repo_creating_home',

                              repo_name=form_result['repo_name_full'],

# 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.lib.celerylib

~~~~~~~~~~~~~~~~~~~~~~~

celery libs 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: Nov 27, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import logging

import os

from hashlib import sha1

from decorator import decorator

from tg import config

import kallithea

from kallithea.lib.pidlock import DaemonLock, LockHeld

from kallithea.lib.utils2 import safe_bytes

from kallithea.model import meta

log = logging.getLogger(__name__)

class FakeTask(object):

    """Fake a sync result to make it look like a finished task"""

    def __init__(self, result):

        self.result = result

    def failed(self):

        return False

    traceback = None # if failed

def task(f_org):

    """Wrapper of celery.task.task, running async if CELERY_APP

    """

    if kallithea.CELERY_APP:

        def f_async(*args, **kwargs):

            log.info('executing %s task', f_org.__name__)

            try:

                f_org(*args, **kwargs)

            finally:

                meta.Session.remove()  # prevent reuse of auto created db sessions

                log.info('executed %s task', f_org.__name__)

        runner = kallithea.CELERY_APP.task(name=f_org.__name__, ignore_result=True)(f_async)

        def f_wrapped(*args, **kwargs):

            t = runner.apply_async(args=args, kwargs=kwargs)

            log.info('executing task %s in async mode - id %s', f_org, t.task_id)

            return t

    else:

        def f_wrapped(*args, **kwargs):

            log.info('executing task %s in sync', f_org.__name__)

            try:

                f_org(*args, **kwargs)

            except Exception as e:

                log.error('exception executing sync task %s in sync: %r', f_org.__name__, e)

                raise # TODO: return this in FakeTask as with async tasks?

            return FakeTask(None)

    return f_wrapped

def __get_lockkey(func, *fargs, **fkwargs):

    params = list(fargs)

    params.extend(['%s-%s' % ar for ar in fkwargs.items()])

    func_name = str(func.__name__) if hasattr(func, '__name__') else str(func)

    lockkey = 'task_%s.lock' % \

        sha1(safe_bytes(func_name + '-' + '-'.join(str(x) for x in params))).hexdigest()

    return lockkey

def locked_task(func):

    def __wrapper(func, *fargs, **fkwargs):

        lockkey = __get_lockkey(func, *fargs, **fkwargs)

        log.info('running task with lockkey %s', lockkey)

        try:

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

<%inherit file="/base/base.html"/>

## don't trigger flash messages on this page

<%def name="flash_msg()">

</%def>

<%block name="title">

    ${_('%s Creating Repository') % c.repo_name}

</%block>

<%def name="breadcrumbs_links()">

    ${_('Creating repository')} ${c.repo}

</%def>

<%block name="header_menu">

    ${self.menu('repositories')}

</%block>

<%def name="main()">

<div class="panel panel-primary">

    <div class="panel-heading clearfix">

        ${self.breadcrumbs()}

    </div>

    <div class="panel-body">

            <h4 class="text-center">

                ${_('Repository "%(repo_name)s" is being created, you will be redirected when this process is finished.' % {'repo_name':c.repo_name})}

            </h4>

        <div id="progress">

            <div class="progress progress-striped active">

                <div class="progress-bar" role="progressbar"

                    aria-valuenow="100" aria-valuemin="0" aria-valuemax="100">

                </div>

            </div>

        </div>

        <div id="progress_error" style="display: none;">

            <div class="alert alert-danger">

                ${_("We're sorry but error occurred during this operation. Please check your Kallithea server logs, or contact administrator.")}

            </div>

        </div>

    </div>

</div>

<script>

'use strict';

(function worker() {

  $.ajax({

    success: function(data) {

      if(data.result === true){

          //redirect to created fork if our ajax loop tells us to do so.

          window.location = ${h.js(h.url('summary_home', repo_name = c.repo))};

      }

    },

    complete: function(resp) {

      if (resp.status == 200){

          // Schedule the next request when the current one's complete

          setTimeout(worker, 1000);

      }

      else{

          $("#progress").html($('#progress_error').html());

      }

    }

  });

})();

</script>

</%def>

<%text>##</%text> extra filenames for indexing, space separated

#index.filenames =

#    .dockerignore

#    .editorconfig

#    INSTALL

#    CHANGELOG

<%text>##</%text>##################################

<%text>##</%text>            SSH CONFIG          ##

<%text>##</%text>##################################

<%text>##</%text> SSH is disabled by default, until an Administrator decides to enable it.

ssh_enabled = false

<%text>##</%text> File where users' SSH keys will be stored *if* ssh_enabled is true.

#ssh_authorized_keys = /home/kallithea/.ssh/authorized_keys

%if user_home_path:

ssh_authorized_keys = ${user_home_path}/.ssh/authorized_keys

%endif

<%text>##</%text> Path to be used in ssh_authorized_keys file to invoke kallithea-cli with ssh-serve.

#kallithea_cli_path = /srv/kallithea/venv/bin/kallithea-cli

%if kallithea_cli_path:

kallithea_cli_path = ${kallithea_cli_path}

%endif

<%text>##</%text> Locale to be used in the ssh-serve command.

<%text>##</%text> This is needed because an SSH client may try to use its own locale

<%text>##</%text> settings, which may not be available on the server.

<%text>##</%text> See `locale -a` for valid values on this system.

#ssh_locale = C.UTF-8

%if ssh_locale:

ssh_locale = ${ssh_locale}

%endif

<%text>##</%text>##################################

<%text>##</%text>         CELERY CONFIG          ##

<%text>##</%text>##################################

<%text>##</%text> Note: Celery doesn't support Windows.

use_celery = false

<%text>##</%text> Celery config settings from https://docs.celeryproject.org/en/4.4.0/userguide/configuration.html prefixed with 'celery.'.

<%text>##</%text> Example: use the message queue on the local virtual host 'kallitheavhost' as the RabbitMQ user 'kallithea':

celery.broker_url = amqp://kallithea:thepassword@localhost:5672/kallitheavhost

celery.worker_concurrency = 2

celery.worker_max_tasks_per_child = 100

<%text>##</%text>##################################

<%text>##</%text>          BEAKER CACHE          ##

<%text>##</%text>##################################

beaker.cache.data_dir = %(here)s/data/cache/data

beaker.cache.lock_dir = %(here)s/data/cache/lock

beaker.cache.regions = long_term,long_term_file

beaker.cache.long_term.type = memory

beaker.cache.long_term.expire = 36000

beaker.cache.long_term.key_length = 256

beaker.cache.long_term_file.type = file

beaker.cache.long_term_file.expire = 604800

beaker.cache.long_term_file.key_length = 256

<%text>##</%text>##################################

<%text>##</%text>        BEAKER SESSION          ##

<%text>##</%text>##################################

<%text>##</%text> Name of session cookie. Should be unique for a given host and path, even when running

<%text>##</%text> on different ports. Otherwise, cookie sessions will be shared and messed up.

session.key = kallithea

<%text>##</%text> Sessions should always only be accessible by the browser, not directly by JavaScript.

session.httponly = true

<%text>##</%text> Session lifetime. 2592000 seconds is 30 days.

session.timeout = 2592000

<%text>##</%text> Server secret used with HMAC to ensure integrity of cookies.

session.secret = ${uuid()}

<%text>##</%text> Further, encrypt the data with AES.

#session.encrypt_key = <key_for_encryption>

#session.validate_key = <validation_key>

<%text>##</%text> Type of storage used for the session, current types are

<%text>##</%text> dbm, file, memcached, database, and memory.

<%text>##</%text> File system storage of session data. (default)

#session.type = file

<%text>##</%text> Cookie only, store all session data inside the cookie. Requires secure secrets.

#session.type = cookie

<%text>##</%text> Database storage of session data.

                                  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)

    def test_api_create_repo(self):

        repo_name = 'api-repo'

        id_, params = _build_data(self.apikey, 'create_repo',

                                  repo_name=repo_name,