q = q.order_by(func.lower(Repository.repo_name))

        return q

    @classmethod

    def url_sep(cls):

        return URL_SEP

    @classmethod

    def normalize_repo_name(cls, repo_name):

        """

        Normalizes os specific repo_name to the format internally stored inside

        database using URL_SEP

        :param cls:

        :param repo_name:

        """

        return cls.url_sep().join(repo_name.split(os.sep))

    @classmethod

    def guess_instance(cls, value):

        return super(Repository, cls).guess_instance(value, Repository.get_by_repo_name)

    @classmethod

        q = q.options(joinedload(Repository.fork)) \

                .options(joinedload(Repository.owner)) \

                .options(joinedload(Repository.group))

        return q.scalar()

    @classmethod

    def get_by_full_path(cls, repo_full_path):

        base_full_path = os.path.realpath(cls.base_path())

        repo_full_path = os.path.realpath(repo_full_path)

        assert repo_full_path.startswith(base_full_path + os.path.sep)

        repo_name = repo_full_path[len(base_full_path) + 1:]

        repo_name = cls.normalize_repo_name(repo_name)

        return cls.get_by_repo_name(repo_name.strip(URL_SEP))

    @classmethod

    def get_repo_forks(cls, repo_id):

        return cls.query().filter(Repository.fork_id == repo_id)

    @classmethod

    def base_path(cls):

        """

        Returns base path where all repos are stored

        :param cls:

# -*- 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/>.

"""

Set of generic validators

