@HasPermissionAnyDecorator('hg.admin')

    def grant_user_permission(self, apiuser, repoid, userid, perm):

        """

        Grant permission for user on given repository, or update existing one

        if found

        :param repoid:

        :param userid:

        :param perm:

        """

        repo = get_repo_or_error(repoid)

        user = get_user_or_error(userid)

        perm = get_perm_or_error(perm)

        try:

            RepoModel().grant_user_permission(repo=repo, user=user, perm=perm)

            Session().commit()

            return dict(

                msg='Granted perm: `%s` for user: `%s` in repo: `%s`' % (

                    perm.permission_name, user.username, repo.repo_name

                ),

                success=True

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to edit permission for user: `%s` in repo: `%s`' % (

                    userid, repoid

                )

            )

    @HasPermissionAnyDecorator('hg.admin')

    def revoke_user_permission(self, apiuser, repoid, userid):

        """

        Revoke permission for user on given repository

        :param apiuser:

        :param repoid:

        :param userid:

        """

        repo = get_repo_or_error(repoid)

        user = get_user_or_error(userid)

        try:

            RepoModel().revoke_user_permission(repo=repo, user=user)

            Session().commit()

            return dict(

                msg='Revoked perm for user: `%s` in repo: `%s`' % (

                    user.username, repo.repo_name

                ),

                success=True

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to edit permission for user: `%s` in repo: `%s`' % (

                    userid, repoid

                )

            )

    @HasPermissionAnyDecorator('hg.admin')

    def grant_users_group_permission(self, apiuser, repoid, usersgroupid,

                                     perm):

        """

        Grant permission for users group on given repository, or update

        existing one if found

        :param apiuser:

        :param repoid:

        :param usersgroupid:

        :param perm:

        """

        repo = get_repo_or_error(repoid)

        perm = get_perm_or_error(perm)

        users_group = get_users_group_or_error(usersgroupid)

        try:

            RepoModel().grant_users_group_permission(repo=repo,

                                                     group_name=users_group,

                                                     perm=perm)

            Session().commit()

            return dict(

                msg='Granted perm: `%s` for users group: `%s` in '

                    'repo: `%s`' % (

                    perm.permission_name, users_group.users_group_name,

                    repo.repo_name

                ),

                success=True

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to edit permission for users group: `%s` in '

                'repo: `%s`' % (

                    usersgroupid, repo.repo_name

                )

            )

    @HasPermissionAnyDecorator('hg.admin')

    def revoke_users_group_permission(self, apiuser, repoid, usersgroupid):

        """

        Revoke permission for users group on given repository

        :param apiuser:

        :param repoid:

        :param usersgroupid:

        """

        repo = get_repo_or_error(repoid)

        users_group = get_users_group_or_error(usersgroupid)

        try:

            RepoModel().revoke_users_group_permission(repo=repo,

                                                      group_name=users_group)

            Session().commit()

            return dict(

                msg='Revoked perm for users group: `%s` in repo: `%s`' % (

                    users_group.users_group_name, repo.repo_name

                ),

                success=True

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to edit permission for users group: `%s` in '

                'repo: `%s`' % (

                    users_group.users_group_name, repo.repo_name

                )

            )

        self.cache_key = cache_key

        self.cache_args = cache_args

        self.cache_active = False

    def __unicode__(self):

        return u"<%s('%s:%s')>" % (self.__class__.__name__,

                                  self.cache_id, self.cache_key)

    @property

    def prefix(self):

        _split = self.cache_key.split(self.cache_args, 1)

        if _split and len(_split) == 2:

            return _split[0]

        return ''

    @classmethod

    def clear_cache(cls):

        cls.query().delete()

    @classmethod

    def _get_key(cls, key):

        """

        Wrapper for generating a key, together with a prefix

        :param key:

        """

        import rhodecode

        prefix = ''

        org_key = key

        iid = rhodecode.CONFIG.get('instance_id')

        if iid:

            prefix = iid

        return "%s%s" % (prefix, key), prefix, org_key

    @classmethod

    def get_by_key(cls, key):

        return cls.query().filter(cls.cache_key == key).scalar()

    @classmethod

    def get_by_repo_name(cls, repo_name):

        return cls.query().filter(cls.cache_args == repo_name).all()

    @classmethod

    def _get_or_create_key(cls, key, repo_name, commit=True):

        inv_obj = Session().query(cls).filter(cls.cache_key == key).scalar()

        if not inv_obj:

            try:

                inv_obj = CacheInvalidation(key, repo_name)

                Session().add(inv_obj)

                if commit:

                    Session().commit()

            except Exception:

                log.error(traceback.format_exc())

                Session().rollback()

        return inv_obj

    @classmethod

    def invalidate(cls, key):

        """

        Returns Invalidation object if this given key should be invalidated

        None otherwise. `cache_active = False` means that this cache

        state is not valid and needs to be invalidated

        :param key:

        """

        repo_name = key

        repo_name = remove_suffix(repo_name, '_README')

        repo_name = remove_suffix(repo_name, '_RSS')

        repo_name = remove_suffix(repo_name, '_ATOM')

        # adds instance prefix

        key, _prefix, _org_key = cls._get_key(key)

        inv = cls._get_or_create_key(key, repo_name)

        if inv and inv.cache_active is False:

            return inv

    @classmethod

    def set_invalidate(cls, key=None, repo_name=None):

        """

        Mark this Cache key for invalidation, either by key or whole

        cache sets based on repo_name

        :param key:

        """

        if key:

            key, _prefix, _org_key = cls._get_key(key)

            inv_objs = Session().query(cls).filter(cls.cache_key == key).all()

        elif repo_name:

            inv_objs = Session().query(cls).filter(cls.cache_args == repo_name).all()

        log.debug('marking %s key[s] for invalidation based on key=%s,repo_name=%s'

                  % (len(inv_objs), key, repo_name))

        try:

            for inv_obj in inv_objs:

                inv_obj.cache_active = False

                Session().add(inv_obj)

            Session().commit()

        except Exception:

            log.error(traceback.format_exc())

            Session().rollback()

    @classmethod

    def set_valid(cls, key):

        """

        Mark this cache key as active and currently cached

        :param key:

        """

        inv_obj = cls.get_by_key(key)

        inv_obj.cache_active = True

        Session().add(inv_obj)

        Session().commit()

    @classmethod

    def get_cache_map(cls):

        class cachemapdict(dict):

            def __init__(self, *args, **kwargs):

                fixkey = kwargs.get('fixkey')

                if fixkey:

                    del kwargs['fixkey']

                self.fixkey = fixkey

                super(cachemapdict, self).__init__(*args, **kwargs)

            def __getattr__(self, name):

                key = name

                if self.fixkey:

                    key, _prefix, _org_key = cls._get_key(key)

                if key in self.__dict__:

                    return self.__dict__[key]

                else:

                    return self[key]

            def __getitem__(self, key):

                if self.fixkey:

                    key, _prefix, _org_key = cls._get_key(key)

                try:

                    return super(cachemapdict, self).__getitem__(key)

                except KeyError:

                    return

        cache_map = cachemapdict(fixkey=True)

        for obj in cls.query().all():

            cache_map[obj.cache_key] = cachemapdict(obj.get_dict())

        return cache_map

