kallithea Changeset - 67e5b90801aa

Changeset - 67e5b90801aa

Parent rev.

Child rev.

[Not reviewed]

default

0 32 0

Mads Kiilerich (mads) - 5 years ago 2020-11-01 04:59:46
mads@kiilerich.com

lib: move webhelpers2 and friends to webutils

Gives less of the unfortunate use of helpers - especially in low level libs.

16 files changed:

kallithea/controllers/admin/auth_settings.py

kallithea/controllers/admin/defaults.py

kallithea/controllers/admin/gists.py

kallithea/controllers/admin/my_account.py

kallithea/controllers/admin/permissions.py

kallithea/controllers/admin/repo_groups.py

kallithea/controllers/admin/repos.py

kallithea/controllers/admin/settings.py

kallithea/controllers/admin/user_groups.py

kallithea/controllers/admin/users.py

kallithea/controllers/changelog.py

kallithea/controllers/changeset.py

kallithea/controllers/compare.py

kallithea/controllers/files.py

kallithea/controllers/forks.py

kallithea/controllers/login.py

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)

kallithea/controllers/admin/auth_settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.auth_settings
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 pluggable authentication controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 26, 2010
 :author: akesterson
 """
 import logging
 import traceback
 import formencode.htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.lib import auth_modules
 from kallithea.lib import helpers as h
 from kallithea.lib import auth_modules, webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import AuthSettingsForm
 log = logging.getLogger(__name__)
 class AuthSettingsController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(AuthSettingsController, self)._before(*args, **kwargs)
     def __load_defaults(self):
         c.available_plugins = [
             'kallithea.lib.auth_modules.auth_internal',
             'kallithea.lib.auth_modules.auth_container',
             'kallithea.lib.auth_modules.auth_ldap',
             'kallithea.lib.auth_modules.auth_crowd',
             'kallithea.lib.auth_modules.auth_pam'
+        ]
         self.enabled_plugins = auth_modules.get_auth_plugins()
         c.enabled_plugin_names = [plugin.__class__.__module__ for plugin in self.enabled_plugins]
     def __render(self, defaults, errors):
         c.defaults = {}
         c.plugin_settings = {}
         c.plugin_shortnames = {}
         for plugin in self.enabled_plugins:
             module = plugin.__class__.__module__
             c.plugin_shortnames[module] = plugin.name
             c.plugin_settings[module] = plugin.plugin_settings()
             for v in c.plugin_settings[module]:
                 fullname = "auth_%s_%s" % (plugin.name, v["name"])
                 if "default" in v:
                     c.defaults[fullname] = v["default"]
                 # Current values will be the default on the form, if there are any
                 setting = db.Setting.get_by_name(fullname)
                 if setting is not None:
                     c.defaults[fullname] = setting.app_settings_value
         if defaults:
             c.defaults.update(defaults)
         # we want to show , separated list of enabled plugins
         c.defaults['auth_plugins'] = ','.join(c.enabled_plugin_names)
         log.debug('defaults: %s', defaults)
         return formencode.htmlfill.render(
             render('admin/auth/auth_settings.html'),
             defaults=c.defaults,
             errors=errors,
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     def index(self):
         self.__load_defaults()
         return self.__render(defaults=None, errors=None)
     def auth_settings(self):
         """POST create and store auth settings"""
         self.__load_defaults()
         log.debug("POST Result: %s", dict(request.POST))
         # First, parse only the plugin list (not the plugin settings).
         _auth_plugins_validator = AuthSettingsForm([]).fields['auth_plugins']
         try:
             new_enabled_plugins = _auth_plugins_validator.to_python(request.POST.get('auth_plugins'))
         except formencode.Invalid:
             # User provided an invalid plugin list. Just fall back to
             # the list of currently enabled plugins. (We'll re-validate
             # and show an error message to the user, below.)
             pass
         else:
             # Hide plugins that the user has asked to be disabled, but
             # do not show plugins that the user has asked to be enabled
             # (yet), since that'll cause validation errors and/or wrong
             # settings being applied (e.g. checkboxes being cleared),
             # since the plugin settings will not be in the POST data.
             c.enabled_plugin_names = [p for p in c.enabled_plugin_names if p in new_enabled_plugins]
         # Next, parse everything including plugin settings.
         _form = AuthSettingsForm(c.enabled_plugin_names)()
         try:
             form_result = _form.to_python(dict(request.POST))
             for k, v in form_result.items():
                 if k == 'auth_plugins':
                     # we want to store it comma separated inside our settings
                     v = ','.join(v)
                 log.debug("%s = %s", k, str(v))
                 setting = db.Setting.create_or_update(k, v)
             meta.Session().commit()
-            h.flash(_('Auth settings updated successfully'),
+            webutils.flash(_('Auth settings updated successfully'),
                        category='success')
         except formencode.Invalid as errors:
             log.error(traceback.format_exc())
             e = errors.error_dict or {}
             return self.__render(
                 defaults=errors.value,
                 errors=e,
+            )
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('error occurred during update of auth settings'),
+            webutils.flash(_('error occurred during update of auth settings'),
                     category='error')
         raise HTTPFound(location=url('auth_home'))

kallithea/controllers/admin/defaults.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.defaults
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 default settings controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
-from kallithea.lib import helpers as h
+from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import DefaultsForm
 log = logging.getLogger(__name__)
 class DefaultsController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(DefaultsController, self)._before(*args, **kwargs)
     def index(self, format='html'):
         defaults = db.Setting.get_default_repo_settings()
         return htmlfill.render(
             render('admin/defaults/defaults.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     def update(self, id):
         _form = DefaultsForm()()
         try:
             form_result = _form.to_python(dict(request.POST))
             for k, v in form_result.items():
                 setting = db.Setting.create_or_update(k, v)
             meta.Session().commit()
-            h.flash(_('Default settings updated successfully'),
+            webutils.flash(_('Default settings updated successfully'),
                     category='success')
         except formencode.Invalid as errors:
             defaults = errors.value
             return htmlfill.render(
                 render('admin/defaults/defaults.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during update of defaults'),
+            webutils.flash(_('Error occurred during update of defaults'),
                     category='error')
         raise HTTPFound(location=url('defaults'))

kallithea/controllers/admin/gists.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.gists
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 gist controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 9, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode.htmlfill
 from sqlalchemy.sql.expression import or_
 from tg import request, response
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPForbidden, HTTPFound, HTTPNotFound
 from kallithea.lib import auth
 from kallithea.lib import helpers as h
 from kallithea.lib import auth, webutils
 from kallithea.lib.auth import LoginRequired
 from kallithea.lib.base import BaseController, jsonify, render
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int, safe_str, time_to_datetime
 from kallithea.lib.vcs.exceptions import NodeNotChangedError, VCSError
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import GistForm
 from kallithea.model.gist import GistModel
 log = logging.getLogger(__name__)
 class GistsController(BaseController):
     def __load_defaults(self, extra_values=None):
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         if extra_values:
             c.lifetime_values.append(extra_values)
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
     @LoginRequired(allow_default_user=True)
     def index(self):
         not_default_user = not request.authuser.is_default_user
         c.show_private = request.GET.get('private') and not_default_user
         c.show_public = request.GET.get('public') and not_default_user
         url_params = {}
         if c.show_public:
             url_params['public'] = 1
         elif c.show_private:
             url_params['private'] = 1
         gists = db.Gist().query() \
             .filter_by(is_expired=False) \
             .order_by(db.Gist.created_on.desc())
         # MY private
         if c.show_private and not c.show_public:
             gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PRIVATE) \
                              .filter(db.Gist.owner_id == request.authuser.user_id)
         # MY public
         elif c.show_public and not c.show_private:
             gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PUBLIC) \
                              .filter(db.Gist.owner_id == request.authuser.user_id)
         # MY public+private
         elif c.show_private and c.show_public:
             gists = gists.filter(or_(db.Gist.gist_type == db.Gist.GIST_PUBLIC,
                                      db.Gist.gist_type == db.Gist.GIST_PRIVATE)) \
                              .filter(db.Gist.owner_id == request.authuser.user_id)
         # default show ALL public gists
         if not c.show_public and not c.show_private:
             gists = gists.filter(db.Gist.gist_type == db.Gist.GIST_PUBLIC)
         c.gists = gists
         p = safe_int(request.GET.get('page'), 1)
         c.gists_pager = Page(c.gists, page=p, items_per_page=10,
                              **url_params)
         return render('admin/gists/index.html')
     @LoginRequired()
     def create(self):
         self.__load_defaults()
         gist_form = GistForm([x[0] for x in c.lifetime_values])()
         try:
             form_result = gist_form.to_python(dict(request.POST))
             # TODO: multiple files support, from the form
             filename = form_result['filename'] or db.Gist.DEFAULT_FILENAME
             nodes = {
                 filename: {
                     'content': form_result['content'],
                     'lexer': form_result['mimetype']  # None is autodetect
+                }
+            }
             _public = form_result['public']
             gist_type = db.Gist.GIST_PUBLIC if _public else db.Gist.GIST_PRIVATE
             gist = GistModel().create(
                 description=form_result['description'],
                 owner=request.authuser.user_id,
                 ip_addr=request.ip_addr,
                 gist_mapping=nodes,
                 gist_type=gist_type,
                 lifetime=form_result['lifetime']
+            )
             meta.Session().commit()
             new_gist_id = gist.gist_access_id
         except formencode.Invalid as errors:
             defaults = errors.value
             return formencode.htmlfill.render(
                 render('admin/gists/new.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception as e:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during gist creation'), category='error')
+            webutils.flash(_('Error occurred during gist creation'), category='error')
             raise HTTPFound(location=url('new_gist'))
         raise HTTPFound(location=url('gist', gist_id=new_gist_id))
     @LoginRequired()
     def new(self, format='html'):
         self.__load_defaults()
         return render('admin/gists/new.html')
     @LoginRequired()
     def delete(self, gist_id):
         gist = GistModel().get_gist(gist_id)
         owner = gist.owner_id == request.authuser.user_id
         if auth.HasPermissionAny('hg.admin')() or owner:
             GistModel().delete(gist)
             meta.Session().commit()
-            h.flash(_('Deleted gist %s') % gist.gist_access_id, category='success')
+            webutils.flash(_('Deleted gist %s') % gist.gist_access_id, category='success')
         else:
             raise HTTPForbidden()
         raise HTTPFound(location=url('gists'))
     @LoginRequired(allow_default_user=True)
     def show(self, gist_id, revision='tip', format='html', f_path=None):
         c.gist = db.Gist.get_or_404(gist_id)
         if c.gist.is_expired:
             log.error('Gist expired at %s',
                       time_to_datetime(c.gist.gist_expires))
             raise HTTPNotFound()
         try:
             c.file_changeset, c.files = GistModel().get_gist_files(gist_id,
                                                             revision=revision)
         except VCSError:
             log.error(traceback.format_exc())
             raise HTTPNotFound()
         if format == 'raw':
             content = '\n\n'.join(
                 safe_str(f.content)
                 for f in c.files if (f_path is None or f.path == f_path)
+            )
             response.content_type = 'text/plain'
             return content
         return render('admin/gists/show.html')
     @LoginRequired()
     def edit(self, gist_id, format='html'):
         c.gist = db.Gist.get_or_404(gist_id)
         if c.gist.is_expired:
             log.error('Gist expired at %s',
                       time_to_datetime(c.gist.gist_expires))
             raise HTTPNotFound()
         try:
             c.file_changeset, c.files = GistModel().get_gist_files(gist_id)
         except VCSError:
             log.error(traceback.format_exc())
             raise HTTPNotFound()
         self.__load_defaults(extra_values=('0', _('Unmodified')))
         rendered = render('admin/gists/edit.html')
         if request.POST:
             rpost = request.POST
             nodes = {}
             for org_filename, filename, mimetype, content in zip(
                                                     rpost.getall('org_files'),
                                                     rpost.getall('files'),
                                                     rpost.getall('mimetypes'),
                                                     rpost.getall('contents')):
                 nodes[org_filename] = {
                     'org_filename': org_filename,
                     'filename': filename,
                     'content': content,
                     'lexer': mimetype,
+                }
             try:
                 GistModel().update(
                     gist=c.gist,
                     description=rpost['description'],
                     owner=c.gist.owner, # FIXME: request.authuser.user_id ?
                     ip_addr=request.ip_addr,
                     gist_mapping=nodes,
                     gist_type=c.gist.gist_type,
                     lifetime=rpost['lifetime']
+                )
                 meta.Session().commit()
-                h.flash(_('Successfully updated gist content'), category='success')
+                webutils.flash(_('Successfully updated gist content'), category='success')
             except NodeNotChangedError:
                 # raised if nothing was changed in repo itself. We anyway then
                 # store only DB stuff for gist
                 meta.Session().commit()
-                h.flash(_('Successfully updated gist data'), category='success')
+                webutils.flash(_('Successfully updated gist data'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during update of gist %s') % gist_id,
+                webutils.flash(_('Error occurred during update of gist %s') % gist_id,
                         category='error')
             raise HTTPFound(location=url('gist', gist_id=gist_id))
         return rendered
     @LoginRequired()
     @jsonify
     def check_revision(self, gist_id):
         c.gist = db.Gist.get_or_404(gist_id)
         last_rev = c.gist.scm_instance.get_changeset()
         success = True
         revision = request.POST.get('revision')
         # TODO: maybe move this to model ?
         if revision != last_rev.raw_id:
             log.error('Last revision %s is different than submitted %s',
                       revision, last_rev)
             # our gist has newer version than we
             success = False
         return {'success': success}

kallithea/controllers/admin/my_account.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.my_account
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 my account controller for Kallithea admin
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: August 20, 2013
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.lib import auth_modules
 from kallithea.lib import helpers as h
 from kallithea.lib import auth_modules, webutils
 from kallithea.lib.auth import AuthUser, LoginRequired
 from kallithea.lib.base import BaseController, IfSshEnabled, render
 from kallithea.lib.utils2 import generate_api_key, safe_int
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.forms import PasswordChangeForm, UserForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.ssh_key import SshKeyModel, SshKeyModelException
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class MyAccountController(BaseController):
     @LoginRequired()
     def _before(self, *args, **kwargs):
         super(MyAccountController, self)._before(*args, **kwargs)
     def __load_data(self):
         c.user = db.User.get(request.authuser.user_id)
         if c.user.is_default_user:
-            h.flash(_("You can't edit this user since it's"
+            webutils.flash(_("You can't edit this user since it's"
                       " crucial for entire application"), category='warning')
             raise HTTPFound(location=url('users'))
     def _load_my_repos_data(self, watched=False):
         if watched:
             admin = False
             repos_list = meta.Session().query(db.Repository) \
                          .join(db.UserFollowing) \
                          .filter(db.UserFollowing.user_id ==
                                  request.authuser.user_id).all()
         else:
             admin = True
             repos_list = meta.Session().query(db.Repository) \
                          .filter(db.Repository.owner_id ==
                                  request.authuser.user_id).all()
         return RepoModel().get_repos_as_dict(repos_list, admin=admin)
     def my_account(self):
         c.active = 'profile'
         self.__load_data()
         c.perm_user = AuthUser(user_id=request.authuser.user_id)
         managed_fields = auth_modules.get_managed_fields(c.user)
         def_user_perms = AuthUser(dbuser=db.User.get_default_user()).global_permissions
         if 'hg.register.none' in def_user_perms:
             managed_fields.extend(['username', 'firstname', 'lastname', 'email'])
         c.readonly = lambda n: 'readonly' if n in managed_fields else None
         defaults = c.user.get_dict()
         update = False
         if request.POST:
             _form = UserForm(edit=True,
                              old_data={'user_id': request.authuser.user_id,
                                        'email': request.authuser.email})()
             form_result = {}
             try:
                 post_data = dict(request.POST)
                 post_data['new_password'] = ''
                 post_data['password_confirmation'] = ''
                 form_result = _form.to_python(post_data)
                 # skip updating those attrs for my account
                 skip_attrs = ['admin', 'active', 'extern_type', 'extern_name',
                               'new_password', 'password_confirmation',
                              ] + managed_fields
                 UserModel().update(request.authuser.user_id, form_result,
                                    skip_attrs=skip_attrs)
-                h.flash(_('Your account was updated successfully'),
+                webutils.flash(_('Your account was updated successfully'),
                         category='success')
                 meta.Session().commit()
                 update = True
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/my_account/my_account.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during update of user %s')
+                webutils.flash(_('Error occurred during update of user %s')
                         % form_result.get('username'), category='error')
         if update:
             raise HTTPFound(location='my_account')
         return htmlfill.render(
             render('admin/my_account/my_account.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def my_account_password(self):
         c.active = 'password'
         self.__load_data()
         managed_fields = auth_modules.get_managed_fields(c.user)
         c.can_change_password = 'password' not in managed_fields
         if request.POST and c.can_change_password:
             _form = PasswordChangeForm(request.authuser.username)()
             try:
                 form_result = _form.to_python(request.POST)
                 UserModel().update(request.authuser.user_id, form_result)
                 meta.Session().commit()
-                h.flash(_("Successfully updated password"), category='success')
+                webutils.flash(_("Successfully updated password"), category='success')
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/my_account/my_account.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during update of user password'),
+                webutils.flash(_('Error occurred during update of user password'),
                         category='error')
         return render('admin/my_account/my_account.html')
     def my_account_repos(self):
         c.active = 'repos'
         self.__load_data()
         # data used to render the grid
         c.data = self._load_my_repos_data()
         return render('admin/my_account/my_account.html')
     def my_account_watched(self):
         c.active = 'watched'
         self.__load_data()
         # data used to render the grid
         c.data = self._load_my_repos_data(watched=True)
         return render('admin/my_account/my_account.html')
     def my_account_perms(self):
         c.active = 'perms'
         self.__load_data()
         c.perm_user = AuthUser(user_id=request.authuser.user_id)
         return render('admin/my_account/my_account.html')
     def my_account_emails(self):
         c.active = 'emails'
         self.__load_data()
         c.user_email_map = db.UserEmailMap.query() \
             .filter(db.UserEmailMap.user == c.user).all()
         return render('admin/my_account/my_account.html')
     def my_account_emails_add(self):
         email = request.POST.get('new_email')
         try:
             UserModel().add_extra_email(request.authuser.user_id, email)
             meta.Session().commit()
-            h.flash(_("Added email %s to user") % email, category='success')
+            webutils.flash(_("Added email %s to user") % email, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['email']
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during email saving'),
+            webutils.flash(_('An error occurred during email saving'),
                     category='error')
         raise HTTPFound(location=url('my_account_emails'))
     def my_account_emails_delete(self):
         email_id = request.POST.get('del_email_id')
         user_model = UserModel()
         user_model.delete_extra_email(request.authuser.user_id, email_id)
         meta.Session().commit()
-        h.flash(_("Removed email from user"), category='success')
+        webutils.flash(_("Removed email from user"), category='success')
         raise HTTPFound(location=url('my_account_emails'))
     def my_account_api_keys(self):
         c.active = 'api_keys'
         self.__load_data()
         show_expired = True
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
         c.user_api_keys = ApiKeyModel().get_api_keys(request.authuser.user_id,
                                                      show_expired=show_expired)
         return render('admin/my_account/my_account.html')
     def my_account_api_keys_add(self):
         lifetime = safe_int(request.POST.get('lifetime'), -1)
         description = request.POST.get('description')
         ApiKeyModel().create(request.authuser.user_id, description, lifetime)
         meta.Session().commit()
-        h.flash(_("API key successfully created"), category='success')
+        webutils.flash(_("API key successfully created"), category='success')
         raise HTTPFound(location=url('my_account_api_keys'))
     def my_account_api_keys_delete(self):
         api_key = request.POST.get('del_api_key')
         if request.POST.get('del_api_key_builtin'):
             user = db.User.get(request.authuser.user_id)
             user.api_key = generate_api_key()
             meta.Session().commit()
-            h.flash(_("API key successfully reset"), category='success')
+            webutils.flash(_("API key successfully reset"), category='success')
         elif api_key:
             ApiKeyModel().delete(api_key, request.authuser.user_id)
             meta.Session().commit()
-            h.flash(_("API key successfully deleted"), category='success')
+            webutils.flash(_("API key successfully deleted"), category='success')
         raise HTTPFound(location=url('my_account_api_keys'))
     @IfSshEnabled
     def my_account_ssh_keys(self):
         c.active = 'ssh_keys'
         self.__load_data()
         c.user_ssh_keys = SshKeyModel().get_ssh_keys(request.authuser.user_id)
         return render('admin/my_account/my_account.html')
     @IfSshEnabled
     def my_account_ssh_keys_add(self):
         description = request.POST.get('description')
         public_key = request.POST.get('public_key')
         try:
             new_ssh_key = SshKeyModel().create(request.authuser.user_id,
                                                description, public_key)
             meta.Session().commit()
             SshKeyModel().write_authorized_keys()
-            h.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
+            webutils.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
         except SshKeyModelException as e:
-            h.flash(e.args[0], category='error')
+            webutils.flash(e.args[0], category='error')
         raise HTTPFound(location=url('my_account_ssh_keys'))
     @IfSshEnabled
     def my_account_ssh_keys_delete(self):
         fingerprint = request.POST.get('del_public_key_fingerprint')
         try:
             SshKeyModel().delete(fingerprint, request.authuser.user_id)
             meta.Session().commit()
             SshKeyModel().write_authorized_keys()
-            h.flash(_("SSH key successfully deleted"), category='success')
+            webutils.flash(_("SSH key successfully deleted"), category='success')
         except SshKeyModelException as e:
-            h.flash(e.args[0], category='error')
+            webutils.flash(e.args[0], category='error')
         raise HTTPFound(location=url('my_account_ssh_keys'))

kallithea/controllers/admin/permissions.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.permissions
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 permissions controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 27, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
-from kallithea.lib import helpers as h
+from kallithea.lib import webutils
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import DefaultPermissionsForm
 from kallithea.model.permission import PermissionModel
 log = logging.getLogger(__name__)
 class PermissionsController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(PermissionsController, self)._before(*args, **kwargs)
     def __load_data(self):
         # Permissions for the Default user on new repositories
         c.repo_perms_choices = [('repository.none', _('None'),),
                                    ('repository.read', _('Read'),),
                                    ('repository.write', _('Write'),),
                                    ('repository.admin', _('Admin'),)]
         # Permissions for the Default user on new repository groups
         c.group_perms_choices = [('group.none', _('None'),),
                                  ('group.read', _('Read'),),
                                  ('group.write', _('Write'),),
                                  ('group.admin', _('Admin'),)]
         # Permissions for the Default user on new user groups
         c.user_group_perms_choices = [('usergroup.none', _('None'),),
                                       ('usergroup.read', _('Read'),),
                                       ('usergroup.write', _('Write'),),
                                       ('usergroup.admin', _('Admin'),)]
         # Registration - allow new Users to create an account
         c.register_choices = [
             ('hg.register.none',
                 _('Disabled')),
             ('hg.register.manual_activate',
                 _('Allowed with manual account activation')),
             ('hg.register.auto_activate',
                 _('Allowed with automatic account activation')), ]
         # External auth account activation
         c.extern_activate_choices = [
             ('hg.extern_activate.manual', _('Manual activation of external account')),
             ('hg.extern_activate.auto', _('Automatic activation of external account')),
+        ]
         # Top level repository creation
         c.repo_create_choices = [('hg.create.none', _('Disabled')),
                                  ('hg.create.repository', _('Enabled'))]
         # User group creation
         c.user_group_create_choices = [('hg.usergroup.create.false', _('Disabled')),
                                        ('hg.usergroup.create.true', _('Enabled'))]
         # Repository forking:
         c.fork_choices = [('hg.fork.none', _('Disabled')),
                           ('hg.fork.repository', _('Enabled'))]
     def permission_globals(self):
         c.active = 'globals'
         self.__load_data()
         if request.POST:
             _form = DefaultPermissionsForm(
                 [x[0] for x in c.repo_perms_choices],
                 [x[0] for x in c.group_perms_choices],
                 [x[0] for x in c.user_group_perms_choices],
                 [x[0] for x in c.repo_create_choices],
                 [x[0] for x in c.user_group_create_choices],
                 [x[0] for x in c.fork_choices],
                 [x[0] for x in c.register_choices],
                 [x[0] for x in c.extern_activate_choices])()
             try:
                 form_result = _form.to_python(dict(request.POST))
                 form_result.update({'perm_user_name': 'default'})
                 PermissionModel().update(form_result)
                 meta.Session().commit()
-                h.flash(_('Global permissions updated successfully'),
+                webutils.flash(_('Global permissions updated successfully'),
                         category='success')
             except formencode.Invalid as errors:
                 defaults = errors.value
                 return htmlfill.render(
                     render('admin/permissions/permissions.html'),
                     defaults=defaults,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during update of permissions'),
+                webutils.flash(_('Error occurred during update of permissions'),
                         category='error')
             raise HTTPFound(location=url('admin_permissions'))
         c.user = db.User.get_default_user()
         defaults = {'anonymous': c.user.active}
         for p in c.user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 defaults['default_repo_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('group.'):
                 defaults['default_group_perm'] = p.permission.permission_name
             if p.permission.permission_name.startswith('usergroup.'):
                 defaults['default_user_group_perm'] = p.permission.permission_name
             elif p.permission.permission_name.startswith('hg.create.'):
                 defaults['default_repo_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.usergroup.'):
                 defaults['default_user_group_create'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.register.'):
                 defaults['default_register'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.extern_activate.'):
                 defaults['default_extern_activate'] = p.permission.permission_name
             if p.permission.permission_name.startswith('hg.fork.'):
                 defaults['default_fork'] = p.permission.permission_name
         return htmlfill.render(
             render('admin/permissions/permissions.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def permission_ips(self):
         c.active = 'ips'
         c.user = db.User.get_default_user()
         c.user_ip_map = db.UserIpMap.query() \
                         .filter(db.UserIpMap.user == c.user).all()
         return render('admin/permissions/permissions.html')
     def permission_perms(self):
         c.active = 'perms'
         c.user = db.User.get_default_user()
         c.perm_user = AuthUser(dbuser=c.user)
         return render('admin/permissions/permissions.html')

kallithea/controllers/admin/repo_groups.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.repo_groups
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repository groups controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Mar 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoGroupPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.utils2 import safe_int
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import RepoGroupForm, RepoGroupPermsForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.scm import AvailableRepoGroupChoices, RepoGroupList
 log = logging.getLogger(__name__)
 class RepoGroupsController(BaseController):
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(RepoGroupsController, self)._before(*args, **kwargs)
     def __load_defaults(self, extras=(), exclude=()):
         """extras is used for keeping current parent ignoring permissions
         exclude is used for not moving group to itself TODO: also exclude descendants
         Note: only admin can create top level groups
         """
         repo_groups = AvailableRepoGroupChoices('admin', extras)
         exclude_group_ids = set(rg.group_id for rg in exclude)
         c.repo_groups = [rg for rg in repo_groups
                          if rg[0] not in exclude_group_ids]
     def __load_data(self, group_id):
         """
         Load defaults settings for edit, and update
         :param group_id:
         """
         repo_group = db.RepoGroup.get_or_404(group_id)
         data = repo_group.get_dict()
         data['group_name'] = repo_group.name
         # fill repository group users
         for p in repo_group.repo_group_to_perm:
             data.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
         # fill repository group groups
         for p in repo_group.users_group_to_perm:
             data.update({'g_perm_%s' % p.users_group.users_group_name:
                              p.permission.permission_name})
         return data
     def _revoke_perms_on_yourself(self, form_result):
         _up = [u for u in form_result['perms_updates'] if request.authuser.username == u[0]]
         _new = [u for u in form_result['perms_new'] if request.authuser.username == u[0]]
         if _new and _new[0][1] != 'group.admin' or _up and _up[0][1] != 'group.admin':
             return True
         return False
     def index(self, format='html'):
         _list = db.RepoGroup.query(sorted=True).all()
         group_iter = RepoGroupList(_list, perm_level='admin')
         repo_groups_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         def repo_group_name(repo_group_name, children_groups):
             return template.get_def("repo_group_name") \
                 .render_unicode(repo_group_name, children_groups, _=_, h=h, c=c)
         def repo_group_actions(repo_group_id, repo_group_name, gr_count):
             return template.get_def("repo_group_actions") \
                 .render_unicode(repo_group_id, repo_group_name, gr_count, _=_, h=h, c=c,
                         ungettext=ungettext)
         for repo_gr in group_iter:
             children_groups = [g.name for g in repo_gr.parents] + [repo_gr.name]
             repo_count = repo_gr.repositories.count()
             repo_groups_data.append({
                 "raw_name": repo_gr.group_name,
                 "group_name": repo_group_name(repo_gr.group_name, children_groups),
-                "desc": h.escape(repo_gr.group_description),
+                "desc": webutils.escape(repo_gr.group_description),
                 "repos": repo_count,
                 "owner": h.person(repo_gr.owner),
                 "action": repo_group_actions(repo_gr.group_id, repo_gr.group_name,
                                              repo_count)
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": repo_groups_data
+        }
         return render('admin/repo_groups/repo_groups.html')
     def create(self):
         self.__load_defaults()
         # permissions for can create group based on parent_id are checked
         # here in the Form
         repo_group_form = RepoGroupForm(repo_groups=c.repo_groups)
         form_result = None
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             gr = RepoGroupModel().create(
                 group_name=form_result['group_name'],
                 group_description=form_result['group_description'],
                 parent=form_result['parent_group_id'],
                 owner=request.authuser.user_id, # TODO: make editable
                 copy_permissions=form_result['group_copy_permissions']
+            )
             meta.Session().commit()
             # TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/repo_groups/repo_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during creation of repository group %s')
+            webutils.flash(_('Error occurred during creation of repository group %s')
                     % request.POST.get('group_name'), category='error')
             if form_result is None:
                 raise
             parent_group_id = form_result['parent_group_id']
             # TODO: maybe we should get back to the main view, not the admin one
             raise HTTPFound(location=url('repos_groups', parent_group=parent_group_id))
-        h.flash(_('Created repository group %s') % gr.group_name,
+        webutils.flash(_('Created repository group %s') % gr.group_name,
                 category='success')
         raise HTTPFound(location=url('repos_group_home', group_name=gr.group_name))
     def new(self):
         if HasPermissionAny('hg.admin')('group create'):
             # we're global admin, we're ok and we can create TOP level groups
             pass
         else:
             # we pass in parent group into creation form, thus we know
             # what would be the group, we can check perms here !
             group_id = safe_int(request.GET.get('parent_group'))
             group = db.RepoGroup.get(group_id) if group_id else None
             group_name = group.group_name if group else None
             if HasRepoGroupPermissionLevel('admin')(group_name, 'group create'):
                 pass
             else:
                 raise HTTPForbidden()
         self.__load_defaults()
         return render('admin/repo_groups/repo_group_add.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update(self, group_name):
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
         # TODO: kill allow_empty_group - it is only used for redundant form validation!
         if HasPermissionAny('hg.admin')('group edit'):
             # we're global admin, we're ok and we can create TOP level groups
             allow_empty_group = True
         elif not c.repo_group.parent_group:
             allow_empty_group = True
         else:
             allow_empty_group = False
         repo_group_form = RepoGroupForm(
             edit=True,
             old_data=c.repo_group.get_dict(),
             repo_groups=c.repo_groups,
             can_create_in_root=allow_empty_group,
         )()
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             new_gr = RepoGroupModel().update(group_name, form_result)
             meta.Session().commit()
-            h.flash(_('Updated repository group %s')
+            webutils.flash(_('Updated repository group %s')
                     % form_result['group_name'], category='success')
             # we now have new name !
             group_name = new_gr.group_name
             # TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             c.active = 'settings'
             return htmlfill.render(
                 render('admin/repo_groups/repo_group_edit.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during update of repository group %s')
+            webutils.flash(_('Error occurred during update of repository group %s')
                     % request.POST.get('group_name'), category='error')
         raise HTTPFound(location=url('edit_repo_group', group_name=group_name))
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete(self, group_name):
         gr = c.repo_group = db.RepoGroup.guess_instance(group_name)
         repos = gr.repositories.all()
         if repos:
-            h.flash(_('This group contains %s repositories and cannot be '
+            webutils.flash(_('This group contains %s repositories and cannot be '
                       'deleted') % len(repos), category='warning')
             raise HTTPFound(location=url('repos_groups'))
         children = gr.children.all()
         if children:
-            h.flash(_('This group contains %s subgroups and cannot be deleted'
+            webutils.flash(_('This group contains %s subgroups and cannot be deleted'
                       % (len(children))), category='warning')
             raise HTTPFound(location=url('repos_groups'))
         try:
             RepoGroupModel().delete(group_name)
             meta.Session().commit()
-            h.flash(_('Removed repository group %s') % group_name,
+            webutils.flash(_('Removed repository group %s') % group_name,
                     category='success')
             # TODO: in future action_logger(, '', '', '')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during deletion of repository group %s')
+            webutils.flash(_('Error occurred during deletion of repository group %s')
                     % group_name, category='error')
         if gr.parent_group:
             raise HTTPFound(location=url('repos_group_home', group_name=gr.parent_group.group_name))
         raise HTTPFound(location=url('repos_groups'))
     def show_by_name(self, group_name):
         """
         This is a proxy that does a lookup group_name -> id, and shows
         the group by id view instead
         """
         group_name = group_name.rstrip('/')
         id_ = db.RepoGroup.get_by_group_name(group_name)
         if id_:
             return self.show(group_name)
         raise HTTPNotFound
     @HasRepoGroupPermissionLevelDecorator('read')
     def show(self, group_name):
         c.active = 'settings'
         c.group = c.repo_group = db.RepoGroup.guess_instance(group_name)
         groups = db.RepoGroup.query(sorted=True).filter_by(parent_group=c.group).all()
         repo_groups_list = self.scm_model.get_repo_groups(groups)
         repos_list = db.Repository.query(sorted=True).filter_by(group=c.group).all()
         c.data = RepoModel().get_repos_as_dict(repos_list,
                                                repo_groups_list=repo_groups_list,
                                                short_name=True)
         return render('admin/repo_groups/repo_group_show.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit(self, group_name):
         c.active = 'settings'
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults(extras=[c.repo_group.parent_group],
                              exclude=[c.repo_group])
         defaults = self.__load_data(c.repo_group.group_id)
         return htmlfill.render(
             render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_advanced(self, group_name):
         c.active = 'advanced'
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         return render('admin/repo_groups/repo_group_edit.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit_repo_group_perms(self, group_name):
         c.active = 'perms'
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         self.__load_defaults()
         defaults = self.__load_data(c.repo_group.group_id)
         return htmlfill.render(
             render('admin/repo_groups/repo_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasRepoGroupPermissionLevelDecorator('admin')
     def update_perms(self, group_name):
         """
         Update permissions for given repository group
         :param group_name:
         """
         c.repo_group = db.RepoGroup.guess_instance(group_name)
         valid_recursive_choices = ['none', 'repos', 'groups', 'all']
         form_result = RepoGroupPermsForm(valid_recursive_choices)().to_python(request.POST)
         if not request.authuser.is_admin:
             if self._revoke_perms_on_yourself(form_result):
                 msg = _('Cannot revoke permission for yourself as admin')
-                h.flash(msg, category='warning')
+                webutils.flash(msg, category='warning')
                 raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
         recursive = form_result['recursive']
         # iterate over all members(if in recursive mode) of this groups and
         # set the permissions !
         # this can be potentially heavy operation
         RepoGroupModel()._update_permissions(c.repo_group,
                                              form_result['perms_new'],
                                              form_result['perms_updates'],
                                              recursive)
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         meta.Session().commit()
-        h.flash(_('Repository group permissions updated'), category='success')
+        webutils.flash(_('Repository group permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_group_perms', group_name=group_name))
     @HasRepoGroupPermissionLevelDecorator('admin')
     def delete_perms(self, group_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if not request.authuser.is_admin:
                 if obj_type == 'user' and request.authuser.user_id == obj_id:
                     msg = _('Cannot revoke permission for yourself as admin')
-                    h.flash(msg, category='warning')
+                    webutils.flash(msg, category='warning')
                     raise Exception('revoke admin permission on self')
             recursive = request.POST.get('recursive', 'none')
             if obj_type == 'user':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id, obj_type='user',
                                                    recursive=recursive)
             elif obj_type == 'user_group':
                 RepoGroupModel().delete_permission(repo_group=group_name,
                                                    obj=obj_id,
                                                    obj_type='user_group',
                                                    recursive=recursive)
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during revoking of permission'),
+            webutils.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()

kallithea/controllers/admin/repos.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.repos
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Repositories controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 7, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import celery.result
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPForbidden, HTTPFound, HTTPInternalServerError, HTTPNotFound
 import kallithea
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired, NotAnonymous
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import safe_int
 from kallithea.lib.vcs import RepositoryError
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import RepoFieldForm, RepoForm, RepoPermsForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import AvailableRepoGroupChoices, RepoList, ScmModel
 log = logging.getLogger(__name__)
 class ReposController(BaseRepoController):
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(ReposController, self)._before(*args, **kwargs)
     def _load_repo(self):
         repo_obj = c.db_repo
         if repo_obj is None:
             raise HTTPNotFound()
         return repo_obj
     def __load_defaults(self, repo=None):
         extras = [] if repo is None else [repo.group]
         c.repo_groups = AvailableRepoGroupChoices('write', extras)
         c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs(repo)
     def __load_data(self):
         """
         Load defaults settings for edit, and update
         """
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         defaults = RepoModel()._get_defaults(c.repo_name)
         defaults['clone_uri'] = c.repo_info.clone_uri_hidden # don't show password
         defaults['permanent_url'] = c.repo_info.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True)
         return defaults
     def index(self, format='html'):
         repos_list = RepoList(db.Repository.query(sorted=True).all(), perm_level='admin')
         # the repo list will be filtered to only show repos where the user has read permissions
         repos_data = RepoModel().get_repos_as_dict(repos_list, admin=True)
         # data used to render the grid
         c.data = repos_data
         return render('admin/repos/repos.html')
     @NotAnonymous()
     def create(self):
         self.__load_defaults()
         try:
             # CanWriteGroup validators checks permissions of this POST
             form_result = RepoForm(repo_groups=c.repo_groups,
                                    landing_revs=c.landing_revs_choices)() \
                             .to_python(dict(request.POST))
         except formencode.Invalid as errors:
             log.info(errors)
             return htmlfill.render(
                 render('admin/repos/repo_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 force_defaults=False,
                 encoding="UTF-8")
         try:
             # create is done sometimes async on celery, db transaction
             # management is handled there.
             task = RepoModel().create(form_result, request.authuser.user_id)
             task_id = task.task_id
         except Exception:
             log.error(traceback.format_exc())
             msg = (_('Error creating repository %s')
                    % form_result.get('repo_name'))
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPFound(location=url('home'))
         raise HTTPFound(location=webutils.url('repo_creating_home',
                               repo_name=form_result['repo_name_full'],
                               task_id=task_id))
     @NotAnonymous()
     def create_repository(self):
         self.__load_defaults()
         if not c.repo_groups:
             raise HTTPForbidden
         parent_group = request.GET.get('parent_group')
         ## apply the defaults from defaults page
         defaults = db.Setting.get_default_repo_settings(strip_prefix=True)
         if parent_group:
             prg = db.RepoGroup.get(parent_group)
             if prg is None or not any(rgc[0] == prg.group_id
                                       for rgc in c.repo_groups):
                 raise HTTPForbidden
             defaults.update({'repo_group': parent_group})
         return htmlfill.render(
             render('admin/repos/repo_add.html'),
             defaults=defaults,
             errors={},
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     @LoginRequired()
     def repo_creating(self, repo_name):
         c.repo = repo_name
         c.task_id = request.GET.get('task_id')
         if not c.repo:
             raise HTTPNotFound()
         return render('admin/repos/repo_creating.html')
     @LoginRequired()
     @jsonify
     def repo_check(self, repo_name):
         c.repo = repo_name
         task_id = request.GET.get('task_id')
         if task_id and task_id not in ['None']:
             if kallithea.CELERY_APP:
                 task_result = celery.result.AsyncResult(task_id, app=kallithea.CELERY_APP)
                 if task_result.failed():
                     raise HTTPInternalServerError(task_result.traceback)
         repo = db.Repository.get_by_repo_name(repo_name)
         if repo and repo.repo_state == db.Repository.STATE_CREATED:
             if repo.clone_uri:
-                h.flash(_('Created repository %s from %s')
+                webutils.flash(_('Created repository %s from %s')
                         % (repo.repo_name, repo.clone_uri_hidden), category='success')
             else:
-                repo_url = h.link_to(repo.repo_name,
+                repo_url = webutils.link_to(repo.repo_name,
                                      webutils.url('summary_home',
                                            repo_name=repo.repo_name))
                 fork = repo.fork
                 if fork is not None:
                     fork_name = fork.repo_name
-                    h.flash(h.HTML(_('Forked repository %s as %s'))
+                    webutils.flash(webutils.HTML(_('Forked repository %s as %s'))
                             % (fork_name, repo_url), category='success')
                 else:
-                    h.flash(h.HTML(_('Created repository %s')) % repo_url,
+                    webutils.flash(webutils.HTML(_('Created repository %s')) % repo_url,
                             category='success')
             return {'result': True}
         return {'result': False}
     @HasRepoPermissionLevelDecorator('admin')
     def update(self, repo_name):
         c.repo_info = self._load_repo()
         self.__load_defaults(c.repo_info)
         c.active = 'settings'
         c.repo_fields = db.RepositoryField.query() \
             .filter(db.RepositoryField.repository == c.repo_info).all()
         repo_model = RepoModel()
         changed_name = repo_name
         repo = db.Repository.get_by_repo_name(repo_name)
         old_data = {
             'repo_name': repo_name,
             'repo_group': repo.group.get_dict() if repo.group else {},
             'repo_type': repo.repo_type,
+        }
         _form = RepoForm(edit=True, old_data=old_data,
                          repo_groups=c.repo_groups,
                          landing_revs=c.landing_revs_choices)()
         try:
             form_result = _form.to_python(dict(request.POST))
             repo = repo_model.update(repo_name, **form_result)
             ScmModel().mark_for_invalidation(repo_name)
-            h.flash(_('Repository %s updated successfully') % repo_name,
+            webutils.flash(_('Repository %s updated successfully') % repo_name,
                     category='success')
             changed_name = repo.repo_name
             action_logger(request.authuser, 'admin_updated_repo',
                 changed_name, request.ip_addr)
             meta.Session().commit()
         except formencode.Invalid as errors:
             log.info(errors)
             defaults = self.__load_data()
             defaults.update(errors.value)
             return htmlfill.render(
                 render('admin/repos/repo_edit.html'),
                 defaults=defaults,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during update of repository %s')
+            webutils.flash(_('Error occurred during update of repository %s')
                     % repo_name, category='error')
         raise HTTPFound(location=url('edit_repo', repo_name=changed_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete(self, repo_name):
         repo_model = RepoModel()
         repo = repo_model.get_by_repo_name(repo_name)
         if not repo:
             raise HTTPNotFound()
         try:
             _forks = repo.forks.count()
             handle_forks = None
             if _forks and request.POST.get('forks'):
                 do = request.POST['forks']
                 if do == 'detach_forks':
                     handle_forks = 'detach'
-                    h.flash(_('Detached %s forks') % _forks, category='success')
+                    webutils.flash(_('Detached %s forks') % _forks, category='success')
                 elif do == 'delete_forks':
                     handle_forks = 'delete'
-                    h.flash(_('Deleted %s forks') % _forks, category='success')
+                    webutils.flash(_('Deleted %s forks') % _forks, category='success')
             repo_model.delete(repo, forks=handle_forks)
             action_logger(request.authuser, 'admin_deleted_repo',
                 repo_name, request.ip_addr)
             ScmModel().mark_for_invalidation(repo_name)
-            h.flash(_('Deleted repository %s') % repo_name, category='success')
+            webutils.flash(_('Deleted repository %s') % repo_name, category='success')
             meta.Session().commit()
         except AttachedForksError:
-            h.flash(_('Cannot delete repository %s which still has forks')
+            webutils.flash(_('Cannot delete repository %s which still has forks')
                         % repo_name, category='warning')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during deletion of %s') % repo_name,
+            webutils.flash(_('An error occurred during deletion of %s') % repo_name,
                     category='error')
         if repo.group:
             raise HTTPFound(location=url('repos_group_home', group_name=repo.group.group_name))
         raise HTTPFound(location=url('repos'))
     @HasRepoPermissionLevelDecorator('admin')
     def edit(self, repo_name):
         defaults = self.__load_data()
         c.repo_fields = db.RepositoryField.query() \
             .filter(db.RepositoryField.repository == c.repo_info).all()
         c.active = 'settings'
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'permissions'
         defaults = RepoModel()._get_defaults(repo_name)
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_update(self, repo_name):
         form = RepoPermsForm()().to_python(request.POST)
         RepoModel()._update_permissions(repo_name, form['perms_new'],
                                         form['perms_updates'])
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         meta.Session().commit()
-        h.flash(_('Repository permissions updated'), category='success')
+        webutils.flash(_('Repository permissions updated'), category='success')
         raise HTTPFound(location=url('edit_repo_perms', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_permissions_revoke(self, repo_name):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             else:
                 assert False
             if obj_type == 'user':
                 RepoModel().revoke_user_permission(repo=repo_name, user=obj_id)
             elif obj_type == 'user_group':
                 RepoModel().revoke_user_group_permission(
                     repo=repo_name, group_name=obj_id
+                )
             else:
                 assert False
             # TODO: implement this
             #action_logger(request.authuser, 'admin_revoked_repo_permissions',
             #              repo_name, request.ip_addr)
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during revoking of permission'),
+            webutils.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()
         return []
     @HasRepoPermissionLevelDecorator('admin')
     def edit_fields(self, repo_name):
         c.repo_info = self._load_repo()
         c.repo_fields = db.RepositoryField.query() \
             .filter(db.RepositoryField.repository == c.repo_info).all()
         c.active = 'fields'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_fields'))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def create_repo_field(self, repo_name):
         try:
             form_result = RepoFieldForm()().to_python(dict(request.POST))
             new_field = db.RepositoryField()
             new_field.repository = db.Repository.get_by_repo_name(repo_name)
             new_field.field_key = form_result['new_field_key']
             new_field.field_type = form_result['new_field_type']  # python type
             new_field.field_value = form_result['new_field_value']  # set initial blank value
             new_field.field_desc = form_result['new_field_desc']
             new_field.field_label = form_result['new_field_label']
             meta.Session().add(new_field)
             meta.Session().commit()
         except formencode.Invalid as e:
-            h.flash(_('Field validation error: %s') % e.msg, category='error')
+            webutils.flash(_('Field validation error: %s') % e.msg, category='error')
         except Exception as e:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during creation of field: %r') % e, category='error')
+            webutils.flash(_('An error occurred during creation of field: %r') % e, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def delete_repo_field(self, repo_name, field_id):
         field = db.RepositoryField.get_or_404(field_id)
         try:
             meta.Session().delete(field)
             meta.Session().commit()
         except Exception as e:
             log.error(traceback.format_exc())
             msg = _('An error occurred during removal of field')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
         raise HTTPFound(location=url('edit_repo_fields', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced(self, repo_name):
         c.repo_info = self._load_repo()
         c.default_user_id = kallithea.DEFAULT_USER_ID
         c.in_public_journal = db.UserFollowing.query() \
             .filter(db.UserFollowing.user_id == c.default_user_id) \
             .filter(db.UserFollowing.follows_repository == c.repo_info).scalar()
         _repos = db.Repository.query(sorted=True).all()
         read_access_repos = RepoList(_repos, perm_level='read')
         c.repos_list = [(None, _('-- Not a fork --'))]
         c.repos_list += [(x.repo_id, x.repo_name)
                          for x in read_access_repos
                          if x.repo_id != c.repo_info.repo_id
                          and x.repo_type == c.repo_info.repo_type]
         defaults = {
             'id_fork_of': c.repo_info.fork_id if c.repo_info.fork_id else ''
+        }
         c.active = 'advanced'
         if request.POST:
             raise HTTPFound(location=url('repo_edit_advanced'))
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_journal(self, repo_name):
         """
         Sets this repository to be visible in public journal,
         in other words asking default user to follow this repo
         :param repo_name:
         """
         try:
             repo_id = db.Repository.get_by_repo_name(repo_name).repo_id
             user_id = kallithea.DEFAULT_USER_ID
             self.scm_model.toggle_following_repo(repo_id, user_id)
-            h.flash(_('Updated repository visibility in public journal'),
+            webutils.flash(_('Updated repository visibility in public journal'),
                     category='success')
             meta.Session().commit()
         except Exception:
-            h.flash(_('An error occurred during setting this'
+            webutils.flash(_('An error occurred during setting this'
                       ' repository in public journal'),
                     category='error')
         raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_advanced_fork(self, repo_name):
         """
         Mark given repository as a fork of another
         :param repo_name:
         """
         try:
             fork_id = request.POST.get('id_fork_of')
             repo = ScmModel().mark_as_fork(repo_name, fork_id,
                                            request.authuser.username)
             fork = repo.fork.repo_name if repo.fork else _('Nothing')
             meta.Session().commit()
-            h.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),
+            webutils.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),
                     category='success')
         except RepositoryError as e:
             log.error(traceback.format_exc())
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
         except Exception as e:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during this operation'),
+            webutils.flash(_('An error occurred during this operation'),
                     category='error')
         raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))
     @HasRepoPermissionLevelDecorator('admin')
     def edit_remote(self, repo_name):
         c.repo_info = self._load_repo()
         c.active = 'remote'
         if request.POST:
             try:
                 ScmModel().pull_changes(repo_name, request.authuser.username, request.ip_addr)
-                h.flash(_('Pulled from remote location'), category='success')
+                webutils.flash(_('Pulled from remote location'), category='success')
             except Exception as e:
                 log.error(traceback.format_exc())
-                h.flash(_('An error occurred during pull from remote location'),
+                webutils.flash(_('An error occurred during pull from remote location'),
                         category='error')
             raise HTTPFound(location=url('edit_repo_remote', repo_name=c.repo_name))
         return render('admin/repos/repo_edit.html')
     @HasRepoPermissionLevelDecorator('admin')
     def edit_statistics(self, repo_name):
         c.repo_info = self._load_repo()
         repo = c.repo_info.scm_instance
         if c.repo_info.stats:
             # this is on what revision we ended up so we add +1 for count
             last_rev = c.repo_info.stats.stat_on_revision + 1
         else:
             last_rev = 0
         c.stats_revision = last_rev
         c.repo_last_rev = repo.count() if repo.revisions else 0
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) / c.repo_last_rev) * 100)
         c.active = 'statistics'
         if request.POST:
             try:
                 RepoModel().delete_stats(repo_name)
                 meta.Session().commit()
             except Exception as e:
                 log.error(traceback.format_exc())
-                h.flash(_('An error occurred during deletion of repository stats'),
+                webutils.flash(_('An error occurred during deletion of repository stats'),
                         category='error')
             raise HTTPFound(location=url('edit_repo_statistics', repo_name=c.repo_name))
         return render('admin/repos/repo_edit.html')

kallithea/controllers/admin/settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.settings
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 settings controller for Kallithea admin
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jul 14, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import config, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.celerylib import tasks
 from kallithea.lib.utils import repo2db_mapper, set_app_settings
 from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs import VCSError
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm
 from kallithea.model.notification import EmailNotificationModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 class SettingsController(BaseController):
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(SettingsController, self)._before(*args, **kwargs)
     def _get_hg_ui_settings(self):
         ret = db.Ui.query().all()
         settings = {}
         for each in ret:
             k = each.ui_section + '_' + each.ui_key
             v = each.ui_value
             if k == 'paths_/':
                 k = 'paths_root_path'
             k = k.replace('.', '_')
             if each.ui_section in ['hooks', 'extensions']:
                 v = each.ui_active
             settings[k] = v
         return settings
     @HasPermissionAnyDecorator('hg.admin')
     def settings_vcs(self):
         c.active = 'vcs'
         if request.POST:
             application_form = ApplicationUiSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8",
                      force_defaults=False)
             try:
                 if c.visual.allow_repo_location_change:
                     sett = db.Ui.get_by_key('paths', '/')
                     sett.ui_value = form_result['paths_root_path']
                 # HOOKS
                 sett = db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE)
                 sett.ui_active = form_result['hooks_changegroup_update']
                 sett = db.Ui.get_by_key('hooks', db.Ui.HOOK_REPO_SIZE)
                 sett.ui_active = form_result['hooks_changegroup_repo_size']
                 ## EXTENSIONS
                 sett = db.Ui.get_or_create('extensions', 'largefiles')
                 sett.ui_active = form_result['extensions_largefiles']
 #                sett = db.Ui.get_or_create('extensions', 'hggit')
 #                sett.ui_active = form_result['extensions_hggit']
                 meta.Session().commit()
-                h.flash(_('Updated VCS settings'), category='success')
+                webutils.flash(_('Updated VCS settings'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred while updating '
+                webutils.flash(_('Error occurred while updating '
                           'application settings'), category='error')
         defaults = db.Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_mapping(self):
         c.active = 'mapping'
         if request.POST:
             rm_obsolete = request.POST.get('destroy', False)
             install_git_hooks = request.POST.get('hooks', False)
             overwrite_git_hooks = request.POST.get('hooks_overwrite', False)
             invalidate_cache = request.POST.get('invalidate', False)
             log.debug('rescanning repo location with destroy obsolete=%s, '
                       'install git hooks=%s and '
                       'overwrite git hooks=%s' % (rm_obsolete, install_git_hooks, overwrite_git_hooks))
             filesystem_repos = ScmModel().repo_scan()
             added, removed = repo2db_mapper(filesystem_repos, rm_obsolete,
                                             install_git_hooks=install_git_hooks,
                                             user=request.authuser.username,
                                             overwrite_git_hooks=overwrite_git_hooks)
             added_msg = h.HTML(', ').join(
                 h.link_to(safe_str(repo_name), webutils.url('summary_home', repo_name=repo_name)) for repo_name in added
             added_msg = webutils.HTML(', ').join(
                 webutils.link_to(safe_str(repo_name), webutils.url('summary_home', repo_name=repo_name)) for repo_name in added
             ) or '-'
-            removed_msg = h.HTML(', ').join(
+            removed_msg = webutils.HTML(', ').join(
                 safe_str(repo_name) for repo_name in removed
             ) or '-'
-            h.flash(h.HTML(_('Repositories successfully rescanned. Added: %s. Removed: %s.')) %
+            webutils.flash(webutils.HTML(_('Repositories successfully rescanned. Added: %s. Removed: %s.')) %
                     (added_msg, removed_msg), category='success')
             if invalidate_cache:
                 log.debug('invalidating all repositories cache')
                 i = 0
                 for repo in db.Repository.query():
                     try:
                         ScmModel().mark_for_invalidation(repo.repo_name)
                         i += 1
                     except VCSError as e:
                         log.warning('VCS error invalidating %s: %s', repo.repo_name, e)
-                h.flash(_('Invalidated %s repositories') % i, category='success')
+                webutils.flash(_('Invalidated %s repositories') % i, category='success')
             raise HTTPFound(location=url('admin_settings_mapping'))
         defaults = db.Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             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(
                     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)
-                h.flash(_('Updated application settings'), category='success')
+                webutils.flash(_('Updated application settings'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred while updating '
+                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(
             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(
                     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)
-                h.flash(_('Updated visualisation settings'),
+                webutils.flash(_('Updated visualisation settings'),
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during updating '
+                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(
             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:
-                h.flash(_('Please enter email address'), category='error')
+                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
             tasks.send_email(recipients, test_email_subj,
                              test_email_txt_body, test_email_html_body)
-            h.flash(_('Send email task created'), category='success')
+            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())
         import kallithea
         c.ini = kallithea.CONFIG
         return htmlfill.render(
             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()):
-                        h.flash(_('Hook already exists'), category='error')
+                        webutils.flash(_('Hook already exists'), category='error')
                     elif ui_key in (x.ui_key for x in db.Ui.get_builtin_hooks()):
-                        h.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
+                        webutils.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
                     elif ui_value and ui_key:
                         db.Ui.create_or_update_hook(ui_key, ui_value)
-                        h.flash(_('Added new hook'), category='success')
+                        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:
-                        h.flash(_('Updated hooks'), category='success')
+                        webutils.flash(_('Updated hooks'), category='success')
                     meta.Session().commit()
                 except Exception:
                     log.error(traceback.format_exc())
-                    h.flash(_('Error occurred during hook creation'),
+                    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(
             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)
             tasks.whoosh_index(repo_location, full_index)
-            h.flash(_('Whoosh reindex task scheduled'), category='success')
+            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(
             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())
         import kallithea
         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(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)

kallithea/controllers/admin/user_groups.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.user_groups
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 User Groups crud controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql.expression import func
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPInternalServerError
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, HasUserGroupPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.exceptions import RepoGroupAssignmentError, UserGroupsAssignedException
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import CustomDefaultPermissionsForm, UserGroupForm, UserGroupPermsForm
 from kallithea.model.scm import UserGroupList
 from kallithea.model.user_group import UserGroupModel
 log = logging.getLogger(__name__)
 class UserGroupsController(BaseController):
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(UserGroupsController, self)._before(*args, **kwargs)
     def __load_data(self, user_group_id):
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
         c.available_members = sorted(((x.user_id, x.username) for x in
                                       db.User.query().all()),
                                      key=lambda u: u[1].lower())
     def __load_defaults(self, user_group_id):
         """
         Load defaults settings for edit, and update
         :param user_group_id:
         """
         user_group = db.UserGroup.get_or_404(user_group_id)
         data = user_group.get_dict()
         return data
     def index(self, format='html'):
         _list = db.UserGroup.query() \
                         .order_by(func.lower(db.UserGroup.users_group_name)) \
                         .all()
         group_iter = UserGroupList(_list, perm_level='admin')
         user_groups_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         def user_group_name(user_group_id, user_group_name):
             return template.get_def("user_group_name") \
                 .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
         def user_group_actions(user_group_id, user_group_name):
             return template.get_def("user_group_actions") \
                 .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
         for user_gr in group_iter:
             user_groups_data.append({
                 "raw_name": user_gr.users_group_name,
                 "group_name": user_group_name(user_gr.users_group_id,
                                               user_gr.users_group_name),
-                "desc": h.escape(user_gr.user_group_description),
+                "desc": webutils.escape(user_gr.user_group_description),
                 "members": len(user_gr.members),
                 "active": h.boolicon(user_gr.users_group_active),
                 "owner": h.person(user_gr.owner.username),
                 "action": user_group_actions(user_gr.users_group_id, user_gr.users_group_name)
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": user_groups_data
+        }
         return render('admin/user_groups/user_groups.html')
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create(self):
         users_group_form = UserGroupForm()()
         try:
             form_result = users_group_form.to_python(dict(request.POST))
             ug = UserGroupModel().create(name=form_result['users_group_name'],
                                          description=form_result['user_group_description'],
                                          owner=request.authuser.user_id,
                                          active=form_result['users_group_active'])
             gr = form_result['users_group_name']
             action_logger(request.authuser,
                           'admin_created_users_group:%s' % gr,
                           None, request.ip_addr)
-            h.flash(h.HTML(_('Created user group %s')) % h.link_to(gr, url('edit_users_group', id=ug.users_group_id)),
+            webutils.flash(webutils.HTML(_('Created user group %s')) % webutils.link_to(gr, url('edit_users_group', id=ug.users_group_id)),
                 category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/user_groups/user_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during creation of user group %s')
+            webutils.flash(_('Error occurred during creation of user group %s')
                     % request.POST.get('users_group_name'), category='error')
         raise HTTPFound(location=url('users_groups'))
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def new(self, format='html'):
         return render('admin/user_groups/user_group_add.html')
     @HasUserGroupPermissionLevelDecorator('admin')
     def update(self, id):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'settings'
         self.__load_data(id)
         available_members = [safe_str(x[0]) for x in c.available_members]
         users_group_form = UserGroupForm(edit=True,
                                          old_data=c.user_group.get_dict(),
                                          available_members=available_members)()
         try:
             form_result = users_group_form.to_python(request.POST)
             UserGroupModel().update(c.user_group, form_result)
             gr = form_result['users_group_name']
             action_logger(request.authuser,
                           'admin_updated_users_group:%s' % gr,
                           None, request.ip_addr)
-            h.flash(_('Updated user group %s') % gr, category='success')
+            webutils.flash(_('Updated user group %s') % gr, category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             ug_model = UserGroupModel()
             defaults = errors.value
             e = errors.error_dict or {}
             defaults.update({
                 'create_repo_perm': ug_model.has_perm(id,
                                                       'hg.create.repository'),
                 'fork_repo_perm': ug_model.has_perm(id,
                                                     'hg.fork.repository'),
             })
             return htmlfill.render(
                 render('admin/user_groups/user_group_edit.html'),
                 defaults=defaults,
                 errors=e,
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during update of user group %s')
+            webutils.flash(_('Error occurred during update of user group %s')
                     % request.POST.get('users_group_name'), category='error')
         raise HTTPFound(location=url('edit_users_group', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def delete(self, id):
         usr_gr = db.UserGroup.get_or_404(id)
         try:
             UserGroupModel().delete(usr_gr)
             meta.Session().commit()
-            h.flash(_('Successfully deleted user group'), category='success')
+            webutils.flash(_('Successfully deleted user group'), category='success')
         except UserGroupsAssignedException as e:
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during deletion of user group'),
+            webutils.flash(_('An error occurred during deletion of user group'),
                     category='error')
         raise HTTPFound(location=url('users_groups'))
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit(self, id, format='html'):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'settings'
         self.__load_data(id)
         defaults = self.__load_defaults(id)
         return htmlfill.render(
             render('admin/user_groups/user_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_perms(self, id):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'perms'
         defaults = {}
         # fill user group users
         for p in c.user_group.user_user_group_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
         for p in c.user_group.user_group_user_group_to_perm:
             defaults.update({'g_perm_%s' % p.user_group.users_group_name:
                              p.permission.permission_name})
         return htmlfill.render(
             render('admin/user_groups/user_group_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def update_perms(self, id):
         """
         grant permission for given usergroup
         :param id:
         """
         user_group = db.UserGroup.get_or_404(id)
         form = UserGroupPermsForm()().to_python(request.POST)
         # set the permissions !
         try:
             UserGroupModel()._update_permissions(user_group, form['perms_new'],
                                                  form['perms_updates'])
         except RepoGroupAssignmentError:
-            h.flash(_('Target group cannot be the same'), category='error')
+            webutils.flash(_('Target group cannot be the same'), category='error')
             raise HTTPFound(location=url('edit_user_group_perms', id=id))
         # TODO: implement this
         #action_logger(request.authuser, 'admin_changed_repo_permissions',
         #              repo_name, request.ip_addr)
         meta.Session().commit()
-        h.flash(_('User group permissions updated'), category='success')
+        webutils.flash(_('User group permissions updated'), category='success')
         raise HTTPFound(location=url('edit_user_group_perms', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def delete_perms(self, id):
         try:
             obj_type = request.POST.get('obj_type')
             obj_id = None
             if obj_type == 'user':
                 obj_id = safe_int(request.POST.get('user_id'))
             elif obj_type == 'user_group':
                 obj_id = safe_int(request.POST.get('user_group_id'))
             if not request.authuser.is_admin:
                 if obj_type == 'user' and request.authuser.user_id == obj_id:
                     msg = _('Cannot revoke permission for yourself as admin')
-                    h.flash(msg, category='warning')
+                    webutils.flash(msg, category='warning')
                     raise Exception('revoke admin permission on self')
             if obj_type == 'user':
                 UserGroupModel().revoke_user_permission(user_group=id,
                                                         user=obj_id)
             elif obj_type == 'user_group':
                 UserGroupModel().revoke_user_group_permission(target_user_group=id,
                                                               user_group=obj_id)
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during revoking of permission'),
+            webutils.flash(_('An error occurred during revoking of permission'),
                     category='error')
             raise HTTPInternalServerError()
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_default_perms(self, id):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'default_perms'
         permissions = {
             'repositories': {},
             'repositories_groups': {}
+        }
         ugroup_repo_perms = db.UserGroupRepoToPerm.query() \
             .options(joinedload(db.UserGroupRepoToPerm.permission)) \
             .options(joinedload(db.UserGroupRepoToPerm.repository)) \
             .filter(db.UserGroupRepoToPerm.users_group_id == id) \
             .all()
         for gr in ugroup_repo_perms:
             permissions['repositories'][gr.repository.repo_name]  \
                 = gr.permission.permission_name
         ugroup_group_perms = db.UserGroupRepoGroupToPerm.query() \
             .options(joinedload(db.UserGroupRepoGroupToPerm.permission)) \
             .options(joinedload(db.UserGroupRepoGroupToPerm.group)) \
             .filter(db.UserGroupRepoGroupToPerm.users_group_id == id) \
             .all()
         for gr in ugroup_group_perms:
             permissions['repositories_groups'][gr.group.group_name] \
                 = gr.permission.permission_name
         c.permissions = permissions
         ug_model = UserGroupModel()
         defaults = c.user_group.get_dict()
         defaults.update({
             'create_repo_perm': ug_model.has_perm(c.user_group,
                                                   '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
+        )
     @HasUserGroupPermissionLevelDecorator('admin')
     def update_default_perms(self, id):
         user_group = db.UserGroup.get_or_404(id)
         try:
             form = CustomDefaultPermissionsForm()()
             form_result = form.to_python(request.POST)
             usergroup_model = UserGroupModel()
             defs = db.UserGroupToPerm.query() \
                 .filter(db.UserGroupToPerm.users_group == user_group) \
                 .all()
             for ug in defs:
                 meta.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')
+            webutils.flash(_("Updated permissions"), category='success')
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during permissions saving'),
+            webutils.flash(_('An error occurred during permissions saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_group_default_perms', id=id))
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_advanced(self, id):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'advanced'
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         return render('admin/user_groups/user_group_edit.html')
     @HasUserGroupPermissionLevelDecorator('admin')
     def edit_members(self, id):
         c.user_group = db.UserGroup.get_or_404(id)
         c.active = 'members'
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
         return render('admin/user_groups/user_group_edit.html')

kallithea/controllers/admin/users.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.users
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Users crud controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from sqlalchemy.sql.expression import func
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 import kallithea
 from kallithea.lib import auth_modules
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, IfSshEnabled, render
 from kallithea.lib.exceptions import DefaultUserException, UserCreationError, UserOwnsReposException
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import datetime_to_time, generate_api_key, safe_int
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.forms import CustomDefaultPermissionsForm, UserForm
 from kallithea.model.ssh_key import SshKeyModel, SshKeyModelException
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class UsersController(BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(UsersController, self)._before(*args, **kwargs)
     def index(self, format='html'):
         c.users_list = db.User.query().order_by(db.User.username) \
                         .filter_by(is_default_user=False) \
                         .order_by(func.lower(db.User.username)) \
                         .all()
         users_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         grav_tmpl = '<div class="gravatar">%s</div>'
         def username(user_id, username):
             return template.get_def("user_name") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         def user_actions(user_id, username):
             return template.get_def("user_actions") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         for user in c.users_list:
             users_data.append({
                 "gravatar": grav_tmpl % h.gravatar(user.email, size=20),
                 "raw_name": user.username,
                 "username": username(user.user_id, user.username),
                 "firstname": h.escape(user.name),
                 "lastname": h.escape(user.lastname),
                 "firstname": webutils.escape(user.name),
                 "lastname": webutils.escape(user.lastname),
                 "last_login": h.fmt_date(user.last_login),
                 "last_login_raw": datetime_to_time(user.last_login),
                 "active": h.boolicon(user.active),
                 "admin": h.boolicon(user.admin),
                 "extern_type": user.extern_type,
                 "extern_name": user.extern_name,
                 "action": user_actions(user.user_id, user.username),
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": users_data
+        }
         return render('admin/users/users.html')
     def create(self):
         c.default_extern_type = db.User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         user_model = UserModel()
         user_form = UserForm()()
         try:
             form_result = user_form.to_python(dict(request.POST))
             user = user_model.create(form_result)
             action_logger(request.authuser, 'admin_created_user:%s' % user.username,
                           None, request.ip_addr)
-            h.flash(_('Created user %s') % user.username,
+            webutils.flash(_('Created user %s') % user.username,
                     category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/users/user_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except UserCreationError as e:
-            h.flash(e, 'error')
+            webutils.flash(e, 'error')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during creation of user %s')
+            webutils.flash(_('Error occurred during creation of user %s')
                     % request.POST.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=user.user_id))
     def new(self, format='html'):
         c.default_extern_type = db.User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         return render('admin/users/user_add.html')
     def update(self, id):
         user_model = UserModel()
         user = user_model.get(id)
         _form = UserForm(edit=True, old_data={'user_id': id,
                                               'email': user.email})()
         form_result = {}
         try:
             form_result = _form.to_python(dict(request.POST))
             skip_attrs = ['extern_type', 'extern_name',
                          ] + auth_modules.get_managed_fields(user)
             user_model.update(id, form_result, skip_attrs=skip_attrs)
             usr = form_result['username']
             action_logger(request.authuser, 'admin_updated_user:%s' % usr,
                           None, request.ip_addr)
-            h.flash(_('User updated successfully'), category='success')
+            webutils.flash(_('User updated successfully'), category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             defaults = errors.value
             e = errors.error_dict or {}
             defaults.update({
                 'create_repo_perm': user_model.has_perm(id,
                                                         'hg.create.repository'),
                 'fork_repo_perm': user_model.has_perm(id, 'hg.fork.repository'),
             })
             return htmlfill.render(
                 self._render_edit_profile(user),
                 defaults=defaults,
                 errors=e,
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('Error occurred during update of user %s')
+            webutils.flash(_('Error occurred during update of user %s')
                     % form_result.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=id))
     def delete(self, id):
         usr = db.User.get_or_404(id)
         has_ssh_keys = bool(usr.ssh_keys)
         try:
             UserModel().delete(usr)
             meta.Session().commit()
-            h.flash(_('Successfully deleted user'), category='success')
+            webutils.flash(_('Successfully deleted user'), category='success')
         except (UserOwnsReposException, DefaultUserException) as e:
-            h.flash(e, category='warning')
+            webutils.flash(e, category='warning')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during deletion of user'),
+            webutils.flash(_('An error occurred during deletion of user'),
                     category='error')
         else:
             if has_ssh_keys:
                 SshKeyModel().write_authorized_keys()
         raise HTTPFound(location=url('users'))
     def _get_user_or_raise_if_default(self, id):
         try:
             return db.User.get_or_404(id, allow_default=False)
         except DefaultUserException:
-            h.flash(_("The default user cannot be edited"), category='warning')
+            webutils.flash(_("The default user cannot be edited"), category='warning')
             raise HTTPNotFound
     def _render_edit_profile(self, user):
         c.user = user
         c.active = 'profile'
         c.perm_user = AuthUser(dbuser=user)
         managed_fields = auth_modules.get_managed_fields(user)
         c.readonly = lambda n: 'readonly' if n in managed_fields else None
         return render('admin/users/user_edit.html')
     def edit(self, id, format='html'):
         user = self._get_user_or_raise_if_default(id)
         defaults = user.get_dict()
         return htmlfill.render(
             self._render_edit_profile(user),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def edit_advanced(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'advanced'
         c.perm_user = AuthUser(dbuser=c.user)
         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 edit_api_keys(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'api_keys'
         show_expired = True
         c.lifetime_values = [
             (str(-1), _('Forever')),
             (str(5), _('5 minutes')),
             (str(60), _('1 hour')),
             (str(60 * 24), _('1 day')),
             (str(60 * 24 * 30), _('1 month')),
+        ]
         c.lifetime_options = [(c.lifetime_values, _("Lifetime"))]
         c.user_api_keys = ApiKeyModel().get_api_keys(c.user.user_id,
                                                      show_expired=show_expired)
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         lifetime = safe_int(request.POST.get('lifetime'), -1)
         description = request.POST.get('description')
         ApiKeyModel().create(c.user.user_id, description, lifetime)
         meta.Session().commit()
-        h.flash(_("API key successfully created"), category='success')
+        webutils.flash(_("API key successfully created"), category='success')
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
     def delete_api_key(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         api_key = request.POST.get('del_api_key')
         if request.POST.get('del_api_key_builtin'):
             c.user.api_key = generate_api_key()
             meta.Session().commit()
-            h.flash(_("API key successfully reset"), category='success')
+            webutils.flash(_("API key successfully reset"), category='success')
         elif api_key:
             ApiKeyModel().delete(api_key, c.user.user_id)
             meta.Session().commit()
-            h.flash(_("API key successfully deleted"), category='success')
+            webutils.flash(_("API key successfully deleted"), category='success')
         raise HTTPFound(location=url('edit_user_api_keys', id=c.user.user_id))
     def update_account(self, id):
         pass
     def edit_perms(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'perms'
         c.perm_user = AuthUser(dbuser=c.user)
         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)
             user_model = UserModel()
             defs = db.UserToPerm.query() \
                 .filter(db.UserToPerm.user == user) \
                 .all()
             for ug in defs:
                 meta.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')
+            webutils.flash(_("Updated permissions"), category='success')
             meta.Session().commit()
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during permissions saving'),
+            webutils.flash(_('An error occurred during permissions saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_perms', id=id))
     def edit_emails(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'emails'
         c.user_email_map = db.UserEmailMap.query() \
             .filter(db.UserEmailMap.user == c.user).all()
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email = request.POST.get('new_email')
         user_model = UserModel()
         try:
             user_model.add_extra_email(id, email)
             meta.Session().commit()
-            h.flash(_("Added email %s to user") % email, category='success')
+            webutils.flash(_("Added email %s to user") % email, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['email']
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during email saving'),
+            webutils.flash(_('An error occurred during email saving'),
                     category='error')
         raise HTTPFound(location=url('edit_user_emails', id=id))
     def delete_email(self, id):
         user = self._get_user_or_raise_if_default(id)
         email_id = request.POST.get('del_email_id')
         user_model = UserModel()
         user_model.delete_extra_email(id, email_id)
         meta.Session().commit()
-        h.flash(_("Removed email from user"), category='success')
+        webutils.flash(_("Removed email from user"), category='success')
         raise HTTPFound(location=url('edit_user_emails', id=id))
     def edit_ips(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'ips'
         c.user_ip_map = db.UserIpMap.query() \
             .filter(db.UserIpMap.user == c.user).all()
         c.default_user_ip_map = db.UserIpMap.query() \
             .filter(db.UserIpMap.user_id == kallithea.DEFAULT_USER_ID).all()
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     def add_ip(self, id):
         ip = request.POST.get('new_ip')
         user_model = UserModel()
         try:
             user_model.add_extra_ip(id, ip)
             meta.Session().commit()
-            h.flash(_("Added IP address %s to user whitelist") % ip, category='success')
+            webutils.flash(_("Added IP address %s to user whitelist") % ip, category='success')
         except formencode.Invalid as error:
             msg = error.error_dict['ip']
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred while adding IP address'),
+            webutils.flash(_('An error occurred while adding IP address'),
                     category='error')
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))
     def delete_ip(self, id):
         ip_id = request.POST.get('del_ip_id')
         user_model = UserModel()
         user_model.delete_extra_ip(id, ip_id)
         meta.Session().commit()
-        h.flash(_("Removed IP address from user whitelist"), category='success')
+        webutils.flash(_("Removed IP address from user whitelist"), category='success')
         if 'default_user' in request.POST:
             raise HTTPFound(location=url('admin_permissions_ips'))
         raise HTTPFound(location=url('edit_user_ips', id=id))
     @IfSshEnabled
     def edit_ssh_keys(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         c.active = 'ssh_keys'
         c.user_ssh_keys = SshKeyModel().get_ssh_keys(c.user.user_id)
         defaults = c.user.get_dict()
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @IfSshEnabled
     def ssh_keys_add(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         description = request.POST.get('description')
         public_key = request.POST.get('public_key')
         try:
             new_ssh_key = SshKeyModel().create(c.user.user_id,
                                                description, public_key)
             meta.Session().commit()
             SshKeyModel().write_authorized_keys()
-            h.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
+            webutils.flash(_("SSH key %s successfully added") % new_ssh_key.fingerprint, category='success')
         except SshKeyModelException as e:
-            h.flash(e.args[0], category='error')
+            webutils.flash(e.args[0], category='error')
         raise HTTPFound(location=url('edit_user_ssh_keys', id=c.user.user_id))
     @IfSshEnabled
     def ssh_keys_delete(self, id):
         c.user = self._get_user_or_raise_if_default(id)
         fingerprint = request.POST.get('del_public_key_fingerprint')
         try:
             SshKeyModel().delete(fingerprint, c.user.user_id)
             meta.Session().commit()
             SshKeyModel().write_authorized_keys()
-            h.flash(_("SSH key successfully deleted"), category='success')
+            webutils.flash(_("SSH key successfully deleted"), category='success')
         except SshKeyModelException as e:
-            h.flash(e.args[0], category='error')
+            webutils.flash(e.args[0], category='error')
         raise HTTPFound(location=url('edit_user_ssh_keys', id=c.user.user_id))

kallithea/controllers/changelog.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.changelog
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 changelog controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 from tg import request, session
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPFound, HTTPNotFound
 import kallithea.lib.helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, NodeDoesNotExistError, RepositoryError
 from kallithea.lib.webutils import url
 log = logging.getLogger(__name__)
 class ChangelogController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(ChangelogController, self)._before(*args, **kwargs)
         c.affected_files_cut_off = 60
     @staticmethod
     def __get_cs(rev, repo):
         """
         Safe way to get changeset. If error occur fail with error message.
         :param rev: revision to fetch
         :param repo: repo instance
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
-            h.flash(_('There are no changesets yet'), category='error')
+            webutils.flash(_('There are no changesets yet'), category='error')
         except RepositoryError as e:
             log.error(traceback.format_exc())
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
         raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name, revision=None, f_path=None):
         limit = 2000
         default = 100
         if request.GET.get('size'):
             c.size = max(min(safe_int(request.GET.get('size')), limit), 1)
             session['changelog_size'] = c.size
             session.save()
         else:
             c.size = int(session.get('changelog_size', default))
         # min size must be 1
         c.size = max(c.size, 1)
         p = safe_int(request.GET.get('page'), 1)
         branch_name = request.GET.get('branch', None)
         if (branch_name and
             branch_name not in c.db_repo_scm_instance.branches and
             branch_name not in c.db_repo_scm_instance.closed_branches and
             not revision
         ):
             raise HTTPFound(location=url('changelog_file_home', repo_name=c.repo_name,
                                     revision=branch_name, f_path=f_path or ''))
         if revision == 'tip':
             revision = None
         c.changelog_for_path = f_path
         try:
             if f_path:
                 log.debug('generating changelog for path %s', f_path)
                 # get the history for the file !
                 tip_cs = c.db_repo_scm_instance.get_changeset()
                 try:
                     collection = tip_cs.get_file_history(f_path)
                 except (NodeDoesNotExistError, ChangesetError):
                     # this node is not present at tip !
                     try:
                         cs = self.__get_cs(revision, repo_name)
                         collection = cs.get_file_history(f_path)
                     except RepositoryError as e:
