def changeset_cache(self):

        try:

            cs_cache = json.loads(self._changeset_cache) # might raise on bad data

            cs_cache['raw_id'] # verify data, raise exception on error

            return cs_cache

        except (TypeError, KeyError, ValueError):

            return EmptyChangeset().__json__()

    @changeset_cache.setter

    def changeset_cache(self, val):

        try:

            self._changeset_cache = json.dumps(val)

        except Exception:

            log.error(traceback.format_exc())

    @classmethod

    def query(cls, sorted=False):

        """Add Repository-specific helpers for common query constructs.

        sorted: if True, apply the default ordering (name, case insensitive).

        """

        q = super(Repository, cls).query()

        if sorted:

            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 = Session().query(cls).filter(cls.repo_name == repo_name)

        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:

        """

        q = Session().query(Ui) \

            .filter(Ui.ui_key == cls.url_sep())

        q = q.options(FromCache("sql_cache_short", "repository_repo_path"))

        return q.one().ui_value

    @property

    def forks(self):

        """

        Return forks of this repo

        """

        return Repository.get_repo_forks(self.repo_id)

    @property

    def parent(self):

        """

        Returns fork parent

        """

        return self.fork

    @property

    def just_name(self):

        return self.repo_name.split(Repository.url_sep())[-1]

# -*- 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():

    class _UniqueListFromString(formencode.FancyValidator):

        """

        Split value on ',' and make unique while preserving order

        """

        messages = dict(

            empty=_('Value cannot be an empty list'),

            missing_value=_('Value cannot be an empty list'),

        )

        def _to_python(self, value, state):

            value = aslist(value, ',')

            seen = set()

            return [c for c in value if not (c in seen or seen.add(c))]

        def empty_value(self, value):

            return []

    return _UniqueListFromString

def ValidUsername(edit=False, old_data=None):

    old_data = old_data or {}

    class _validator(formencode.validators.FancyValidator):

        messages = {

            if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:

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

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(users_group_name=msg)

                )

    return _validator

def ValidRepoGroup(edit=False, old_data=None):

    old_data = old_data or {}

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'parent_group_id': _('Cannot assign this group as parent'),

            'group_exists': _('Group "%(group_name)s" already exists'),

            'repo_exists':

                _('Repository with name "%(group_name)s" already exists')

        }

        def validate_python(self, value, state):

            # TODO WRITE VALIDATIONS

            group_name = value.get('group_name')

            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

def ValidOldPassword(username):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_password': _('Invalid old password')

        }

        def validate_python(self, value, state):

            from kallithea.lib import auth_modules

            if auth_modules.authenticate(username, value, '') is None:

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

                raise formencode.Invalid(msg, value, state,

                    error_dict=dict(current_password=msg)

                )

    return _validator

def ValidPasswordsMatch(password_field, password_confirmation_field):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'password_mismatch': _('Passwords do not match'),

        }

        def validate_python(self, value, state):

    class _validator(formencode.validators.FancyValidator):

        messages = {

            'invalid_repo_name':

                _('Repository name %(repo)s is not allowed'),

            'repository_exists':

                _('Repository named %(repo)s already exists'),

            'repository_in_group_exists': _('Repository "%(repo)s" already '

                                            'exists in group "%(group)s"'),

            'same_group_exists': _('Repository group with name "%(repo)s" '

                                   'already exists')

        }

        def _to_python(self, value, state):

            repo_name = repo_name_slug(value.get('repo_name', ''))

            repo_group = value.get('repo_group')

            if repo_group:

                gr = RepoGroup.get(repo_group)

                group_path = gr.full_path

                group_name = gr.group_name

                # value needs to be aware of group name in order to check

                # db key This is an actual just the name to store in the

                # database

                repo_name_full = group_path + RepoGroup.url_sep() + repo_name

            else:

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

    from kallithea.lib.utils import make_ui

    def url_handler(repo_type, url, ui):

        if repo_type == 'hg':

            from kallithea.lib.vcs.backends.hg.repository import MercurialRepository

            if url.startswith('http') or url.startswith('ssh'):

                # initially check if it's at least the proper URL

                # or does it pass basic auth

                MercurialRepository._check_url(url, ui)

            elif url.startswith('svn+http'):

                from hgsubversion.svnrepo import svnremoterepo

                svnremoterepo(ui, url).svn.uuid

            elif url.startswith('git+http'):

                raise NotImplementedError()

            else:

                raise Exception('clone from URI %s not allowed' % (url,))

        elif repo_type == 'git':

            from kallithea.lib.vcs.backends.git.repository import GitRepository

            if url.startswith('http') or url.startswith('git'):

                # initially check if it's at least the proper URL

                # or does it pass basic auth

                GitRepository._check_url(url)

            elif url.startswith('svn+http'):

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

    def test_index(self):

        self.log_user()

        response = self.app.get(url('repos'))

    def test_create(self):

        self.log_user()

        repo_name = self.NEW_REPO

        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,

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

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

        self.checkSessionFlash(response,

                               '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,

                               '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(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(safe_str(os.path.join(Ui.get_by_key('paths', '/').ui_value, repo_name_full)))

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

import mock

from kallithea import CONFIG

from kallithea.config.conf import INDEX_FILENAMES

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.repo_group import RepoGroupModel

from kallithea.tests.base import *

from kallithea.tests.fixture import create_test_index, Fixture

fixture = Fixture()

def init_indexing_test(repo):

    prev = fixture.commit_change(repo.repo_name,

                                 filename='this_should_be_unique_filename.txt',

                                 content='this_should_be_unique_content\n',

                                 message='this_should_be_unique_commit_log',

                                 vcs_type='hg',

                                 newfile=True)

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

    @classmethod

    def setup_class(cls):

        for reponame, init_or_fork, groupname in repos:

            if groupname and groupname not in groupids:

                group = fixture.create_repo_group(groupname)

                groupids[groupname] = group.group_id

            if callable(init_or_fork):

                repo = fixture.create_repo(reponame,

                                           repo_group=groupname)

                init_or_fork(repo)

            else:

                repo = fixture.create_fork(init_or_fork, reponame,

                                           repo_group=groupname)

            repoids[reponame] = repo.repo_id

        # treat "it" as indexable filename

        filenames_mock = list(INDEX_FILENAMES)

        filenames_mock.append('it')

        with mock.patch('kallithea.lib.indexers.daemon.INDEX_FILENAMES',

                        filenames_mock):

            rebuild_index(full_index=False) # only for newly added repos

    @classmethod

    def teardown_class(cls):

                                        force_delete=True)

        Session().commit()

        Session.remove()

        rebuild_index(full_index=True) # rebuild fully for subsequent tests

    @parametrize('reponame', [

        (u'indexing_test'),

        (u'indexing_test-fork'),

        (u'group/indexing_test'),

        (u'this-is-it'),

        (u'*-fork'),

        (u'group/*'),

    ])

    @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_repository_tokenization(self, reponame, searchtype, query, hit):

        self.log_user()

        q = 'repository:%s %s' % (reponame, query)

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

        # matching against both 2 revisions

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

        ('commit', 'owner:"this-is-it"', 0),

        # matching against only 1 revision

        ('commit', 'author:"this is it"', 1),

        ('commit', 'author:"this-is-it"', 1),

    ])

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

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

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

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