'hg.create.repository'),

            'create_user_group_perm': ug_model.has_perm(c.user_group,

                                                        'hg.usergroup.create.true'),

            'fork_repo_perm': ug_model.has_perm(c.user_group,

                                                'hg.fork.repository'),

        })

        return htmlfill.render(

            render('admin/user_groups/user_group_edit.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False

        )

    @HasUserGroupPermissionAnyDecorator('usergroup.admin')

    def update_default_perms(self, id):

        user_group = UserGroup.get_or_404(id)

        try:

            form = CustomDefaultPermissionsForm()()

            form_result = form.to_python(request.POST)

            inherit_perms = form_result['inherit_default_permissions']

            user_group.inherit_default_permissions = inherit_perms

            usergroup_model = UserGroupModel()

            defs = UserGroupToPerm.query() \

                .filter(UserGroupToPerm.users_group == user_group) \

                .all()

            for ug in defs:

                Session().delete(ug)

            if form_result['create_repo_perm']:

                usergroup_model.grant_perm(id, 'hg.create.repository')

            else:

                usergroup_model.grant_perm(id, 'hg.create.none')

            if form_result['create_user_group_perm']:

                usergroup_model.grant_perm(id, 'hg.usergroup.create.true')

            else:

                usergroup_model.grant_perm(id, 'hg.usergroup.create.false')

            if form_result['fork_repo_perm']:

                usergroup_model.grant_perm(id, 'hg.fork.repository')

            else:

                usergroup_model.grant_perm(id, 'hg.fork.none')

            h.flash(_("Updated permissions"), category='success')

            Session().commit()

        except Exception:

        umodel = UserModel()

        defaults = c.user.get_dict()

        defaults.update({

            'create_repo_perm': umodel.has_perm(c.user, 'hg.create.repository'),

            'create_user_group_perm': umodel.has_perm(c.user,

                                                      'hg.usergroup.create.true'),

            'fork_repo_perm': umodel.has_perm(c.user, 'hg.fork.repository'),

        })

        return htmlfill.render(

            render('admin/users/user_edit.html'),

            defaults=defaults,

            encoding="UTF-8",

            force_defaults=False)

    def update_perms(self, id):

        user = self._get_user_or_raise_if_default(id)

        try:

            form = CustomDefaultPermissionsForm()()

            form_result = form.to_python(request.POST)

            inherit_perms = form_result['inherit_default_permissions']

            user.inherit_default_permissions = inherit_perms

            user_model = UserModel()

            defs = UserToPerm.query() \

                .filter(UserToPerm.user == user) \

                .all()

            for ug in defs:

                Session().delete(ug)

            if form_result['create_repo_perm']:

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

            else:

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

            if form_result['create_user_group_perm']:

                user_model.grant_perm(id, 'hg.usergroup.create.true')

            else:

                user_model.grant_perm(id, 'hg.usergroup.create.false')

            if form_result['fork_repo_perm']:

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

            else:

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

            h.flash(_("Updated permissions"), category='success')

            Session().commit()

        except Exception:

            log.error(traceback.format_exc())

        self.create_default_options()

        log.info('created ui config')

    def create_user(self, username, password, email='', admin=False):

        log.info('creating user %s', username)

        UserModel().create_or_update(username, password, email,

                                     firstname=u'Kallithea', lastname=u'Admin',

                                     active=True, admin=admin,

                                     extern_type=User.DEFAULT_AUTH_TYPE)

    def create_default_user(self):

        log.info('creating default user')

        # create default user for handling default permissions.

        user = UserModel().create_or_update(username=User.DEFAULT_USER,

                                            password=str(uuid.uuid1())[:20],

                                            email='anonymous@kallithea-scm.org',

                                            firstname=u'Anonymous',

                                            lastname=u'User')

        # based on configuration options activate/deactivate this user which

        # controls anonymous access

        if self.cli_args.get('public_access') is False:

            log.info('Public access disabled')

            user.active = False

            Session().commit()

    def create_permissions(self):

        """

        Creates all permissions defined in the system

        """

        # module.(access|create|change|delete)_[name]

        # module.(none|read|write|admin)

        log.info('creating permissions')

        PermissionModel(self.sa).create_permissions()

    def populate_default_permissions(self):

        """

        Populate default permissions. It will create only the default

        permissions that are missing, and not alter already defined ones

        """

        log.info('creating default user permissions')

        PermissionModel(self.sa).create_default_permissions(user=User.DEFAULT_USER)

    @classmethod

    def get_from_cs_author(cls, author):

        """

        Tries to get User objects out of commit author string

        :param author:

        """

        from kallithea.lib.helpers import email, author_name

        # Valid email in the attribute passed, see if they're in the system

        _email = email(author)

        if _email:

            user = cls.get_by_email(_email)

            if user is not None:

                return user

        # Maybe we can match by username?

        _author = author_name(author)

        user = cls.get_by_username(_author, case_insensitive=True)

        if user is not None:

            return user

    def update_lastlogin(self):

        """Update user lastlogin"""

        self.last_login = datetime.datetime.now()

        log.debug('updated user %s lastlogin', self.username)

    @classmethod

    def get_first_admin(cls):

        user = User.query().filter(User.admin == True).first()

        if user is None:

            raise Exception('Missing administrative account!')

        return user

    @classmethod

    def get_default_user(cls, cache=False):

        user = User.get_by_username(User.DEFAULT_USER, cache=cache)

        if user is None:

            raise Exception('Missing default account!')

        return user

    def get_api_data(self, details=False):

        """

        Common function for generating user related data for API

        """

        user = self

        data = dict(

            user_id=user.user_id,

            username=user.username,

            owner=repo.user.username,

            fork_of=repo.fork.repo_name if repo.fork else None,

            enable_statistics=repo.enable_statistics,

            enable_locking=repo.enable_locking,

            enable_downloads=repo.enable_downloads,

            last_changeset=repo.changeset_cache,

            locked_by=User.get(self.locked[0]).get_api_data() \

                if self.locked[0] else None,

            locked_date=time_to_datetime(self.locked[1]) \

                if self.locked[1] else None

        )

        rc_config = Setting.get_app_settings()

        repository_fields = str2bool(rc_config.get('repository_fields'))

        if repository_fields:

            for f in self.extra_fields:

                data[f.field_key_prefixed] = f.field_value

        return data

    @classmethod

    def lock(cls, repo, user_id, lock_time=None):

        if lock_time is not None:

            lock_time = time.time()

        repo.locked = [user_id, lock_time]

        Session().commit()

    @classmethod

    def unlock(cls, repo):

        repo.locked = None

        Session().commit()

    @classmethod

    def getlock(cls, repo):

        return repo.locked

    @property

    def last_db_change(self):

        return self.updated_on

    @property

    def clone_uri_hidden(self):

        clone_uri = self.clone_uri

        if clone_uri:

            import urlobject

            url_obj = urlobject.URLObject(self.clone_uri)

            if url_obj.password:

                clone_uri = url_obj.with_password('*****')

        return clone_uri

    def clone_url(self, **override):

        import kallithea.lib.helpers as h

        qualified_home_url = h.canonical_url('home')

            del override['with_id']

        if 'uri_tmpl' in override:

            uri_tmpl = override['uri_tmpl']

            del override['uri_tmpl']

        # we didn't override our tmpl from **overrides

        if not uri_tmpl:

            uri_tmpl = self.DEFAULT_CLONE_URI

            try:

                from pylons import tmpl_context as c

                uri_tmpl = c.clone_uri_tmpl

            except AttributeError:

                # in any case if we call this outside of request context,

                # ie, not having tmpl_context set up

                pass

        return get_clone_url(uri_tmpl=uri_tmpl,

                             qualified_home_url=qualified_home_url,

                             repo_name=self.repo_name,

                             repo_id=self.repo_id, **override)

    def set_state(self, state):

        self.repo_state = state

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

    # SCM PROPERTIES

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

    def get_changeset(self, rev=None):

        return get_changeset_safe(self.scm_instance, rev)

    def get_landing_changeset(self):

        """

        Returns landing changeset, or if that doesn't exist returns the tip

        """

        _rev_type, _rev = self.landing_rev

        cs = self.get_changeset(_rev)

        if isinstance(cs, EmptyChangeset):

            return self.get_changeset()

        return cs

    def update_changeset_cache(self, cs_cache=None):

        """

        Update cache of last changeset for repository, keys should be::

            short_id

            raw_id

            revision

            message

            date

            author

        :param cs_cache:

        """

        from kallithea.lib.vcs.backends.base import BaseChangeset

        if cs_cache is None:

            cs_cache = EmptyChangeset()

            # use no-cache version here

            scm_repo = self.scm_instance_no_cache()

            if scm_repo:

                cs_cache = scm_repo.get_changeset()

        if isinstance(cs_cache, BaseChangeset):

            cs_cache = cs_cache.__json__()

        if (not self.changeset_cache or cs_cache['raw_id'] != self.changeset_cache['raw_id']):

            _default = datetime.datetime.fromtimestamp(0)

            last_change = cs_cache.get('date') or _default

            log.debug('updated repo %s with new cs cache %s',

                      self.repo_name, cs_cache)

            self.updated_on = last_change

            self.changeset_cache = cs_cache

            Session().commit()

        else:

            log.debug('changeset_cache for %s already up to date with %s',

                      self.repo_name, cs_cache['raw_id'])

    @property

    def tip(self):

        return self.get_changeset('tip')

    @property

    def author(self):

        return self.tip.author

    @property

    def last_change(self):

        return self.scm_instance.last_change

    def get_comments(self, revisions=None):

        """

        Returns comments for this repository grouped by revisions

        :param revisions: filter query by revisions only

        """

        cmts = ChangesetComment.query() \

        for inv_obj in inv_objs:

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

                      inv_obj, safe_str(repo_name))

            Session().delete(inv_obj)

        Session().commit()

    @classmethod

    def test_and_set_valid(cls, repo_name, kind, valid_cache_keys=None):

        """

        Mark this cache key as active and currently cached.

        Return True if the existing cache registration still was valid.

        Return False to indicate that it had been invalidated and caches should be refreshed.

        """

        key = (repo_name + '_' + kind) if kind else repo_name

        cache_key = cls._get_cache_key(key)

        if valid_cache_keys and cache_key in valid_cache_keys:

            return True

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

        if inv_obj is None:

            inv_obj = cls(cache_key, repo_name)

        elif inv_obj.cache_active:

            return True

        inv_obj.cache_active = True

        try:

            Session().commit()

        except sqlalchemy.exc.IntegrityError:

            log.error('commit of CacheInvalidation failed - retrying')

            Session().rollback()

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

            if inv_obj is None:

                log.error('failed to create CacheInvalidation entry')

                # TODO: fail badly?

            # else: TOCTOU - another thread added the key at the same time; no further action required

        return False

    @classmethod

    def get_valid_cache_keys(cls):

        """

        Return opaque object with information of which caches still are valid

        and can be used without checking for invalidation.

        """

        return set(inv_obj.cache_key for inv_obj in cls.query().filter(cls.cache_active).all())

