# -*- 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 os

import random

import re

import mock

import pytest

from kallithea.lib import ext_json

from kallithea.lib.auth import AuthUser

from kallithea.lib.utils2 import ascii_bytes

from kallithea.model.changeset_status import ChangesetStatusModel

from kallithea.model.db import ChangesetStatus, PullRequest, RepoGroup, Repository, Setting, Ui, User

from kallithea.model.gist import GistModel

from kallithea.model.meta import Session

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

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)

    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)

    Session().commit()

    return gr

class _BaseTestApi(object):

    REPO = None

    REPO_TYPE = None

    @classmethod

    def setup_class(cls):

        cls.usr = 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='test@example.com',

            firstname='first',

            lastname='last'

        )

        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_Optional_object(self):

        from kallithea.controllers.api.api import Optional

        option1 = Optional(None)

        assert '<Optional:%s>' % None == repr(option1)

        assert option1() is None

        assert 1 == Optional.extract(Optional(1))

        assert 'trololo' == Optional.extract('trololo')

    def test_Optional_OAttr(self):

        from kallithea.controllers.api.api import Optional, OAttr

        option1 = Optional(OAttr('apiuser'))

        assert 'apiuser' == Optional.extract(option1)

    def test_OAttr_object(self):

        from kallithea.controllers.api.api import OAttr

        oattr1 = OAttr('apiuser')

        assert '<OptionalAttr:apiuser>' == repr(oattr1)

        assert oattr1() == oattr1

    def test_api_wrong_key(self):

        id_, params = _build_data('trololo', 'get_user')

        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 = 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 mis-feature - 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(Ui.get_by_key('paths', '/').ui_value, self.REPO)

        Session().commit()

        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(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 Repository.query().filter(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(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)

    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='test@example.com',

                                  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 = 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 = 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 = 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_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='u232@example.com',

                                           firstname='u1', lastname='u1')

        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)

    def test_api_delete_user_when_exception_happened(self):

        usr = UserModel().create_or_update(username='test_user',

                                           password='qweqwe',

                                           email='u232@example.com',

                                           firstname='u1', lastname='u1')

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

    def test_api_update_user_when_exception_happens(self):

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

        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)

        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 = []

        fixture.destroy_repo(repo_name)

        # Expected and arguably more correct result:

        #expected = 'failed to create repository `%s`' % repo_name

        #self._compare_error(id_, expected, given=response.body)

        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,

            'task': None,

        }

        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,

            'task': None,

        }

        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)

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

        response = api_call(self, params)

        if changing_attr == 'name':

            repo_name = updates['name']

        if changing_attr == 'repo_group':

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

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

    @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 mis-feature - 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)

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