('hooks', Ui.HOOK_REPO_SIZE, 'python:kallithea.lib.hooks.repo_size', True),

            ('extensions', 'largefiles', '', True),

            ('largefiles', 'usercache', os.path.join(repo_root_path, '.cache', 'largefiles'), True),

            ('extensions', 'hgsubversion', '', False),

            ('extensions', 'hggit', '', False),

        ]

        for ui_section, ui_key, ui_value, ui_active in ui_config:

            ui_conf = Ui(

                ui_section=ui_section,

                ui_key=ui_key,

                ui_value=ui_value,

                ui_active=ui_active)

            self.sa.add(ui_conf)

        settings = [

            ('realm', 'Kallithea', 'unicode'),

            ('title', '', 'unicode'),

            ('ga_code', '', 'unicode'),

            ('show_public_icon', True, 'bool'),

            ('show_private_icon', True, 'bool'),

            ('stylify_metalabels', False, 'bool'),

            ('dashboard_items', 100, 'int'), # TODO: call it page_size

            ('admin_grid_items', 25, 'int'),

            ('show_version', True, 'bool'),

            ('use_gravatar', True, 'bool'),

            ('gravatar_url', User.DEFAULT_GRAVATAR_URL, 'unicode'),

            ('clone_uri_tmpl', Repository.DEFAULT_CLONE_URI, 'unicode'),

            ('clone_ssh_tmpl', Repository.DEFAULT_CLONE_SSH, 'unicode'),

        ]

        for key, val, type_ in settings:

            sett = Setting(key, val, type_)

            self.sa.add(sett)

        self.create_auth_plugin_options()

        self.create_default_options()

        log.info('Populated Ui and Settings defaults')

    def create_user(self, username, password, email='', admin=False):

        log.info('creating user %s', username)

        UserModel().create_or_update(username, password, email,

                                     firstname='Kallithea', lastname='Admin',

                                     active=True, admin=admin,

                                     extern_type=User.DEFAULT_AUTH_TYPE)

    def create_default_user(self):

        log.info('creating default user')

        # create default user for handling default permissions.

                                            password=str(uuid.uuid1())[:20],

                                            email='anonymous@kallithea-scm.org',

                                            firstname='Anonymous',

                                            lastname='User')

        # based on configuration options activate/deactivate this user which

        # controls anonymous access

        if self.cli_args.get('public_access') is False:

            log.info('Public access disabled')

            user.active = False

            Session().commit()

    def create_permissions(self):

        """

        Creates all permissions defined in the system

        """

        # module.(access|create|change|delete)_[name]

        # module.(none|read|write|admin)

        log.info('creating permissions')

        PermissionModel().create_permissions()

    def populate_default_permissions(self):

        """

        Populate default permissions. It will create only the default

        permissions that are missing, and not alter already defined ones

        """

        log.info('creating default user permissions')

    def get_or_create(cls, section, key):

        """ Return specified Ui object, creating it if necessary. """

        setting = cls.get_by_key(section, key)

        if setting is None:

            setting = cls(ui_section=section, ui_key=key)

            Session().add(setting)

        return setting

    @classmethod

    def get_builtin_hooks(cls):

        q = cls.query()

        q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))

        q = q.filter(cls.ui_section == 'hooks')

        q = q.order_by(cls.ui_section, cls.ui_key)

        return q.all()

    @classmethod

    def get_custom_hooks(cls):

        q = cls.query()

        q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))

        q = q.filter(cls.ui_section == 'hooks')

        q = q.order_by(cls.ui_section, cls.ui_key)

        return q.all()

    @classmethod

    def get_repos_location(cls):

        return cls.get_by_key('paths', '/').ui_value

    @classmethod

    def create_or_update_hook(cls, key, val):

        new_ui = cls.get_or_create('hooks', key)

        new_ui.ui_active = True

        new_ui.ui_value = val

    def __repr__(self):

        return '<%s %s.%s=%r>' % (

            self.__class__.__name__,

            self.ui_section, self.ui_key, self.ui_value)

