# BytesIO

#==============================================================================

try:

    from io import BytesIO

except ImportError:

    from cStringIO import StringIO as BytesIO

#==============================================================================

# bytes

#==============================================================================

if __py_version__ >= (2, 6):

    _bytes = bytes

else:

    # in py2.6 bytes is a synonim for str

    _bytes = str

#==============================================================================

# deque

#==============================================================================

if __py_version__ >= (2, 6):

    from collections import deque

else:

    #need to implement our own deque with maxlen

    class deque(object):

        def __init__(self, iterable=(), maxlen= -1):

            if not hasattr(self, 'data'):

                self.left = self.right = 0

                self.data = {}

            self.maxlen = maxlen or -1

            self.extend(iterable)

        def append(self, x):

            self.data[self.right] = x

            self.right += 1

            if self.maxlen != -1 and len(self) > self.maxlen:

                self.popleft()

        def appendleft(self, x):

            self.left -= 1

            self.data[self.left] = x

            if self.maxlen != -1 and len(self) > self.maxlen:

                self.pop()

        def pop(self):

            if self.left == self.right:

                raise IndexError('cannot pop from empty deque')

            self.right -= 1

            elem = self.data[self.right]

            del self.data[self.right]

            return elem

        def popleft(self):

            if self.left == self.right:

                raise IndexError('cannot pop from empty deque')

            elem = self.data[self.left]

            del self.data[self.left]

            self.left += 1

            return elem

        def clear(self):

            self.data.clear()

            self.left = self.right = 0

        def extend(self, iterable):

            for elem in iterable:

                self.append(elem)

        def extendleft(self, iterable):

            for elem in iterable:

                self.appendleft(elem)

        def rotate(self, n=1):

            if self:

                n %= len(self)

                for i in xrange(n):

                    self.appendleft(self.pop())

        def __getitem__(self, i):

            if i < 0:

                i += len(self)

            try:

                return self.data[i + self.left]

            except KeyError:

                raise IndexError

        def __setitem__(self, i, value):

            if i < 0:

                i += len(self)

            try:

                self.data[i + self.left] = value

            except KeyError:

                raise IndexError

        def __delitem__(self, i):

            size = len(self)

            if not (-size <= i < size):

                raise IndexError

            data = self.data

            if i < 0:

                i += size

            for j in xrange(self.left + i, self.right - 1):

                data[j] = data[j + 1]

            self.pop()

        def __len__(self):

            return self.right - self.left

        def __cmp__(self, other):

            if type(self) != type(other):

                return cmp(type(self), type(other))

            return cmp(list(self), list(other))

        def __repr__(self, _track=[]):

            if id(self) in _track:

                return '...'

            _track.append(id(self))

            r = 'deque(%r, maxlen=%s)' % (list(self), self.maxlen)

            _track.remove(id(self))

            return r

        def __getstate__(self):

            return (tuple(self),)

        def __setstate__(self, s):

            self.__init__(s[0])

        def __hash__(self):

            raise TypeError

        def __copy__(self):

            return self.__class__(self)

        def __deepcopy__(self, memo={}):

            from copy import deepcopy

            result = self.__class__()

            memo[id(self)] = result

            result.__init__(deepcopy(tuple(self), memo))

            return result

#==============================================================================

# threading.Event

#==============================================================================

if __py_version__ >= (2, 6):

    from threading import Event, Thread

else:

    from threading import _Verbose, Condition, Lock, Thread

    def Event(*args, **kwargs):

        return _Event(*args, **kwargs)

    class _Event(_Verbose):

        # After Tim Peters' event class (without is_posted())

        def __init__(self, verbose=None):

            _Verbose.__init__(self, verbose)

            self.__cond = Condition(Lock())

            self.__flag = False

        def isSet(self):

            return self.__flag

        is_set = isSet

        def set(self):

            self.__cond.acquire()

            try:

                self.__flag = True

                self.__cond.notify_all()

            finally:

                self.__cond.release()

        def clear(self):

            self.__cond.acquire()

            try:

                self.__flag = False

            finally:

                self.__cond.release()

        def wait(self, timeout=None):

            self.__cond.acquire()

            try:

                if not self.__flag:

                    self.__cond.wait(timeout)

            finally:

                self.__cond.release()

        """

        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:

        """

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

        inv = cls._get_or_create_key(key, _prefix, _org_key)

        if inv and inv.cache_active is False:

            return inv

    @classmethod

    def set_invalidate(cls, key):

        """

        Mark this Cache key for invalidation

        :param key:

        """

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

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

        log.debug('marking %s key[s] %s for invalidation' % (len(inv_objs),

                                                             _org_key))

        try:

            for inv_obj in inv_objs:

                if inv_obj:

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

