Changeset - 79567af5523e
[Not reviewed]
default
0 4 0
Mads Kiilerich (mads) - 5 years ago 2021-01-10 22:10:34
mads@kiilerich.com
Grafted from: 32003970f8c5
hooks: drop UI display of internal hooks - these hooks are just an implementation detail
4 files changed with 1 insertions and 24 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/admin/settings.py
Show inline comments
 
@@ -175,237 +175,236 @@ class SettingsController(base.BaseContro
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_global(self):
 
        c.active = 'global'
 
        if request.POST:
 
            application_form = ApplicationSettingsForm()()
 
            try:
 
                form_result = application_form.to_python(dict(request.POST))
 
            except formencode.Invalid as errors:
 
                return htmlfill.render(
 
                    base.render('admin/settings/settings.html'),
 
                    defaults=errors.value,
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8",
 
                    force_defaults=False)
 

	
 
            try:
 
                for setting in (
 
                    'title',
 
                    'realm',
 
                    'ga_code',
 
                    'captcha_public_key',
 
                    'captcha_private_key',
 
                ):
 
                    db.Setting.create_or_update(setting, form_result[setting])
 

	
 
                meta.Session().commit()
 
                set_app_settings(config)
 
                webutils.flash(_('Updated application settings'), category='success')
 

	
 
            except Exception:
 
                log.error(traceback.format_exc())
 
                webutils.flash(_('Error occurred while updating '
 
                          'application settings'),
 
                          category='error')
 

	
 
            raise HTTPFound(location=url('admin_settings_global'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_visual(self):
 
        c.active = 'visual'
 
        if request.POST:
 
            application_form = ApplicationVisualisationForm()()
 
            try:
 
                form_result = application_form.to_python(dict(request.POST))
 
            except formencode.Invalid as errors:
 
                return htmlfill.render(
 
                    base.render('admin/settings/settings.html'),
 
                    defaults=errors.value,
 
                    errors=errors.error_dict or {},
 
                    prefix_error=False,
 
                    encoding="UTF-8",
 
                    force_defaults=False)
 

	
 
            try:
 
                settings = [
 
                    ('show_public_icon', 'show_public_icon', 'bool'),
 
                    ('show_private_icon', 'show_private_icon', 'bool'),
 
                    ('stylify_metalabels', 'stylify_metalabels', 'bool'),
 
                    ('repository_fields', 'repository_fields', 'bool'),
 
                    ('dashboard_items', 'dashboard_items', 'int'),
 
                    ('admin_grid_items', 'admin_grid_items', 'int'),
 
                    ('show_version', 'show_version', 'bool'),
 
                    ('use_gravatar', 'use_gravatar', 'bool'),
 
                    ('gravatar_url', 'gravatar_url', 'unicode'),
 
                    ('clone_uri_tmpl', 'clone_uri_tmpl', 'unicode'),
 
                    ('clone_ssh_tmpl', 'clone_ssh_tmpl', 'unicode'),
 
                ]
 
                for setting, form_key, type_ in settings:
 
                    db.Setting.create_or_update(setting, form_result[form_key], type_)
 

	
 
                meta.Session().commit()
 
                set_app_settings(config)
 
                webutils.flash(_('Updated visualisation settings'),
 
                        category='success')
 

	
 
            except Exception:
 
                log.error(traceback.format_exc())
 
                webutils.flash(_('Error occurred during updating '
 
                          'visualisation settings'),
 
                        category='error')
 

	
 
            raise HTTPFound(location=url('admin_settings_visual'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_email(self):
 
        c.active = 'email'
 
        if request.POST:
 
            test_email = request.POST.get('test_email')
 
            test_email_subj = 'Kallithea test email'
 
            test_body = ('Kallithea Email test, '
 
                               'Kallithea version: %s' % c.kallithea_version)
 
            if not test_email:
 
                webutils.flash(_('Please enter email address'), category='error')
 
                raise HTTPFound(location=url('admin_settings_email'))
 

	
 
            test_email_txt_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'txt', body=test_body)
 
            test_email_html_body = EmailNotificationModel() \
 
                .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
 
                                'html', body=test_body)
 

	
 
            recipients = [test_email] if test_email else None
 

	
 
            notification.send_email(recipients, test_email_subj,
 
                             test_email_txt_body, test_email_html_body)
 

	
 
            webutils.flash(_('Send email task created'), category='success')
 
            raise HTTPFound(location=url('admin_settings_email'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.ini = kallithea.CONFIG
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_hooks(self):
 
        c.active = 'hooks'
 
        if request.POST:
 
            if c.visual.allow_custom_hooks_settings:
 
                ui_key = request.POST.get('new_hook_ui_key')
 
                ui_value = request.POST.get('new_hook_ui_value')
 

	
 
                hook_id = request.POST.get('hook_id')
 

	
 
                try:
 
                    ui_key = ui_key and ui_key.strip()
 
                    if ui_key in (x.ui_key for x in db.Ui.get_custom_hooks()):
 
                        webutils.flash(_('Hook already exists'), category='error')
 
                    elif ui_key and '.kallithea_' in ui_key:
 
                        webutils.flash(_('Hook names with ".kallithea_" are reserved for internal use. Please use another hook name.'), category='error')
 
                    elif ui_value and ui_key:
 
                        db.Ui.create_or_update_hook(ui_key, ui_value)
 
                        webutils.flash(_('Added new hook'), category='success')
 
                    elif hook_id:
 
                        db.Ui.delete(hook_id)
 
                        meta.Session().commit()
 

	
 
                    # check for edits
 
                    update = False
 
                    _d = request.POST.dict_of_lists()
 
                    for k, v, ov in zip(_d.get('hook_ui_key', []),
 
                                        _d.get('hook_ui_value_new', []),
 
                                        _d.get('hook_ui_value', [])):
 
                        if v != ov:
 
                            db.Ui.create_or_update_hook(k, v)
 
                            update = True
 

	
 
                    if update:
 
                        webutils.flash(_('Updated hooks'), category='success')
 
                    meta.Session().commit()
 
                except Exception:
 
                    log.error(traceback.format_exc())
 
                    webutils.flash(_('Error occurred during hook creation'),
 
                            category='error')
 

	
 
                raise HTTPFound(location=url('admin_settings_hooks'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.hooks = db.Ui.get_builtin_hooks()
 
        c.custom_hooks = db.Ui.get_custom_hooks()
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_search(self):
 
        c.active = 'search'
 
        if request.POST:
 
            repo_location = self._get_hg_ui_settings()['paths_root_path']
 
            full_index = request.POST.get('full_index', False)
 
            kallithea.lib.indexers.daemon.whoosh_index(repo_location, full_index)
 
            webutils.flash(_('Whoosh reindex task scheduled'), category='success')
 
            raise HTTPFound(location=url('admin_settings_search'))
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def settings_system(self):
 
        c.active = 'system'
 

	
 
        defaults = db.Setting.get_app_settings()
 
        defaults.update(self._get_hg_ui_settings())
 

	
 
        c.ini = kallithea.CONFIG
 
        server_info = db.Setting.get_server_info()
 
        for key, val in server_info.items():
 
            setattr(c, key, val)
 

	
 
        return htmlfill.render(
 
            base.render('admin/settings/settings.html'),
 
            defaults=defaults,
 
            encoding="UTF-8",
 
            force_defaults=False)
kallithea/model/db.py
Show inline comments
 
@@ -165,392 +165,384 @@ class BaseDbModel(object):
 
        return '<DB:%s>' % (self.__class__.__name__)
 

	
 

	
 
_table_args_default_dict = {'extend_existing': True,
 
                            'mysql_engine': 'InnoDB',
 
                            'sqlite_autoincrement': True,
 
                           }
 

	
 
class Setting(meta.Base, BaseDbModel):
 
    __tablename__ = 'settings'
 
    __table_args__ = (
 
        _table_args_default_dict,
 
    )
 

	
 
    SETTINGS_TYPES = {
 
        'str': safe_bytes,
 
        'int': safe_int,
 
        'unicode': safe_str,
 
        'bool': asbool,
 
        'list': functools.partial(aslist, sep=',')
 
    }
 

	
 
    app_settings_id = Column(Integer(), primary_key=True)
 
    app_settings_name = Column(String(255), nullable=False, unique=True)
 
    _app_settings_value = Column("app_settings_value", Unicode(4096), nullable=False)
 
    _app_settings_type = Column("app_settings_type", String(255), nullable=True) # FIXME: not nullable?
 

	
 
    def __init__(self, key='', val='', type='unicode'):
 
        self.app_settings_name = key
 
        self.app_settings_value = val
 
        self.app_settings_type = type
 

	
 
    @validates('_app_settings_value')
 
    def validate_settings_value(self, key, val):
 
        assert isinstance(val, str)
 
        return val
 

	
 
    @hybrid_property
 
    def app_settings_value(self):
 
        v = self._app_settings_value
 
        _type = self.app_settings_type
 
        converter = self.SETTINGS_TYPES.get(_type) or self.SETTINGS_TYPES['unicode']
 
        return converter(v)
 

	
 
    @app_settings_value.setter
 
    def app_settings_value(self, val):
 
        """
 
        Setter that will always make sure we use str in app_settings_value
 
        """
 
        self._app_settings_value = safe_str(val)
 

	
 
    @hybrid_property
 
    def app_settings_type(self):
 
        return self._app_settings_type
 

	
 
    @app_settings_type.setter
 
    def app_settings_type(self, val):
 
        if val not in self.SETTINGS_TYPES:
 
            raise Exception('type must be one of %s got %s'
 
                            % (list(self.SETTINGS_TYPES), val))
 
        self._app_settings_type = val
 

	
 
    def __repr__(self):
 
        return "<%s %s.%s=%r>" % (
 
            self.__class__.__name__,
 
            self.app_settings_name, self.app_settings_type, self.app_settings_value
 
        )
 

	
 
    @classmethod
 
    def get_by_name(cls, key):
 
        return cls.query() \
 
            .filter(cls.app_settings_name == key).scalar()
 

	
 
    @classmethod
 
    def get_by_name_or_create(cls, key, val='', type='unicode'):
 
        res = cls.get_by_name(key)
 
        if res is None:
 
            res = cls(key, val, type)
 
        return res
 

	
 
    @classmethod
 
    def create_or_update(cls, key, val=None, type=None):
 
        """
 
        Creates or updates Kallithea setting. If updates are triggered, it will only
 
        update parameters that are explicitly set. 'None' values will be skipped.
 

	
 
        :param key:
 
        :param val:
 
        :param type:
 
        :return:
 
        """
 
        res = cls.get_by_name(key)
 
        if res is None:
 
            # new setting
 
            val = val if val is not None else ''
 
            type = type if type is not None else 'unicode'
 
            res = cls(key, val, type)
 
            meta.Session().add(res)
 
        else:
 
            if val is not None:
 
                # update if set
 
                res.app_settings_value = val
 
            if type is not None:
 
                # update if set
 
                res.app_settings_type = type
 
        return res
 

	
 
    @classmethod
 
    def get_app_settings(cls):
 

	
 
        ret = cls.query()
 
        if ret is None:
 
            raise Exception('Could not get application settings !')
 
        settings = {}
 
        for each in ret:
 
            settings[each.app_settings_name] = \
 
                each.app_settings_value
 

	
 
        return settings
 

	
 
    @classmethod
 
    def get_auth_settings(cls):
 
        ret = cls.query() \
 
                .filter(cls.app_settings_name.startswith('auth_')).all()
 
        fd = {}
 
        for row in ret:
 
            fd[row.app_settings_name] = row.app_settings_value
 
        return fd
 

	
 
    @classmethod
 
    def get_default_repo_settings(cls, strip_prefix=False):
 
        ret = cls.query() \
 
                .filter(cls.app_settings_name.startswith('default_')).all()
 
        fd = {}
 
        for row in ret:
 
            key = row.app_settings_name
 
            if strip_prefix:
 
                key = remove_prefix(key, prefix='default_')
 
            fd.update({key: row.app_settings_value})
 

	
 
        return fd
 

	
 
    @classmethod
 
    def get_server_info(cls):
 
        import platform
 

	
 
        import pkg_resources
 

	
 
        mods = [(p.project_name, p.version) for p in pkg_resources.working_set]
 
        info = {
 
            'modules': sorted(mods, key=lambda k: k[0].lower()),
 
            'py_version': platform.python_version(),
 
            'platform': platform.platform(),
 
            'kallithea_version': kallithea.__version__,
 
            'git_version': str(check_git_version()),
 
            'git_path': kallithea.CONFIG.get('git_path')
 
        }
 
        return info
 

	
 

	
 
class Ui(meta.Base, BaseDbModel):
 
    __tablename__ = 'ui'
 
    __table_args__ = (
 
        Index('ui_ui_section_ui_key_idx', 'ui_section', 'ui_key'),
 
        UniqueConstraint('ui_section', 'ui_key'),
 
        _table_args_default_dict,
 
    )
 

	
 
    HOOK_UPDATE = 'changegroup.kallithea_update'
 
    HOOK_REPO_SIZE = 'changegroup.kallithea_repo_size'
 

	
 
    ui_id = Column(Integer(), primary_key=True)
 
    ui_section = Column(String(255), nullable=False)
 
    ui_key = Column(String(255), nullable=False)
 
    ui_value = Column(String(255), nullable=True) # FIXME: not nullable?
 
    ui_active = Column(Boolean(), nullable=False, default=True)
 

	
 
    @classmethod
 
    def get_by_key(cls, section, key):
 
        """ Return specified Ui object, or None if not found. """
 
        return cls.query().filter_by(ui_section=section, ui_key=key).scalar()
 

	
 
    @classmethod
 
    def get_or_create(cls, section, key):
 
        """ Return specified Ui object, creating it if necessary. """
 
        setting = cls.get_by_key(section, key)
 
        if setting is None:
 
            setting = cls(ui_section=section, ui_key=key)
 
            meta.Session().add(setting)
 
        return setting
 

	
 
    @classmethod
 
    def get_builtin_hooks(cls):
 
        q = cls.query()
 
        q = q.filter(cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))
 
        q = q.filter(cls.ui_section == 'hooks')
 
        q = q.order_by(cls.ui_section, cls.ui_key)
 
        return q.all()
 

	
 
    @classmethod
 
    def get_custom_hooks(cls):
 
        q = cls.query()
 
        q = q.filter(~cls.ui_key.in_([cls.HOOK_UPDATE, cls.HOOK_REPO_SIZE]))
 
        q = q.filter(cls.ui_section == 'hooks')
 
        q = q.order_by(cls.ui_section, cls.ui_key)
 
        return q.all()
 

	
 
    @classmethod
 
    def get_repos_location(cls):
 
        return cls.get_by_key('paths', '/').ui_value
 

	
 
    @classmethod
 
    def create_or_update_hook(cls, key, val):
 
        new_ui = cls.get_or_create('hooks', key)
 
        new_ui.ui_active = True
 
        new_ui.ui_value = val
 

	
 
    def __repr__(self):
 
        return '<%s %s.%s=%r>' % (
 
            self.__class__.__name__,
 
            self.ui_section, self.ui_key, self.ui_value)
 

	
 

	
 
class User(meta.Base, BaseDbModel):
 
    __tablename__ = 'users'
 
    __table_args__ = (
 
        Index('u_username_idx', 'username'),
 
        Index('u_email_idx', 'email'),
 
        _table_args_default_dict,
 
    )
 

	
 
    DEFAULT_USER_NAME = 'default'
 
    DEFAULT_GRAVATAR_URL = 'https://secure.gravatar.com/avatar/{md5email}?d=identicon&s={size}'
 
    # The name of the default auth type in extern_type, 'internal' lives in auth_internal.py
 
    DEFAULT_AUTH_TYPE = 'internal'
 

	
 
    user_id = Column(Integer(), primary_key=True)
 
    username = Column(String(255), nullable=False, unique=True)
 
    password = Column(String(255), nullable=False)
 
    active = Column(Boolean(), nullable=False, default=True)
 
    admin = Column(Boolean(), nullable=False, default=False)
 
    name = Column("firstname", Unicode(255), nullable=False)
 
    lastname = Column(Unicode(255), nullable=False)
 
    _email = Column("email", String(255), nullable=True, unique=True) # FIXME: not nullable?
 
    last_login = Column(DateTime(timezone=False), nullable=True)
 
    extern_type = Column(String(255), nullable=True) # FIXME: not nullable?
 
    extern_name = Column(String(255), nullable=True) # FIXME: not nullable?
 
    api_key = Column(String(255), nullable=False)
 
    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)
 
    _user_data = Column("user_data", LargeBinary(), nullable=True)  # JSON data # FIXME: not nullable?
 

	
 
    user_log = relationship('UserLog')
 
    user_perms = relationship('UserToPerm', primaryjoin="User.user_id==UserToPerm.user_id", cascade='all')
 

	
 
    repositories = relationship('Repository')
 
    repo_groups = relationship('RepoGroup')
 
    user_groups = relationship('UserGroup')
 
    user_followers = relationship('UserFollowing', primaryjoin='UserFollowing.follows_user_id==User.user_id', cascade='all')
 
    followings = relationship('UserFollowing', primaryjoin='UserFollowing.user_id==User.user_id', cascade='all')
 

	
 
    repo_to_perm = relationship('UserRepoToPerm', primaryjoin='UserRepoToPerm.user_id==User.user_id', cascade='all')
 
    repo_group_to_perm = relationship('UserRepoGroupToPerm', primaryjoin='UserRepoGroupToPerm.user_id==User.user_id', cascade='all')
 

	
 
    group_member = relationship('UserGroupMember', cascade='all')
 

	
 
    # comments created by this user
 
    user_comments = relationship('ChangesetComment', cascade='all')
 
    # extra emails for this user
 
    user_emails = relationship('UserEmailMap', cascade='all')
 
    # extra API keys
 
    user_api_keys = relationship('UserApiKeys', cascade='all')
 
    ssh_keys = relationship('UserSshKeys', cascade='all')
 

	
 
    @hybrid_property
 
    def email(self):
 
        return self._email
 

	
 
    @email.setter
 
    def email(self, val):
 
        self._email = val.lower() if val else None
 

	
 
    @property
 
    def firstname(self):
 
        # alias for future
 
        return self.name
 

	
 
    @property
 
    def emails(self):
 
        other = UserEmailMap.query().filter(UserEmailMap.user == self).all()
 
        return [self.email] + [x.email for x in other]
 

	
 
    @property
 
    def api_keys(self):
 
        other = UserApiKeys.query().filter(UserApiKeys.user == self).all()
 
        return [self.api_key] + [x.api_key for x in other]
 

	
 
    @property
 
    def ip_addresses(self):
 
        ret = UserIpMap.query().filter(UserIpMap.user == self).all()
 
        return [x.ip_addr for x in ret]
 

	
 
    @property
 
    def full_name(self):
 
        return '%s %s' % (self.firstname, self.lastname)
 

	
 
    @property
 
    def full_name_or_username(self):
 
        """
 
        Show full name.
 
        If full name is not set, fall back to username.
 
        """
 
        return ('%s %s' % (self.firstname, self.lastname)
 
                if (self.firstname and self.lastname) else self.username)
 

	
 
    @property
 
    def full_name_and_username(self):
 
        """
 
        Show full name and username as 'Firstname Lastname (username)'.
 
        If full name is not set, fall back to username.
 
        """
 
        return ('%s %s (%s)' % (self.firstname, self.lastname, self.username)
 
                if (self.firstname and self.lastname) else self.username)
 

	
 
    @property
 
    def full_contact(self):
 
        return '%s %s <%s>' % (self.firstname, self.lastname, self.email)
 

	
 
    @property
 
    def short_contact(self):
 
        return '%s %s' % (self.firstname, self.lastname)
 

	
 
    @property
 
    def is_admin(self):
 
        return self.admin
 

	
 
    @hybrid_property
 
    def is_default_user(self):
 
        return self.username == User.DEFAULT_USER_NAME
 

	
 
    @hybrid_property
 
    def user_data(self):
 
        if not self._user_data:
 
            return {}
 

	
 
        try:
 
            return ext_json.loads(self._user_data)
 
        except TypeError:
 
            return {}
 

	
 
    @user_data.setter
 
    def user_data(self, val):
 
        try:
 
            self._user_data = ascii_bytes(ext_json.dumps(val))
 
        except Exception:
 
            log.error(traceback.format_exc())
 

	
 
    def __repr__(self):
 
        return "<%s %s: %r>" % (self.__class__.__name__, self.user_id, self.username)
 

	
 
    @classmethod
 
    def guess_instance(cls, value):
 
        return super(User, cls).guess_instance(value, User.get_by_username)
 

	
 
    @classmethod
 
    def get_or_404(cls, id_, allow_default=True):
 
        '''
 
        Overridden version of BaseDbModel.get_or_404, with an extra check on
 
        the default user.
 
        '''
 
        user = super(User, cls).get_or_404(id_)
 
        if not allow_default and user.is_default_user:
 
            raise DefaultUserException()
 
        return user
 

	
 
    @classmethod
 
    def get_by_username_or_email(cls, username_or_email, case_insensitive=True):
 
        """
 
        For anything that looks like an email address, look up by the email address (matching
 
        case insensitively).
 
        For anything else, try to look up by the user name.
 

	
 
        This assumes no normal username can have '@' symbol.
 
        """
 
        if '@' in username_or_email:
 
            return User.get_by_email(username_or_email)
 
        else:
 
            return User.get_by_username(username_or_email, case_insensitive=case_insensitive)
 

	
 
    @classmethod
 
    def get_by_username(cls, username, case_insensitive=False):
 
        if case_insensitive:
 
            q = cls.query().filter(sqlalchemy.func.lower(cls.username) == sqlalchemy.func.lower(username))
kallithea/templates/admin/settings/settings_hooks.html
Show inline comments
 
<div class="form">
 
      <div class="form-group">
 
      <h4>${_('Built-in Mercurial Hooks (Read-Only)')}</h4>
 
      % for hook in c.hooks:
 
        <% input_id = hook.ui_key.replace('.', '_') %>
 
            <label class="control-label" for="${input_id}" title="${hook.ui_key}">${hook.ui_key}</label>
 
            <div>
 
              ${h.text(hook.ui_key,hook.ui_value,id=input_id,size=60,readonly="readonly",class_='form-control')}
 
            </div>
 
      % endfor
 
      </div>
 
</div>
 

	
 
% if c.visual.allow_custom_hooks_settings:
 
${h.form(url('admin_settings_hooks'), method='post')}
 
<div class="form">
 
      <div class="form-group">
 
        <h4>${_('Custom Hooks')}</h4>
 
        <span class="help-block">${_('Hooks can be used to trigger actions on certain events such as push / pull. They can trigger Python functions or external applications.')}</span>
 
        %for hook in c.custom_hooks:
 
            <div class="form-group form-inline" id="${'id%s' % hook.ui_id }">
 
                <% input_id = hook.ui_key.replace('.', '_') %>
 
                    <label class="control-label" for="${input_id}" title="${hook.ui_key}">${hook.ui_key}</label>
 
                    <div>
 
                        ${h.hidden('hook_ui_key',hook.ui_key,id='hook_ui_key_'+input_id)}
 
                        ${h.hidden('hook_ui_value',hook.ui_value,id='hook_ui_value_'+input_id)}
 
                        ${h.text('hook_ui_value_new',hook.ui_value,id=input_id,size=50,class_='form-control')}
 
                        <button type="button" class="btn btn-default btn-xs"
 
                            onclick="delete_hook(${hook.ui_id},'${'id%s' % hook.ui_id }')">
 
                            <i class="icon-trashcan"></i>
 
                            ${_('Delete')}
 
                        </button>
 
                    </div>
 
            </div>
 
        %endfor
 

	
 
        <div class="form-group form-inline">
 
            <label>
 
                ${h.text('new_hook_ui_key',size=15,class_='form-control')}
 
            </label>
 
            <div>
 
                ${h.text('new_hook_ui_value',size=50,class_='form-control')}
 
            </div>
 
        </div>
 
        <div class="form-group">
 
            <div class="buttons">
 
                ${h.submit('save',_('Save'),class_="btn btn-default")}
 
            </div>
 
        </div>
 
      </div>
 
</div>
 
${h.end_form()}
 
% endif
 

	
 
<script>
 
'use strict';
 
function delete_hook(hook_id, field_id) {
 
    var sUrl = ${h.js(h.url('admin_settings_hooks_delete'))};
 
    function success() {
 
            $('#' + field_id).remove();
 
        }
 
    function failure() {
 
            alert(${h.js(_('Failed to remove hook'))});
 
        }
 
    var postData = {'hook_id': hook_id};
 
    ajaxPOST(sUrl, postData, success, failure);
 
}
 
</script>
kallithea/tests/functional/test_admin_settings.py
Show inline comments
 
# -*- coding: utf-8 -*-
 

	
 
from kallithea.model import db
 
from kallithea.tests import base
 
from kallithea.tests.fixture import Fixture
 

	
 

	
 
fixture = Fixture()
 

	
 

	
 
class TestAdminSettingsController(base.TestController):
 

	
 
    def test_index_main(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings'))
 

	
 
    def test_index_mapping(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_mapping'))
 

	
 
    def test_index_global(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_global'))
 

	
 
    def test_index_visual(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_visual'))
 

	
 
    def test_index_email(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_email'))
 

	
 
    def test_index_hooks(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_hooks'))
 
        response.mustcontain(no='.kallithea_')
 

	
 
    def test_create_custom_hook(self):
 
        self.log_user()
 
        response = self.app.post(base.url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_1',
 
                                            new_hook_ui_value='cd %s' % base.TESTS_TMP_PATH,
 
                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        self.checkSessionFlash(response, 'Added new hook')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('cd %s' % base.TESTS_TMP_PATH)
 

	
 
        # test_edit_custom_hook
 
        response = self.app.post(base.url('admin_settings_hooks'),
 
                                params=dict(hook_ui_key='test_hooks_1',
 
                                            hook_ui_value='old_value_of_hook_1',
 
                                            hook_ui_value_new='new_value_of_hook_1',
 
                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('new_value_of_hook_1')
 

	
 
        # test_add_existing_custom_hook
 
        response = self.app.post(base.url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_1',
 
                                            new_hook_ui_value='attempted_new_value',
 
                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        self.checkSessionFlash(response, 'Hook already exists')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_1')
 
        response.mustcontain('new_value_of_hook_1')
 

	
 
    def test_create_custom_hook_delete(self):
 
        self.log_user()
 
        response = self.app.post(base.url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='test_hooks_2',
 
                                            new_hook_ui_value='cd %s2' % base.TESTS_TMP_PATH,
 
                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        self.checkSessionFlash(response, 'Added new hook')
 
        response = response.follow()
 
        response.mustcontain('test_hooks_2')
 
        response.mustcontain('cd %s2' % base.TESTS_TMP_PATH)
 

	
 
        hook_id = db.Ui.get_by_key('hooks', 'test_hooks_2').ui_id
 
        ## delete
 
        self.app.post(base.url('admin_settings_hooks'),
 
                        params=dict(hook_id=hook_id, _session_csrf_secret_token=self.session_csrf_secret_token()))
 
        response = self.app.get(base.url('admin_settings_hooks'))
 
        response.mustcontain(no=['test_hooks_2'])
 
        response.mustcontain(no=['cd %s2' % base.TESTS_TMP_PATH])
 

	
 
    def test_add_existing_builtin_hook(self):
 
        self.log_user()
 
        response = self.app.post(base.url('admin_settings_hooks'),
 
                                params=dict(new_hook_ui_key='changegroup.kallithea_update',
 
                                            new_hook_ui_value='attempted_new_value',
 
                                            _session_csrf_secret_token=self.session_csrf_secret_token()))
 

	
 
        self.checkSessionFlash(response, 'reserved for internal use')
 
        response = response.follow()
 
        response.mustcontain('changegroup.kallithea_update')
 

	
 
    def test_index_search(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_search'))
 

	
 
    def test_index_system(self):
 
        self.log_user()
 
        response = self.app.get(base.url('admin_settings_system'))
 

	
 
    def test_ga_code_active(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = 'ga-test-123456789'
 
        response = self.app.post(base.url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 

	
 
        assert db.Setting.get_app_settings()['ga_code'] == new_ga_code
 

	
 
        response = response.follow()
 
        response.mustcontain("""_gaq.push(['_setAccount', '%s']);""" % new_ga_code)
 

	
 
    def test_ga_code_inactive(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(base.url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert db.Setting.get_app_settings()['ga_code'] == new_ga_code
 

	
 
        response = response.follow()
 
        response.mustcontain(no=["_gaq.push(['_setAccount', '%s']);" % new_ga_code])
 

	
 
    def test_captcha_activate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(base.url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='1234567890',
 
                                 captcha_public_key='1234567890',
 
                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert db.Setting.get_app_settings()['captcha_private_key'] == '1234567890'
 

	
 
        response = self.app.get(base.url('register'))
 
        response.mustcontain('captcha')
 

	
 
    def test_captcha_deactivate(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        old_realm = 'Kallithea authentication'
 
        new_ga_code = ''
 
        response = self.app.post(base.url('admin_settings_global'),
 
                        params=dict(title=old_title,
 
                                 realm=old_realm,
 
                                 ga_code=new_ga_code,
 
                                 captcha_private_key='',
 
                                 captcha_public_key='1234567890',
 
                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
 
                                 ))
 

	
 
        self.checkSessionFlash(response, 'Updated application settings')
 
        assert db.Setting.get_app_settings()['captcha_private_key'] == ''
 

	
 
        response = self.app.get(base.url('register'))
 
        response.mustcontain(no=['captcha'])
 

	
 
    def test_title_change(self):
 
        self.log_user()
 
        old_title = 'Kallithea'
 
        new_title = old_title + '_changed'
 
        old_realm = 'Kallithea authentication'
 

	
 
        for new_title in ['Changed', 'Żółwik', old_title]:
 
            response = self.app.post(base.url('admin_settings_global'),
 
                        params=dict(title=new_title,
 
                                 realm=old_realm,
 
                                 ga_code='',
 
                                 captcha_private_key='',
 
                                 captcha_public_key='',
 
                                 _session_csrf_secret_token=self.session_csrf_secret_token(),
 
                                ))
 

	
 
            self.checkSessionFlash(response, 'Updated application settings')
 
            assert db.Setting.get_app_settings()['title'] == new_title
 

	
 
            response = response.follow()
 
            response.mustcontain("""<span class="branding">%s</span>""" % new_title)
0 comments (0 inline, 0 general)