class ChangesetComment(Base, BaseModel):

    __tablename__ = 'changeset_comments'

    __table_args__ = (

    @property

    def description(self):

        from kallithea.model.notification import NotificationModel

        return NotificationModel().make_description(self)

class UserNotification(Base, BaseModel):

    __tablename__ = 'user_to_notification'

    __table_args__ = (

        UniqueConstraint('user_id', 'notification_id'),

        _table_args_default_dict,

    )

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

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

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

    sent_on = Column(DateTime(timezone=False), nullable=True) # FIXME: not nullable?

    user = relationship('User')

    notification = relationship('Notification')

    def mark_as_read(self):

        self.read = True

class Gist(Base, BaseModel):

    __tablename__ = 'gists'

    __table_args__ = (

        Index('g_gist_access_id_idx', 'gist_access_id'),

        Index('g_created_on_idx', 'created_on'),

        _table_args_default_dict,

    )

    GIST_PUBLIC = u'public'

    GIST_PRIVATE = u'private'

    DEFAULT_FILENAME = u'gistfile1.txt'

    gist_id = Column(Integer(), primary_key=True)

    gist_access_id = Column(Unicode(250), nullable=False)

    gist_description = Column(UnicodeText(), nullable=False)

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

    gist_expires = Column(Float(53), nullable=False)

    gist_type = Column(Unicode(128), nullable=False)

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

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

    owner = relationship('User')

        q = UserNotification.query() \

            .filter(UserNotification.user == user) \

            .join((Notification, UserNotification.notification_id ==

                                 Notification.notification_id)) \

            .options(joinedload('notification')) \

            .options(subqueryload('notification.created_by_user')) \

            .order_by(Notification.created_on.desc())

        if filter_:

            q = q.filter(Notification.type_.in_(filter_))

        return q

    def mark_read(self, user, notification):

        try:

            notification = Notification.guess_instance(notification)

            user = self._get_user(user)

            if notification and user:

                obj = UserNotification.query() \

                        .filter(UserNotification.user == user) \

                        .filter(UserNotification.notification

                                == notification) \

                        .one()

                obj.read = True

                return True

        except Exception:

            log.error(traceback.format_exc())

            raise

    def mark_all_read_for_user(self, user, filter_=None):

        user = self._get_user(user)

        q = UserNotification.query() \

            .filter(UserNotification.user == user) \

            .filter(UserNotification.read == False) \

            .join((Notification, UserNotification.notification_id ==

                                 Notification.notification_id))

        if filter_:

            q = q.filter(Notification.type_.in_(filter_))

        # this is a little inefficient but sqlalchemy doesn't support

        # update on joined tables :(

            obj.read = True

    def get_unread_cnt_for_user(self, user):

        user = self._get_user(user)

        return UserNotification.query() \

                .filter(UserNotification.read == False) \

                .filter(UserNotification.user == user).count()

    def get_unread_for_user(self, user):

        user = self._get_user(user)

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

                .filter(UserNotification.read == False) \

                .filter(UserNotification.user == user).all()]

    def get_user_notification(self, user, notification):

        user = self._get_user(user)

        notification = Notification.guess_instance(notification)

        return UserNotification.query() \

            .filter(UserNotification.notification == notification) \

            .filter(UserNotification.user == user).scalar()

    def make_description(self, notification, show_age=True):

        """

        Creates a human readable description based on properties

        new_reviewer_users = set(self._get_valid_reviewers(reviewers_ids))

        to_add = new_reviewer_users - current_reviewer_users

        to_remove = current_reviewer_users - new_reviewer_users

        if not to_add and not to_remove:

            return # all done

        log.debug("Adding %s reviewers", to_add)

        self.__add_reviewers(user, pull_request, to_add, set())

        log.debug("Removing %s reviewers", to_remove)

        for prr in current_reviewers:

            if prr.user in to_remove:

                Session().delete(prr)

    def delete(self, pull_request):

        pull_request = PullRequest.guess_instance(pull_request)

        Session().delete(pull_request)

    def close_pull_request(self, pull_request):

        pull_request = PullRequest.guess_instance(pull_request)

        pull_request.status = PullRequest.STATUS_CLOSED

        pull_request.updated_on = datetime.datetime.now()

        if token_age < 0:

            log.debug('timestamp is from the future')

            return False

        if token_age > UserModel.password_reset_token_lifetime:

            log.debug('password reset token expired')

            return False

        expected_token = self.get_reset_password_token(user,

                                                       timestamp,

                                                       h.authentication_token())

        log.debug('computed password reset token: %s', expected_token)

        log.debug('received password reset token: %s', token)

        return expected_token == token

    def reset_password(self, user_email, new_passwd):

        from kallithea.lib.celerylib import tasks

        from kallithea.lib import auth

        user = User.get_by_email(user_email)

        if user is not None:

            if not self.can_change_password(user):

                raise Exception('trying to change password for external user')

            user.password = auth.get_crypt_password(new_passwd)

            Session().commit()

            log.info('change password for %s', user_email)

        if new_passwd is None:

            raise Exception('unable to set new password')

        tasks.send_email([user_email],

                 _('Password reset notification'),

                 _('The password to your account %s has been changed using password reset form.') % (user.username,))

        log.info('send password reset mail to %s', user_email)

        return True

    def has_perm(self, user, perm):

        perm = self._get_perm(perm)

        user = self._get_user(user)

        return UserToPerm.query().filter(UserToPerm.user == user) \

            .filter(UserToPerm.permission == perm).scalar() is not None

    def grant_perm(self, user, perm):

        """

        Grant user global permissions

        :param user:

    def test_api_get_user_without_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user')

        response = api_call(self, params)

        usr = User.get_by_username(self.TEST_USER_LOGIN)

        ret = usr.get_api_data()

        ret['permissions'] = AuthUser(dbuser=usr).permissions

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_user_with_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user',

                                  userid=self.TEST_USER_LOGIN)

        response = api_call(self, params)

        expected = 'userid is not the same as your user'

        self._compare_error(id_, expected, given=response.body)

    def test_api_pull(self):

        repo_name = u'test_pull'

        r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        r.clone_uri = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)

        Session.commit()

        id_, params = _build_data(self.apikey, 'pull',

                                  repoid=repo_name,)

        response = api_call(self, params)

        expected = {'msg': 'Pulled from `%s`' % repo_name,

                    'repository': repo_name}

        self._compare_ok(id_, expected, given=response.body)

        fixture.destroy_repo(repo_name)

    def test_api_pull_error(self):

        id_, params = _build_data(self.apikey, 'pull',

                                  repoid=self.REPO, )

        response = api_call(self, params)

        expected = 'Unable to pull changes from `%s`' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_rescan_repos(self):

        id_, params = _build_data(self.apikey, 'rescan_repos')

        response = api_call(self, params)

