kallithea Changeset - a36a8804e7be

Changeset - a36a8804e7be

Parent rev.

Child rev.

[Not reviewed]

default

0 12 0

Mads Kiilerich (mads) - 5 years ago 2020-12-18 22:24:48
mads@kiilerich.com

Grafted from: f7c6ee9b26f5

lib: use webutils for special mako templates - helpers is only for controller web templates

12 files changed with 48 insertions and 49 deletions:

kallithea/controllers/admin/repo_groups.py

kallithea/controllers/admin/user_groups.py

kallithea/controllers/admin/users.py

kallithea/model/notification.py

kallithea/model/repo.py

kallithea/templates/data_table/_dt_elements.html

kallithea/templates/email/changeset_comment.html

kallithea/templates/email/changeset_comment.txt

kallithea/templates/email/pull_request.html

kallithea/templates/email/pull_request.txt

kallithea/tests/models/test_notifications.py

scripts/deps.py

0 comments (0 inline, 0 general)

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
 import kallithea.lib.helpers as h
 from kallithea.controllers import base
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoGroupPermissionLevelDecorator, LoginRequired
 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(base.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)
+                .render_unicode(repo_group_name, children_groups, _=_, webutils=webutils, 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,
+                .render_unicode(repo_group_id, repo_group_name, gr_count, _=_, webutils=webutils, 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": webutils.escape(repo_gr.group_name),
                 "group_name": repo_group_name(repo_gr.group_name, children_groups),
                 "desc": webutils.escape(repo_gr.group_description),
                 "repos": repo_count,
                 "owner": repo_gr.owner.username,
                 "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 base.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(
                 base.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())
             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))
         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):
         parent_group_id = safe_int(request.GET.get('parent_group') or '-1')
         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 = db.RepoGroup.get(parent_group_id) if parent_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 htmlfill.render(
             base.render('admin/repo_groups/repo_group_add.html'),
             defaults={'parent_group_id': parent_group_id},
             errors={},
             prefix_error=False,
             encoding="UTF-8",
             force_defaults=False)
     @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()
             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(
                 base.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())
             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:
             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:
             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()
             webutils.flash(_('Removed repository group %s') % group_name,
                     category='success')
             # TODO: in future action_logger(, '', '', '')
         except Exception:
             log.error(traceback.format_exc())
             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 base.render('admin/repo_groups/repo_group_show.html')
     @HasRepoGroupPermissionLevelDecorator('admin')
     def edit(self, group_name):
         c.active = 'settings'

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
 import kallithea.lib.helpers as h
 from kallithea.controllers import base
 from kallithea.lib import webutils
 from kallithea.lib.auth import HasPermissionAnyDecorator, HasUserGroupPermissionLevelDecorator, LoginRequired
 from kallithea.lib.exceptions import RepoGroupAssignmentError, UserGroupsAssignedException
 from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.lib.webutils import url
 from kallithea.model import db, meta, userlog
 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(base.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)