class ChangesetComment(Base, BaseModel):

    __tablename__ = 'changeset_comments'

    __table_args__ = (

        {'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=False)

    line_no = Column('line_no', Unicode(10), 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', Unicode(25000), nullable=False)

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

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

    repo = relationship('Repository')

    @classmethod

    def get_users(cls, revision):

        """

        Returns user associated with this changesetComment. ie those

        who actually commented

        :param cls:

        :param revision:

        """

        return Session.query(User)\

                .filter(cls.revision == revision)\

                .join(ChangesetComment.author).all()

class Notification(Base, BaseModel):

    __tablename__ = 'notifications'

    __table_args__ = (

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

         'mysql_charset': 'utf8'},

    )

    TYPE_CHANGESET_COMMENT = u'cs_comment'

    TYPE_MESSAGE = u'message'

    TYPE_MENTION = u'mention'

    TYPE_REGISTRATION = u'registration'

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

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

    body = Column('body', Unicode(50000), nullable=True)

    created_by = Column("created_by", Integer(), ForeignKey('users.user_id'), nullable=True)

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

    type_ = Column('type', Unicode(256))

    created_by_user = relationship('User')

    notifications_to_users = relationship('UserNotification', lazy='joined',

                                          cascade="all, delete, delete-orphan")

    @property

    def recipients(self):

        return [x.user for x in UserNotification.query()\

                .filter(UserNotification.notification == self).all()]

    @classmethod

    def create(cls, created_by, subject, body, recipients, type_=None):

        if type_ is None:

            type_ = Notification.TYPE_MESSAGE

        notification = cls()

        notification.created_by_user = created_by

        notification.subject = subject

        notification.body = body

        notification.type_ = type_

        notification.created_on = datetime.datetime.now()

        for u in recipients:

            assoc = UserNotification()

            assoc.notification = notification

            u.notifications.append(assoc)

        Session.add(notification)

        return notification

    @property

    def description(self):

        from rhodecode.model.notification import NotificationModel

        return NotificationModel().make_description(self)

class UserNotification(Base, BaseModel):

    __tablename__ = 'user_to_notification'

    __table_args__ = (

        UniqueConstraint('user_id', 'notification_id'),

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

         'mysql_charset': 'utf8'}

    )

    user_id = Column('user_id', Integer(), ForeignKey('users.user_id'), primary_key=True)

    notification_id = Column("notification_id", Integer(), ForeignKey('notifications.notification_id'), primary_key=True)

    read = Column('read', Boolean, default=False)

    sent_on = Column('sent_on', DateTime(timezone=False), nullable=True, unique=None)

    user = relationship('User', lazy="joined")

    notification = relationship('Notification', lazy="joined",

                                order_by=lambda: Notification.created_on.desc(),)

    def mark_as_read(self):

        self.read = True

        Session.add(self)

class DbMigrateVersion(Base, BaseModel):

    __tablename__ = 'db_migrate_version'

    __table_args__ = (

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

         'mysql_charset': 'utf8'},

    )

    repository_id = Column('repository_id', String(250), primary_key=True)

    repository_path = Column('repository_path', Text)

    version = Column('version', Integer)

## this is migration from 1_4_0, but now it's here to overcome a problem of

## attaching a FK to this from 1_3_0 !

class PullRequest(Base, BaseModel):

    __tablename__ = 'pull_requests'

    __table_args__ = (

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

         'mysql_charset': 'utf8'},

    )

    STATUS_NEW = u'new'

    STATUS_OPEN = u'open'

    STATUS_CLOSED = u'closed'

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

    title = Column('title', Unicode(256), nullable=True)

    description = Column('description', UnicodeText(10240), nullable=True)

    status = Column('status', Unicode(256), nullable=False, default=STATUS_NEW)

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

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

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

    _revisions = Column('revisions', UnicodeText(20500))  # 500 revisions max

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

    org_ref = Column('org_ref', Unicode(256), nullable=False)

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

"""

Mercurial libs compatibility

"""

from mercurial import archival, merge as hg_merge, patch, ui

from mercurial.commands import clone, nullid, pull

from mercurial.context import memctx, memfilectx

from mercurial.error import RepoError, RepoLookupError, Abort

from mercurial.hgweb.common import get_contact

from mercurial.localrepo import localrepository

from mercurial.match import match

from mercurial.mdiff import diffopts

from mercurial.node import hex

from mercurial.encoding import tolocal

from mercurial import discovery

