raise JSONRPCError("repo `%s` already exist" % repo_name)

        defs = Setting.get_default_repo_settings(strip_prefix=True)

        if isinstance(private, Optional):

            private = defs.get('repo_private') or Optional.extract(private)

        if isinstance(repo_type, Optional):

            repo_type = defs.get('repo_type')

        if isinstance(enable_statistics, Optional):

            enable_statistics = defs.get('repo_enable_statistics')

        if isinstance(enable_locking, Optional):

            enable_locking = defs.get('repo_enable_locking')

        if isinstance(enable_downloads, Optional):

            enable_downloads = defs.get('repo_enable_downloads')

        clone_uri = Optional.extract(clone_uri)

        description = Optional.extract(description)

        landing_rev = Optional.extract(landing_rev)

        copy_permissions = Optional.extract(copy_permissions)

        try:

            repo_name_cleaned = repo_name.split('/')[-1]

            # create structure of groups and return the last group

            repo_group = map_groups(repo_name)

            data = dict(

                repo_name=repo_name_cleaned,

                repo_name_full=repo_name,

                repo_type=repo_type,

                repo_description=description,

                owner=owner,

                repo_private=private,

                clone_uri=clone_uri,

                repo_group=repo_group,

                repo_landing_rev=landing_rev,

                enable_statistics=enable_statistics,

                enable_locking=enable_locking,

                enable_downloads=enable_downloads,

                repo_copy_permissions=copy_permissions,

            )

            task = RepoModel().create(form_data=data, cur_user=owner)

            from celery.result import BaseAsyncResult

            task_id = None

            if isinstance(task, BaseAsyncResult):

                task_id = task.task_id

            # 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

                task=task_id

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to create repository `%s`' % (repo_name,))

    # permission check inside

    def update_repo(self, apiuser, repoid, name=Optional(None),

                    owner=Optional(OAttr('apiuser')),

                    group=Optional(None),

                    description=Optional(''), private=Optional(False),

                    clone_uri=Optional(None), landing_rev=Optional('rev:tip'),

                    enable_statistics=Optional(False),

                    enable_locking=Optional(False),

                    enable_downloads=Optional(False)):

        """

        Updates repo

        :param apiuser: filled automatically from apikey

        :type apiuser: AuthUser

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

        :param enable_downloads:

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAnyApi('hg.admin')(user=apiuser):

            # check if we have admin permission for this repo !

            if not HasRepoPermissionAnyApi('repository.admin')(user=apiuser,

                                                               repo_name=repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

            if (name != repo.repo_name and

                not HasPermissionAnyApi('hg.create.repository')(user=apiuser)

                ):

                raise JSONRPCError('no permission to create (or move) repositories')

        updates = {

            # update function requires this.

            'repo_name': repo.repo_name

        }

        repo_group = group

        if not isinstance(repo_group, Optional):

            repo_group = get_repo_group_or_error(repo_group)

            repo_group = repo_group.group_id

        try:

            store_update(updates, name, 'repo_name')

            store_update(updates, repo_group, 'repo_group')

            store_update(updates, owner, 'user')

            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_locking, 'repo_enable_locking')

            store_update(updates, enable_downloads, 'repo_enable_downloads')

            RepoModel().update(repo, **updates)

            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)

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

    def fork_repo(self, apiuser, repoid, fork_name,

                  owner=Optional(OAttr('apiuser')),

                  description=Optional(''), copy_permissions=Optional(False),

                  private=Optional(False), landing_rev=Optional('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 apiuser: filled automatically from apikey

        :type apiuser: AuthUser

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

            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,

                      "task": "<celery task id or None if done sync>"

                    }

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

        if HasPermissionAnyApi('hg.admin')(user=apiuser):

            pass

        elif HasRepoPermissionAnyApi('repository.admin',

                                     'repository.write',

                                     'repository.read')(user=apiuser,

                                                        repo_name=repo.repo_name):

            if not isinstance(owner, Optional):

                #forbid setting owner for non-admins

                raise JSONRPCError(

                    'Only Kallithea admin can specify `owner` param'

            repo_name = updates['name']

        if changing_attr == 'repo_group':

            repo_name = '/'.join([updates['group'], repo_name])

        try:

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

    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 = {'active': False}

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

    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=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_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) 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_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', crash):

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

        )

        response = api_call(self, params)

        ret = {

            'msg': 'Created fork of `%s` as `%s`' % (self.REPO,

                                                     fork_name),

            'success': True,

            'task': None,

        }

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

        fixture.destroy_repo(fork_name)

    def test_api_fork_repo_non_admin(self):

        fork_name = 'api-repo-fork'

        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,

            'task': None,

        }

        expected = ret