+                .render_unicode(user_group_id, user_group_name, _=_, webutils=webutils, 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)
+                .render_unicode(user_group_id, user_group_name, _=_, webutils=webutils, 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": webutils.escape(user_gr.user_group_description),
                 "members": len(user_gr.members),
                 "active": h.boolicon(user_gr.users_group_active),
                 "owner": 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 base.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']
             userlog.action_logger(request.authuser,
                           'admin_created_users_group:%s' % gr,
                           None, request.ip_addr)
             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(
                 base.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())
             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 base.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']
             userlog.action_logger(request.authuser,
                           'admin_updated_users_group:%s' % gr,
                           None, request.ip_addr)
             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(
                 base.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())
             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()
             webutils.flash(_('Successfully deleted user group'), category='success')
         except UserGroupsAssignedException as e:
             webutils.flash(e, category='error')
         except Exception:
             log.error(traceback.format_exc())
             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(
             base.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(
             base.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:
             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()
         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')

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
 import kallithea.lib.helpers as h
 from kallithea.controllers import base
 from kallithea.lib import auth_modules, webutils
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.exceptions import DefaultUserException, UserCreationError, UserOwnsReposException
 from kallithea.lib.utils2 import datetime_to_time, generate_api_key, safe_int
 from kallithea.lib.webutils import fmt_date, url
 from kallithea.model import db, meta, userlog
 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(base.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)
+                .render_unicode(user_id, username, _=_, webutils=webutils, c=c)
         def user_actions(user_id, username):
             return template.get_def("user_actions") \
-                .render_unicode(user_id, username, _=_, h=h, c=c)
+                .render_unicode(user_id, username, _=_, webutils=webutils, 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": webutils.escape(user.name),
                 "lastname": webutils.escape(user.lastname),
                 "last_login": 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 base.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)
             userlog.action_logger(request.authuser, 'admin_created_user:%s' % user.username,
                           None, request.ip_addr)
             webutils.flash(_('Created user %s') % user.username,
                     category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 base.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:
             webutils.flash(e, 'error')
         except Exception:
             log.error(traceback.format_exc())
             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 base.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']
             userlog.action_logger(request.authuser, 'admin_updated_user:%s' % usr,
                           None, request.ip_addr)
             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())
             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()
             webutils.flash(_('Successfully deleted user'), category='success')
         except (UserOwnsReposException, DefaultUserException) as e:
             webutils.flash(e, category='warning')
         except Exception:
             log.error(traceback.format_exc())
             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:
             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 base.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(
             base.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(
             base.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()
         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)

kallithea/model/notification.py

➞

Show inline comments

@@ @@ -10,223 +10,221 @@ @@
 # 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.model.notification
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Model for notifications
 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 20, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 from tg import app_globals
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from kallithea.lib import webutils
 from kallithea.model import async_tasks, db
 log = logging.getLogger(__name__)
 class NotificationModel(object):
     TYPE_CHANGESET_COMMENT = 'cs_comment'
     TYPE_MESSAGE = 'message'
     TYPE_MENTION = 'mention' # not used
     TYPE_REGISTRATION = 'registration'
     TYPE_PULL_REQUEST = 'pull_request'
     TYPE_PULL_REQUEST_COMMENT = 'pull_request_comment'
     def create(self, created_by, body, recipients=None,
                type_=TYPE_MESSAGE, with_email=True,
                email_kwargs=None, repo_name=None):
         """
         Creates notification of given type
         :param created_by: int, str or User instance. User who created this
             notification
         :param body:
         :param recipients: list of int, str or User objects, when None
             is given send to all admins
         :param type_: type of notification
         :param with_email: send email with this notification
         :param email_kwargs: additional dict to pass as args to email template
         """
         email_kwargs = email_kwargs or {}
         if recipients and not getattr(recipients, '__iter__', False):
             raise Exception('recipients must be a list or iterable')
         created_by_obj = db.User.guess_instance(created_by)
         recipients_objs = set()
         if recipients:
             for u in recipients:
                 obj = db.User.guess_instance(u)
                 if obj is not None:
                     recipients_objs.add(obj)
                 else:
                     # TODO: inform user that requested operation couldn't be completed
                     log.error('cannot email unknown user %r', u)
             log.debug('sending notifications %s to %s',
                 type_, recipients_objs
+            )
         elif recipients is None:
             # empty recipients means to all admins
             recipients_objs = db.User.query().filter(db.User.admin == True).all()
             log.debug('sending notifications %s to admins: %s',
                 type_, recipients_objs
+            )
         #else: silently skip notification mails?
         if not with_email:
             return
         headers = {}
         headers['X-Kallithea-Notification-Type'] = type_
         if 'threading' in email_kwargs:
             headers['References'] = ' '.join('<%s>' % x for x in email_kwargs['threading'])
         # this is passed into template
         created_on = webutils.fmt_date(datetime.datetime.now())
         html_kwargs = {
                   'body': None if body is None else webutils.render_w_mentions(body, repo_name),
                   'when': created_on,
                   'user': created_by_obj.username,
+                  }
         txt_kwargs = {
                   'body': body,
                   'when': created_on,
                   'user': created_by_obj.username,
+                  }
         html_kwargs.update(email_kwargs)
         txt_kwargs.update(email_kwargs)
         email_subject = EmailNotificationModel() \
                             .get_email_description(type_, **txt_kwargs)
         email_txt_body = EmailNotificationModel() \
                             .get_email_tmpl(type_, 'txt', **txt_kwargs)
         email_html_body = EmailNotificationModel() \
                             .get_email_tmpl(type_, 'html', **html_kwargs)
         # don't send email to the person who caused the notification, except for
         # notifications about new pull requests where the author is explicitly
         # added.
         rec_mails = set(obj.email for obj in recipients_objs)
         if type_ == NotificationModel.TYPE_PULL_REQUEST:
             rec_mails.add(created_by_obj.email)
         else:
             rec_mails.discard(created_by_obj.email)
         # send email with notification to participants
         for rec_mail in sorted(rec_mails):
             async_tasks.send_email([rec_mail], email_subject, email_txt_body,
                      email_html_body, headers,
                      from_name=created_by_obj.full_name_or_username)
 class EmailNotificationModel(object):
     TYPE_CHANGESET_COMMENT = NotificationModel.TYPE_CHANGESET_COMMENT
     TYPE_MESSAGE = NotificationModel.TYPE_MESSAGE # only used for testing
     # NotificationModel.TYPE_MENTION is not used
     TYPE_PASSWORD_RESET = 'password_link'
     TYPE_REGISTRATION = NotificationModel.TYPE_REGISTRATION
     TYPE_PULL_REQUEST = NotificationModel.TYPE_PULL_REQUEST
     TYPE_PULL_REQUEST_COMMENT = NotificationModel.TYPE_PULL_REQUEST_COMMENT
     TYPE_DEFAULT = 'default'
     def __init__(self):
         super(EmailNotificationModel, self).__init__()
         self._tmpl_lookup = app_globals.mako_lookup
         self.email_types = {
             self.TYPE_CHANGESET_COMMENT: 'changeset_comment',
             self.TYPE_PASSWORD_RESET: 'password_reset',
             self.TYPE_REGISTRATION: 'registration',
             self.TYPE_DEFAULT: 'default',
             self.TYPE_PULL_REQUEST: 'pull_request',
             self.TYPE_PULL_REQUEST_COMMENT: 'pull_request_comment',
+        }
         self._subj_map = {
             self.TYPE_CHANGESET_COMMENT: _('[Comment] %(repo_name)s changeset %(short_id)s "%(message_short)s" on %(branch)s by %(cs_author_username)s'),
             self.TYPE_MESSAGE: 'Test Message',
             # self.TYPE_PASSWORD_RESET
             self.TYPE_REGISTRATION: _('New user %(new_username)s registered'),
             # self.TYPE_DEFAULT
             self.TYPE_PULL_REQUEST: _('[Review] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),
             self.TYPE_PULL_REQUEST_COMMENT: _('[Comment] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),
+        }
     def get_email_description(self, type_, **kwargs):
         """
         return subject for email based on given type
         """
         tmpl = self._subj_map[type_]
         try:
             subj = tmpl % kwargs
         except KeyError as e:
             log.error('error generating email subject for %r from %s: %s', type_, ', '.join(self._subj_map), e)
             raise
         # gmail doesn't do proper threading but will ignore leading square
         # bracket content ... so that is where we put status info
         bracket_tags = []
         status_change = kwargs.get('status_change')
         if status_change:
             bracket_tags.append(str(status_change))  # apply str to evaluate LazyString before .join
         if kwargs.get('closing_pr'):
             bracket_tags.append(_('Closing'))
         if bracket_tags:
             if subj.startswith('['):
                 subj = '[' + ', '.join(bracket_tags) + ': ' + subj[1:]
             else:
                 subj = '[' + ', '.join(bracket_tags) + '] ' + subj
         return subj
     def get_email_tmpl(self, type_, content_type, **kwargs):
         """
         return generated template for email based on given type
         """
         import kallithea.lib.helpers as h
         base = 'email/' + self.email_types.get(type_, self.email_types[self.TYPE_DEFAULT]) + '.' + content_type
         email_template = self._tmpl_lookup.get_template(base)
         # translator and helpers inject
         _kwargs = {'_': _,
-                   'h': h,
+                   'webutils': webutils,
                    'c': c}
         _kwargs.update(kwargs)
         if content_type == 'html':
             _kwargs.update({
                 "color_text": "#202020",
                 "color_emph": "#395fa0",
                 "color_link": "#395fa0",
                 "color_border": "#ddd",
                 "color_background_grey": "#f9f9f9",
                 "color_button": "#395fa0",
                 "monospace_style": "font-family:Lucida Console,Consolas,Monaco,Inconsolata,Liberation Mono,monospace",
                 "sans_style": "font-family:Helvetica,Arial,sans-serif",
                 })
             _kwargs.update({
                 "default_style": "%(sans_style)s;font-weight:200;font-size:14px;line-height:17px;color:%(color_text)s" % _kwargs,
                 "comment_style": "%(monospace_style)s;white-space:pre-wrap" % _kwargs,
                 "data_style": "border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,
                 "emph_style": "font-weight:600;color:%(color_emph)s" % _kwargs,
                 "link_style": "color:%(color_link)s;text-decoration:none" % _kwargs,
                 "link_text_style": "color:%(color_text)s;text-decoration:none;border:%(color_border)s 1px solid;background:%(color_background_grey)s" % _kwargs,
                 })
         log.debug('rendering tmpl %s with kwargs %s', base, _kwargs)
         return email_template.render_unicode(**_kwargs)

kallithea/model/repo.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.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model 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: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import shutil
 import traceback
 from datetime import datetime
 import kallithea.lib.utils2
 from kallithea.lib import hooks, webutils
 from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.lib.utils import is_valid_repo_uri, make_ui
 from kallithea.lib.utils2 import LazyProperty, get_current_authuser, obfuscate_url_pw, remove_prefix
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.model import db, meta, scm
 log = logging.getLogger(__name__)
 class RepoModel(object):
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = db.User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = db.UserRepoToPerm()
         repo_to_perm.permission = db.Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         meta.Session().add(repo_to_perm)
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = db.Ui.query().filter(db.Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_id == repo_id)
         return repo.scalar()
     def get_repo(self, repository):
         return db.Repository.guess_instance(repository)
     def get_by_repo_name(self, repo_name):
         repo = db.Repository.query() \
             .filter(db.Repository.repo_name == repo_name)
         return repo.scalar()
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         from tg import app_globals, request
         from tg import tmpl_context as c
         from tg.i18n import ugettext as _
         import kallithea.lib.helpers as h
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render_unicode(*args, **kwargs)
         return tmpl.render_unicode(
             *args,
             _=_,
             webutils=webutils,
             c=c,
             request=request,
             **kwargs)
     def get_repos_as_dict(self, repos_list, repo_groups_list=None,
                           admin=False,
                           short_name=False):
         """Return repository list for use by DataTable.
         repos_list: list of repositories - but will be filtered for read permission.
         repo_groups_list: added at top of list without permission check.
         admin: return data for action column.
         """
         _render = self._render_datatable
         from tg import request
         from tg import tmpl_context as c
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name)
         def following(repo_id, is_following):
             return _render('following', repo_id, is_following)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return webutils.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name)
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
         repos_data = []
         for gr in repo_groups_list or []:
             repos_data.append(dict(
                 raw_name='\0' + webutils.html_escape(gr.name),  # sort before repositories
                 just_name=webutils.html_escape(gr.name),
                 name=_render('group_name_html', group_name=gr.group_name, name=gr.name),
                 desc=desc(gr.group_description)))
         for repo in repos_list:
             if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):
                 continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": webutils.html_escape(repo.repo_name),
                 "just_name": webutils.html_escape(repo.just_name),
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "following": following(
                     repo.repo_id,
                     scm.ScmModel().is_following_repo(repo.repo_name, request.authuser.user_id),
                 ),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": repo.owner.username,
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id, repo.owner.username)
                 })
             repos_data.append(row)
         return {
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = db.Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None
         defaults = repo_info.get_dict()
         defaults['repo_name'] = repo_info.just_name
         defaults['repo_group'] = repo_info.group_id
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
         # fill owner
         if repo_info.owner:
             defaults.update({'owner': repo_info.owner.username})
         else:
             replacement_user = db.User.query().filter(db.User.admin ==
                                                    True).first().username
             defaults.update({'owner': replacement_user})
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
         return defaults
     def update(self, repo, **kwargs):
         try:
             cur_repo = db.Repository.guess_instance(repo)
             org_repo_name = cur_repo.repo_name
             if 'owner' in kwargs:
                 cur_repo.owner = db.User.get_by_username(kwargs['owner'])
             if 'repo_group' in kwargs:
                 assert kwargs['repo_group'] != '-1', kwargs # RepoForm should have converted to None
                 cur_repo.group = db.RepoGroup.get(kwargs['repo_group'])
                 cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 # clone_uri is modified - if given a value, check it is valid
                 if clone_uri != '':
                     # will raise exception on error
                     is_valid_repo_uri(cur_repo.repo_type, clone_uri, make_ui())
                 cur_repo.clone_uri = clone_uri
             if 'repo_name' in kwargs:
                 repo_name = kwargs['repo_name']
                 if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                     raise Exception('invalid repo name %s' % repo_name)
                 cur_repo.repo_name = cur_repo.get_new_name(repo_name)
             # if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
+                )
                 # handle extra fields
             for field in [k for k in kwargs if k.startswith(db.RepositoryField.PREFIX)]:
                 k = db.RepositoryField.un_prefix_key(field)
                 ex_field = db.RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
             if org_repo_name != cur_repo.repo_name:
                 # rename repository