class User(Base, BaseDbModel):

    __tablename__ = 'users'

    __table_args__ = (

        Index('u_username_idx', 'username'),

        Index('u_email_idx', 'email'),

        _table_args_default_dict,

    )

    DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'

    # The name of the default auth type in extern_type, 'internal' lives in auth_internal.py

    DEFAULT_AUTH_TYPE = 'internal'

    user_id = Column(Integer(), primary_key=True)

    username = Column(String(255), nullable=False, unique=True)

    password = Column(String(255), nullable=False)

    active = Column(Boolean(), nullable=False, default=True)

    admin = Column(Boolean(), nullable=False, default=False)

    name = Column("firstname", Unicode(255), nullable=False)

    lastname = Column(Unicode(255), nullable=False)

    _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?

    last_login = Column(DateTime(timezone=False), nullable=True)

    extern_type = Column(String(255), nullable=True) # FIXME: not nullable?

    extern_name = Column(String(255), nullable=True) # FIXME: not nullable?

    api_key = Column(String(255), nullable=False)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?

    user_log = relationship('UserLog')

    user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')

    repositories = relationship('Repository')

    repo_groups = relationship('RepoGroup')

    user_groups = relationship('UserGroup')

    user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')

    followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')

    repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')

    repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')

    group_member = relationship('UserGroupMember', cascade='all')

    # comments created by this user

    user_comments = relationship('ChangesetComment', cascade='all')

    # extra emails for this user

    user_emails = relationship('UserEmailMap', cascade='all')

    # extra API keys

    user_api_keys = relationship('UserApiKeys', cascade='all')

    ssh_keys = relationship('UserSshKeys', cascade='all')

    @hybrid_property

    def email(self):

        return self._email

    @email.setter

    def email(self, val):

        self._email = val.lower() if val else None

        return [self.email] + [x.email for x in other]

    @property

    def api_keys(self):

        other = UserApiKeys.query().filter(UserApiKeys.user == self).all()

        return [self.api_key] + [x.api_key for x in other]

    @property

    def ip_addresses(self):

        ret = UserIpMap.query().filter(UserIpMap.user == self).all()

        return [x.ip_addr for x in ret]

    @property

    def full_name(self):

        return '%s %s' % (self.firstname, self.lastname)

    @property

    def full_name_or_username(self):

        """

        Show full name.

        If full name is not set, fall back to username.

        """

        return ('%s %s' % (self.firstname, self.lastname)

                if (self.firstname and self.lastname) else self.username)

    @property

    def full_name_and_username(self):

        """

        Show full name and username as 'Firstname Lastname (username)'.

        If full name is not set, fall back to username.

        """

        return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)

                if (self.firstname and self.lastname) else self.username)

    @property

    def full_contact(self):

        return '%s %s <%s>' % (self.firstname, self.lastname, self.email)

    @property

    def short_contact(self):

        return '%s %s' % (self.firstname, self.lastname)

    @property

    def is_admin(self):

        return self.admin

    @hybrid_property

    def is_default_user(self):

    @hybrid_property

    def user_data(self):

        if not self._user_data:

            return {}

        try:

            return ext_json.loads(self._user_data)

        except TypeError:

            return {}

    @user_data.setter

    def user_data(self, val):

        try:

            self._user_data = ascii_bytes(ext_json.dumps(val))

        except Exception:

            log.error(traceback.format_exc())

    def __repr__(self):

        return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)

    @classmethod

    def guess_instance(cls, value):

        return super(User, cls).guess_instance(value, User.get_by_username)

    @classmethod

    def get_or_404(cls, id_, allow_default=True):

        '''

        Overridden version of BaseDbModel.get_or_404, with an extra check on

        the default user.

        '''

        user = super(User, cls).get_or_404(id_)

        if not allow_default and user.is_default_user:

            raise DefaultUserException()

        return user

    @classmethod

    def get_by_username_or_email(cls, username_or_email, case_insensitive=True):

        """

        For anything that looks like an email address, look up by the email address (matching

        case insensitively).

        For anything else, try to look up by the user name.

        This assumes no normal username can have '@' symbol.

        """

        if '@' in username_or_email:

            return User.get_by_email(username_or_email)

        else:

    @classmethod

    def get_by_email(cls, email, cache=False):

        q = cls.query().filter(sqlalchemy.func.lower(cls.email) == sqlalchemy.func.lower(email))

        ret = q.scalar()

        if ret is None:

            q = UserEmailMap.query()

            # try fetching in alternate email map

            q = q.filter(sqlalchemy.func.lower(UserEmailMap.email) == sqlalchemy.func.lower(email))

            q = q.options(joinedload(UserEmailMap.user))

            ret = getattr(q.scalar(), 'user', None)

        return ret

    @classmethod

    def get_from_cs_author(cls, author):

        """

        Tries to get User objects out of commit author string

        :param author:

        """

        from kallithea.lib.helpers import email, author_name

        # Valid email in the attribute passed, see if they're in the system

        _email = email(author)

        if _email:

            user = cls.get_by_email(_email)

            if user is not None:

                return user

        # Maybe we can match by username?

        _author = author_name(author)

        user = cls.get_by_username(_author, case_insensitive=True)

        if user is not None:

            return user

    def update_lastlogin(self):

        """Update user lastlogin"""

        self.last_login = datetime.datetime.now()

        log.debug('updated user %s lastlogin', self.username)

    @classmethod

    def get_first_admin(cls):

        user = User.query().filter(User.admin == True).first()

        if user is None:

            raise Exception('Missing administrative account!')

        return user

    @classmethod

    def get_default_user(cls):

        if user is None:

            raise Exception('Missing default account!')

        return user

    def get_api_data(self, details=False):

        """

        Common function for generating user related data for API

        """

        user = self

        data = dict(

            user_id=user.user_id,

            username=user.username,

            firstname=user.name,

            lastname=user.lastname,

            email=user.email,

            emails=user.emails,

            active=user.active,

            admin=user.admin,

        )

        if details:

            data.update(dict(

                extern_type=user.extern_type,

                extern_name=user.extern_name,

                api_key=user.api_key,

                api_keys=user.api_keys,

                last_login=user.last_login,

                ip_addresses=user.ip_addresses

                ))

        return data

    def __json__(self):

        data = dict(

            full_name=self.full_name,

            full_name_or_username=self.full_name_or_username,

            short_contact=self.short_contact,

            full_contact=self.full_contact

        )

        data.update(self.get_api_data())

        return data

