response = api_call(self, params)

        repo = RepoModel().get_by_repo_name(repo_name)

        self.assertNotEqual(repo, 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)

    @parameterized.expand([

        (u'',),

        (u'.',),

        (u'..',),

        (u':',),

        (u'/',),

        (u'<test>',),

    ])

    def test_api_create_repo_bad_names(self, repo_name):

        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)

        if repo_name == '/':

            expected = "repo group `` not found"

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

        elif repo_name in [':', '<test>']:

            # FIXME: special characters and XSS injection should not be allowed

            expected = {

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

                'success': True,

                'task': None,

            }

            self._compare_ok(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 repo was a mis-feature - it would bypass access control

        # 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 = 'my_gr'

        repo_name = '%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)

        self.assertNotEqual(repo, 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_name = '%s/api-repo' % 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 = '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)

        self.assertNotEqual(repo, 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)

        self.assertNotEqual(repo, 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_copy_permissions=False,

            repo_state=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_group_create_params(self, **custom):

        defs = dict(

            group_name=None,

            group_description='DESC',

            group_parent_id=None,

            perms_updates=[],

            perms_new=[],

            enable_locking=False,

            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, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            r = Repository.get_by_repo_name(name)

            if r:

                return r

        if isinstance(kwargs.get('repo_group'), RepoGroup):

            kwargs['repo_group'] = kwargs['repo_group'].group_id

        form_data = self._get_repo_create_params(repo_name=name, **kwargs)

        cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create(form_data, cur_user)

        Session().commit()

        return Repository.get_by_repo_name(name)

    def create_fork(self, repo_to_fork, fork_name, **kwargs):

        repo_to_fork = Repository.get_by_repo_name(repo_to_fork)

        form_data = self._get_repo_create_params(repo_name=fork_name,

                                            fork_parent_id=repo_to_fork,

                                            repo_type=repo_to_fork.repo_type,

                                            **kwargs)

        form_data['update_after_clone'] = False

        #TODO: fix it !!

        form_data['description'] = form_data['repo_description']

        form_data['private'] = form_data['repo_private']

        form_data['landing_rev'] = form_data['repo_landing_rev']

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create_fork(form_data, cur_user=owner)

        Session().commit()

        r = Repository.get_by_repo_name(fork_name)

        assert r

        return r

    def destroy_repo(self, repo_name, **kwargs):

        RepoModel().delete(repo_name, **kwargs)

        Session().commit()

    def create_repo_group(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            gr = RepoGroup.get_by_group_name(group_name=name)

            if gr:

                return gr

        form_data = self._get_group_create_params(group_name=name, **kwargs)

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        gr = RepoGroupModel().create(

            group_name=form_data['group_name'],

            group_description=form_data['group_name'],

            owner=owner, parent=form_data['group_parent_id'])

        Session().commit()

        gr = RepoGroup.get_by_group_name(gr.group_name)

        return gr

    def destroy_repo_group(self, repogroupid):

        RepoGroupModel().delete(repogroupid)

        Session().commit()

    def create_user(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            user = User.get_by_username(name)

            if user:

                return user

        form_data = self._get_user_create_params(name, **kwargs)

        user = UserModel().create(form_data)

        Session().commit()

        user = User.get_by_username(user.username)

        return user

    def destroy_user(self, userid):

        UserModel().delete(userid)

        Session().commit()

    def create_user_group(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            gr = UserGroup.get_by_group_name(group_name=name)

            if gr:

                return gr

        form_data = self._get_user_group_create_params(name, **kwargs)

        owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        user_group = UserGroupModel().create(

            name=form_data['users_group_name'],

            description=form_data['user_group_description'],

            owner=owner, active=form_data['users_group_active'],

            group_data=form_data['user_group_data'])

        Session().commit()

        user_group = UserGroup.get_by_group_name(user_group.users_group_name)

        return user_group

    def destroy_user_group(self, usergroupid):

        UserGroupModel().delete(user_group=usergroupid, force=True)

        Session().commit()

    def create_gist(self, **kwargs):

        form_data = {

            'description': u'new-gist',

            'owner': TEST_USER_ADMIN_LOGIN,

            'gist_type': GistModel.cls.GIST_PUBLIC,

            'lifetime': -1,

            'gist_mapping': {'filename1.txt':{'content':'hello world'},}

        }

        form_data.update(kwargs)

        gist = GistModel().create(

            description=form_data['description'],owner=form_data['owner'],

            gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],

            lifetime=form_data['lifetime']

        )

        Session().commit()

        return gist

    def destroy_gists(self, gistid=None):

        for g in GistModel.cls.get_all():

            if gistid:

                if gistid == g.gist_access_id:

                    GistModel().delete(g)

            else:

                GistModel().delete(g)

        Session().commit()

    def load_resource(self, resource_name, strip=True):

        with open(os.path.join(FIXTURES, resource_name)) as f:

            source = f.read()

            if strip:

                source = source.strip()

        return source

    def commit_change(self, repo, filename, content, message, vcs_type, parent=None, newfile=False):

        repo = Repository.get_by_repo_name(repo)

        _cs = parent

        if not parent:

            _cs = EmptyChangeset(alias=vcs_type)