kallithea/templates/data_table/_dt_elements.html

➞

Show inline comments

 ## DATA TABLE RE USABLE ELEMENTS
 ## usage:
 ## <%namespace name="dt" file="/data_table/_dt_elements.html"/>
 <%namespace name="base" file="/base/base.html"/>
 <%def name="repo_name(name,rtype,rstate,private,fork_of,short_name=False)">
     <%
     def get_name(name,short_name=short_name):
       if short_name:
         return name.split('/')[-1]
       else:
         return name
     %>
   <div class="dt_repo ${'dt_repo_pending' if rstate == 'repo_state_pending' else ''}">
     ${base.repolabel(rtype)}
-    <a href="${h.url('summary_home', repo_name=name)}">
+    <a href="${webutils.url('summary_home', repo_name=name)}">
         ${get_name(name)}
     </a>
     %if private and c.visual.show_private_icon:
       <i class="icon-lock" title="${_('Private repository')}"></i>
     %elif not private and c.visual.show_public_icon:
       <i class="icon-globe" title="${_('Public repository')}"></i>
     %endif
     %if fork_of:
-      <a href="${h.url('summary_home',repo_name=fork_of.repo_name)}"><i class="icon-fork"></i></a>
+      <a href="${webutils.url('summary_home',repo_name=fork_of.repo_name)}"><i class="icon-fork"></i></a>
     %endif
     %if rstate == 'repo_state_pending':
       <i class="icon-wrench" title="${_('Repository creation in progress...')}"></i>
     %endif
   </div>
 </%def>
 <%def name="following(repo_id, repo_following)">
   %if request.authuser.username != 'default':
     <a href="#" class="${'following' if repo_following else 'follow'}" onclick="return toggleFollowingRepo(this, ${repo_id});"><i class="list-extra icon-heart-empty show-follow" title="${_('Follow')}"></i><i class="list-extra icon-heart show-following" title="${_('Unfollow')}"></i></a>
   %endif
 </%def>
 <%def name="last_change(last_change)">