from mercurial import localrepo

from mercurial import scmutil

            v.UnicodeString(strip=False, min=6, not_empty=True)

        )

        active = v.StringBoolean(if_missing=False)

        firstname = v.UnicodeString(strip=True, min=1, not_empty=False)

        lastname = v.UnicodeString(strip=True, min=1, not_empty=False)

        email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))

        chained_validators = [v.ValidPasswordsMatch()]

    return _RegisterForm

def PasswordResetForm():

    class _PasswordResetForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        email = All(v.ValidSystemEmail(), v.Email(not_empty=True))

    return _PasswordResetForm

def RepoForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),

             repo_groups=[], landing_revs=[]):

    class _RepoForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),

                        v.SlugifyName())

        clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))

        repo_group = v.OneOf(repo_groups, hideList=True)

        repo_type = v.OneOf(supported_backends)

        description = v.UnicodeString(strip=True, min=1, not_empty=False)

        private = v.StringBoolean(if_missing=False)

        enable_statistics = v.StringBoolean(if_missing=False)

        enable_downloads = v.StringBoolean(if_missing=False)

        enable_locking = v.StringBoolean(if_missing=False)

        landing_rev = v.OneOf(landing_revs, hideList=True)

        if edit:

            #this is repo owner

            user = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())

        chained_validators = [v.ValidCloneUri(),

                              v.ValidRepoName(edit, old_data),

                              v.ValidPerms()]

    return _RepoForm

def RepoForkForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),

                 repo_groups=[], landing_revs=[]):

    class _RepoForkForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),

                        v.SlugifyName())

        repo_group = v.OneOf(repo_groups, hideList=True)

        repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))

        description = v.UnicodeString(strip=True, min=1, not_empty=True)

        private = v.StringBoolean(if_missing=False)

        copy_permissions = v.StringBoolean(if_missing=False)

        update_after_clone = v.StringBoolean(if_missing=False)

        fork_parent_id = v.UnicodeString()

        chained_validators = [v.ValidForkName(edit, old_data)]

        landing_rev = v.OneOf(landing_revs, hideList=True)

    return _RepoForkForm

def RepoSettingsForm(edit=False, old_data={},

                     supported_backends=BACKENDS.keys(), repo_groups=[],

                     landing_revs=[]):

    class _RepoForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),

                        v.SlugifyName())

        description = v.UnicodeString(strip=True, min=1, not_empty=True)

        repo_group = v.OneOf(repo_groups, hideList=True)

        private = v.StringBoolean(if_missing=False)

        landing_rev = v.OneOf(landing_revs, hideList=True)

        chained_validators = [v.ValidRepoName(edit, old_data), v.ValidPerms(),

                              v.ValidSettings()]

    return _RepoForm

def ApplicationSettingsForm():

    class _ApplicationSettingsForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        rhodecode_title = v.UnicodeString(strip=True, min=1, not_empty=True)

        rhodecode_realm = v.UnicodeString(strip=True, min=1, not_empty=True)

        rhodecode_ga_code = v.UnicodeString(strip=True, min=1, not_empty=False)

    return _ApplicationSettingsForm

def ApplicationVisualisationForm():

    class _ApplicationVisualisationForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        rhodecode_show_public_icon = v.StringBoolean(if_missing=False)

        rhodecode_show_private_icon = v.StringBoolean(if_missing=False)

        rhodecode_stylify_metatags = v.StringBoolean(if_missing=False)

    return _ApplicationVisualisationForm

def ApplicationUiSettingsForm():

    class _ApplicationUiSettingsForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = False

        web_push_ssl = v.StringBoolean(if_missing=False)

        paths_root_path = All(

            v.ValidPath(),

            v.UnicodeString(strip=True, min=1, not_empty=True)

        )

        hooks_changegroup_update = v.StringBoolean(if_missing=False)

        hooks_changegroup_repo_size = v.StringBoolean(if_missing=False)

        hooks_changegroup_push_logger = v.StringBoolean(if_missing=False)

        hooks_outgoing_pull_logger = v.StringBoolean(if_missing=False)

        extensions_largefiles = v.StringBoolean(if_missing=False)

        extensions_hgsubversion = v.StringBoolean(if_missing=False)

        extensions_hggit = v.StringBoolean(if_missing=False)

    return _ApplicationUiSettingsForm