class UserApiKeys(Base, BaseDbModel):

    __tablename__ = 'user_api_keys'

    __table_args__ = (

        Index('uak_api_key_idx', 'api_key'),

        Index('uak_api_key_expires_idx', 'api_key', 'expires'),

        _table_args_default_dict,

    )

import traceback

from sqlalchemy.exc import DatabaseError

from kallithea.lib.utils2 import str2bool

from kallithea.model.db import Permission, Session, User, UserRepoGroupToPerm, UserRepoToPerm, UserToPerm, UserUserGroupToPerm

log = logging.getLogger(__name__)

class PermissionModel(object):

    """

    Permissions model for Kallithea

    """

    def create_permissions(self):

        """

        Create permissions for whole system

        """

        for p in Permission.PERMS:

            if not Permission.get_by_key(p[0]):

                new_perm = Permission()

                new_perm.permission_name = p[0]

                Session().add(new_perm)

    def create_default_permissions(self, user, force=False):

        """

        Create missing default permissions for user. If force is set, the default

        permissions for the user are reset, otherwise only missing permissions are

        created.

        :param user:

        """

        user = User.guess_instance(user)

        def _make_perm(perm):

            new_perm = UserToPerm()

            new_perm.user = user

            new_perm.permission = Permission.get_by_key(perm)

            return new_perm

        def _get_group(perm_name):

            return '.'.join(perm_name.split('.')[:1])

        perms = UserToPerm.query().filter(UserToPerm.user == user).all()

        defined_perms_groups = set(_get_group(x.permission.permission_name) for x in perms)

        log.debug('GOT ALREADY DEFINED:%s', perms)

        if force:

            for perm in perms:

                Session().delete(perm)

            Session().commit()

            defined_perms_groups = []

        # For every default permission that needs to be created, we check if

        # its group is already defined. If it's not, we create default permission.

            gr = _get_group(perm_name)

            if gr not in defined_perms_groups:

                log.debug('GR:%s not found, creating permission %s',

                          gr, perm_name)

                new_perm = _make_perm(perm_name)

                Session().add(new_perm)

    def update(self, form_result):

        perm_user = User.get_by_username(username=form_result['perm_user_name'])

        try:

            # stage 1 set anonymous access

            if perm_user.is_default_user:

                perm_user.active = str2bool(form_result['anonymous'])

            # stage 2 reset defaults and set them from form data

            def _make_new(usr, perm_name):

                log.debug('Creating new permission:%s', perm_name)

                new = UserToPerm()

                new.user = usr

                new.permission = Permission.get_by_key(perm_name)

                return new

            # clear current entries, to make this function idempotent

            # it will fix even if we define more permissions or permissions

            # are somehow missing

            u2p = UserToPerm.query() \

                .filter(UserToPerm.user == perm_user) \

                .all()

            for p in u2p:

                Session().delete(p)

            # create fresh set of permissions

            for def_perm_key in ['default_repo_perm',

                                 'default_group_perm',

                                 'default_user_group_perm',

                                 'default_repo_create',

                                 'create_on_write', # special case for create repos on write access to group

                                 #'default_repo_group_create', # not implemented yet

                                 'default_user_group_create',

                                 'default_fork',

                                 'default_register',

                                 'default_extern_activate']:

                p = _make_new(perm_user, form_result[def_perm_key])

                Session().add(p)

            # stage 3 update all default permissions for repos if checked

            if form_result['overwrite_default_repo']:

                _def_name = form_result['default_repo_perm'].split('repository.')[-1]

                _def = Permission.get_by_key('repository.' + _def_name)

                )

    return _validator

