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

                )

            )

                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

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