-  <span data-toggle="tooltip" title="${h.fmt_date(last_change)}" date="${last_change}">${h.age(last_change)}</span>
+  <span data-toggle="tooltip" title="${webutils.fmt_date(last_change)}" date="${last_change}">${webutils.age(last_change)}</span>
 </%def>
 <%def name="revision(name,rev,tip,author,last_msg)">
   %if rev >= 0:
-      <a data-toggle="popover" title="${author | entity}" data-content="${last_msg | entity}" class="changeset_hash" href="${h.url('changeset_home',repo_name=name,revision=tip)}">${'r%s:%s' % (rev,h.short_id(tip))}</a>
+      <a data-toggle="popover" title="${author | entity}" data-content="${last_msg | entity}" class="changeset_hash" href="${webutils.url('changeset_home',repo_name=name,revision=tip)}">${'r%s:%s' % (rev,webutils.short_id(tip))}</a>
   %else:
       ${_('No changesets yet')}
   %endif
 </%def>
 <%def name="rss(name)">
   %if request.authuser.username != 'default':
-    <a title="${_('Subscribe to %s rss feed')% name}" href="${h.url('rss_feed_home',repo_name=name,api_key=request.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
+    <a title="${_('Subscribe to %s rss feed')% name}" href="${webutils.url('rss_feed_home',repo_name=name,api_key=request.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
   %else:
-    <a title="${_('Subscribe to %s rss feed')% name}" href="${h.url('rss_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
+    <a title="${_('Subscribe to %s rss feed')% name}" href="${webutils.url('rss_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
   %endif
 </%def>
 <%def name="atom(name)">
   %if request.authuser.username != 'default':
-    <a title="${_('Subscribe to %s atom feed')% name}" href="${h.url('atom_feed_home',repo_name=name,api_key=request.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
+    <a title="${_('Subscribe to %s atom feed')% name}" href="${webutils.url('atom_feed_home',repo_name=name,api_key=request.authuser.api_key)}"><i class="icon-rss-squared"></i></a>
   %else:
-    <a title="${_('Subscribe to %s atom feed')% name}" href="${h.url('atom_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
+    <a title="${_('Subscribe to %s atom feed')% name}" href="${webutils.url('atom_feed_home',repo_name=name)}"><i class="icon-rss-squared"></i></a>
   %endif
 </%def>
 <%def name="repo_actions(repo_name)">
-      <a href="${h.url('edit_repo',repo_name=repo_name)}" title="${_('Edit')}" class="btn btn-default btn-xs">
+      <a href="${webutils.url('edit_repo',repo_name=repo_name)}" title="${_('Edit')}" class="btn btn-default btn-xs">
         <i class="icon-pencil"></i>${_('Edit')}
       </a>
-      ${h.form(h.url('delete_repo', repo_name=repo_name))}
+      ${webutils.form(webutils.url('delete_repo', repo_name=repo_name))}
         <button name="${'remove_%s' % repo_name}" class="btn btn-default btn-xs"
             onclick="return confirm('${_('Confirm to delete this repository: %s') % repo_name}');">
           <i class="icon-trashcan"></i>${_('Delete')}
         </button>
-      ${h.end_form()}
+      ${webutils.end_form()}
 </%def>
 <%def name="repo_state(repo_state)">
     %if repo_state == u'repo_state_pending':
         <div class="label label-info">${_('Creating')}</div>
     %elif repo_state == u'repo_state_created':
         <div class="label label-success">${_('Created')}</div>
     %else:
         <div class="label label-danger" title="${repo_state}">invalid</div>
     %endif
 </%def>
 <%def name="user_actions(user_id, username)">
-   <a href="${h.url('edit_user',id=user_id)}" title="${_('Edit')}" class="btn btn-default btn-xs">
+   <a href="${webutils.url('edit_user',id=user_id)}" title="${_('Edit')}" class="btn btn-default btn-xs">
      <i class="icon-pencil"></i>${_('Edit')}
    </a>
-  ${h.form(h.url('delete_user', id=user_id))}
+  ${webutils.form(webutils.url('delete_user', id=user_id))}
     <button id="${'remove_user_%s' % user_id}" name="${'remove_user_%s' % repo_name}" class="btn btn-default btn-xs" title="${_('Delete')}"
         onclick="return confirm('${_('Confirm to delete this user: %s') % username}');">
       <i class="icon-trashcan"></i>${_('Delete')}
     </button>
-  ${h.end_form()}
+  ${webutils.end_form()}
 </%def>
 <%def name="user_group_actions(user_group_id, user_group_name)">
-    <a href="${h.url('edit_users_group', id=user_group_id)}" title="${_('Edit')}" class="btn btn-default btn-xs">
+    <a href="${webutils.url('edit_users_group', id=user_group_id)}" title="${_('Edit')}" class="btn btn-default btn-xs">
       <i class="icon-pencil"></i>${_('Edit')}
     </a>
-    ${h.form(h.url('delete_users_group', id=user_group_id))}
+    ${webutils.form(webutils.url('delete_users_group', id=user_group_id))}
       <button id="${'remove_group_%s' % user_group_id}" name="${'remove_user_%s' % repo_name}" class="btn btn-default btn-xs" title="${_('Delete')}"
           onclick="return confirm('${_('Confirm to delete this user group: %s') % user_group_name}');">
         <i class="icon-trashcan"></i>${_('Delete')}
       </button>
-    ${h.end_form()}
+    ${webutils.end_form()}
 </%def>
 <%def name="group_name_html(group_name,name)">
   <div class="dt_repo">
     <i class="icon-folder"></i>
-    <a href="${h.url('repos_group_home',group_name=group_name)}">${name}</a>
+    <a href="${webutils.url('repos_group_home',group_name=group_name)}">${name}</a>
   </div>
 </%def>
 <%def name="repo_group_actions(repo_group_id, repo_group_name, gr_count)">
-    <a href="${h.url('edit_repo_group',group_name=repo_group_name)}" title="${_('Edit')}" class="btn btn-default btn-xs">
+    <a href="${webutils.url('edit_repo_group',group_name=repo_group_name)}" title="${_('Edit')}" class="btn btn-default btn-xs">
       <i class="icon-pencil"></i>${_('Edit')}
     </a>
