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,

                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.

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

    state = form_data.get('repo_state', db.Repository.STATE_PENDING)

    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,

${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">

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

            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}),

        ('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)

# 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_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='%s+test@example.com' % 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):