def DefaultPermissionsForm(perms_choices, register_choices, create_choices,

                           fork_choices):

    class _DefaultPermissionsForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        overwrite_default = v.StringBoolean(if_missing=False)

        anonymous = v.StringBoolean(if_missing=False)

        default_perm = v.OneOf(perms_choices)

        default_register = v.OneOf(register_choices)

        default_create = v.OneOf(create_choices)

        default_fork = v.OneOf(fork_choices)

    return _DefaultPermissionsForm

def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,

                     tls_kind_choices):

    class _LdapSettingsForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        #pre_validators = [LdapLibValidator]

        ldap_active = v.StringBoolean(if_missing=False)

        ldap_host = v.UnicodeString(strip=True,)

        ldap_port = v.Number(strip=True,)

        ldap_tls_kind = v.OneOf(tls_kind_choices)

        ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)

        ldap_dn_user = v.UnicodeString(strip=True,)

        ldap_dn_pass = v.UnicodeString(strip=True,)

        ldap_base_dn = v.UnicodeString(strip=True,)

        ldap_filter = v.UnicodeString(strip=True,)

        ldap_search_scope = v.OneOf(search_scope_choices)

        ldap_attr_login = All(

            v.AttrLoginValidator(),

            v.UnicodeString(strip=True,)

        )

        ldap_attr_firstname = v.UnicodeString(strip=True,)

        ldap_attr_lastname = v.UnicodeString(strip=True,)

        ldap_attr_email = v.UnicodeString(strip=True,)

    return _LdapSettingsForm

def UserExtraEmailForm():

    class _UserExtraEmailForm(formencode.Schema):

        email = All(v.UniqSystemEmail(), v.Email)

    return _UserExtraEmailForm