def ValidPerms(type_='repo'):

    if type_ == 'repo_group':

        EMPTY_PERM = 'group.none'

    elif type_ == 'repo':

        EMPTY_PERM = 'repository.none'

    elif type_ == 'user_group':

        EMPTY_PERM = 'usergroup.none'

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'perm_new_member_name':

                _('This username or user group name is not valid')

        }

        def to_python(self, value, state):

            perms_update = OrderedSet()

            perms_new = OrderedSet()

            # build a list of permission to update and new permission to create

            # CLEAN OUT ORG VALUE FROM NEW MEMBERS, and group them using

            new_perms_group = defaultdict(dict)

            for k, v in value.copy().items():

                if k.startswith('perm_new_member'):

                    del value[k]

                    _type, part = k.split('perm_new_member_')

                    args = part.split('_')

                    if len(args) == 1:

                        new_perms_group[args[0]]['perm'] = v

                    elif len(args) == 2:

                        _key, pos = args

                        new_perms_group[pos][_key] = v

            # fill new permissions in order of how they were added

            for k in sorted(new_perms_group, key=lambda k: int(k)):

                perm_dict = new_perms_group[k]

                new_member = perm_dict.get('name')

                new_perm = perm_dict.get('perm')

                new_type = perm_dict.get('type')

                if new_member and new_perm and new_type:

                    perms_new.add((new_member, new_perm, new_type))

            for k, v in value.items():

                if k.startswith('u_perm_') or k.startswith('g_perm_'):

                    t = {'u': 'user',

                         'g': 'users_group'

                    }[k[0]]

                        if str2bool(value.get('repo_private')):

                            # set none for default when updating to

                            # private repo protects against form manipulation

                            v = EMPTY_PERM

            value['perms_updates'] = list(perms_update)

            value['perms_new'] = list(perms_new)

            # update permissions

            for k, v, t in perms_new:

                try:

                    if t == 'user':

                        _user_db = User.query() \

                            .filter(User.active == True) \

                            .filter(User.username == k).one()

                    if t == 'users_group':

                        _user_db = UserGroup.query() \

                            .filter(UserGroup.users_group_active == True) \

                            .filter(UserGroup.users_group_name == k).one()

                except Exception as e:

                    log.warning('Error validating %s permission %s', t, k)

                    msg = self.message('perm_new_member_type', state)

                    raise formencode.Invalid(msg, value, state,

                        error_dict=dict(perm_new_member_name=msg)

                    )

            return value

    return _validator

