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

        ret = repo.get_api_data()

        members = []

        followers = []

        assert 2 == len(repo.repo_to_perm)

        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)

        for user in repo.followers:

            followers.append(user.user.get_api_data())

        ret['members'] = members

        ret['followers'] = followers

        expected = ret

        try:

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

        finally:

            RepoModel().revoke_user_permission(self.REPO, self.TEST_USER_LOGIN)

    def test_api_get_repo_by_non_admin_no_permission_to_repo(self):

        RepoModel().grant_user_permission(repo=self.REPO,

                                          user=self.TEST_USER_LOGIN,

                                          perm='repository.none')

        id_, params = _build_data(self.apikey_regular, 'get_repo',

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = 'repository `%s` does not exist' % (self.REPO)

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

    def test_api_get_repo_that_doesn_not_exist(self):

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

                                  repoid='no-such-repo')

        response = api_call(self, params)

        ret = 'repository `%s` does not exist' % 'no-such-repo'

        expected = ret

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

    def test_api_get_repos(self):

        id_, params = _build_data(self.apikey, 'get_repos')

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

            for repo in Repository.query()

        ])

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

    def test_api_get_repos_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_repos')

        response = api_call(self, params)

        expected = jsonify([

            repo.get_api_data()

            for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN)

        ])

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

    @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(['files', 'dirs', 'all'])))

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

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

        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 = u'api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

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

        # TODO: introduce other test coverage of actual remote cloning

        clone_uri = os.path.join(TESTS_TMP_PATH, self.REPO)

        repo_name = u'api-repo'

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,

                                  clone_uri=clone_uri,

        )

        response = api_call(self, params)

        expected = "failed to create repository `%s`" % repo_name

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

        fixture.destroy_repo(repo_name)

    def test_api_create_repo_and_repo_group(self):

        repo_group_name = u'my_gr'

        repo_name = u'%s/api-repo' % repo_group_name

        # repo creation can no longer also create repo group

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = u'repo group `%s` not found' % repo_group_name

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

        assert RepoModel().get_by_repo_name(repo_name) is None

        # create group before creating repo

        rg = fixture.create_repo_group(repo_group_name)

        Session().commit()

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

                                  repo_name=repo_name,

                                  owner=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = {

            'msg': 'Created new repository `%s`' % repo_name,

            'success': True,

            'task': None,

        }

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

        repo = RepoModel().get_by_repo_name(repo_name)

        assert repo is not None

        fixture.destroy_repo(repo_name)

        fixture.destroy_repo_group(repo_group_name)

    def test_api_create_repo_in_repo_group_without_permission(self):

        repo_group_name = u'%s/api-repo-repo' % TEST_REPO_GROUP

        repo_name = u'%s/api-repo' % repo_group_name

        rg = fixture.create_repo_group(repo_group_name)

        Session().commit()

        RepoGroupModel().grant_user_permission(repo_group_name,

                                               self.TEST_USER_LOGIN,

                                               'group.none')

        Session().commit()

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

                                  repo_name=repo_name,

                                  repo_type=self.REPO_TYPE,

        )

        response = api_call(self, params)

        # Current result when API access control is different from Web:

        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)

        # 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 = u'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 = u'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 = u'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 = u'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=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=TEST_USER_ADMIN_LOGIN,

                                  repo_type=self.REPO_TYPE,)

        response = api_call(self, params)

        expected = u'repo group `%s` not found' % group_name

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

        fixture.destroy_repo(repo_name)

    @mock.patch.object(RepoModel, 'create', crash)

    def test_api_create_repo_exception_occurred(self):

        repo_name = u'api-repo'

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

                                  repo_name=repo_name,

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

    @parametrize('changing_attr,updates', [

        ('owner', {'owner': TEST_USER_REGULAR_LOGIN}),

        ('description', {'description': u'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_locking', {'enable_locking': True}),

        ('enable_downloads', {'enable_downloads': True}),

        ('name', {'name': u'new_repo_name'}),