-    ${h.form(h.url('delete_repo_group', group_name=repo_group_name))}
+    ${webutils.form(webutils.url('delete_repo_group', group_name=repo_group_name))}
       <button id="${'remove_%s' % repo_group_name}" name="${'remove_%s' % repo_group_name}" class="btn btn-default btn-xs" title="${_('Delete')}"
           onclick="return confirm('${ungettext('Confirm to delete this group: %s with %s repository','Confirm to delete this group: %s with %s repositories',gr_count) % (repo_group_name, gr_count)}')">
         <i class="icon-trashcan"></i>${_('Delete')}
       </button>
-    ${h.end_form()}
+    ${webutils.end_form()}
 </%def>
 <%def name="user_name(user_id, username)">
-    ${h.link_to(username,h.url('edit_user', id=user_id))}
+    ${webutils.link_to(username,webutils.url('edit_user', id=user_id))}
 </%def>
 <%def name="repo_group_name(repo_group_name, children_groups)">
   <div class="text-nowrap">
   <a href="${h.url('repos_group_home',group_name=repo_group_name)}">
     <i class="icon-folder" title="${_('Repository group')}"></i>${h.literal(' &raquo; ').join(children_groups)}</a>
   <a href="${webutils.url('repos_group_home',group_name=repo_group_name)}">
     <i class="icon-folder" title="${_('Repository group')}"></i>${webutils.literal(' &raquo; ').join(children_groups)}</a>
   </div>
 </%def>
 <%def name="user_group_name(user_group_id, user_group_name)">
   <div class="text-nowrap">
-  <a href="${h.url('edit_users_group', id=user_group_id)}">
+  <a href="${webutils.url('edit_users_group', id=user_group_id)}">
     <i class="icon-users" title="${_('User group')}"></i>${user_group_name}</a>
   </div>
 </%def>

kallithea/templates/email/changeset_comment.html

➞

Show inline comments

 <%inherit file="main.html"/>\
+\
 <%block name="header">\
-<% title = _('Mention in Comment on Changeset "%s"') % h.shorter(message, 200, firstline=True) if is_mention else _('Comment on Changeset "%s"') % h.shorter(message, 200, firstline=True) %>\
+<% title = _('Mention in Comment on Changeset "%s"') % webutils.shorter(message, 200, firstline=True) if is_mention else _('Comment on Changeset "%s"') % webutils.shorter(message, 200, firstline=True) %>\
 <%include file="header.html" args="title=title,link=cs_comment_url"/>\
 </%block>\
+\
 <table cellpadding="0" cellspacing="0" border="0" width="100%">
     <tr>
         <td>
 <%include file="comment.html" args="text=body,author=cs_comment_user,status=status_change"/>\
         </td>
     </tr>
     <tr>
         <td height="30px" style="height:30px"></td>
     </tr>
     <tr>
         <td>
             <div>
                 ${_('Changeset on')}
                 <a style="${link_text_style}"
                    href="${cs_target_repo}">${cs_target_repo}</a>
                 ${_('branch')}
                 <span style="${data_style}">${branch}</span>:
             </div>
             <div>
                 "<a style="${link_style}"
-                   href="${cs_url}">${h.shorter(message, 60, firstline=True)}</a>"
+                   href="${cs_url}">${webutils.shorter(message, 60, firstline=True)}</a>"
                 ${_('by')}
                 <span style="${data_style}">${cs_author.full_name_and_username}</span>.
             </div>
         </td>
     </tr>
     <tr>
         <td>
 <% title = _('View Comment') %>\
 <%include file="button.html" args="url=cs_comment_url,title=title,padding_bottom=False"/>\
         </td>
     </tr>
 </table>

kallithea/templates/email/changeset_comment.txt

➞

Show inline comments

 <%block name="header">\
-<% title = _('Mention in Comment on Changeset "%s"') % h.shorter(message, 200, firstline=True) if is_mention else _('Comment on Changeset "%s"') % h.shorter(message, 200, firstline=True) %>\
+<% title = _('Mention in Comment on Changeset "%s"') % webutils.shorter(message, 200, firstline=True) if is_mention else _('Comment on Changeset "%s"') % webutils.shorter(message, 200, firstline=True) %>\
 <%include file="header.txt" args="title=title,link=cs_comment_url"/>\
 </%block>\
 <%include file="comment.txt" args="text=body,author=cs_comment_user,status=status_change"/>\
 ${_('Changeset on')|n,unicode} \
 ${cs_target_repo|n,unicode} \
 ${_('branch')|n,unicode} \
 ${branch|n,unicode}:
-"${h.shorter(message, 60, firstline=True)|n,unicode}" \
+"${webutils.shorter(message, 60, firstline=True)|n,unicode}" \
 ${_('by')|n,unicode} \
 ${cs_author.full_name_and_username|n,unicode}.
 <% title = _('View Comment') %>\
 <%include file="button.txt" args="url=cs_comment_url,title=title"/>\

kallithea/templates/email/pull_request.html

➞

Show inline comments

 <%inherit file="main.html"/>\
+\
 <%block name="header">\
 <% title = _('Mention on Pull Request %s "%s" by %s') % (pr_nice_id, pr_title, pr_user_created) if is_mention else _('Added as Reviewer of Pull Request %s "%s" by %s') % (pr_nice_id, pr_title, pr_user_created) %>\
 <%include file="header.html" args="title=title,link=pr_url"/>\
 </%block>\