def ValidSettings():

    class _validator(formencode.validators.FancyValidator):

        def _convert_to_python(self, value, state):

            # settings  form for users that are not admin

            # can't edit certain parameters, it's extra backup if they mangle

            # with forms

            forbidden_params = [

                'user', 'repo_type',

                'repo_enable_downloads', 'repo_enable_statistics'

            ]

            for param in forbidden_params:

                if param in value:

                    del value[param]

            return value

        def _validate_python(self, value, state):

            pass

    return _validator

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

        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,

                                          perm='repository.none')

        try:

            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)

        finally:

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

                                              perm='repository.read')

    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)

    @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

        )

        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)

    @base.parametrize('fork_name', [

        'api-repo-fork',

        '%s/api-repo-fork' % TEST_REPO_GROUP,

    ])

    def test_api_fork_repo_non_admin(self, fork_name):

        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

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

        fixture.destroy_repo(fork_name)

    def test_api_fork_repo_non_admin_specify_owner(self):

        fork_name = 'api-repo-fork'

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

                                  repoid=self.REPO,

                                  fork_name=fork_name,

                                  owner=base.TEST_USER_ADMIN_LOGIN,

        )

        response = api_call(self, params)

        expected = 'Only Kallithea admin can specify `owner` param'

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

        fixture.destroy_repo(fork_name)

    def test_api_fork_repo_non_admin_no_permission_to_fork(self):

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

                                          perm='repository.none')

        try:

            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)

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

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

        finally:

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

                                              perm='repository.read')

            fixture.destroy_repo(fork_name)

    @base.parametrize('name,perm', [

        ('read', 'repository.read'),

        ('write', 'repository.write'),

        ('admin', 'repository.admin'),

    ])

    def test_api_fork_repo_non_admin_no_create_repo_permission(self, name, perm):

        fork_name = 'api-repo-fork'

        # regardless of base repository permission, forking is disallowed

        # when repository creation is disabled

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

                                          user=self.TEST_USER_LOGIN,

                                          perm=perm)

        UserModel().revoke_perm('default', 'hg.create.repository')

        UserModel().grant_perm('default', 'hg.create.none')

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

                                  repoid=self.REPO,

                                  fork_name=fork_name,

        )

        response = api_call(self, params)

        expected = 'no permission to create repositories'

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

        fixture.destroy_repo(fork_name)

    def test_api_fork_repo_unknown_owner(self):

        fork_name = 'api-repo-fork'

        owner = 'i-dont-exist'

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

                                  repoid=self.REPO,

                                  fork_name=fork_name,

                                  owner=owner,

        )

        response = api_call(self, params)

        expected = 'user `%s` does not exist' % owner

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

    def test_api_fork_repo_fork_exists(self):

        fork_name = 'api-repo-fork'

        fixture.create_fork(self.REPO, fork_name)

        try:

            fork_name = 'api-repo-fork'

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

                                      repoid=self.REPO,

                                      fork_name=fork_name,

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=repo_name,

                                                repo_type=self.REPO_TYPE,

                                                repo_description=description,

                                                repo_group=gr.group_id,

                                                _session_csrf_secret_token=self.session_csrf_secret_token()))

        ## run the check page that triggers the flash message

        response = self.app.get(base.url('repo_check_home', repo_name=repo_name_full))

        assert response.json == {'result': True}

        self.checkSessionFlash(response,

                               'Created repository <a href="/%s">%s</a>'

                               % (repo_name_full, repo_name_full))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_full).one()

        new_repo_id = new_repo.repo_id

        assert new_repo.repo_name == repo_name_full

        assert new_repo.description == description

        # test if the repository is visible in the list ?

        response = self.app.get(base.url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        assert len(inherited_perms) == 1

        # test if the repository was created on filesystem

        try:

            vcs.get_repo(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)