-                        h.flash(e, category='warning')
+                        webutils.flash(e, category='warning')
                         raise HTTPFound(location=webutils.url('changelog_home', repo_name=repo_name))
             else:
                 collection = c.db_repo_scm_instance.get_changesets(start=0, end=revision,
                                                         branch_name=branch_name, reverse=True)
             c.total_cs = len(collection)
             c.cs_pagination = Page(collection, page=p, item_count=c.total_cs, items_per_page=c.size,
                                    branch=branch_name)
             page_revisions = [x.raw_id for x in c.cs_pagination]
             c.cs_comments = c.db_repo.get_comments(page_revisions)
             c.cs_statuses = c.db_repo.statuses(page_revisions)
         except EmptyRepositoryError as e:
-            h.flash(e, category='warning')
+            webutils.flash(e, category='warning')
             raise HTTPFound(location=url('summary_home', repo_name=c.repo_name))
         except (RepositoryError, ChangesetDoesNotExistError, Exception) as e:
             log.error(traceback.format_exc())
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
             raise HTTPFound(location=url('changelog_home', repo_name=c.repo_name))
         c.branch_name = branch_name
         c.branch_filters = [('', _('None'))] + \
             [(k, k) for k in c.db_repo_scm_instance.branches]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.branch_filters += [('-', '-')] + \
                 [(k, prefix + k) for k in c.db_repo_scm_instance.closed_branches]
         revs = []
         if not f_path:
             revs = [x.revision for x in c.cs_pagination]
         c.jsdata = graph_data(c.db_repo_scm_instance, revs)
         c.revision = revision # requested revision ref
         c.first_revision = c.cs_pagination[0] # pagination is never empty here!
         return render('changelog/changelog.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def changelog_details(self, cs):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             c.cs = c.db_repo_scm_instance.get_changeset(cs)
             return render('changelog/changelog_details.html')
         raise HTTPNotFound()

kallithea/controllers/changeset.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.changeset
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 changeset controller showing changes between revisions
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 25, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import binascii
 import logging
 import traceback
 from collections import OrderedDict
 from tg import request, response
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPForbidden, HTTPNotFound
 import kallithea.lib.helpers as h
 from kallithea.lib import auth, diffs, webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import ascii_str, safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError
 from kallithea.model import db, meta
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.comment import ChangesetCommentsModel
 from kallithea.model.pull_request import PullRequestModel
 log = logging.getLogger(__name__)
 def create_cs_pr_comment(repo_name, revision=None, pull_request=None, allowed_to_change_status=True):
     """
     Add a comment to the specified changeset or pull request, using POST values
     from the request.
     Comments can be inline (when a file path and line number is specified in
     POST) or general comments.
     A comment can be accompanied by a review status change (accepted, rejected,
     etc.). Pull requests can be closed or deleted.
     Parameter 'allowed_to_change_status' is used for both status changes and
     closing of pull requests. For deleting of pull requests, more specific
     checks are done.
     """
     assert request.environ.get('HTTP_X_PARTIAL_XHR')
     if pull_request:
         pull_request_id = pull_request.pull_request_id
     else:
         pull_request_id = None
     status = request.POST.get('changeset_status')
     close_pr = request.POST.get('save_close')
     delete = request.POST.get('save_delete')
     f_path = request.POST.get('f_path')
     line_no = request.POST.get('line')
     if (status or close_pr or delete) and (f_path or line_no):
         # status votes and closing is only possible in general comments
         raise HTTPBadRequest()
     if not allowed_to_change_status:
         if status or close_pr:
-            h.flash(_('No permission to change status'), 'error')
+            webutils.flash(_('No permission to change status'), 'error')
             raise HTTPForbidden()
     if pull_request and delete == "delete":
         if (pull_request.owner_id == request.authuser.user_id or
             auth.HasPermissionAny('hg.admin')() or
             auth.HasRepoPermissionLevel('admin')(pull_request.org_repo.repo_name) or
             auth.HasRepoPermissionLevel('admin')(pull_request.other_repo.repo_name)
         ) and not pull_request.is_closed():
             PullRequestModel().delete(pull_request)
             meta.Session().commit()
-            h.flash(_('Successfully deleted pull request %s') % pull_request_id,
+            webutils.flash(_('Successfully deleted pull request %s') % pull_request_id,
                     category='success')
             return {
                'location': webutils.url('my_pullrequests'), # or repo pr list?
+            }
         raise HTTPForbidden()
     text = request.POST.get('text', '').strip()
     comment = ChangesetCommentsModel().create(
         text=text,
         repo=c.db_repo.repo_id,
         author=request.authuser.user_id,
         revision=revision,
         pull_request=pull_request_id,
         f_path=f_path or None,
         line_no=line_no or None,
         status_change=db.ChangesetStatus.get_status_lbl(status) if status else None,
         closing_pr=close_pr,
+    )
     if status:
         ChangesetStatusModel().set_status(
             c.db_repo.repo_id,
             status,
             request.authuser.user_id,
             comment,
             revision=revision,
             pull_request=pull_request_id,
+        )
     if pull_request:
         action = 'user_commented_pull_request:%s' % pull_request_id
     else:
         action = 'user_commented_revision:%s' % revision
     action_logger(request.authuser, action, c.db_repo, request.ip_addr)
     if pull_request and close_pr:
         PullRequestModel().close_pull_request(pull_request_id)
         action_logger(request.authuser,
                       'user_closed_pull_request:%s' % pull_request_id,
                       c.db_repo, request.ip_addr)
     meta.Session().commit()
     data = {
-       'target_id': h.safeid(request.POST.get('f_path')),
+       'target_id': webutils.safeid(request.POST.get('f_path')),
+    }
     if comment is not None:
         c.comment = comment
         data.update(comment.get_dict())
         data.update({'rendered_text':
                      render('changeset/changeset_comment_block.html')})
     return data
 def delete_cs_pr_comment(repo_name, comment_id):
     """Delete a comment from a changeset or pull request"""
     co = db.ChangesetComment.get_or_404(comment_id)
     if co.repo.repo_name != repo_name:
         raise HTTPNotFound()
     if co.pull_request and co.pull_request.is_closed():
         # don't allow deleting comments on closed pull request
         raise HTTPForbidden()
     owner = co.author_id == request.authuser.user_id
     repo_admin = auth.HasRepoPermissionLevel('admin')(repo_name)
     if auth.HasPermissionAny('hg.admin')() or repo_admin or owner:
         ChangesetCommentsModel().delete(comment=co)
         meta.Session().commit()
         return True
     else:
         raise HTTPForbidden()
 class ChangesetController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(ChangesetController, self)._before(*args, **kwargs)
         c.affected_files_cut_off = 60
     def _index(self, revision, method):
         c.pull_request = None
         c.fulldiff = request.GET.get('fulldiff') # for reporting number of changed files
         # get ranges of revisions if preset
         rev_range = revision.split('...')[:2]
         c.cs_repo = c.db_repo
         try:
             if len(rev_range) == 2:
                 rev_start = rev_range[0]
                 rev_end = rev_range[1]
                 rev_ranges = c.db_repo_scm_instance.get_changesets(start=rev_start,
                                                              end=rev_end)
             else:
                 rev_ranges = [c.db_repo_scm_instance.get_changeset(revision)]
             c.cs_ranges = list(rev_ranges)
             if not c.cs_ranges:
                 raise RepositoryError('Changeset range returned empty result')
         except (ChangesetDoesNotExistError, EmptyRepositoryError):
             log.debug(traceback.format_exc())
             msg = _('Such revision does not exist for this repository')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPNotFound()
         c.changes = OrderedDict()
         c.lines_added = 0  # count of lines added
         c.lines_deleted = 0  # count of lines removes
         c.changeset_statuses = db.ChangesetStatus.STATUSES
         comments = dict()
         c.statuses = []
         c.inline_comments = []
         c.inline_cnt = 0
         # Iterate over ranges (default changeset view is always one changeset)
         for changeset in c.cs_ranges:
             if method == 'show':
                 c.statuses.extend([ChangesetStatusModel().get_status(
                             c.db_repo.repo_id, changeset.raw_id)])
                 # Changeset comments
                 comments.update((com.comment_id, com)
                                 for com in ChangesetCommentsModel()
                                 .get_comments(c.db_repo.repo_id,
                                               revision=changeset.raw_id))
                 # Status change comments - mostly from pull requests
                 comments.update((st.comment_id, st.comment)
                                 for st in ChangesetStatusModel()
                                 .get_statuses(c.db_repo.repo_id,
                                               changeset.raw_id, with_revisions=True)
                                 if st.comment_id is not None)
                 inlines = ChangesetCommentsModel() \
                             .get_inline_comments(c.db_repo.repo_id,
                                                  revision=changeset.raw_id)
                 c.inline_comments.extend(inlines)
             cs2 = changeset.raw_id
             cs1 = changeset.parents[0].raw_id if changeset.parents else EmptyChangeset().raw_id
             ignore_whitespace_diff = h.get_ignore_whitespace_diff(request.GET)
             diff_context_size = h.get_diff_context_size(request.GET)
             raw_diff = diffs.get_diff(c.db_repo_scm_instance, cs1, cs2,
                 ignore_whitespace=ignore_whitespace_diff, context=diff_context_size)
             diff_limit = None if c.fulldiff else self.cut_off_limit
             file_diff_data = []
             if method == 'show':
                 diff_processor = diffs.DiffProcessor(raw_diff,
                                                      vcs=c.db_repo_scm_instance.alias,
                                                      diff_limit=diff_limit)
                 c.limited_diff = diff_processor.limited_diff
                 for f in diff_processor.parsed:
                     st = f['stats']
                     c.lines_added += st['added']
                     c.lines_deleted += st['deleted']
                     filename = f['filename']
                     fid = h.FID(changeset.raw_id, filename)
                     url_fid = h.FID('', filename)
                     html_diff = diffs.as_html(parsed_lines=[f])
                     file_diff_data.append((fid, url_fid, f['operation'], f['old_filename'], filename, html_diff, st))
             else:
                 # downloads/raw we only need RAW diff nothing else
                 file_diff_data.append(('', None, None, None, raw_diff, None))
             c.changes[changeset.raw_id] = (cs1, cs2, file_diff_data)
         # sort comments in creation order
         c.comments = [com for com_id, com in sorted(comments.items())]
         # count inline comments
         for __, lines in c.inline_comments:
             for comments in lines.values():
                 c.inline_cnt += len(comments)
         if len(c.cs_ranges) == 1:
             c.changeset = c.cs_ranges[0]
             c.parent_tmpl = ''.join(['# Parent  %s\n' % x.raw_id
                                      for x in c.changeset.parents])
             c.changeset_graft_source_hash = ascii_str(c.changeset.extra.get(b'source', b''))
             c.changeset_transplant_source_hash = ascii_str(binascii.hexlify(c.changeset.extra.get(b'transplant_source', b'')))
         if method == 'download':
             response.content_type = 'text/plain'
             response.content_disposition = 'attachment; filename=%s.diff' \
                                             % revision[:12]
             return raw_diff
         elif method == 'patch':
             response.content_type = 'text/plain'
             c.diff = safe_str(raw_diff)
             return render('changeset/patch_changeset.html')
         elif method == 'raw':
             response.content_type = 'text/plain'
             return raw_diff
         elif method == 'show':
             if len(c.cs_ranges) == 1:
                 return render('changeset/changeset.html')
             else:
                 c.cs_ranges_org = None
                 c.cs_comments = {}
                 revs = [ctx.revision for ctx in reversed(c.cs_ranges)]
                 c.jsdata = graph_data(c.db_repo_scm_instance, revs)
                 return render('changeset/changeset_range.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, revision, method='show'):
         return self._index(revision, method=method)
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def changeset_raw(self, revision):
         return self._index(revision, method='raw')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def changeset_patch(self, revision):
         return self._index(revision, method='patch')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def changeset_download(self, revision):
         return self._index(revision, method='download')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def comment(self, repo_name, revision):
         return create_cs_pr_comment(repo_name, revision=revision)
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         return delete_cs_pr_comment(repo_name, comment_id)
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def changeset_info(self, repo_name, revision):
         if request.is_xhr:
             try:
                 return c.db_repo_scm_instance.get_changeset(revision)
             except ChangesetDoesNotExistError as e:
                 return EmptyChangeset(message=str(e))
         else:
             raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def changeset_children(self, repo_name, revision):
         if request.is_xhr:
             changeset = c.db_repo_scm_instance.get_changeset(revision)
             result = {"results": []}
             if changeset.children:
                 result = {"results": changeset.children}
             return result
         else:
             raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def changeset_parents(self, repo_name, revision):
         if request.is_xhr:
             changeset = c.db_repo_scm_instance.get_changeset(revision)
             result = {"results": []}
             if changeset.parents:
                 result = {"results": changeset.parents}
             return result
         else:
             raise HTTPBadRequest()

kallithea/controllers/compare.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.compare
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 compare controller showing differences between two
 repos, branches, bookmarks or tips
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 6, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPFound, HTTPNotFound
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.graphmod import graph_data
 from kallithea.lib.webutils import url
 from kallithea.model import db
 log = logging.getLogger(__name__)
 class CompareController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(CompareController, self)._before(*args, **kwargs)
         # The base repository has already been retrieved.
         c.a_repo = c.db_repo
         # Retrieve the "changeset" repository (default: same as base).
         other_repo = request.GET.get('other_repo', None)
         if other_repo is None:
             c.cs_repo = c.a_repo
         else:
             c.cs_repo = db.Repository.get_by_repo_name(other_repo)
             if c.cs_repo is None:
                 msg = _('Could not find other repository %s') % other_repo
-                h.flash(msg, category='error')
+                webutils.flash(msg, category='error')
                 raise HTTPFound(location=url('compare_home', repo_name=c.a_repo.repo_name))
         # Verify that it's even possible to compare these two repositories.
         if c.a_repo.scm_instance.alias != c.cs_repo.scm_instance.alias:
             msg = _('Cannot compare repositories of different types')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPFound(location=url('compare_home', repo_name=c.a_repo.repo_name))
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name):
         c.compare_home = True
         c.a_ref_name = c.cs_ref_name = None
         return render('compare/compare_diff.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def compare(self, repo_name, org_ref_type, org_ref_name, other_ref_type, other_ref_name):
         org_ref_name = org_ref_name.strip()
         other_ref_name = other_ref_name.strip()
         # If merge is True:
         #   Show what org would get if merged with other:
         #   List changesets that are ancestors of other but not of org.
         #   New changesets in org is thus ignored.
         #   Diff will be from common ancestor, and merges of org to other will thus be ignored.
         # If merge is False:
         #   Make a raw diff from org to other, no matter if related or not.
         #   Changesets in one and not in the other will be ignored
         merge = bool(request.GET.get('merge'))
         # fulldiff disables cut_off_limit
         fulldiff = request.GET.get('fulldiff')
         # partial uses compare_cs.html template directly
         partial = request.environ.get('HTTP_X_PARTIAL_XHR')
         # is_ajax_preview puts hidden input field with changeset revisions
         c.is_ajax_preview = partial and request.GET.get('is_ajax_preview')
         # swap url for compare_diff page - never partial and never is_ajax_preview
         c.swap_url = webutils.url('compare_url',
             repo_name=c.cs_repo.repo_name,
             org_ref_type=other_ref_type, org_ref_name=other_ref_name,
             other_repo=c.a_repo.repo_name,
             other_ref_type=org_ref_type, other_ref_name=org_ref_name,
             merge=merge or '')
         ignore_whitespace_diff = h.get_ignore_whitespace_diff(request.GET)
         diff_context_size = h.get_diff_context_size(request.GET)
         c.a_rev = self._get_ref_rev(c.a_repo, org_ref_type, org_ref_name,
             returnempty=True)
         c.cs_rev = self._get_ref_rev(c.cs_repo, other_ref_type, other_ref_name)
         c.compare_home = False
         c.a_ref_name = org_ref_name
         c.a_ref_type = org_ref_type
         c.cs_ref_name = other_ref_name
         c.cs_ref_type = other_ref_type
         c.cs_ranges, c.cs_ranges_org, c.ancestors = c.a_repo.scm_instance.get_diff_changesets(
             c.a_rev, c.cs_repo.scm_instance, c.cs_rev)
         raw_ids = [x.raw_id for x in c.cs_ranges]
         c.cs_comments = c.cs_repo.get_comments(raw_ids)
         c.cs_statuses = c.cs_repo.statuses(raw_ids)
         revs = [ctx.revision for ctx in reversed(c.cs_ranges)]
         c.jsdata = graph_data(c.cs_repo.scm_instance, revs)
         if partial:
             return render('compare/compare_cs.html')
         org_repo = c.a_repo
         other_repo = c.cs_repo
         if merge:
             rev1 = msg = None
             if not c.cs_ranges:
                 msg = _('Cannot show empty diff')
             elif not c.ancestors:
                 msg = _('No ancestor found for merge diff')
             elif len(c.ancestors) == 1:
                 rev1 = c.ancestors[0]
             else:
                 msg = _('Multiple merge ancestors found for merge compare')
             if rev1 is None:
-                h.flash(msg, category='error')
+                webutils.flash(msg, category='error')
                 log.error(msg)
                 raise HTTPNotFound
             # case we want a simple diff without incoming changesets,
             # previewing what will be merged.
             # Make the diff on the other repo (which is known to have other_rev)
             log.debug('Using ancestor %s as rev1 instead of %s',
                       rev1, c.a_rev)
             org_repo = other_repo
         else: # comparing tips, not necessarily linearly related
             if org_repo != other_repo:
                 # TODO: we could do this by using hg unionrepo
                 log.error('cannot compare across repos %s and %s', org_repo, other_repo)
-                h.flash(_('Cannot compare repositories without using common ancestor'), category='error')
+                webutils.flash(_('Cannot compare repositories without using common ancestor'), category='error')
                 raise HTTPBadRequest
             rev1 = c.a_rev
         diff_limit = None if fulldiff else self.cut_off_limit
         log.debug('running diff between %s and %s in %s',
                   rev1, c.cs_rev, org_repo.scm_instance.path)
         raw_diff = diffs.get_diff(org_repo.scm_instance, rev1=rev1, rev2=c.cs_rev,
                                       ignore_whitespace=ignore_whitespace_diff,
                                       context=diff_context_size)
         diff_processor = diffs.DiffProcessor(raw_diff, diff_limit=diff_limit)
         c.limited_diff = diff_processor.limited_diff
         c.file_diff_data = []
         c.lines_added = 0
         c.lines_deleted = 0
         for f in diff_processor.parsed:
             st = f['stats']
             c.lines_added += st['added']
             c.lines_deleted += st['deleted']
             filename = f['filename']
             fid = h.FID('', filename)
             html_diff = diffs.as_html(parsed_lines=[f])
             c.file_diff_data.append((fid, None, f['operation'], f['old_filename'], filename, html_diff, st))
         return render('compare/compare_diff.html')

kallithea/controllers/files.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.files
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Files controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 21, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import posixpath
 import shutil
 import tempfile
 import traceback
 from collections import OrderedDict
 from tg import request, response
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 import kallithea
 from kallithea.lib import diffs
 from kallithea.lib import helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.exceptions import NonRelativePathError
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import asbool, convert_line_endings, detect_mode, safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import (ChangesetDoesNotExistError, ChangesetError, EmptyRepositoryError, ImproperArchiveTypeError, NodeAlreadyExistsError,
                                           NodeDoesNotExistError, NodeError, RepositoryError, VCSError)
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.lib.vcs.utils import author_email
 from kallithea.lib.webutils import url
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 class FilesController(BaseRepoController):
     def _before(self, *args, **kwargs):
         super(FilesController, self)._before(*args, **kwargs)
     def __get_cs(self, rev, silent_empty=False):
         """
         Safe way to get changeset if error occur it redirects to tip with
         proper message
         :param rev: revision to fetch
         :silent_empty: return None if repository is empty
         """
         try:
             return c.db_repo_scm_instance.get_changeset(rev)
         except EmptyRepositoryError as e:
             if silent_empty:
                 return None
             url_ = url('files_add_home',
                        repo_name=c.repo_name,
                        revision=0, f_path='', anchor='edit')
             add_new = h.link_to(_('Click here to add new file'), url_, class_="alert-link")
             h.flash(_('There are no files yet.') + ' ' + add_new, category='warning')
             add_new = webutils.link_to(_('Click here to add new file'), url_, class_="alert-link")
             webutils.flash(_('There are no files yet.') + ' ' + add_new, category='warning')
             raise HTTPNotFound()
         except (ChangesetDoesNotExistError, LookupError):
             msg = _('Such revision does not exist for this repository')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
             raise HTTPNotFound()
     def __get_filenode(self, cs, path):
         """
         Returns file_node or raise HTTP error.
         :param cs: given changeset
         :param path: path to lookup
         """
         try:
             file_node = cs.get_node(path)
             if file_node.is_dir():
                 raise RepositoryError('given path is a directory')
         except ChangesetDoesNotExistError:
             msg = _('Such revision does not exist for this repository')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPNotFound()
         except RepositoryError as e:
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
             raise HTTPNotFound()
         return file_node
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name, revision, f_path, annotate=False):
         # redirect to given revision from form if given
         post_revision = request.POST.get('at_rev', None)
         if post_revision:
             cs = self.__get_cs(post_revision) # FIXME - unused!
         c.revision = revision
         c.changeset = self.__get_cs(revision)
         c.branch = request.GET.get('branch', None)
         c.f_path = f_path
         c.annotate = annotate
         cur_rev = c.changeset.revision
         # used in files_source.html:
         c.cut_off_limit = self.cut_off_limit
         c.fulldiff = request.GET.get('fulldiff')
         # prev link
         try:
             prev_rev = c.db_repo_scm_instance.get_changeset(cur_rev).prev(c.branch)
             c.url_prev = url('files_home', repo_name=c.repo_name,
                          revision=prev_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_prev += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_prev = '#'
         # next link
         try:
             next_rev = c.db_repo_scm_instance.get_changeset(cur_rev).next(c.branch)
             c.url_next = url('files_home', repo_name=c.repo_name,
                      revision=next_rev.raw_id, f_path=f_path)
             if c.branch:
                 c.url_next += '?branch=%s' % c.branch
         except (ChangesetDoesNotExistError, VCSError):
             c.url_next = '#'
         # files or dirs
         try:
             c.file = c.changeset.get_node(f_path)
             if c.file.is_submodule():
                 raise HTTPFound(location=c.file.url)
             elif c.file.is_file():
                 c.load_full_history = False
                 # determine if we're on branch head
                 _branches = c.db_repo_scm_instance.branches
                 c.on_branch_head = revision in _branches or revision in _branches.values()
                 _hist = []
                 c.file_history = []
                 if c.load_full_history:
                     c.file_history, _hist = self._get_node_history(c.changeset, f_path)
                 c.authors = []
                 for a in set([x.author for x in _hist]):
                     c.authors.append((author_email(a), h.person(a)))
             else:
                 c.authors = c.file_history = []
         except RepositoryError as e:
-            h.flash(e, category='error')
+            webutils.flash(e, category='error')
             raise HTTPNotFound()
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('files/files_ypjax.html')
         # TODO: tags and bookmarks?
         c.revision_options = [(c.changeset.raw_id,
                               _('%s at %s') % (b, c.changeset.short_id)) for b in c.changeset.branches] + \
             [(n, b) for b, n in c.db_repo_scm_instance.branches.items()]
         if c.db_repo_scm_instance.closed_branches:
             prefix = _('(closed)') + ' '
             c.revision_options += [('-', '-')] + \
                 [(n, prefix + b) for b, n in c.db_repo_scm_instance.closed_branches.items()]
         return render('files/files.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def history(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, f_path)
             res = []
             for obj in file_history:
                 res.append({
                     'text': obj[1],
                     'children': [{'id': o[0], 'text': o[1]} for o in obj[0]]
                 })
             data = {
                 'more': False,
                 'results': res
+            }
             return data
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def authors(self, repo_name, revision, f_path):
         changeset = self.__get_cs(revision)
         _file = changeset.get_node(f_path)
         if _file.is_file():
             file_history, _hist = self._get_node_history(changeset, f_path)
             c.authors = []
             for a in set([x.author for x in _hist]):
                 c.authors.append((author_email(a), h.person(a)))
             return render('files/files_history_box.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def rawfile(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, f_path)
         response.content_disposition = \
             'attachment; filename=%s' % f_path.split(kallithea.URL_SEP)[-1]
         response.content_type = file_node.mimetype
         return file_node.content
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def raw(self, repo_name, revision, f_path):
         cs = self.__get_cs(revision)
         file_node = self.__get_filenode(cs, f_path)
         raw_mimetype_mapping = {
             # map original mimetype to a mimetype used for "show as raw"
             # you can also provide a content-disposition to override the
             # default "attachment" disposition.
             # orig_type: (new_type, new_dispo)
             # show images inline:
             'image/x-icon': ('image/x-icon', 'inline'),
             'image/png': ('image/png', 'inline'),
             'image/gif': ('image/gif', 'inline'),
             'image/jpeg': ('image/jpeg', 'inline'),
             'image/svg+xml': ('image/svg+xml', 'inline'),
+        }
         mimetype = file_node.mimetype
         try:
             mimetype, dispo = raw_mimetype_mapping[mimetype]
         except KeyError:
             # we don't know anything special about this, handle it safely
             if file_node.is_binary:
                 # do same as download raw for binary files
                 mimetype, dispo = 'application/octet-stream', 'attachment'
             else:
                 # do not just use the original mimetype, but force text/plain,
                 # otherwise it would serve text/html and that might be unsafe.
                 # Note: underlying vcs library fakes text/plain mimetype if the
                 # mimetype can not be determined and it thinks it is not
                 # binary.This might lead to erroneous text display in some
                 # cases, but helps in other cases, like with text files
                 # without extension.
                 mimetype, dispo = 'text/plain', 'inline'
         if dispo == 'attachment':
             dispo = 'attachment; filename=%s' % f_path.split(os.sep)[-1]
         response.content_disposition = dispo
         response.content_type = mimetype
         return file_node.content
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def delete(self, repo_name, revision, f_path):
         repo = c.db_repo
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches and revision not in _branches.values():
-            h.flash(_('You can only delete files with revision '
+            webutils.flash(_('You can only delete files with revision '
                       'being a valid branch'), category='warning')
             raise HTTPFound(location=webutils.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         c.default_message = _('Deleted file %s via Kallithea') % (f_path)
         c.f_path = f_path
         node_path = f_path
         author = request.authuser.full_contact
         if r_post:
             message = r_post.get('message') or c.default_message
             try:
                 nodes = {
                     node_path: {
                         'content': ''
+                    }
+                }
                 self.scm_model.delete_nodes(
                     user=request.authuser.user_id,
                     ip_addr=request.ip_addr,
                     repo=c.db_repo,
                     message=message,
                     nodes=nodes,
                     parent_cs=c.cs,
                     author=author,
+                )
-                h.flash(_('Successfully deleted file %s') % f_path,
+                webutils.flash(_('Successfully deleted file %s') % f_path,
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during commit'), category='error')
+                webutils.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_delete.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def edit(self, repo_name, revision, f_path):
         repo = c.db_repo
         # check if revision is a branch identifier- basically we cannot
         # create multiple heads via file editing
         _branches = repo.scm_instance.branches
         # check if revision is a branch name or branch hash
         if revision not in _branches and revision not in _branches.values():
-            h.flash(_('You can only edit files with revision '
+            webutils.flash(_('You can only edit files with revision '
                       'being a valid branch'), category='warning')
             raise HTTPFound(location=webutils.url('files_home',
                                   repo_name=repo_name, revision='tip',
                                   f_path=f_path))
         r_post = request.POST
         c.cs = self.__get_cs(revision)
         c.file = self.__get_filenode(c.cs, f_path)
         if c.file.is_binary:
             raise HTTPFound(location=url('files_home', repo_name=c.repo_name,
                             revision=c.cs.raw_id, f_path=f_path))
         c.default_message = _('Edited file %s via Kallithea') % (f_path)
         c.f_path = f_path
         if r_post:
             old_content = safe_str(c.file.content)
             sl = old_content.splitlines(1)
             first_line = sl[0] if sl else ''
             # modes:  0 - Unix, 1 - Mac, 2 - DOS
             mode = detect_mode(first_line, 0)
             content = convert_line_endings(r_post.get('content', ''), mode)
             message = r_post.get('message') or c.default_message
             author = request.authuser.full_contact
             if content == old_content:
-                h.flash(_('No changes'), category='warning')
+                webutils.flash(_('No changes'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             try:
                 self.scm_model.commit_change(repo=c.db_repo_scm_instance,
                                              repo_name=repo_name, cs=c.cs,
                                              user=request.authuser.user_id,
                                              ip_addr=request.ip_addr,
                                              author=author, message=message,
                                              content=content, f_path=f_path)
-                h.flash(_('Successfully committed to %s') % f_path,
+                webutils.flash(_('Successfully committed to %s') % f_path,
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during commit'), category='error')
+                webutils.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_edit.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('write')
     def add(self, repo_name, revision, f_path):
         repo = c.db_repo
         r_post = request.POST
         c.cs = self.__get_cs(revision, silent_empty=True)
         if c.cs is None:
             c.cs = EmptyChangeset(alias=c.db_repo_scm_instance.alias)
         c.default_message = (_('Added file via Kallithea'))
         c.f_path = f_path
         if r_post:
             unix_mode = 0
             content = convert_line_endings(r_post.get('content', ''), unix_mode)
             message = r_post.get('message') or c.default_message
             filename = r_post.get('filename')
             location = r_post.get('location', '')
             file_obj = r_post.get('upload_file', None)
             if file_obj is not None and hasattr(file_obj, 'filename'):
                 filename = file_obj.filename
                 content = file_obj.file
                 if hasattr(content, 'file'):
                     # non posix systems store real file under file attr
                     content = content.file
             if not content:
-                h.flash(_('No content'), category='warning')
+                webutils.flash(_('No content'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             if not filename:
-                h.flash(_('No filename'), category='warning')
+                webutils.flash(_('No filename'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             # strip all crap out of file, just leave the basename
             filename = os.path.basename(filename)
             node_path = posixpath.join(location, filename)
             author = request.authuser.full_contact
             try:
                 nodes = {
                     node_path: {
                         'content': content
+                    }
+                }
                 self.scm_model.create_nodes(
                     user=request.authuser.user_id,
                     ip_addr=request.ip_addr,
                     repo=c.db_repo,
                     message=message,
                     nodes=nodes,
                     parent_cs=c.cs,
                     author=author,
+                )
-                h.flash(_('Successfully committed to %s') % node_path,
+                webutils.flash(_('Successfully committed to %s') % node_path,
                         category='success')
             except NonRelativePathError as e:
-                h.flash(_('Location must be relative path and must not '
+                webutils.flash(_('Location must be relative path and must not '
                           'contain .. in path'), category='warning')
                 raise HTTPFound(location=url('changeset_home', repo_name=c.repo_name,
                                     revision='tip'))
             except (NodeError, NodeAlreadyExistsError) as e:
-                h.flash(_(e), category='error')
+                webutils.flash(_(e), category='error')
             except Exception:
                 log.error(traceback.format_exc())
-                h.flash(_('Error occurred during commit'), category='error')
+                webutils.flash(_('Error occurred during commit'), category='error')
             raise HTTPFound(location=url('changeset_home',
                                 repo_name=c.repo_name, revision='tip'))
         return render('files/files_add.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def archivefile(self, repo_name, fname):
         fileformat = None
         revision = None
         ext = None
         subrepos = request.GET.get('subrepos') == 'true'
         for a_type, ext_data in settings.ARCHIVE_SPECS.items():
             archive_spec = fname.split(ext_data[1])
             if len(archive_spec) == 2 and archive_spec[1] == '':
                 fileformat = a_type or ext_data[1]
                 revision = archive_spec[0]
                 ext = ext_data[1]
         try:
             dbrepo = RepoModel().get_by_repo_name(repo_name)
             if not dbrepo.enable_downloads:
                 return _('Downloads disabled') # TODO: do something else?
             if c.db_repo_scm_instance.alias == 'hg':
                 # patch and reset hooks section of UI config to not run any
                 # hooks on fetching archives with subrepos
                 for k, v in c.db_repo_scm_instance._repo.ui.configitems('hooks'):
                     c.db_repo_scm_instance._repo.ui.setconfig('hooks', k, None)
             cs = c.db_repo_scm_instance.get_changeset(revision)
             content_type = settings.ARCHIVE_SPECS[fileformat][0]
         except ChangesetDoesNotExistError:
             return _('Unknown revision %s') % revision
         except EmptyRepositoryError:
             return _('Empty repository')
         except (ImproperArchiveTypeError, KeyError):
             return _('Unknown archive type')
         rev_name = cs.raw_id[:12]
         archive_name = '%s-%s%s' % (repo_name.replace('/', '_'), rev_name, ext)
         archive_path = None
         cached_archive_path = None
         archive_cache_dir = kallithea.CONFIG.get('archive_cache_dir')
         if archive_cache_dir and not subrepos: # TODO: subrepo caching?
             if not os.path.isdir(archive_cache_dir):
                 os.makedirs(archive_cache_dir)
             cached_archive_path = os.path.join(archive_cache_dir, archive_name)
             if os.path.isfile(cached_archive_path):
                 log.debug('Found cached archive in %s', cached_archive_path)
                 archive_path = cached_archive_path
             else:
                 log.debug('Archive %s is not yet cached', archive_name)
         if archive_path is None:
             # generate new archive
             fd, archive_path = tempfile.mkstemp()
             log.debug('Creating new temp archive in %s', archive_path)
             with os.fdopen(fd, 'wb') as stream:
                 cs.fill_archive(stream=stream, kind=fileformat, subrepos=subrepos)
                 # stream (and thus fd) has been closed by cs.fill_archive
             if cached_archive_path is not None:
                 # we generated the archive - move it to cache
                 log.debug('Storing new archive in %s', cached_archive_path)
                 shutil.move(archive_path, cached_archive_path)
                 archive_path = cached_archive_path
         def get_chunked_archive(archive_path):
             stream = open(archive_path, 'rb')
             while True:
                 data = stream.read(16 * 1024)
                 if not data:
                     break
                 yield data
             stream.close()
             if archive_path != cached_archive_path:
                 log.debug('Destroying temp archive %s', archive_path)
                 os.remove(archive_path)
         action_logger(user=request.authuser,
                       action='user_downloaded_archive:%s' % (archive_name),
                       repo=repo_name, ipaddr=request.ip_addr, commit=True)
         response.content_disposition = str('attachment; filename=%s' % (archive_name))
         response.content_type = str(content_type)
         return get_chunked_archive(archive_path)
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def diff(self, repo_name, f_path):
         ignore_whitespace_diff = h.get_ignore_whitespace_diff(request.GET)
         diff_context_size = h.get_diff_context_size(request.GET)
         diff2 = request.GET.get('diff2', '')
         diff1 = request.GET.get('diff1', '') or diff2
         c.action = request.GET.get('diff')
         c.no_changes = diff1 == diff2
         c.f_path = f_path
         c.big_diff = False
         fulldiff = request.GET.get('fulldiff')
         c.changes = OrderedDict()
         c.changes[diff2] = []
         # special case if we want a show rev only, it's impl here
         # to reduce JS and callbacks
         if request.GET.get('show_rev'):
             if asbool(request.GET.get('annotate', 'False')):
                 _url = url('files_annotate_home', repo_name=c.repo_name,
                            revision=diff1, f_path=c.f_path)
             else:
                 _url = url('files_home', repo_name=c.repo_name,
                            revision=diff1, f_path=c.f_path)
             raise HTTPFound(location=_url)
         try:
             if diff1 not in ['', None, 'None', '0' * 12, '0' * 40]:
                 c.changeset_1 = c.db_repo_scm_instance.get_changeset(diff1)
                 try:
                     node1 = c.changeset_1.get_node(f_path)
                     if node1.is_dir():
                         raise NodeError('%s path is a %s not a file'
                                         % (node1, type(node1)))
                 except NodeDoesNotExistError:
                     c.changeset_1 = EmptyChangeset(cs=diff1,
                                                    revision=c.changeset_1.revision,
                                                    repo=c.db_repo_scm_instance)
                     node1 = FileNode(f_path, '', changeset=c.changeset_1)
             else:
                 c.changeset_1 = EmptyChangeset(repo=c.db_repo_scm_instance)
                 node1 = FileNode(f_path, '', changeset=c.changeset_1)
             if diff2 not in ['', None, 'None', '0' * 12, '0' * 40]:
                 c.changeset_2 = c.db_repo_scm_instance.get_changeset(diff2)
                 try:
                     node2 = c.changeset_2.get_node(f_path)
                     if node2.is_dir():
                         raise NodeError('%s path is a %s not a file'
                                         % (node2, type(node2)))
                 except NodeDoesNotExistError:
                     c.changeset_2 = EmptyChangeset(cs=diff2,
                                                    revision=c.changeset_2.revision,
                                                    repo=c.db_repo_scm_instance)
                     node2 = FileNode(f_path, '', changeset=c.changeset_2)
             else:
                 c.changeset_2 = EmptyChangeset(repo=c.db_repo_scm_instance)
                 node2 = FileNode(f_path, '', changeset=c.changeset_2)
         except (RepositoryError, NodeError):
             log.error(traceback.format_exc())
             raise HTTPFound(location=url('files_home', repo_name=c.repo_name,
                                 f_path=f_path))
         if c.action == 'download':
             raw_diff = diffs.get_gitdiff(node1, node2,
                                       ignore_whitespace=ignore_whitespace_diff,
                                       context=diff_context_size)
             diff_name = '%s_vs_%s.diff' % (diff1, diff2)
             response.content_type = 'text/plain'
             response.content_disposition = (
                 'attachment; filename=%s' % diff_name
+            )
             return raw_diff
         elif c.action == 'raw':
             raw_diff = diffs.get_gitdiff(node1, node2,
                                       ignore_whitespace=ignore_whitespace_diff,
                                       context=diff_context_size)
             response.content_type = 'text/plain'
             return raw_diff
         else:
             fid = h.FID(diff2, node2.path)
             diff_limit = None if fulldiff else self.cut_off_limit
             c.a_rev, c.cs_rev, a_path, diff, st, op = diffs.wrapped_diff(filenode_old=node1,
                                          filenode_new=node2,
                                          diff_limit=diff_limit,
                                          ignore_whitespace=ignore_whitespace_diff,
                                          line_context=diff_context_size)
             c.file_diff_data = [(fid, fid, op, a_path, node2.path, diff, st)]
             return render('files/file_diff.html')
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def diff_2way(self, repo_name, f_path):
         diff1 = request.GET.get('diff1', '')
         diff2 = request.GET.get('diff2', '')
         try:
             if diff1 not in ['', None, 'None', '0' * 12, '0' * 40]:
                 c.changeset_1 = c.db_repo_scm_instance.get_changeset(diff1)
                 try:
                     node1 = c.changeset_1.get_node(f_path)
                     if node1.is_dir():
                         raise NodeError('%s path is a %s not a file'
                                         % (node1, type(node1)))
                 except NodeDoesNotExistError:
                     c.changeset_1 = EmptyChangeset(cs=diff1,
                                                    revision=c.changeset_1.revision,
                                                    repo=c.db_repo_scm_instance)
                     node1 = FileNode(f_path, '', changeset=c.changeset_1)
             else:
                 c.changeset_1 = EmptyChangeset(repo=c.db_repo_scm_instance)
                 node1 = FileNode(f_path, '', changeset=c.changeset_1)
             if diff2 not in ['', None, 'None', '0' * 12, '0' * 40]:
                 c.changeset_2 = c.db_repo_scm_instance.get_changeset(diff2)
                 try:
                     node2 = c.changeset_2.get_node(f_path)
                     if node2.is_dir():
                         raise NodeError('%s path is a %s not a file'
                                         % (node2, type(node2)))
                 except NodeDoesNotExistError:
                     c.changeset_2 = EmptyChangeset(cs=diff2,
                                                    revision=c.changeset_2.revision,
                                                    repo=c.db_repo_scm_instance)
                     node2 = FileNode(f_path, '', changeset=c.changeset_2)
             else:
                 c.changeset_2 = EmptyChangeset(repo=c.db_repo_scm_instance)
                 node2 = FileNode(f_path, '', changeset=c.changeset_2)
         except ChangesetDoesNotExistError as e:
             msg = _('Such revision does not exist for this repository')
-            h.flash(msg, category='error')
+            webutils.flash(msg, category='error')
             raise HTTPNotFound()
         c.node1 = node1
         c.node2 = node2
         c.cs1 = c.changeset_1
         c.cs2 = c.changeset_2
         return render('files/diff_2way.html')
     def _get_node_history(self, cs, f_path, changesets=None):
         """
         get changesets history for given node
         :param cs: changeset to calculate history
         :param f_path: path for node to calculate history for
         :param changesets: if passed don't calculate history and take
             changesets defined in this list
         """
         # calculate history based on tip
         tip_cs = c.db_repo_scm_instance.get_changeset()
         if changesets is None:
             try:
                 changesets = tip_cs.get_file_history(f_path)
             except (NodeDoesNotExistError, ChangesetError):
                 # this node is not present at tip !
                 changesets = cs.get_file_history(f_path)
         hist_l = []
         changesets_group = ([], _("Changesets"))
         branches_group = ([], _("Branches"))
         tags_group = ([], _("Tags"))
         for chs in changesets:
             # TODO: loop over chs.branches ... but that will not give all the bogus None branches for Git ...
             _branch = chs.branch
             n_desc = '%s (%s)' % (h.show_id(chs), _branch)
             changesets_group[0].append((chs.raw_id, n_desc,))
         hist_l.append(changesets_group)
         for name, chs in c.db_repo_scm_instance.branches.items():
             branches_group[0].append((chs, name),)
         hist_l.append(branches_group)
         for name, chs in c.db_repo_scm_instance.tags.items():
             tags_group[0].append((chs, name),)
         hist_l.append(tags_group)
         return hist_l, changesets
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def nodelist(self, repo_name, revision, f_path):
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             cs = self.__get_cs(revision)
             _d, _f = ScmModel().get_nodes(repo_name, cs.raw_id, f_path,
                                           flat=False)
             return {'nodes': _d + _f}

kallithea/controllers/forks.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.forks
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 forks controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 23, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 import kallithea
 import kallithea.lib.helpers as h
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, HasRepoPermissionLevel, HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, render
 from kallithea.lib.page import Page
 from kallithea.lib.utils2 import safe_int
 from kallithea.model import db
 from kallithea.model.forms import RepoForkForm
 from kallithea.model.repo import RepoModel
 from kallithea.model.scm import AvailableRepoGroupChoices, ScmModel
 log = logging.getLogger(__name__)
 class ForksController(BaseRepoController):
     def __load_defaults(self):
         c.repo_groups = AvailableRepoGroupChoices('write')
         c.landing_revs_choices, c.landing_revs = ScmModel().get_repo_landing_revs()
         c.can_update = db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE).ui_active
     def __load_data(self):
         """
         Load defaults settings for edit, and update
         """
         self.__load_defaults()
         c.repo_info = c.db_repo
         repo = c.db_repo.scm_instance
         if c.repo_info is None:
             raise HTTPNotFound()
         c.default_user_id = kallithea.DEFAULT_USER_ID
         c.in_public_journal = db.UserFollowing.query() \
             .filter(db.UserFollowing.user_id == c.default_user_id) \
             .filter(db.UserFollowing.follows_repository == c.repo_info).scalar()
         if c.repo_info.stats:
             last_rev = c.repo_info.stats.stat_on_revision + 1
         else:
             last_rev = 0
         c.stats_revision = last_rev
         c.repo_last_rev = repo.count() if repo.revisions else 0
         if last_rev == 0 or c.repo_last_rev == 0:
             c.stats_percentage = 0
         else:
             c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                             c.repo_last_rev) * 100)
         defaults = RepoModel()._get_defaults(c.repo_name)
         # alter the description to indicate a fork
         defaults['description'] = ('fork of repository: %s \n%s'
                                    % (defaults['repo_name'],
                                       defaults['description']))
         # add suffix to fork
         defaults['repo_name'] = '%s-fork' % defaults['repo_name']
         return defaults
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def forks(self, repo_name):
         p = safe_int(request.GET.get('page'), 1)
         repo_id = c.db_repo.repo_id
         d = []
         for r in db.Repository.get_repo_forks(repo_id):
             if not HasRepoPermissionLevel('read')(r.repo_name, 'get forks check'):
                 continue
             d.append(r)
         c.forks_pager = Page(d, page=p, items_per_page=20)
         if request.environ.get('HTTP_X_PARTIAL_XHR'):
             return render('/forks/forks_data.html')
         return render('/forks/forks.html')
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionLevelDecorator('read')
     def fork(self, repo_name):
         c.repo_info = db.Repository.get_by_repo_name(repo_name)
         if not c.repo_info:
             raise HTTPNotFound()
         defaults = self.__load_data()
         return htmlfill.render(
             render('forks/fork.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
     @HasRepoPermissionLevelDecorator('read')
     def fork_create(self, repo_name):
         self.__load_defaults()
         c.repo_info = db.Repository.get_by_repo_name(repo_name)
         _form = RepoForkForm(old_data={'repo_type': c.repo_info.repo_type},
                              repo_groups=c.repo_groups,
                              landing_revs=c.landing_revs_choices)()
         form_result = {}
         task_id = None
         try:
             form_result = _form.to_python(dict(request.POST))
             # an approximation that is better than nothing
             if not db.Ui.get_by_key('hooks', db.Ui.HOOK_UPDATE).ui_active:
                 form_result['update_after_clone'] = False
             # create fork is done sometimes async on celery, db transaction
             # management is handled there.
             task = RepoModel().create_fork(form_result, request.authuser.user_id)
             task_id = task.task_id
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('forks/fork.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
-            h.flash(_('An error occurred during repository forking %s') %
+            webutils.flash(_('An error occurred during repository forking %s') %
                     repo_name, category='error')
         raise HTTPFound(location=webutils.url('repo_creating_home',
                               repo_name=form_result['repo_name_full'],
                               task_id=task_id))

kallithea/controllers/login.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.login
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Login controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 22, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import re
 import formencode
 from formencode import htmlfill
 from tg import request, session
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest, HTTPFound
-import kallithea.lib.helpers as h
+from kallithea.lib import webutils
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator
 from kallithea.lib.base import BaseController, log_in_user, render
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta
 from kallithea.model.forms import LoginForm, PasswordResetConfirmationForm, PasswordResetRequestForm, RegisterForm
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class LoginController(BaseController):
     def _validate_came_from(self, came_from,
             _re=re.compile(r"/(?!/)[-!#$%&'()*+,./:;=?@_~0-9A-Za-z]*$")):
         """Return True if came_from is valid and can and should be used.
         Determines if a URI reference is valid and relative to the origin;
         or in RFC 3986 terms, whether it matches this production:
           origin-relative-ref = path-absolute [ "?" query ] [ "#" fragment ]
         with the exception that '%' escapes are not validated and '#' is
         allowed inside the fragment part.
         """
         return _re.match(came_from) is not None
     def index(self):
         c.came_from = request.GET.get('came_from', '')
         if c.came_from:
             if not self._validate_came_from(c.came_from):
                 log.error('Invalid came_from (not server-relative): %r', c.came_from)
                 raise HTTPBadRequest()
         else:
             c.came_from = url('home')
         if request.POST:
             # import Login Form validator class
             login_form = LoginForm()()
             try:
                 # login_form will check username/password using ValidAuth and report failure to the user
                 c.form_result = login_form.to_python(dict(request.POST))
                 username = c.form_result['username']
                 user = db.User.get_by_username_or_email(username)
                 assert user is not None  # the same user get just passed in the form validation
             except formencode.Invalid as errors:
                 defaults = errors.value
                 # remove password from filling in form again
                 defaults.pop('password', None)
                 return htmlfill.render(
                     render('/login.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw this exception signaling that there's issue
                 # with user creation, explanation should be provided in
                 # Exception itself
-                h.flash(e, 'error')
+                webutils.flash(e, 'error')
             else:
                 # login_form already validated the password - now set the session cookie accordingly
                 auth_user = log_in_user(user, c.form_result['remember'], is_external_auth=False, ip_addr=request.ip_addr)
                 if auth_user:
                     raise HTTPFound(location=c.came_from)
-                h.flash(_('Authentication failed.'), 'error')
+                webutils.flash(_('Authentication failed.'), 'error')
         else:
             # redirect if already logged in
             if not request.authuser.is_anonymous:
                 raise HTTPFound(location=c.came_from)
             # continue to show login to default user
         return render('/login.html')
     @HasPermissionAnyDecorator('hg.admin', 'hg.register.auto_activate',
                                'hg.register.manual_activate')
     def register(self):
         def_user_perms = AuthUser(dbuser=db.User.get_default_user()).global_permissions
         c.auto_active = 'hg.register.auto_activate' in def_user_perms
         settings = db.Setting.get_app_settings()
         captcha_private_key = settings.get('captcha_private_key')
         c.captcha_active = bool(captcha_private_key)
         c.captcha_public_key = settings.get('captcha_public_key')
         if request.POST:
             register_form = RegisterForm()()
             try:
                 form_result = register_form.to_python(dict(request.POST))
                 form_result['active'] = c.auto_active
                 if c.captcha_active:
                     from kallithea.lib.recaptcha import submit
                     response = submit(request.POST.get('g-recaptcha-response'),
                                       private_key=captcha_private_key,
                                       remoteip=request.ip_addr)
                     if not response.is_valid:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': _msg}
                         raise formencode.Invalid(_msg, _value, None,
                                                  error_dict=error_dict)
                 UserModel().create_registration(form_result)
-                h.flash(_('You have successfully registered with %s') % (c.site_name or 'Kallithea'),
+                webutils.flash(_('You have successfully registered with %s') % (c.site_name or 'Kallithea'),
                         category='success')
                 meta.Session().commit()
                 raise HTTPFound(location=url('login_home'))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('/register.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             except UserCreationError as e:
                 # container auth or other auth functions that create users on
                 # the fly can throw this exception signaling that there's issue
                 # with user creation, explanation should be provided in
                 # Exception itself
-                h.flash(e, 'error')
+                webutils.flash(e, 'error')
         return render('/register.html')
     def password_reset(self):
         settings = db.Setting.get_app_settings()
         captcha_private_key = settings.get('captcha_private_key')
         c.captcha_active = bool(captcha_private_key)
         c.captcha_public_key = settings.get('captcha_public_key')
         if request.POST:
             password_reset_form = PasswordResetRequestForm()()
             try:
                 form_result = password_reset_form.to_python(dict(request.POST))
                 if c.captcha_active:
                     from kallithea.lib.recaptcha import submit
                     response = submit(request.POST.get('g-recaptcha-response'),
                                       private_key=captcha_private_key,
                                       remoteip=request.ip_addr)
                     if not response.is_valid:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': _msg}
                         raise formencode.Invalid(_msg, _value, None,
                                                  error_dict=error_dict)
                 redirect_link = UserModel().send_reset_password_email(form_result)
-                h.flash(_('A password reset confirmation code has been sent'),
+                webutils.flash(_('A password reset confirmation code has been sent'),
                             category='success')
                 raise HTTPFound(location=redirect_link)
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('/password_reset.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
         return render('/password_reset.html')
     def password_reset_confirmation(self):
         # This controller handles both GET and POST requests, though we
         # only ever perform the actual password change on POST (since
         # GET requests are not allowed to have side effects, and do not
         # receive automatic CSRF protection).
         # The template needs the email address outside of the form.
         c.email = request.params.get('email')
         c.timestamp = request.params.get('timestamp') or ''
         c.token = request.params.get('token') or ''
         if not request.POST:
             return render('/password_reset_confirmation.html')
         form = PasswordResetConfirmationForm()()
         try:
             form_result = form.to_python(dict(request.POST))
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding='UTF-8')
         if not UserModel().verify_reset_password_token(
             form_result['email'],
             form_result['timestamp'],
             form_result['token'],
         ):
             return htmlfill.render(
                 render('/password_reset_confirmation.html'),
                 defaults=form_result,
                 errors={'token': _('Invalid password reset token')},
                 prefix_error=False,
                 encoding='UTF-8')
         UserModel().reset_password(form_result['email'], form_result['password'])
-        h.flash(_('Successfully updated password'), category='success')
+        webutils.flash(_('Successfully updated password'), category='success')
         raise HTTPFound(location=url('login_home'))
     def logout(self):
         session.delete()
         log.info('Logging out and deleting session for user')
         raise HTTPFound(location=url('home'))
     def session_csrf_secret_token(self):
         """Return the CSRF protection token for the session - just like it
         could have been screen scraped from a page with a form.
         Only intended for testing but might also be useful for other kinds
         of automation.
         """
-        return h.session_csrf_secret_token()
+        return webutils.session_csrf_secret_token()

Changeset was too big and was cut off... Show full diff anyway

0 comments (0 inline, 0 general)