+\
 <table cellpadding="0" cellspacing="0" border="0" width="100%">
     <tr>
         <td>
             <div>
                 ${_('Pull request')}
                 <a style="${link_style}"
                    href="${pr_url}">${pr_nice_id} "${pr_title}"</a>
                 ${_('by')}
                 <span style="${data_style}">${pr_owner.full_name_and_username}</span>.
             </div>
             <div>
                 ${_('from')}
                 <a style="${link_text_style}"
                    href="${pr_source_repo}">${pr_source_repo}</a>
                 ${_('branch')}
                 <span style="${data_style}">${pr_source_branch}</span>
                 <br/>
                 ${_('to')}
                 <a style="${link_text_style}"
                    href="${pr_target_repo}">${pr_target_repo}</a>
                 ${_('branch')}
                 <span style="${data_style}">${pr_target_branch}</span>
             </div>
         </td>
     </tr>
     <tr><td height="10px" style="height:10px"></td></tr>
     <tr>
         <td>
             <div>
                 ${_('Description')}:
             </div>
         </td>
     </tr>
     <tr><td height="10px" style="height:10px"></td></tr>
     <tr>
         <td>
             <table cellpadding="0" cellspacing="0" width="100%" border="0" bgcolor="${color_background_grey}" style="border:1px solid ${color_border};border-radius:4px">
                 <tr>
                     <td height="10px" style="height:10px" colspan="3"></td>
                 </tr>
                 <tr>
                     <td width="20px" style="width:20px"></td>
                     <td>
                         <div style="${comment_style}">${body}</div>
                     </td>
                     <td width="20px" style="width:20px"></td>
                 </tr>
                 <tr>
                     <td height="10px" style="height:10px" colspan="3"></td>
                 </tr>
             </table>
         </td>
     </tr>
     <tr><td height="15px" style="height:15px"></td></tr>
     <tr>
         <td>
             <div>${_('Changesets')}:</div>
         </td>
     </tr>
     <tr><td height="10px" style="height:10px"></td></tr>
     <tr>
         <td style="${sans_style}">
             <ul style="color:${color_link};padding-left:15px;margin:0">
                 %for revision, desc in pr_revisions:
                     <li style="mso-special-format:bullet">
                         <a style="${link_style}"
                            href="${h.canonical_url('changeset_home', repo_name=org_repo_name, revision=revision)}">
                             ${h.shorter(desc, 80, firstline=True)}
                            href="${webutils.canonical_url('changeset_home', repo_name=org_repo_name, revision=revision)}">
                             ${webutils.shorter(desc, 80, firstline=True)}
                         </a>
                     </li>
                 %endfor
             </ul>
         </td>
     </tr>
     <tr>
         <td>
 <% title = _('View Pull Request') %>\
 <%include file="button.html" args="url=pr_url,title=title,padding_bottom=False"/>\
         </td>
     </tr>
 </table>

kallithea/templates/email/pull_request.txt

➞

Show inline comments

 <%block name="header">\
 <% title = _('Mention on Pull Request %s "%s" by %s') % (pr_nice_id, pr_title, pr_user_created) if is_mention else _('Added as Reviewer of Pull Request %s "%s" by %s') % (pr_nice_id, pr_title, pr_user_created) %>\
 <%include file="header.txt" args="title=title,link=pr_url"/>\
 </%block>\
 ${_('Pull request')|n,unicode} \
 ${pr_nice_id|n,unicode} \
 "${pr_title|n,unicode}" \
 ${_('by')|n,unicode} \
 ${pr_owner.full_name_and_username|n,unicode}
 ${_('from')} \
 ${pr_source_repo|n,unicode} \
 ${_('branch')|n,unicode} \
 ${pr_source_branch|n,unicode}
 ${_('to')|n,unicode} \
 ${pr_target_repo|n,unicode} \
 ${_('branch')|n,unicode} \
 ${pr_target_branch|n,unicode}
 ${_('Description')|n,unicode}:
 ${body|n,unicode}
 ${_('Changesets')|n,unicode}:
 %for revision, desc in pr_revisions:
-${h.shorter(desc, 80, firstline=True)|n,unicode}
+${webutils.shorter(desc, 80, firstline=True)|n,unicode}
 %endfor
 <% title = _('View Pull Request') %>\
 <%include file="button.txt" args="url=pr_url,title=title"/>\

kallithea/tests/models/test_notifications.py

➞

Show inline comments

 import os
 import re
 import mock
 from tg.util.webtest import test_context
 import kallithea.lib.celerylib
 import kallithea.lib.helpers as h
 import kallithea.model.async_tasks
 from kallithea.lib import webutils
 from kallithea.model import db, meta
 from kallithea.model.notification import EmailNotificationModel, NotificationModel
 from kallithea.model.user import UserModel
 from kallithea.tests import base
 class TestNotifications(base.TestController):
     def setup_method(self, method):
         meta.Session.remove()
         u1 = UserModel().create_or_update(username='u1',
                                         password='qweqwe',
                                         email='u1@example.com',
                                         firstname='u1', lastname='u1')
         meta.Session().commit()
         self.u1 = u1.user_id
         u2 = UserModel().create_or_update(username='u2',
                                         password='qweqwe',
                                         email='u2@example.com',
                                         firstname='u2', lastname='u3')
         meta.Session().commit()
         self.u2 = u2.user_id
         u3 = UserModel().create_or_update(username='u3',
                                         password='qweqwe',
                                         email='u3@example.com',
                                         firstname='u3', lastname='u3')
         meta.Session().commit()
         self.u3 = u3.user_id
     def test_create_notification(self):
         with test_context(self.app):
             usrs = [self.u1, self.u2]
             def send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):
                 assert recipients == ['u2@example.com']
                 assert subject == 'Test Message'
                 assert body == "hi there"
                 assert '>hi there<' in html_body
                 assert from_name == 'u1 u1'
             with mock.patch.object(kallithea.model.async_tasks, 'send_email', send_email):
                 NotificationModel().create(created_by=self.u1,
                                                    body='hi there',
                                                    recipients=usrs)