def PullRequestForm():

    class _PullRequestForm(formencode.Schema):

        allow_extra_fields = True

        filter_extra_fields = True

        user = v.UnicodeString(strip=True, required=True)

        org_repo = v.UnicodeString(strip=True, required=True)

        org_ref = v.UnicodeString(strip=True, required=True)

        other_repo = v.UnicodeString(strip=True, required=True)

        other_ref = v.UnicodeString(strip=True, required=True)

        revisions = All(v.NotReviewedRevisions()(), v.UniqueList(not_empty=True))

        review_members = v.UniqueList(not_empty=True)

        pullrequest_title = v.UnicodeString(strip=True, required=True, min=3)

        pullrequest_desc = v.UnicodeString(strip=True, required=False)

                                                perm='group.none')

        u1_auth = AuthUser(user_id=self.u1.user_id)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

        # add repo to group

        name = RepoGroup.url_sep().join([self.g1.group_name, 'test_perm'])

        self.test_repo = RepoModel().create_repo(

            repo_name=name,

            repo_type='hg',

            description='',

            repos_group=self.g1,

            owner=self.u1,

        )

        Session().commit()

        u1_auth = AuthUser(user_id=self.u1.user_id)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

        #grant permission for u2 !

        ReposGroupModel().grant_user_permission(repos_group=self.g1,

                                                user=self.u2,

                                                perm='group.read')

        ReposGroupModel().grant_user_permission(repos_group=self.g2,

                                                user=self.u2,

                                                perm='group.read')

        Session().commit()

        self.assertNotEqual(self.u1, self.u2)

        #u1 and anon should have not change perms while u2 should !

        u1_auth = AuthUser(user_id=self.u1.user_id)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

        u2_auth = AuthUser(user_id=self.u2.user_id)

        self.assertEqual(u2_auth.permissions['repositories_groups'],

                 {u'group1': u'group.read', u'group2': u'group.read'})

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                 {u'group1': u'group.none', u'group2': u'group.none'})

    def test_repo_group_user_as_user_group_member(self):

        # create Group1

        self.g1 = _make_group('group1', skip_if_exists=True)

        Session().commit()

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                         {u'group1': u'group.read'})

        # set default permission to none

        ReposGroupModel().grant_user_permission(repos_group=self.g1,

                                                user=self.anon,

                                                perm='group.none')

        # make group

        self.ug1 = UsersGroupModel().create('G1')

        # add user to group

        UsersGroupModel().add_user_to_group(self.ug1, self.u1)

        Session().commit()

        # check if user is in the group

        membrs = [x.user_id for x in UsersGroupModel().get(self.ug1.users_group_id).members]

        self.assertEqual(membrs, [self.u1.user_id])

        # add some user to that group

        # check his permissions

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                         {u'group1': u'group.none'})

        u1_auth = AuthUser(user_id=self.u1.user_id)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                         {u'group1': u'group.none'})

        # grant ug1 read permissions for

        ReposGroupModel().grant_users_group_permission(repos_group=self.g1,

                                                       group_name=self.ug1,

                                                       perm='group.read')

        Session().commit()

        # check if the

        obj = Session().query(UsersGroupRepoGroupToPerm)\

            .filter(UsersGroupRepoGroupToPerm.group == self.g1)\

            .filter(UsersGroupRepoGroupToPerm.users_group == self.ug1)\

            .scalar()

        self.assertEqual(obj.permission.permission_name, 'group.read')

        a1_auth = AuthUser(user_id=self.anon.user_id)

        self.assertEqual(a1_auth.permissions['repositories_groups'],

                         {u'group1': u'group.none'})

        u1_auth = AuthUser(user_id=self.u1.user_id)

        self.assertEqual(u1_auth.permissions['repositories_groups'],

                         {u'group1': u'group.read'})

    def test_inherited_permissions_from_default_on_user_enabled(self):

        user_model = UserModel()

        # enable fork and create on default user

        usr = 'default'

        user_model.revoke_perm(usr, 'hg.create.none')

        user_model.grant_perm(usr, 'hg.create.repository')

        user_model.revoke_perm(usr, 'hg.fork.none')

        user_model.grant_perm(usr, 'hg.fork.repository')

        # make sure inherit flag is turned on

        self.u1.inherit_default_permissions = True

        Session().commit()

        u1_auth = AuthUser(user_id=self.u1.user_id)

        # this user will have inherited permissions from default user

        self.assertEqual(u1_auth.permissions['global'],

                         set(['hg.create.repository', 'hg.fork.repository',

                              'hg.register.manual_activate',

                              'repository.read']))

    def test_inherited_permissions_from_default_on_user_disabled(self):

        user_model = UserModel()

        # disable fork and create on default user

        usr = 'default'

        user_model.revoke_perm(usr, 'hg.create.repository')

        user_model.grant_perm(usr, 'hg.create.none')

        user_model.revoke_perm(usr, 'hg.fork.repository')

        user_model.grant_perm(usr, 'hg.fork.none')

        # make sure inherit flag is turned on

        self.u1.inherit_default_permissions = True

        Session().commit()

        u1_auth = AuthUser(user_id=self.u1.user_id)

        # this user will have inherited permissions from default user

        self.assertEqual(u1_auth.permissions['global'],

                         set(['hg.create.none', 'hg.fork.none',

                              'hg.register.manual_activate',

                              'repository.read']))

    def test_non_inherited_permissions_from_default_on_user_enabled(self):

        user_model = UserModel()

        # enable fork and create on default user

        usr = 'default'

        user_model.revoke_perm(usr, 'hg.create.none')

        user_model.grant_perm(usr, 'hg.create.repository')

        user_model.revoke_perm(usr, 'hg.fork.none')

        user_model.grant_perm(usr, 'hg.fork.repository')

        #disable global perms on specific user

        user_model.revoke_perm(self.u1, 'hg.create.repository')

        user_model.grant_perm(self.u1, 'hg.create.none')

        user_model.revoke_perm(self.u1, 'hg.fork.repository')

        user_model.grant_perm(self.u1, 'hg.fork.none')

        # make sure inherit flag is turned off

        self.u1.inherit_default_permissions = False

        Session().commit()

        u1_auth = AuthUser(user_id=self.u1.user_id)

        # this user will have non inherited permissions from he's

        # explicitly set permissions

        self.assertEqual(u1_auth.permissions['global'],

                         set(['hg.create.none', 'hg.fork.none',

                              'hg.register.manual_activate',

                              'repository.read']))

    def test_non_inherited_permissions_from_default_on_user_disabled(self):

        user_model = UserModel()

        # disable fork and create on default user

        usr = 'default'

        user_model.revoke_perm(usr, 'hg.create.repository')

        user_model.grant_perm(usr, 'hg.create.none')

        user_model.revoke_perm(usr, 'hg.fork.repository')

        user_model.grant_perm(usr, 'hg.fork.none')

        #enable global perms on specific user

        user_model.revoke_perm(self.u1, 'hg.create.none')

        user_model.grant_perm(self.u1, 'hg.create.repository')

        user_model.revoke_perm(self.u1, 'hg.fork.none')

        user_model.grant_perm(self.u1, 'hg.fork.repository')

        # make sure inherit flag is turned off

        self.u1.inherit_default_permissions = False

        Session().commit()

        u1_auth = AuthUser(user_id=self.u1.user_id)

        # this user will have non inherited permissions from he's

        # explicitly set permissions

        self.assertEqual(u1_auth.permissions['global'],

                         set(['hg.create.repository', 'hg.fork.repository',

                              'hg.register.manual_activate',

                              'repository.read']))