class Fixture(object):

    def __init__(self):

        pass

    def anon_access(self, status):

        """

        Context manager for controlling anonymous access.

        Anon access will be set and committed, but restored again when exiting the block.

        Usage:

        fixture = Fixture()

        with fixture.anon_access(False):

            stuff

        """

        class context(object):

            def __enter__(self):

                anon = User.get_default_user()

                self._before = anon.active

                anon.active = status

                Session().commit()

                invalidate_all_caches()

            def __exit__(self, exc_type, exc_val, exc_tb):

                anon = User.get_default_user()

                anon.active = self._before

                Session().commit()

        return context()

    def _get_repo_create_params(self, **custom):

        defs = dict(

            repo_name=None,

            repo_type='hg',

            clone_uri='',

            repo_group=u'-1',

            repo_description=u'DESC',

            repo_private=False,

            repo_landing_rev='rev:tip',

            repo_copy_permissions=False,

            repo_state=Repository.STATE_CREATED,

        )

        defs.update(custom)

        if 'repo_name_full' not in custom:

            defs.update({'repo_name_full': defs['repo_name']})

        # fix the repo name if passed as repo_name_full

        if defs['repo_name']:

            defs['repo_name'] = defs['repo_name'].split('/')[-1]

                                         'filename': 'foo',

                                         'public': 'public',

                                         '_authentication_token': self.authentication_token()},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: foo')

        response.mustcontain('gist test')

        response.mustcontain('<div class="btn btn-mini btn-success disabled">Public Gist</div>')

    def test_create_with_path_with_dirs(self):

        self.log_user()

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

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': '/home/foo',

                                         'public': 'public',

                                         '_authentication_token': self.authentication_token()},

                                 status=200)

        response.mustcontain('Filename cannot be inside a directory')

    def test_access_expired_gist(self):

        self.log_user()

        gist = _create_gist('never-see-me')

        gist.gist_expires = 0  # 1970

        Session().commit()

        response = self.app.get(url('gist', gist_id=gist.gist_access_id), status=404)

    def test_create_private(self):

        self.log_user()

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

                                 params={'lifetime': -1,

                                         'content': 'private gist test',

                                         'filename': 'private-foo',

                                         'private': 'private',

                                         '_authentication_token': self.authentication_token()},

                                 status=302)

        response = response.follow()

        response.mustcontain('added file: private-foo<')

        response.mustcontain('private gist test')

        response.mustcontain('<div class="btn btn-mini btn-warning disabled">Private Gist</div>')

    def test_create_with_description(self):

        self.log_user()

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

                                 params={'lifetime': -1,

                                         'content': 'gist test',

                                         'filename': 'foo-desc',

        assert Repository.get_by_repo_name(self.REPO).private == True

        #now the repo default permission should be None

        perm = _get_permission_for_user(user='default', repo=self.REPO)

        assert len(perm), 1

        assert perm[0].permission.permission_name == 'repository.none'

        response = self.app.post(url('update_repo', repo_name=self.REPO),

                        fixture._get_repo_create_params(repo_private=False,

                                                repo_name=self.REPO,

                                                repo_type=self.REPO_TYPE,

                                                user=TEST_USER_ADMIN_LOGIN,

                                                _authentication_token=self.authentication_token()))

        self.checkSessionFlash(response,

                               msg='Repository %s updated successfully' % (self.REPO))

        assert Repository.get_by_repo_name(self.REPO).private == False

        #we turn off private now the repo default permission should stay None

        perm = _get_permission_for_user(user='default', repo=self.REPO)

        assert len(perm), 1

        assert perm[0].permission.permission_name == 'repository.none'

        #update this permission back

        perm[0].permission = Permission.get_by_key('repository.read')

        Session().commit()

    def test_set_repo_fork_has_no_self_id(self):

        self.log_user()

        repo = Repository.get_by_repo_name(self.REPO)

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

        opt = """<option value="%s">%s</option>""" % (repo.repo_id, self.REPO)

        response.mustcontain(no=[opt])

    def test_set_fork_of_other_repo(self):

        self.log_user()

        other_repo = u'other_%s' % self.REPO_TYPE

        fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)

        repo = Repository.get_by_repo_name(self.REPO)

        repo2 = Repository.get_by_repo_name(other_repo)

        response = self.app.post(url('edit_repo_advanced_fork', repo_name=self.REPO),

                                params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))

        repo = Repository.get_by_repo_name(self.REPO)

        repo2 = Repository.get_by_repo_name(other_repo)

        self.checkSessionFlash(response,

            'Marked repository %s as fork of %s' % (repo.repo_name, repo2.repo_name))

        assert repo.fork == repo2

        response = response.follow()

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

import os

import posixpath

import mimetypes

from kallithea.tests.base import *

from kallithea.model.db import Repository

from kallithea.model.meta import Session

from kallithea.tests.fixture import Fixture

fixture = Fixture()

ARCHIVE_SPECS = {

    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),

    '.tar.gz': ('application/x-gzip', 'tgz', ''),

    '.zip': ('application/zip', 'zip', ''),

}

HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')

GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')

def _set_downloads(repo_name, set_to):

    repo = Repository.get_by_repo_name(repo_name)

    repo.enable_downloads = set_to