-    @mock.patch.object(h, 'canonical_url', (lambda arg, **kwargs: 'http://%s/?%s' % (arg, '&'.join('%s=%s' % (k, v) for (k, v) in sorted(kwargs.items())))))
+    @mock.patch.object(webutils, 'canonical_url', (lambda arg, **kwargs: 'http://%s/?%s' % (arg, '&'.join('%s=%s' % (k, v) for (k, v) in sorted(kwargs.items())))))
     def test_dump_html_mails(self):
         # Exercise all notification types and dump them to one big html file
         l = []
         def send_email(recipients, subject, body='', html_body='', headers=None, from_name=None):
             l.append('<hr/>\n')
             l.append('<h1>%s</h1>\n' % desc) # desc is from outer scope
             l.append('<pre>\n')
             l.append('From: %s <name@example.com>\n' % from_name)
             l.append('To: %s\n' % ' '.join(recipients))
             l.append('Subject: %s\n' % subject)
             l.append('</pre>\n')
             l.append('<hr/>\n')
             l.append('<pre>%s</pre>\n' % body)
             l.append('<hr/>\n')
             l.append(html_body)
             l.append('<hr/>\n')
         with test_context(self.app):
             with mock.patch.object(kallithea.model.async_tasks, 'send_email', send_email):
                 pr_kwargs = dict(
                     pr_nice_id='#7',
                     pr_title='The Title',
                     pr_title_short='The Title',
                     pr_url='http://pr.org/7',
                     pr_target_repo='http://mainline.com/repo',
                     pr_target_branch='trunk',
                     pr_source_repo='https://dev.org/repo',
                     pr_source_branch='devbranch',
                     pr_owner=db.User.get(self.u2),
                     pr_owner_username='u2'
+                    )
                 for type_, body, kwargs in [
                         (NotificationModel.TYPE_CHANGESET_COMMENT,
                          'This is the new \'comment\'.\n\n - and here it ends indented.',
                          dict(
                             short_id='cafe1234',
                             raw_id='cafe1234c0ffeecafe',
                             branch='brunch',
                             cs_comment_user='Opinionated User (jsmith)',
                             cs_comment_url='http://comment.org',
                             is_mention=[False, True],
                             message='This changeset did something clever which is hard to explain',
                             message_short='This changeset did something cl...',
                             status_change=[None, 'Approved'],
                             cs_target_repo='http://example.com/repo_target',
                             cs_url='http://changeset.com',
                             cs_author_username=db.User.get(self.u2).username,
                             cs_author=db.User.get(self.u2))),
                         (NotificationModel.TYPE_MESSAGE,
                          'This is the \'body\' of the "test" message\n - nothing interesting here except indentation.',
                          dict()),
                         #(NotificationModel.TYPE_MENTION, '$body', None), # not used
                         (NotificationModel.TYPE_REGISTRATION,
                          'Registration body',
                          dict(
                             new_username='newbie',
                             registered_user_url='http://newbie.org',
                             new_email='new@email.com',
                             new_full_name='New Full Name')),
                         (NotificationModel.TYPE_PULL_REQUEST,
                          'This PR is \'awesome\' because it does <stuff>\n - please approve indented!',
                          dict(
                             pr_user_created='Requesting User (root)', # pr_owner should perhaps be used for @mention in description ...
                             is_mention=[False, True],
                             pr_revisions=[('123abc'*7, "Introduce one and two\n\nand that's it"), ('567fed'*7, 'Make one plus two equal tree')],
                             org_repo_name='repo_org',
                             **pr_kwargs)),
                         (NotificationModel.TYPE_PULL_REQUEST_COMMENT,
                          'Me too!\n\n - and indented on second line',
                          dict(
                             closing_pr=[False, True],
                             is_mention=[False, True],
                             pr_comment_user='Opinionated User (jsmith)',
                             pr_comment_url='http://pr.org/comment',
                             status_change=[None, 'Under Review'],
                             **pr_kwargs)),
                         ]:
                     kwargs['repo_name'] = 'repo/name'
                     params = [(type_, type_, body, kwargs)]
                     for param_name in ['is_mention', 'status_change', 'closing_pr']: # TODO: inline/general
                         if not isinstance(kwargs.get(param_name), list):
                             continue
                         new_params = []
                         for v in kwargs[param_name]:
                             for desc, type_, body, kwargs in params:
                                 kwargs = dict(kwargs)
                                 kwargs[param_name] = v
                                 new_params.append(('%s, %s=%r' % (desc, param_name, v), type_, body, kwargs))
                         params = new_params
                     for desc, type_, body, kwargs in params:
                         # desc is used as "global" variable
                         NotificationModel().create(created_by=self.u1,
                                                            body=body, email_kwargs=kwargs,
                                                            recipients=[self.u2], type_=type_)
                 # Email type TYPE_PASSWORD_RESET has no corresponding notification type - test it directly:
                 desc = 'TYPE_PASSWORD_RESET'
                 kwargs = dict(user='John Doe', reset_token='decbf64715098db5b0bd23eab44bd792670ab746', reset_url='http://reset.com/decbf64715098db5b0bd23eab44bd792670ab746')
                 kallithea.model.async_tasks.send_email(['john@doe.com'],
                     "Password reset link",
                     EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'txt', **kwargs),
                     EmailNotificationModel().get_email_tmpl(EmailNotificationModel.TYPE_PASSWORD_RESET, 'html', **kwargs),
                     from_name=db.User.get(self.u1).full_name_or_username)
         out = '<!doctype html>\n<html lang="en">\n<head><title>Notifications</title><meta http-equiv="Content-Type" content="text/html; charset=UTF-8"></head>\n<body>\n%s\n</body>\n</html>\n' % \
             re.sub(r'<(/?(?:!doctype|html|head|title|meta|body)\b[^>]*)>', r'<!--\1-->', ''.join(l))
         outfn = os.path.join(os.path.dirname(__file__), 'test_dump_html_mails.out.html')
         reffn = os.path.join(os.path.dirname(__file__), 'test_dump_html_mails.ref.html')
         with open(outfn, 'w') as f:
             f.write(out)
         with open(reffn) as f:
             ref = f.read()
         assert ref == out # copy test_dump_html_mails.out.html to test_dump_html_mails.ref.html to update expectations
         os.unlink(outfn)

scripts/deps.py

➞