class ChangesetComment(Base, BaseModel):

    __tablename__ = 'changeset_comments'

    __table_args__ = (

        Index('cc_revision_idx', 'revision'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'},

    )

    comment_id = Column('comment_id', Integer(), nullable=False, primary_key=True)

    repo_id = Column('repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    revision = Column('revision', String(40), nullable=True)

    pull_request_id = Column("pull_request_id", Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)

    line_no = Column('line_no', Unicode(10), nullable=True)

    hl_lines = Column('hl_lines', Unicode(512), nullable=True)

    f_path = Column('f_path', Unicode(1000), nullable=True)

    user_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)

    text = Column('text', UnicodeText(25000), nullable=False)

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

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

    author = relationship('User', lazy='joined')

    repo = relationship('Repository')

    status_change = relationship('ChangesetStatus', cascade="all, delete, delete-orphan")

    pull_request = relationship('PullRequest', lazy='joined')

    @classmethod

    def get_users(cls, revision=None, pull_request_id=None):

        """

        Returns user associated with this ChangesetComment. ie those

        who actually commented

        :param cls:

        :param revision:

        """

        q = Session().query(User)\

                .join(ChangesetComment.author)

        if revision:

            q = q.filter(cls.revision == revision)

        elif pull_request_id:

            q = q.filter(cls.pull_request_id == pull_request_id)

        return q.all()

from rhodecode.tests import *

from rhodecode.model.db import RhodeCodeSetting

from nose.plugins.skip import SkipTest

skip_ldap_test = False

try:

    import ldap

except ImportError:

    # means that python-ldap is not installed

    skip_ldap_test = True

    pass

class TestLdapSettingsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='admin/ldap_settings',

                                    action='index'))

        self.assertTrue('LDAP administration' in response.body)

    def test_ldap_save_settings(self):

        self.log_user()

        if skip_ldap_test:

            raise SkipTest('skipping due to missing ldap lib')

        test_url = url(controller='admin/ldap_settings',

                       action='ldap_settings')

        response = self.app.post(url=test_url,

            params={'ldap_host' : u'dc.example.com',

                    'ldap_port' : '999',

                    'ldap_tls_kind' : 'PLAIN',

                    'ldap_tls_reqcert' : 'NEVER',

                    'ldap_dn_user':'test_user',

                    'ldap_dn_pass':'test_pass',

                    'ldap_base_dn':'test_base_dn',

                    'ldap_filter':'test_filter',

                    'ldap_search_scope':'BASE',

                    'ldap_attr_login':'test_attr_login',

                    'ldap_attr_firstname':'ima',

                    'ldap_attr_lastname':'tester',

                    'ldap_attr_email':'test@example.com' })

        new_settings = RhodeCodeSetting.get_ldap_settings()

        self.assertEqual(new_settings['ldap_host'], u'dc.example.com',

                         'fail db write compare')

        self.checkSessionFlash(response,

                               'Ldap settings updated successfully')

    def test_ldap_error_form(self):

        self.log_user()

        if skip_ldap_test:

            raise SkipTest('skipping due to missing ldap lib')

        test_url = url(controller='admin/ldap_settings',

                       action='ldap_settings')

        response = self.app.post(url=test_url,

            params={'ldap_host' : '',

                    'ldap_port' : 'i-should-be-number',

                    'ldap_tls_kind' : 'PLAIN',

                    'ldap_tls_reqcert' : 'NEVER',

                    'ldap_dn_user':'',

                    'ldap_dn_pass':'',

                    'ldap_base_dn':'',

                    'ldap_filter':'',

                    'ldap_search_scope':'BASE',

                    'ldap_attr_login':'', #  <----- missing required input

                    'ldap_attr_firstname':'',

                    'ldap_attr_lastname':'',

                    'ldap_attr_email':'' })

        self.assertTrue("""<span class="error-message">The LDAP Login"""

                        """ attribute of the CN must be specified""" in

                        response.body)

        self.assertTrue("""<span class="error-message">Please """

                        """enter a number</span>""" in response.body)

    def test_ldap_login(self):

        pass

    def test_ldap_login_incorrect(self):

        pass