"""

import os

import re

import formencode

import logging

from collections import defaultdict

from tg.i18n import ugettext as _

from webhelpers.pylonslib.secure_form import authentication_token

import sqlalchemy

from formencode.validators import (

    UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,

    NotEmpty, IPAddress, CIDR, String, FancyValidator

)

from kallithea.lib.compat import OrderedSet

from kallithea.lib import ipaddr

from kallithea.lib.utils import repo_name_slug

from kallithea.lib.utils2 import str2bool, aslist

from kallithea.model.db import RepoGroup, Repository, UserGroup, User

from kallithea.lib.exceptions import LdapImportError

from kallithea.config.routing import ADMIN_PREFIX

from kallithea.lib.auth import HasRepoGroupPermissionLevel, HasPermissionAny

# silence warnings and pylint

UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \

    NotEmpty, IPAddress, CIDR, String, FancyValidator

log = logging.getLogger(__name__)

def UniqueListFromString():

            parent_group_id = value.get('parent_group_id')

            # slugify repo group just in case :)

            slug = repo_name_slug(group_name)

            # check for parent of self

            parent_of_self = lambda: (

                old_data['group_id'] == parent_group_id

                if parent_group_id else False

            )

            if edit and parent_of_self():

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

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(parent_group_id=msg)

                )

            old_gname = None

            if edit:

                old_gname = RepoGroup.get(old_data.get('group_id')).group_name

            if old_gname != group_name or not edit:

                # check group

                gr = RepoGroup.query() \

                      .filter(RepoGroup.parent_group_id == parent_group_id) \

                      .scalar()

                if gr is not None:

                    msg = self.message('group_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

                # check for same repo

                repo = Repository.query() \

                      .scalar()

                if repo is not None:

                    msg = self.message('repo_exists', state, group_name=slug)

                    raise formencode.Invalid(msg, value, state,

                            error_dict=dict(group_name=msg)

                    )

    return _validator

def ValidPassword():

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password':

                _('Invalid characters (non-ascii) in password')

        }

        def validate_python(self, value, state):

            try:

                (value or '').decode('ascii')

            except UnicodeError:

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

                raise formencode.Invalid(msg, value, state,)

    return _validator

                group_name = group_path = ''

                repo_name_full = repo_name

            value['repo_name'] = repo_name

            value['repo_name_full'] = repo_name_full

            value['group_path'] = group_path

            value['group_name'] = group_name

            return value

        def validate_python(self, value, state):

            repo_name = value.get('repo_name')

            repo_name_full = value.get('repo_name_full')

            group_path = value.get('group_path')

            group_name = value.get('group_name')

            if repo_name in [ADMIN_PREFIX, '']:

                msg = self.message('invalid_repo_name', state, repo=repo_name)

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(repo_name=msg)

                )

            rename = old_data.get('repo_name') != repo_name_full

            create = not edit

            if rename or create:

                if group_path != '':

                        msg = self.message('repository_in_group_exists', state,

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                        msg = self.message('same_group_exists', state,

                                repo=repo_name)

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

                        msg = self.message('repository_exists', state,

                        raise formencode.Invalid(msg, value, state,

                            error_dict=dict(repo_name=msg)

                        )

            return value

    return _validator

def ValidForkName(*args, **kwargs):

    return ValidRepoName(*args, **kwargs)

def SlugifyName():

    class _validator(formencode.validators.FancyValidator):

        def _to_python(self, value, state):

            return repo_name_slug(value)

        def validate_python(self, value, state):

            pass

    return _validator

def ValidCloneUri():

class TestRepoGroupsController(TestController):

# -*- coding: utf-8 -*-

import os

import mock

import urllib

import pytest

from kallithea.lib import vcs

from kallithea.lib.utils2 import safe_str, safe_unicode

from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \

    Permission, Ui

from kallithea.model.user import UserModel

from kallithea.tests.base import *

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.repo import RepoModel

from kallithea.tests.fixture import Fixture, error_function

fixture = Fixture()

def _get_permission_for_user(user, repo):

    perm = UserRepoToPerm.query() \

                .filter(UserRepoToPerm.repository ==

                        Repository.get_by_repo_name(repo)) \

                .filter(UserRepoToPerm.user == User.get_by_username(user)) \

                .all()

    return perm

class _BaseTestCase(TestController):

    """

    Write all tests here

    """

    REPO = None

    REPO_TYPE = None

    NEW_REPO = None

    OTHER_TYPE_REPO = None

    OTHER_TYPE = None

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

                               % (repo_name, repo_name))

        # test if the repo was created in the database

        new_repo = Session().query(Repository) \

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

        assert new_repo.repo_name == repo_name

        assert new_repo.description == description

        # test if the repository is visible in the list ?

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

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

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

        except vcs.exceptions.VCSError:

            pytest.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name)

        Session().commit()

    def test_create_in_group(self):

        self.log_user()

        ## create GROUP

        group_name = u'sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description=u'test',

                                     owner=TEST_USER_ADMIN_LOGIN)

        Session().commit()

        repo_name = u'ingroup'

        repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])

        description = u'description for newly created repo'

        response = self.app.post(url('repos'),

                        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,

                                                _authentication_token=self.authentication_token()))

        ## run the check page that triggers the flash message

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

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

        self.checkSessionFlash(response,

def init_stopword_test(repo):

    prev = fixture.commit_change(repo.repo_name,

                                 filename='this/is/it',

                                 content='def test\n',

                                 message='bother to ask where - in folder',

                                 vcs_type='hg',

                                 newfile=True)

    prev = fixture.commit_change(repo.repo_name,

                                 filename='join.us',

                                 content='def test\n',

                                 message='bother to ask where - top level',

                                 author='this is it <this-is-it@foo.bar.com>',

                                 vcs_type='hg',

                                 parent=prev,

                                 newfile=True)

repos = [

    # reponame,              init func or fork base, groupname

    (u'indexing_test',       init_indexing_test,     None),

    (u'indexing_test-fork',  u'indexing_test',       None),

    (u'group/indexing_test', u'indexing_test',       u'group'),

    (u'this-is-it',          u'indexing_test',       None),

    (u'indexing_test-foo',   u'indexing_test',       None),

    (u'stopword_test',       init_stopword_test,     None),

]

# map: name => id

repoids = {}

groupids = {}

def rebuild_index(full_index):

    with mock.patch('kallithea.lib.indexers.daemon.log.debug',

                    lambda *args, **kwargs: None):

        # The more revisions managed repositories have, the more

        # memory capturing "log.debug()" output in "indexers.daemon"

        # requires. This may cause unintentional failure of subsequent

        # tests, if ENOMEM at forking "git" prevents from rebuilding

        # index for search.

        # Therefore, "log.debug()" is disabled regardless of logging

        # level while rebuilding index.

        # (FYI, ENOMEM occurs at forking "git" with python 2.7.3,

        # Linux 3.2.78-1 x86_64, 3GB memory, and no ulimit

        # configuration for memory)

        create_test_index(TESTS_TMP_PATH, CONFIG, full_index=full_index)

class TestSearchControllerIndexing(TestController):

        response = self.app.get(url(controller='search', action='index'),

                                {'q': q, 'type': searchtype})

        response.mustcontain('>%d results' % hit)

    @parametrize('reponame', [

        (u'indexing_test'),

        (u'indexing_test-fork'),

        (u'group/indexing_test'),

        (u'this-is-it'),

    ])

    @parametrize('searchtype,query,hit', [

        ('content', 'this_should_be_unique_content', 1),

        ('commit', 'this_should_be_unique_commit_log', 1),

        ('path', 'this_should_be_unique_filename.txt', 1),

    ])

    def test_searching_under_repository(self, reponame, searchtype, query, hit):

        self.log_user()

        response = self.app.get(url(controller='search', action='index',

                                    repo_name=reponame),

                                {'q': query, 'type': searchtype})

        response.mustcontain('>%d results' % hit)

    @parametrize('searchtype,query,hit', [

        ('content', 'path:this/is/it def test', 1),

        ('commit', 'added:this/is/it bother to ask where', 1),

        # this condition matches against files below, because

        # "path:" condition is also applied on "repository path".

        # - "this/is/it" in "stopword_test" repo

        # - "this_should_be_unique_filename.txt" in "this-is-it" repo

        ('path', 'this/is/it', 2),

        ('content', 'extension:us', 1),

        ('path', 'extension:us', 1),

    ])

    def test_filename_stopword(self, searchtype, query, hit):

        response = self.app.get(url(controller='search', action='index'),

                                {'q': query, 'type': searchtype})

        response.mustcontain('>%d results' % hit)

    @parametrize('searchtype,query,hit', [

        # matching against both 2 files

        ('content', 'owner:"this is it"', 0),

        ('content', 'owner:this-is-it', 0),

        ('path', 'owner:"this is it"', 0),

        ('path', 'owner:this-is-it', 0),