Show inline comments

 #!/usr/bin/env python3
 import re
 import sys
 ignored_modules = set('''
 argparse
 base64
 bcrypt
 binascii
 bleach
 calendar
 celery
 celery
 chardet
 click
 collections
 configparser
 copy
 csv
 ctypes
 datetime
 dateutil
 decimal
 decorator
 difflib
 distutils
 docutils
 email
 errno
 fileinput
 functools
 getpass
 grp
 hashlib
 hmac
 html
 http
 imp
 importlib
 inspect
 io
 ipaddr
 IPython
 isapi_wsgi
 itertools
 json
 kajiki
 ldap
 logging
 mako
 markdown
 mimetypes
 mock
 msvcrt
 multiprocessing
 operator
 os
 paginate
 paginate_sqlalchemy
 pam
 paste
 pkg_resources
 platform
 posixpath
 pprint
 pwd
 pyflakes
 pytest
 pytest_localserver
 random
 re
 routes
 setuptools
 shlex
 shutil
 smtplib
 socket
 ssl
 stat
 string
 struct
 subprocess
 sys
 tarfile
 tempfile
 textwrap
 tgext
 threading
 time
 traceback
 traitlets
 types
 urllib
 urlobject
 uuid
 warnings
 webhelpers2
 webob
 webtest
 whoosh
 win32traceutil
 zipfile
 '''.split())
 top_modules = set('''
 kallithea.alembic
 kallithea.bin
 kallithea.config
 kallithea.controllers
 kallithea.templates.py
 scripts
 '''.split())
 bottom_external_modules = set('''
 tg
 mercurial
 sqlalchemy
 alembic
 formencode
 pygments
 dulwich
 beaker
 psycopg2
 docs
 setup
 conftest
 '''.split())
 normal_modules = set('''
 kallithea
 kallithea.controllers.base
 kallithea.lib
 kallithea.lib.auth
 kallithea.lib.auth_modules
 kallithea.lib.celerylib
 kallithea.lib.db_manage
 kallithea.lib.helpers
 kallithea.lib.hooks
 kallithea.lib.indexers
 kallithea.lib.utils
 kallithea.lib.utils2
 kallithea.lib.vcs
 kallithea.lib.webutils
 kallithea.model
 kallithea.model.async_tasks
 kallithea.model.scm
 kallithea.templates.py
 '''.split())
 shown_modules = normal_modules | top_modules
 # break the chains somehow - this is a cleanup TODO list
 known_violations = [
 ('kallithea.lib.auth_modules', 'kallithea.lib.auth'),  # needs base&facade
 ('kallithea.lib.utils', 'kallithea.model'),  # clean up utils
 ('kallithea.lib.utils', 'kallithea.model.db'),
 ('kallithea.lib.utils', 'kallithea.model.scm'),
 ('kallithea.model.async_tasks', 'kallithea.lib.helpers'),
 ('kallithea.model.async_tasks', 'kallithea.lib.hooks'),
 ('kallithea.model.async_tasks', 'kallithea.lib.indexers'),
 ('kallithea.model.async_tasks', 'kallithea.model'),
 ('kallithea.model', 'kallithea.lib.auth'),  # auth.HasXXX
 ('kallithea.model', 'kallithea.lib.auth_modules'),  # validators
 ('kallithea.model', 'kallithea.lib.helpers'),
 ('kallithea.model', 'kallithea.lib.hooks'),  # clean up hooks
 ('kallithea.model', 'kallithea.model.scm'),
 ('kallithea.model.scm', 'kallithea.lib.hooks'),
+]
 extra_edges = [
 ('kallithea.config', 'kallithea.controllers'),  # through TG
 ('kallithea.lib.auth', 'kallithea.lib.auth_modules'),  # custom loader
+]
 def normalize(s):
     """Given a string with dot path, return the string it should be shown as."""
     parts = s.replace('.__init__', '').split('.')
     short_2 = '.'.join(parts[:2])
     short_3 = '.'.join(parts[:3])
     short_4 = '.'.join(parts[:4])
     if parts[0] in ['scripts', 'contributor_data', 'i18n_utils']:
         return 'scripts'
     if short_3 == 'kallithea.model.meta':
         return 'kallithea.model.db'
     if parts[:4] == ['kallithea', 'lib', 'vcs', 'ssh']:
         return 'kallithea.lib.vcs.ssh'
     if short_4 in shown_modules:
         return short_4
     if short_3 in shown_modules:
         return short_3
     if short_2 in shown_modules:
         return short_2
     if short_2 == 'kallithea.tests':
         return None
     if parts[0] in ignored_modules:
         return None
     assert parts[0] in bottom_external_modules, parts
     return parts[0]
 def main(filenames):
     if not filenames or filenames[0].startswith('-'):
         print('''\
 Usage:
     hg files 'set:!binary()&grep("^#!.*python")' 'set:**.py' | xargs scripts/deps.py
     dot -Tsvg deps.dot > deps.svg
         ''')
         raise SystemExit(1)
     files_imports = dict()  # map filenames to its imports
     import_deps = set()  # set of tuples with module name and its imports
     for fn in filenames:
         with open(fn) as f:
             s = f.read()
         dot_name = (fn[:-3] if fn.endswith('.py') else fn).replace('/', '.')
         file_imports = set()
         for m in re.finditer(r'^ *(?:from ([^ ]*) import (?:([a-zA-Z].*)|\(([^)]*)\))|import (.*))$', s, re.MULTILINE):
             m_from, m_from_import, m_from_import2, m_import = m.groups()
             if m_from:
                 pre = m_from + '.'
                 if pre.startswith('.'):
                     pre = dot_name.rsplit('.', 1)[0] + pre
                 importlist = m_from_import or m_from_import2
             else:
                 pre = ''
                 importlist = m_import
             for imp in importlist.split('#', 1)[0].split(','):
                 full_imp = pre + imp.strip().split(' as ', 1)[0]
                 file_imports.add(full_imp)
                 import_deps.add((dot_name, full_imp))
         files_imports[fn] = file_imports
     # dump out all deps for debugging and analysis
     with open('deps.txt', 'w') as f:
         for fn, file_imports in sorted(files_imports.items()):
             for file_import in sorted(file_imports):
                 if file_import.split('.', 1)[0] in ignored_modules:
                     continue
                 f.write('%s: %s\n' % (fn, file_import))
     # find leafs that haven't been ignored - they are the important external dependencies and shown in the bottom row
     only_imported = set(
         set(normalize(b) for a, b in import_deps) -
         set(normalize(a) for a, b in import_deps) -
         set([None, 'kallithea'])
+    )
     normalized_dep_edges = set()
     for dot_name, full_imp in import_deps:
         a = normalize(dot_name)
         b = normalize(full_imp)
         if a is None or b is None or a == b:
             continue
         normalized_dep_edges.add((a, b))
         #print((dot_name, full_imp, a, b))
     normalized_dep_edges.update(extra_edges)
     unseen_shown_modules = shown_modules.difference(a for a, b in normalized_dep_edges).difference(b for a, b in normalized_dep_edges)
     assert not unseen_shown_modules, unseen_shown_modules
     with open('deps.dot', 'w') as f:
         f.write('digraph {\n')
         f.write('subgraph { rank = same; %s}\n' % ''.join('"%s"; ' % s for s in sorted(top_modules)))
         f.write('subgraph { rank = same; %s}\n' % ''.join('"%s"; ' % s for s in sorted(only_imported)))
         for a, b in sorted(normalized_dep_edges):
             f.write('  "%s" -> "%s"%s\n' % (a, b, ' [color=red]' if (a, b) in known_violations else ' [color=green]' if (a, b) in extra_edges else ''))
         f.write('}\n')
     # verify dependencies by untangling dependency chain bottom-up:
     todo = set(normalized_dep_edges)
     for x in known_violations:
         todo.remove(x)
     while todo:
         depending = set(a for a, b in todo)
         depended = set(b for a, b in todo)
         drop = depended - depending
         if not drop:
             print('ERROR: cycles:', len(todo))
             for x in sorted(todo):
                 print('%s,' % (x,))
             raise SystemExit(1)
         #for do_b in sorted(drop):
         #    print('Picking', do_b, '- unblocks:', ' '.join(a for a, b in sorted((todo)) if b == do_b))
         todo = set((a, b) for a, b in todo if b in depending)
         #print()
 if __name__ == '__main__':
     main(sys.argv[1:])

0 comments (0 inline, 0 general)