kallithea Changeset - 08eec03c9485

Changeset - 08eec03c9485

Parent rev.

Child rev.

[Not reviewed]

default

0 26 0

Mads Kiilerich - 6 years ago 2020-01-02 02:22:14
mads@kiilerich.com

Grafted from: b865e216381c

py3: rename all existing safe_unicode to safe_str

18 files changed:

kallithea/controllers/admin/gists.py

kallithea/controllers/admin/settings.py

kallithea/controllers/admin/user_groups.py

kallithea/controllers/changeset.py

kallithea/controllers/feed.py

kallithea/controllers/files.py

kallithea/controllers/summary.py

kallithea/lib/annotate.py

kallithea/lib/base.py

kallithea/lib/diffs.py

kallithea/lib/helpers.py

kallithea/lib/indexers/daemon.py

kallithea/lib/markup_renderer.py

kallithea/lib/middleware/permanent_repo_url.py

kallithea/lib/utils2.py

kallithea/lib/vcs/backends/git/changeset.py

kallithea/lib/vcs/backends/git/inmemory.py

kallithea/lib/vcs/backends/git/repository.py

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

0 comments (0 inline, 0 general)

kallithea/controllers/admin/gists.py

➞

Show inline comments

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

kallithea/controllers/admin/settings.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.settings
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 settings controller for Kallithea admin
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jul 14, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from tg import config, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, render
 from kallithea.lib.celerylib import tasks
 from kallithea.lib.exceptions import HgsubversionImportError
 from kallithea.lib.utils import repo2db_mapper, set_app_settings
-from kallithea.lib.utils2 import safe_unicode
+from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs import VCSError
 from kallithea.model.db import Repository, Setting, Ui
 from kallithea.model.forms import ApplicationSettingsForm, ApplicationUiSettingsForm, ApplicationVisualisationForm
 from kallithea.model.meta import Session
 from kallithea.model.notification import EmailNotificationModel
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 class SettingsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('setting', 'settings', controller='admin/settings',
     #         path_prefix='/admin', name_prefix='admin_')
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(SettingsController, self)._before(*args, **kwargs)
     def _get_hg_ui_settings(self):
         ret = Ui.query().all()
         settings = {}
         for each in ret:
             k = each.ui_section + '_' + each.ui_key
             v = each.ui_value
             if k == 'paths_/':
                 k = 'paths_root_path'
             k = k.replace('.', '_')
             if each.ui_section in ['hooks', 'extensions']:
                 v = each.ui_active
             settings[k] = v
         return settings
     @HasPermissionAnyDecorator('hg.admin')
     def settings_vcs(self):
         c.active = 'vcs'
         if request.POST:
             application_form = ApplicationUiSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                      defaults=errors.value,
                      errors=errors.error_dict or {},
                      prefix_error=False,
                      encoding="UTF-8",
                      force_defaults=False)
             try:
                 if c.visual.allow_repo_location_change:
                     sett = Ui.get_by_key('paths', '/')
                     sett.ui_value = form_result['paths_root_path']
                 # HOOKS
                 sett = Ui.get_by_key('hooks', Ui.HOOK_UPDATE)
                 sett.ui_active = form_result['hooks_changegroup_update']
                 sett = Ui.get_by_key('hooks', Ui.HOOK_REPO_SIZE)
                 sett.ui_active = form_result['hooks_changegroup_repo_size']
                 ## EXTENSIONS
                 sett = Ui.get_or_create('extensions', 'largefiles')
                 sett.ui_active = form_result['extensions_largefiles']
                 sett = Ui.get_or_create('extensions', 'hgsubversion')
                 sett.ui_active = form_result['extensions_hgsubversion']
                 if sett.ui_active:
                     try:
                         import hgsubversion  # pragma: no cover
                     except ImportError:
                         raise HgsubversionImportError
 #                sett = Ui.get_or_create('extensions', 'hggit')
 #                sett.ui_active = form_result['extensions_hggit']
                 Session().commit()
                 h.flash(_('Updated VCS settings'), category='success')
             except HgsubversionImportError:
                 log.error(traceback.format_exc())
                 h.flash(_('Unable to activate hgsubversion support. '
                           'The "hgsubversion" library is missing'),
                         category='error')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'), category='error')
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_mapping(self):
         c.active = 'mapping'
         if request.POST:
             rm_obsolete = request.POST.get('destroy', False)
             install_git_hooks = request.POST.get('hooks', False)
             overwrite_git_hooks = request.POST.get('hooks_overwrite', False)
             invalidate_cache = request.POST.get('invalidate', False)
             log.debug('rescanning repo location with destroy obsolete=%s, '
                       'install git hooks=%s and '
                       'overwrite git hooks=%s' % (rm_obsolete, install_git_hooks, overwrite_git_hooks))
             filesystem_repos = ScmModel().repo_scan()
             added, removed = repo2db_mapper(filesystem_repos, rm_obsolete,
                                             install_git_hooks=install_git_hooks,
                                             user=request.authuser.username,
                                             overwrite_git_hooks=overwrite_git_hooks)
             added_msg = h.HTML(', ').join(
-                h.link_to(safe_unicode(repo_name), h.url('summary_home', repo_name=repo_name)) for repo_name in added
+                h.link_to(safe_str(repo_name), h.url('summary_home', repo_name=repo_name)) for repo_name in added
             ) or '-'
             removed_msg = h.HTML(', ').join(
-                safe_unicode(repo_name) for repo_name in removed
+                safe_str(repo_name) for repo_name in removed
             ) or '-'
             h.flash(h.HTML(_('Repositories successfully rescanned. Added: %s. Removed: %s.')) %
                     (added_msg, removed_msg), category='success')
             if invalidate_cache:
                 log.debug('invalidating all repositories cache')
                 i = 0
                 for repo in Repository.query():
                     try:
                         ScmModel().mark_for_invalidation(repo.repo_name)
                         i += 1
                     except VCSError as e:
                         log.warning('VCS error invalidating %s: %s', repo.repo_name, e)
                 h.flash(_('Invalidated %s repositories') % i, category='success')
             raise HTTPFound(location=url('admin_settings_mapping'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_global(self):
         c.active = 'global'
         if request.POST:
             application_form = ApplicationSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             try:
                 for setting in (
                     'title',
                     'realm',
                     'ga_code',
                     'captcha_public_key',
                     'captcha_private_key',
                 ):
                     Setting.create_or_update(setting, form_result[setting])
                 Session().commit()
                 set_app_settings(config)
                 h.flash(_('Updated application settings'), category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred while updating '
                           'application settings'),
                           category='error')
             raise HTTPFound(location=url('admin_settings_global'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_visual(self):
         c.active = 'visual'
         if request.POST:
             application_form = ApplicationVisualisationForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
             except formencode.Invalid as errors:
                 return htmlfill.render(
                     render('admin/settings/settings.html'),
                     defaults=errors.value,
                     errors=errors.error_dict or {},
                     prefix_error=False,
                     encoding="UTF-8",
                     force_defaults=False)
             try:
                 settings = [
                     ('show_public_icon', 'show_public_icon', 'bool'),
                     ('show_private_icon', 'show_private_icon', 'bool'),
                     ('stylify_metalabels', 'stylify_metalabels', 'bool'),
                     ('repository_fields', 'repository_fields', 'bool'),
                     ('dashboard_items', 'dashboard_items', 'int'),
                     ('admin_grid_items', 'admin_grid_items', 'int'),
                     ('show_version', 'show_version', 'bool'),
                     ('use_gravatar', 'use_gravatar', 'bool'),
                     ('gravatar_url', 'gravatar_url', 'unicode'),
                     ('clone_uri_tmpl', 'clone_uri_tmpl', 'unicode'),
                     ('clone_ssh_tmpl', 'clone_ssh_tmpl', 'unicode'),
+                ]
                 for setting, form_key, type_ in settings:
                     Setting.create_or_update(setting, form_result[form_key], type_)
                 Session().commit()
                 set_app_settings(config)
                 h.flash(_('Updated visualisation settings'),
                         category='success')
             except Exception:
                 log.error(traceback.format_exc())
                 h.flash(_('Error occurred during updating '
                           'visualisation settings'),
                         category='error')
             raise HTTPFound(location=url('admin_settings_visual'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_email(self):
         c.active = 'email'
         if request.POST:
             test_email = request.POST.get('test_email')
             test_email_subj = 'Kallithea test email'
             test_body = ('Kallithea Email test, '
                                'Kallithea version: %s' % c.kallithea_version)
             if not test_email:
                 h.flash(_('Please enter email address'), category='error')
                 raise HTTPFound(location=url('admin_settings_email'))
             test_email_txt_body = EmailNotificationModel() \
                 .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
                                 'txt', body=test_body)
             test_email_html_body = EmailNotificationModel() \
                 .get_email_tmpl(EmailNotificationModel.TYPE_DEFAULT,
                                 'html', body=test_body)
             recipients = [test_email] if test_email else None
             tasks.send_email(recipients, test_email_subj,
                              test_email_txt_body, test_email_html_body)
             h.flash(_('Send email task created'), category='success')
             raise HTTPFound(location=url('admin_settings_email'))
         defaults = Setting.get_app_settings()
         defaults.update(self._get_hg_ui_settings())
         import kallithea
         c.ini = kallithea.CONFIG
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False)
     @HasPermissionAnyDecorator('hg.admin')
     def settings_hooks(self):
         c.active = 'hooks'
         if request.POST:
             if c.visual.allow_custom_hooks_settings:
                 ui_key = request.POST.get('new_hook_ui_key')
                 ui_value = request.POST.get('new_hook_ui_value')
                 hook_id = request.POST.get('hook_id')
                 try:
                     ui_key = ui_key and ui_key.strip()
                     if ui_key in (x.ui_key for x in Ui.get_custom_hooks()):
                         h.flash(_('Hook already exists'), category='error')
                     elif ui_key in (x.ui_key for x in Ui.get_builtin_hooks()):
                         h.flash(_('Builtin hooks are read-only. Please use another hook name.'), category='error')
                     elif ui_value and ui_key:
                         Ui.create_or_update_hook(ui_key, ui_value)
                         h.flash(_('Added new hook'), category='success')
                     elif hook_id:
                         Ui.delete(hook_id)
                         Session().commit()
                     # check for edits
                     update = False

kallithea/controllers/admin/user_groups.py

➞

Show inline comments

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

kallithea/controllers/changeset.py

➞

Show inline comments

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

kallithea/controllers/feed.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.feed
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 Feed controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from beaker.cache import cache_region
 from tg import response
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from kallithea import CONFIG
 from kallithea.lib import feeds
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController
 from kallithea.lib.diffs import DiffProcessor
-from kallithea.lib.utils2 import safe_int, safe_unicode, str2bool
+from kallithea.lib.utils2 import safe_int, safe_str, str2bool
 log = logging.getLogger(__name__)
 class FeedController(BaseRepoController):
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def _before(self, *args, **kwargs):
         super(FeedController, self)._before(*args, **kwargs)
     def _get_title(self, cs):
         return h.shorter(cs.message, 160)
     def __get_desc(self, cs):
         desc_msg = [(_('%s committed on %s')
                      % (h.person(cs.author), h.fmt_date(cs.date))) + '<br/>']
         # branches, tags, bookmarks
         for branch in cs.branches:
             desc_msg.append('branch: %s<br/>' % branch)
         for book in cs.bookmarks:
             desc_msg.append('bookmark: %s<br/>' % book)
         for tag in cs.tags:
             desc_msg.append('tag: %s<br/>' % tag)
         changes = []
         diff_limit = safe_int(CONFIG.get('rss_cut_off_limit', 32 * 1024))
         raw_diff = cs.diff()
         diff_processor = DiffProcessor(raw_diff,
                                        diff_limit=diff_limit,
                                        inline_diff=False)
         for st in diff_processor.parsed:
             st.update({'added': st['stats']['added'],
                        'removed': st['stats']['deleted']})
             changes.append('\n %(operation)s %(filename)s '
                            '(%(added)s lines added, %(removed)s lines removed)'
                             % st)
         if diff_processor.limited_diff:
             changes = changes + ['\n ' +
                                  _('Changeset was too big and was cut off...')]
         # rev link
         _url = h.canonical_url('changeset_home', repo_name=c.db_repo.repo_name,
                    revision=cs.raw_id)
         desc_msg.append('changeset: <a href="%s">%s</a>' % (_url, cs.raw_id[:8]))
         desc_msg.append('<pre>')
         desc_msg.append(h.urlify_text(cs.message))
         desc_msg.append('\n')
         desc_msg.extend(changes)
         if str2bool(CONFIG.get('rss_include_diff', False)):
             desc_msg.append('\n\n')
-            desc_msg.append(safe_unicode(raw_diff))
+            desc_msg.append(safe_str(raw_diff))
         desc_msg.append('</pre>')
         return desc_msg
     def _feed(self, repo_name, feeder):
         """Produce a simple feed"""
         @cache_region('long_term', '_get_feed_from_cache')
         def _get_feed_from_cache(*_cache_keys):  # parameters are not really used - only as caching key
             header = dict(
                 title=_('%s %s feed') % (c.site_name, repo_name),
                 link=h.canonical_url('summary_home', repo_name=repo_name),
                 description=_('Changes on %s repository') % repo_name,
+            )
             rss_items_per_page = safe_int(CONFIG.get('rss_items_per_page', 20))
             entries=[]
             for cs in reversed(list(c.db_repo_scm_instance[-rss_items_per_page:])):
                 entries.append(dict(
                     title=self._get_title(cs),
                     link=h.canonical_url('changeset_home', repo_name=repo_name, revision=cs.raw_id),
                     author_email=cs.author_email,
                     author_name=cs.author_name,
                     description=''.join(self.__get_desc(cs)),
                     pubdate=cs.date,
                 ))
             return feeder.render(header, entries)
         response.content_type = feeder.content_type
         return _get_feed_from_cache(repo_name, feeder.__name__)
     def atom(self, repo_name):
         """Produce a simple atom-1.0 feed"""
         return self._feed(repo_name, feeds.AtomFeed)
     def rss(self, repo_name):
         """Produce a simple rss2 feed"""
         return self._feed(repo_name, feeds.RssFeed)

kallithea/controllers/files.py

➞

Show inline comments

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

kallithea/controllers/summary.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.summary
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Summary controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 18, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import calendar
 import itertools
 import logging
 import traceback
 from datetime import date, timedelta
 from time import mktime
 from beaker.cache import cache_region
 from tg import request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPBadRequest
 import kallithea.lib.helpers as h
 from kallithea.config.conf import ALL_EXTS, ALL_READMES, LANGUAGES_EXTENSIONS_MAP
 from kallithea.lib import ext_json
 from kallithea.lib.auth import HasRepoPermissionLevelDecorator, LoginRequired
 from kallithea.lib.base import BaseRepoController, jsonify, render
 from kallithea.lib.celerylib.tasks import get_commits_stats
 from kallithea.lib.markup_renderer import MarkupRenderer
 from kallithea.lib.page import Page
-from kallithea.lib.utils2 import safe_int, safe_unicode
+from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetError, EmptyRepositoryError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.model.db import Statistics
 log = logging.getLogger(__name__)
 README_FILES = [''.join([x[0][0], x[1][0]]) for x in
                     sorted(list(itertools.product(ALL_READMES, ALL_EXTS)),
                            key=lambda y:y[0][1] + y[1][1])]
 class SummaryController(BaseRepoController):
     def __get_readme_data(self, db_repo):
         repo_name = db_repo.repo_name
         log.debug('Looking for README file')
         @cache_region('long_term', '_get_readme_from_cache')
         def _get_readme_from_cache(*_cache_keys):  # parameters are not really used - only as caching key
             readme_data = None
             readme_file = None
             try:
                 # gets the landing revision! or tip if fails
                 cs = db_repo.get_landing_changeset()
                 if isinstance(cs, EmptyChangeset):
                     raise EmptyRepositoryError()
                 renderer = MarkupRenderer()
                 for f in README_FILES:
                     try:
                         readme = cs.get_node(f)
                         if not isinstance(readme, FileNode):
                             continue
                         readme_file = f
                         log.debug('Found README file `%s` rendering...',
                                   readme_file)
-                        readme_data = renderer.render(safe_unicode(readme.content),
+                        readme_data = renderer.render(safe_str(readme.content),
                                                       filename=f)
                         break
                     except NodeDoesNotExistError:
                         continue
             except ChangesetError:
                 log.error(traceback.format_exc())
                 pass
             except EmptyRepositoryError:
                 pass
             return readme_data, readme_file
         kind = 'README'
         return _get_readme_from_cache(repo_name, kind, c.db_repo.changeset_cache.get('raw_id'))
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def index(self, repo_name):
         p = safe_int(request.GET.get('page'), 1)
         size = safe_int(request.GET.get('size'), 10)
         try:
             collection = c.db_repo_scm_instance.get_changesets(reverse=True)
         except EmptyRepositoryError as e:
             h.flash(e, category='warning')
             collection = []
         c.cs_pagination = Page(collection, page=p, items_per_page=size)
         page_revisions = [x.raw_id for x in list(c.cs_pagination)]
         c.cs_comments = c.db_repo.get_comments(page_revisions)
         c.cs_statuses = c.db_repo.statuses(page_revisions)
         c.ssh_repo_url = None
         if request.authuser.is_default_user:
             username = None
         else:
             username = request.authuser.username
             if c.ssh_enabled:
                 c.ssh_repo_url = c.db_repo.clone_url(clone_uri_tmpl=c.clone_ssh_tmpl)
         c.clone_repo_url = c.db_repo.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=False, username=username)
         c.clone_repo_url_id = c.db_repo.clone_url(clone_uri_tmpl=c.clone_uri_tmpl, with_id=True, username=username)
         if c.db_repo.enable_statistics:
             c.show_stats = True
         else:
             c.show_stats = False
         stats = Statistics.query() \
             .filter(Statistics.repository == c.db_repo) \
             .scalar()
         c.stats_percentage = 0
         if stats and stats.languages:
             lang_stats_d = ext_json.loads(stats.languages)
             lang_stats = [(x, {"count": y,
                                "desc": LANGUAGES_EXTENSIONS_MAP.get(x, '?')})
                           for x, y in lang_stats_d.items()]
             lang_stats.sort(key=lambda k: (-k[1]['count'], k[0]))
             c.trending_languages = lang_stats[:10]
         else:
             c.trending_languages = []
         c.enable_downloads = c.db_repo.enable_downloads
         c.readme_data, c.readme_file = \
             self.__get_readme_data(c.db_repo)
         return render('summary/summary.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def repo_size(self, repo_name):
         if request.is_xhr:
             return c.db_repo._repo_size()
         else:
             raise HTTPBadRequest()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def statistics(self, repo_name):
         if c.db_repo.enable_statistics:
             c.show_stats = True
             c.no_data_msg = _('No data ready yet')
         else:
             c.show_stats = False
             c.no_data_msg = _('Statistics are disabled for this repository')
         td = date.today() + timedelta(days=1)
         td_1m = td - timedelta(days=calendar.monthrange(td.year, td.month)[1])
         td_1y = td - timedelta(days=365)
         ts_min_m = mktime(td_1m.timetuple())
         ts_min_y = mktime(td_1y.timetuple())
         ts_max_y = mktime(td.timetuple())
         c.ts_min = ts_min_m
         c.ts_max = ts_max_y
         stats = Statistics.query() \
             .filter(Statistics.repository == c.db_repo) \
             .scalar()
         c.stats_percentage = 0
         if stats and stats.languages:
             c.commit_data = ext_json.loads(stats.commit_activity)
             c.overview_data = ext_json.loads(stats.commit_activity_combined)
             lang_stats_d = ext_json.loads(stats.languages)
             lang_stats = [(x, {"count": y,
                                "desc": LANGUAGES_EXTENSIONS_MAP.get(x, '?')})
                           for x, y in lang_stats_d.items()]
             lang_stats.sort(key=lambda k: (-k[1]['count'], k[0]))
             c.trending_languages = lang_stats[:10]
             last_rev = stats.stat_on_revision + 1
             c.repo_last_rev = c.db_repo_scm_instance.count() \
                 if c.db_repo_scm_instance.revisions else 0
             if last_rev == 0 or c.repo_last_rev == 0:
                 pass
             else:
                 c.stats_percentage = '%.2f' % ((float((last_rev)) /
                                                 c.repo_last_rev) * 100)
         else:
             c.commit_data = {}
             c.overview_data = ([[ts_min_y, 0], [ts_max_y, 10]])
             c.trending_languages = []
         recurse_limit = 500  # don't recurse more than 500 times when parsing
         get_commits_stats(c.db_repo.repo_name, ts_min_y, ts_max_y, recurse_limit)
         return render('summary/statistics.html')

kallithea/lib/annotate.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.lib.annotate
 ~~~~~~~~~~~~~~~~~~~~~~
 Annotation library for usage in Kallithea, previously part of vcs
 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: Dec 4, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from pygments import highlight
 from pygments.formatters import HtmlFormatter
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.lib.vcs.nodes import FileNode
-from kallithea.lib.vcs.utils import safe_unicode
+from kallithea.lib.vcs.utils import safe_str
 def annotate_highlight(filenode, annotate_from_changeset_func,
         order=None, headers=None, **options):
     """
     Returns html portion containing annotated table with 3 columns: line
     numbers, changeset information and pygmentized line of code.
     :param filenode: FileNode object
     :param annotate_from_changeset_func: function taking changeset and
       returning single annotate cell; needs break line at the end
     :param order: ordered sequence of ``ls`` (line numbers column),
       ``annotate`` (annotate column), ``code`` (code column); Default is
       ``['ls', 'annotate', 'code']``
     :param headers: dictionary with headers (keys are whats in ``order``
       parameter)
     """
     from kallithea.lib.pygmentsutils import get_custom_lexer
     options['linenos'] = True
     formatter = AnnotateHtmlFormatter(filenode=filenode,
         annotate_from_changeset_func=annotate_from_changeset_func, order=order,
         headers=headers, **options)
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
-    highlighted = highlight(safe_unicode(filenode.content), lexer, formatter)
+    highlighted = highlight(safe_str(filenode.content), lexer, formatter)
     return highlighted
 class AnnotateHtmlFormatter(HtmlFormatter):
     def __init__(self, filenode, annotate_from_changeset_func,
             order=None, **options):
         """
         ``annotate_from_changeset_func`` must be a function
         which returns string from the given changeset. For example, we may pass
         following function as ``annotate_from_changeset_func``::
             def changeset_to_anchor(changeset):
                 return '<a href="/changesets/%s/">%s</a>\n' % \
                        (changeset.raw_id, changeset.raw_id)
         :param annotate_from_changeset_func: see above
         :param order: (default: ``['ls', 'annotate', 'code']``); order of
           columns;
         :param options: standard pygment's HtmlFormatter options, there is
           extra option tough, ``headers``. For instance we can pass::
              formatter = AnnotateHtmlFormatter(filenode, headers={
                 'ls': '#',
                 'annotate': 'Annotate',
                 'code': 'Code',
              })
         """
         super(AnnotateHtmlFormatter, self).__init__(**options)
         self.annotate_from_changeset_func = annotate_from_changeset_func
         self.order = order or ('ls', 'annotate', 'code')
         headers = options.pop('headers', None)
         if headers and not ('ls' in headers and 'annotate' in headers and
             'code' in headers
         ):
             raise ValueError("If headers option dict is specified it must "
                 "all 'ls', 'annotate' and 'code' keys")
         self.headers = headers
         if isinstance(filenode, FileNode):
             self.filenode = filenode
         else:
             raise VCSError("This formatter expect FileNode parameter, not %r"
                 % type(filenode))
     def _wrap_tablelinenos(self, inner):
         inner_lines = []
         lncount = 0
         for t, line in inner:
             if t:
                 lncount += 1
             inner_lines.append(line)
         fl = self.linenostart
         mw = len(str(lncount + fl - 1))
         sp = self.linenospecial
         st = self.linenostep
         la = self.lineanchors
         aln = self.anchorlinenos
         if sp:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if i % sp == 0:
                         if aln:
                             lines.append('<a href="#%s-%d" class="special">'
                                          '%*d</a>' %
                                          (la, i, mw, i))
                         else:
                             lines.append('<span class="special">'
                                          '%*d</span>' % (mw, i))
                     else:
                         if aln:
                             lines.append('<a href="#%s-%d">'
                                          '%*d</a>' % (la, i, mw, i))
                         else:
                             lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         else:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if aln:
                         lines.append('<a href="#%s-%d">%*d</a>'
                                      % (la, i, mw, i))
                     else:
                         lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
 #        annotate_changesets = [tup[1] for tup in self.filenode.annotate]
 #        # TODO: not sure what that fixes
 #        # If pygments cropped last lines break we need do that too
 #        ln_cs = len(annotate_changesets)
 #        ln_ = len(ls.splitlines())
 #        if  ln_cs > ln_:
 #            annotate_changesets = annotate_changesets[:ln_ - ln_cs]
         annotate = ''.join((self.annotate_from_changeset_func(el[2]())
                             for el in self.filenode.annotate))
         # in case you wonder about the seemingly redundant <div> here:
         # since the content in the other cell also is wrapped in a div,
         # some browsers in some configurations seem to mess up the formatting.
         '''
         yield 0, ('<table class="%stable">' % self.cssclass +
                   '<tr><td class="linenos"><div class="linenodiv"><pre>' +
                   ls + '</pre></div></td>' +
                   '<td class="code">')
         yield 0, ''.join(inner_lines)
         yield 0, '</td></tr></table>'
         '''
         headers_row = []
         if self.headers:
             headers_row = ['<tr class="annotate-header">']
             for key in self.order:
                 td = ''.join(('<td>', self.headers[key], '</td>'))
                 headers_row.append(td)
             headers_row.append('</tr>')
         body_row_start = ['<tr>']
         for key in self.order:
             if key == 'ls':
                 body_row_start.append(
                     '<td class="linenos"><div class="linenodiv"><pre>' +
                     ls + '</pre></div></td>')
             elif key == 'annotate':
                 body_row_start.append(
                     '<td class="annotate"><div class="annotatediv"><pre>' +
                     annotate + '</pre></div></td>')
             elif key == 'code':
                 body_row_start.append('<td class="code">')
         yield 0, ('<table class="%stable">' % self.cssclass +
                   ''.join(headers_row) +
                   ''.join(body_row_start)
+                  )
         yield 0, ''.join(inner_lines)
         yield 0, '</td></tr></table>'

kallithea/lib/base.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.lib.base
 ~~~~~~~~~~~~~~~~~~
 The base Controller API
 Provides the BaseController class for subclassing. And usage in different
 controllers
 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: Oct 06, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import base64
 import datetime
 import logging
 import traceback
 import warnings
 import decorator
 import paste.auth.basic
 import paste.httpexceptions
 import paste.httpheaders
 import webob.exc
 from tg import TGController, config, render_template, request, response, session
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from kallithea import BACKENDS, __version__
 from kallithea.config.routing import url
 from kallithea.lib import auth_modules, ext_json
 from kallithea.lib.auth import AuthUser, HasPermissionAnyMiddleware
 from kallithea.lib.exceptions import UserCreationError
 from kallithea.lib.utils import get_repo_slug, is_valid_repo
-from kallithea.lib.utils2 import AttributeDict, ascii_bytes, safe_int, safe_unicode, set_hook_environment, str2bool
+from kallithea.lib.utils2 import AttributeDict, ascii_bytes, safe_int, safe_str, set_hook_environment, str2bool
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError
 from kallithea.model import meta
 from kallithea.model.db import PullRequest, Repository, Setting, User
 from kallithea.model.scm import ScmModel
 log = logging.getLogger(__name__)
 def render(template_path):
     return render_template({'url': url}, 'mako', template_path)
 def _filter_proxy(ip):
     """
     HEADERS can have multiple ips inside the left-most being the original
     client, and each successive proxy that passed the request adding the IP
     address where it received the request from.
     :param ip:
     """
     if ',' in ip:
         _ips = ip.split(',')
         _first_ip = _ips[0].strip()
         log.debug('Got multiple IPs %s, using %s', ','.join(_ips), _first_ip)
         return _first_ip
     return ip
 def _get_ip_addr(environ):
     proxy_key = 'HTTP_X_REAL_IP'
     proxy_key2 = 'HTTP_X_FORWARDED_FOR'
     def_key = 'REMOTE_ADDR'
     ip = environ.get(proxy_key)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(proxy_key2)
     if ip:
         return _filter_proxy(ip)
     ip = environ.get(def_key, '0.0.0.0')
     return _filter_proxy(ip)
 def get_path_info(environ):
     """Return unicode PATH_INFO from environ ... using tg.original_request if available.
     """
     org_req = environ.get('tg.original_request')
     if org_req is not None:
         environ = org_req.environ
-    return safe_unicode(environ['PATH_INFO'])
+    return safe_str(environ['PATH_INFO'])
 def log_in_user(user, remember, is_external_auth, ip_addr):
     """
     Log a `User` in and update session and cookies. If `remember` is True,
     the session cookie is set to expire in a year; otherwise, it expires at
     the end of the browser session.
     Returns populated `AuthUser` object.
     """
     # It should not be possible to explicitly log in as the default user.
     assert not user.is_default_user, user
     auth_user = AuthUser.make(dbuser=user, is_external_auth=is_external_auth, ip_addr=ip_addr)
     if auth_user is None:
         return None
     user.update_lastlogin()
     meta.Session().commit()
     # Start new session to prevent session fixation attacks.
     session.invalidate()
     session['authuser'] = cookie = auth_user.to_cookie()
     # If they want to be remembered, update the cookie.
     # NOTE: Assumes that beaker defaults to browser session cookie.
     if remember:
         t = datetime.datetime.now() + datetime.timedelta(days=365)
         session._set_cookie_expires(t)
     session.save()
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user.username, cookie)
     # dumps session attrs back to cookie
     session._update_cookie_out()
     return auth_user
 class BasicAuth(paste.auth.basic.AuthBasicAuthenticator):
     def __init__(self, realm, authfunc, auth_http_code=None):
         self.realm = realm
         self.authfunc = authfunc
         self._rc_auth_http_code = auth_http_code
     def build_authentication(self, environ):
         head = paste.httpheaders.WWW_AUTHENTICATE.tuples('Basic realm="%s"' % self.realm)
         # Consume the whole body before sending a response
         try:
             request_body_size = int(environ.get('CONTENT_LENGTH', 0))
         except (ValueError):
             request_body_size = 0
         environ['wsgi.input'].read(request_body_size)
         if self._rc_auth_http_code and self._rc_auth_http_code == '403':
             # return 403 if alternative http return code is specified in
             # Kallithea config
             return paste.httpexceptions.HTTPForbidden(headers=head)
         return paste.httpexceptions.HTTPUnauthorized(headers=head)
     def authenticate(self, environ):
         authorization = paste.httpheaders.AUTHORIZATION(environ)
         if not authorization:
             return self.build_authentication(environ)
         (authmeth, auth) = authorization.split(' ', 1)
         if 'basic' != authmeth.lower():
             return self.build_authentication(environ)
         auth = base64.b64decode(auth.strip())
         _parts = auth.split(':', 1)
         if len(_parts) == 2:
             username, password = _parts
             if self.authfunc(username, password, environ) is not None:
                 return username
         return self.build_authentication(environ)
     __call__ = authenticate
 class BaseVCSController(object):
     """Base controller for handling Mercurial/Git protocol requests
     (coming from a VCS client, and not a browser).
     """
     scm_alias = None # 'hg' / 'git'
     def __init__(self, application, config):
         self.application = application
         self.config = config
         # base path of repo locations
         self.basepath = self.config['base_path']
         # authenticate this VCS request using the authentication modules
         self.authenticate = BasicAuth('', auth_modules.authenticate,
                                       config.get('auth_ret_code'))
     @classmethod
     def parse_request(cls, environ):
         """If request is parsed as a request for this VCS, return a namespace with the parsed request.
         If the request is unknown, return None.
         """
         raise NotImplementedError()
     def _authorize(self, environ, action, repo_name, ip_addr):
         """Authenticate and authorize user.
         Since we're dealing with a VCS client and not a browser, we only
         support HTTP basic authentication, either directly via raw header
         inspection, or by using container authentication to delegate the
         authentication to the web server.
         Returns (user, None) on successful authentication and authorization.
         Returns (None, wsgi_app) to send the wsgi_app response to the client.
         """
         # Use anonymous access if allowed for action on repo.
         default_user = User.get_default_user(cache=True)
         default_authuser = AuthUser.make(dbuser=default_user, ip_addr=ip_addr)
         if default_authuser is None:
             log.debug('No anonymous access at all') # move on to proper user auth
         else:
             if self._check_permission(action, default_authuser, repo_name):
                 return default_authuser, None
             log.debug('Not authorized to access this repository as anonymous user')
         username = None
         #==============================================================
         # DEFAULT PERM FAILED OR ANONYMOUS ACCESS IS DISABLED SO WE
         # NEED TO AUTHENTICATE AND ASK FOR AUTH USER PERMISSIONS
         #==============================================================
         # try to auth based on environ, container auth methods
         log.debug('Running PRE-AUTH for container based authentication')
         pre_auth = auth_modules.authenticate('', '', environ)
         if pre_auth is not None and pre_auth.get('username'):
             username = pre_auth['username']
         log.debug('PRE-AUTH got %s as username', username)
         # If not authenticated by the container, running basic auth
         if not username:
             self.authenticate.realm = self.config['realm']
             result = self.authenticate(environ)
             if isinstance(result, str):
                 paste.httpheaders.AUTH_TYPE.update(environ, 'basic')
                 paste.httpheaders.REMOTE_USER.update(environ, result)
                 username = result
             else:
                 return None, result.wsgi_application
         #==============================================================
         # CHECK PERMISSIONS FOR THIS REQUEST USING GIVEN USERNAME
         #==============================================================
         try:
             user = User.get_by_username_or_email(username)
         except Exception:
             log.error(traceback.format_exc())
             return None, webob.exc.HTTPInternalServerError()
         authuser = AuthUser.make(dbuser=user, ip_addr=ip_addr)
         if authuser is None:
             return None, webob.exc.HTTPForbidden()
         if not self._check_permission(action, authuser, repo_name):
             return None, webob.exc.HTTPForbidden()
         return user, None
     def _handle_request(self, environ, start_response):
         raise NotImplementedError()
     def _check_permission(self, action, authuser, repo_name):
         """
         Checks permissions using action (push/pull) user and repository
         name
         :param action: 'push' or 'pull' action
         :param user: `User` instance
         :param repo_name: repository name
         """
         if action == 'push':
             if not HasPermissionAnyMiddleware('repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
         else:
             #any other action need at least read permission
             if not HasPermissionAnyMiddleware('repository.read',
                                               'repository.write',
                                               'repository.admin')(authuser,
                                                                   repo_name):
                 return False
         return True

kallithea/lib/diffs.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.lib.diffs
 ~~~~~~~~~~~~~~~~~~~
 Set of diffing helpers, previously part of vcs
 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: Dec 4, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import difflib
 import logging
 import re
 from tg.i18n import ugettext as _
 from kallithea.lib import helpers as h
-from kallithea.lib.utils2 import safe_unicode
+from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.lib.vcs.nodes import FileNode, SubModuleNode
 log = logging.getLogger(__name__)
 def _safe_id(idstring):
     r"""Make a string safe for including in an id attribute.
     The HTML spec says that id attributes 'must begin with
     a letter ([A-Za-z]) and may be followed by any number
     of letters, digits ([0-9]), hyphens ("-"), underscores
     ("_"), colons (":"), and periods (".")'. These regexps
     are slightly over-zealous, in that they remove colons
     and periods unnecessarily.
     Whitespace is transformed into underscores, and then
     anything which is not a hyphen or a character that
     matches \w (alphanumerics and underscore) is removed.
     """
     # Transform all whitespace to underscore
     idstring = re.sub(r'\s', "_", idstring)
     # Remove everything that is not a hyphen or a member of \w
     idstring = re.sub(r'(?!-)\W', "", idstring).lower()
     return idstring
 def as_html(table_class='code-difftable', line_class='line',
             old_lineno_class='lineno old', new_lineno_class='lineno new',
             no_lineno_class='lineno',
             code_class='code', enable_comments=False, parsed_lines=None):
     """
     Return given diff as html table with customized css classes
     """
     def _link_to_if(condition, label, url):
         """
         Generates a link if condition is meet or just the label if not.
         """
         if condition:
             return '''<a href="%(url)s" data-pseudo-content="%(label)s"></a>''' % {
                 'url': url,
                 'label': label
+            }
         else:
             return label
     _html_empty = True
     _html = []
     _html.append('''<table class="%(table_class)s">\n''' % {
         'table_class': table_class
     })
     for diff in parsed_lines:
         for line in diff['chunks']:
             _html_empty = False
             for change in line:
                 _html.append('''<tr class="%(lc)s %(action)s">\n''' % {
                     'lc': line_class,
                     'action': change['action']
                 })
                 anchor_old_id = ''
                 anchor_new_id = ''
                 anchor_old = "%(filename)s_o%(oldline_no)s" % {
                     'filename': _safe_id(diff['filename']),
                     'oldline_no': change['old_lineno']
+                }
                 anchor_new = "%(filename)s_n%(oldline_no)s" % {
                     'filename': _safe_id(diff['filename']),
                     'oldline_no': change['new_lineno']
+                }
                 cond_old = (change['old_lineno'] != '...' and
                             change['old_lineno'])
                 cond_new = (change['new_lineno'] != '...' and
                             change['new_lineno'])
                 no_lineno = (change['old_lineno'] == '...' and
                              change['new_lineno'] == '...')
                 if cond_old:
                     anchor_old_id = 'id="%s"' % anchor_old
                 if cond_new:
                     anchor_new_id = 'id="%s"' % anchor_new
                 ###########################################################
                 # OLD LINE NUMBER
                 ###########################################################
                 _html.append('''\t<td %(a_id)s class="%(olc)s" %(colspan)s>''' % {
                     'a_id': anchor_old_id,
                     'olc': no_lineno_class if no_lineno else old_lineno_class,
                     'colspan': 'colspan="2"' if no_lineno else ''
                 })
                 _html.append('''%(link)s''' % {
                     'link': _link_to_if(not no_lineno, change['old_lineno'],
                                         '#%s' % anchor_old)
                 })
                 _html.append('''</td>\n''')
                 ###########################################################
                 # NEW LINE NUMBER
                 ###########################################################
                 if not no_lineno:
                     _html.append('''\t<td %(a_id)s class="%(nlc)s">''' % {
                         'a_id': anchor_new_id,
                         'nlc': new_lineno_class
                     })
                     _html.append('''%(link)s''' % {
                         'link': _link_to_if(True, change['new_lineno'],
                                             '#%s' % anchor_new)
                     })
                     _html.append('''</td>\n''')
                 ###########################################################
                 # CODE
                 ###########################################################
                 comments = '' if enable_comments else 'no-comment'
                 _html.append('''\t<td class="%(cc)s %(inc)s">''' % {
                     'cc': code_class,
                     'inc': comments
                 })
                 _html.append('''\n\t\t<div class="add-bubble"><div>&nbsp;</div></div><pre>%(code)s</pre>\n''' % {
                     'code': change['line']
                 })
                 _html.append('''\t</td>''')
                 _html.append('''\n</tr>\n''')
     _html.append('''</table>''')
     if _html_empty:
         return None
     return ''.join(_html)
 def wrap_to_table(html):
     """Given a string with html, return it wrapped in a table, similar to what
     DiffProcessor returns."""
     return '''\
               <table class="code-difftable">
                 <tr class="line no-comment">
                 <td class="lineno new"></td>
                 <td class="code no-comment"><pre>%s</pre></td>
                 </tr>
               </table>''' % html
 def wrapped_diff(filenode_old, filenode_new, diff_limit=None,
                 ignore_whitespace=True, line_context=3,
                 enable_comments=False):
     """
     Returns a file diff wrapped into a table.
     Checks for diff_limit and presents a message if the diff is too big.
     """
     if filenode_old is None:
         filenode_old = FileNode(filenode_new.path, '', EmptyChangeset())
     op = None
     a_path = filenode_old.path # default, might be overriden by actual rename in diff
     if filenode_old.is_binary or filenode_new.is_binary:
         html_diff = wrap_to_table(_('Binary file'))
         stats = (0, 0)
     elif diff_limit != -1 and (
             diff_limit is None or
             (filenode_old.size < diff_limit and filenode_new.size < diff_limit)):
         raw_diff = get_gitdiff(filenode_old, filenode_new,
                                 ignore_whitespace=ignore_whitespace,
                                 context=line_context)
         diff_processor = DiffProcessor(raw_diff)
         if diff_processor.parsed: # there should be exactly one element, for the specified file
             f = diff_processor.parsed[0]
             op = f['operation']
             a_path = f['old_filename']
         html_diff = as_html(parsed_lines=diff_processor.parsed, enable_comments=enable_comments)
         stats = diff_processor.stat()
     else:
         html_diff = wrap_to_table(_('Changeset was too big and was cut off, use '
                                'diff menu to display this diff'))
         stats = (0, 0)
     if not html_diff:
         submodules = [o for o in [filenode_new, filenode_old] if isinstance(o, SubModuleNode)]
         if submodules:
             html_diff = wrap_to_table(h.escape('Submodule %r' % submodules[0]))
         else:
             html_diff = wrap_to_table(_('No changes detected'))
     cs1 = filenode_old.changeset.raw_id
     cs2 = filenode_new.changeset.raw_id
@@ @@ -288,385 +288,385 @@ class DiffProcessor(object): @@
         :param diff_limit: define the size of diff that is considered "big"
             based on that parameter cut off will be triggered, set to None
             to show full diff
         """
         if not isinstance(diff, bytes):
             raise Exception('Diff must be bytes - got %s' % type(diff))
         self._diff = memoryview(diff)
         self.adds = 0
         self.removes = 0
         self.diff_limit = diff_limit
         self.limited_diff = False
         self.vcs = vcs
         self.parsed = self._parse_gitdiff(inline_diff=inline_diff)
     def _parse_gitdiff(self, inline_diff):
         """Parse self._diff and return a list of dicts with meta info and chunks for each file.
         Might set limited_diff.
         Optionally, do an extra pass and to extra markup of one-liner changes.
         """
         _files = [] # list of dicts with meta info and chunks
         starts = [m.start() for m in self._diff_git_re.finditer(self._diff)]
         starts.append(len(self._diff))
         for start, end in zip(starts, starts[1:]):
             if self.diff_limit and end > self.diff_limit:
                 self.limited_diff = True
                 continue
             head, diff_lines = _get_header(self.vcs, self._diff[start:end])
             op = None
             stats = {
                 'added': 0,
                 'deleted': 0,
                 'binary': False,
                 'ops': {},
+            }
             if head['deleted_file_mode']:
                 op = 'removed'
                 stats['binary'] = True
                 stats['ops'][DEL_FILENODE] = 'deleted file'
             elif head['new_file_mode']:
                 op = 'added'
                 stats['binary'] = True
                 stats['ops'][NEW_FILENODE] = 'new file %s' % head['new_file_mode']
             else:  # modify operation, can be cp, rename, chmod
                 # CHMOD
                 if head['new_mode'] and head['old_mode']:
                     op = 'modified'
                     stats['binary'] = True
                     stats['ops'][CHMOD_FILENODE] = ('modified file chmod %s => %s'
                                         % (head['old_mode'], head['new_mode']))
                 # RENAME
                 if (head['rename_from'] and head['rename_to']
                       and head['rename_from'] != head['rename_to']):
                     op = 'renamed'
                     stats['binary'] = True
                     stats['ops'][RENAMED_FILENODE] = ('file renamed from %s to %s'
                                     % (head['rename_from'], head['rename_to']))
                 # COPY
                 if head.get('copy_from') and head.get('copy_to'):
                     op = 'modified'
                     stats['binary'] = True
                     stats['ops'][COPIED_FILENODE] = ('file copied from %s to %s'
                                         % (head['copy_from'], head['copy_to']))
                 # FALL BACK: detect missed old style add or remove
                 if op is None:
                     if not head['a_file'] and head['b_file']:
                         op = 'added'
                         stats['binary'] = True
                         stats['ops'][NEW_FILENODE] = 'new file'
                     elif head['a_file'] and not head['b_file']:
                         op = 'removed'
                         stats['binary'] = True
                         stats['ops'][DEL_FILENODE] = 'deleted file'
                 # it's not ADD not DELETE
                 if op is None:
                     op = 'modified'
                     stats['binary'] = True
                     stats['ops'][MOD_FILENODE] = 'modified file'
             # a real non-binary diff
             if head['a_file'] or head['b_file']:
                 chunks, added, deleted = _parse_lines(diff_lines)
                 stats['binary'] = False
                 stats['added'] = added
                 stats['deleted'] = deleted
                 # explicit mark that it's a modified file
                 if op == 'modified':
                     stats['ops'][MOD_FILENODE] = 'modified file'
             else:  # Git binary patch (or empty diff)
                 # Git binary patch
                 if head['bin_patch']:
                     stats['ops'][BIN_FILENODE] = 'binary diff not shown'
                 chunks = []
             if op == 'removed' and chunks:
                 # a way of seeing deleted content could perhaps be nice - but
                 # not with the current UI
                 chunks = []
             chunks.insert(0, [{
                 'old_lineno': '',
                 'new_lineno': '',
                 'action':     'context',
                 'line':       msg,
                 } for _op, msg in stats['ops'].items()
                   if _op not in [MOD_FILENODE]])
             _files.append({
                 'old_filename':     head['a_path'],
                 'filename':         head['b_path'],
                 'old_revision':     head['a_blob_id'],
                 'new_revision':     head['b_blob_id'],
                 'chunks':           chunks,
                 'operation':        op,
                 'stats':            stats,
             })
         if not inline_diff:
             return _files
         # highlight inline changes when one del is followed by one add
         for diff_data in _files:
             for chunk in diff_data['chunks']:
                 lineiter = iter(chunk)
                 try:
                     peekline = next(lineiter)
                     while True:
                         # find a first del line
                         while peekline['action'] != 'del':
                             peekline = next(lineiter)
                         delline = peekline
                         peekline = next(lineiter)
                         # if not followed by add, eat all following del lines
                         if peekline['action'] != 'add':
                             while peekline['action'] == 'del':
                                 peekline = next(lineiter)
                             continue
                         # found an add - make sure it is the only one
                         addline = peekline
                         try:
                             peekline = next(lineiter)
                         except StopIteration:
                             # add was last line - ok
                             _highlight_inline_diff(delline, addline)
                             raise
                         if peekline['action'] != 'add':
                             # there was only one add line - ok
                             _highlight_inline_diff(delline, addline)
                 except StopIteration:
                     pass
         return _files
     def stat(self):
         """
         Returns tuple of added, and removed lines for this instance
         """
         return self.adds, self.removes
 _escape_re = re.compile(r'(&)|(<)|(>)|(\t)|(\r)|(?<=.)( \n| $)')
 def _escaper(string):
     """
     Do HTML escaping/markup
     """
     def substitute(m):
         groups = m.groups()
         if groups[0]:
             return '&amp;'
         if groups[1]:
             return '&lt;'
         if groups[2]:
             return '&gt;'
         if groups[3]:
             return '<u>\t</u>'
         if groups[4]:
             return '<u class="cr"></u>'
         if groups[5]:
             return ' <i></i>'
         assert False
-    return _escape_re.sub(substitute, safe_unicode(string))
+    return _escape_re.sub(substitute, safe_str(string))
 _git_header_re = re.compile(br"""
     ^diff[ ]--git[ ]a/(?P<a_path>.+?)[ ]b/(?P<b_path>.+?)\n
     (?:^old[ ]mode[ ](?P<old_mode>\d+)\n
        ^new[ ]mode[ ](?P<new_mode>\d+)(?:\n|$))?
     (?:^similarity[ ]index[ ](?P<similarity_index>\d+)%\n
        ^rename[ ]from[ ](?P<rename_from>.+)\n
        ^rename[ ]to[ ](?P<rename_to>.+)(?:\n|$))?
     (?:^new[ ]file[ ]mode[ ](?P<new_file_mode>.+)(?:\n|$))?
     (?:^deleted[ ]file[ ]mode[ ](?P<deleted_file_mode>.+)(?:\n|$))?
     (?:^index[ ](?P<a_blob_id>[0-9A-Fa-f]+)
         \.\.(?P<b_blob_id>[0-9A-Fa-f]+)[ ]?(?P<b_mode>.+)?(?:\n|$))?
     (?:^(?P<bin_patch>GIT[ ]binary[ ]patch)(?:\n|$))?
     (?:^---[ ](a/(?P<a_file>.+?)|/dev/null)\t?(?:\n|$))?
     (?:^\+\+\+[ ](b/(?P<b_file>.+?)|/dev/null)\t?(?:\n|$))?
 """, re.VERBOSE | re.MULTILINE)
 _hg_header_re = re.compile(br"""
     ^diff[ ]--git[ ]a/(?P<a_path>.+?)[ ]b/(?P<b_path>.+?)\n
     (?:^old[ ]mode[ ](?P<old_mode>\d+)\n
        ^new[ ]mode[ ](?P<new_mode>\d+)(?:\n|$))?
     (?:^similarity[ ]index[ ](?P<similarity_index>\d+)%(?:\n|$))?
     (?:^rename[ ]from[ ](?P<rename_from>.+)\n
        ^rename[ ]to[ ](?P<rename_to>.+)(?:\n|$))?
     (?:^copy[ ]from[ ](?P<copy_from>.+)\n
        ^copy[ ]to[ ](?P<copy_to>.+)(?:\n|$))?
     (?:^new[ ]file[ ]mode[ ](?P<new_file_mode>.+)(?:\n|$))?
     (?:^deleted[ ]file[ ]mode[ ](?P<deleted_file_mode>.+)(?:\n|$))?
     (?:^index[ ](?P<a_blob_id>[0-9A-Fa-f]+)
         \.\.(?P<b_blob_id>[0-9A-Fa-f]+)[ ]?(?P<b_mode>.+)?(?:\n|$))?
     (?:^(?P<bin_patch>GIT[ ]binary[ ]patch)(?:\n|$))?
     (?:^---[ ](a/(?P<a_file>.+?)|/dev/null)\t?(?:\n|$))?
     (?:^\+\+\+[ ](b/(?P<b_file>.+?)|/dev/null)\t?(?:\n|$))?
 """, re.VERBOSE | re.MULTILINE)
 _header_next_check = re.compile(br'''(?!@)(?!literal )(?!delta )''')
 def _get_header(vcs, diff_chunk):
     """
     Parses a Git diff for a single file (header and chunks) and returns a tuple with:
 . A dict with meta info:
         a_path, b_path, similarity_index, rename_from, rename_to,
         old_mode, new_mode, new_file_mode, deleted_file_mode,
         a_blob_id, b_blob_id, b_mode, a_file, b_file
 . An iterator yielding lines with simple HTML markup.
     """
     match = None
     if vcs == 'git':
         match = _git_header_re.match(diff_chunk)
     elif vcs == 'hg':
         match = _hg_header_re.match(diff_chunk)
     if match is None:
         raise Exception('diff not recognized as valid %s diff' % vcs)
     meta_info = match.groupdict()
     rest = diff_chunk[match.end():]
     if rest and _header_next_check.match(rest):
         raise Exception('cannot parse %s diff header: %r followed by %r' % (vcs, diff_chunk[:match.end()], rest[:1000]))
     diff_lines = (_escaper(m.group(0)) for m in re.finditer(br'.*\n|.+$', rest)) # don't split on \r as str.splitlines do
     return meta_info, diff_lines
 _chunk_re = re.compile(r'^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(.*)')
 _newline_marker = re.compile(r'^\\ No newline at end of file')
 def _parse_lines(diff_lines):
     """
     Given an iterator of diff body lines, parse them and return a dict per
     line and added/removed totals.
     """
     added = deleted = 0
     old_line = old_end = new_line = new_end = None
     chunks = []
     try:
         line = next(diff_lines)
         while True:
             lines = []
             chunks.append(lines)
             match = _chunk_re.match(line)
             if not match:
                 raise Exception('error parsing diff @@ line %r' % line)
             gr = match.groups()
             (old_line, old_end,
              new_line, new_end) = [int(x or 1) for x in gr[:-1]]
             old_line -= 1
             new_line -= 1
             context = len(gr) == 5
             old_end += old_line
             new_end += new_line
             if context:
                 # skip context only if it's first line
                 if int(gr[0]) > 1:
                     lines.append({
                         'old_lineno': '...',
                         'new_lineno': '...',
                         'action':     'context',
                         'line':       line,
                     })
             line = next(diff_lines)
             while old_line < old_end or new_line < new_end:
                 if not line:
                     raise Exception('error parsing diff - empty line at -%s+%s' % (old_line, new_line))
                 affects_old = affects_new = False
                 command = line[0]
                 if command == '+':
                     affects_new = True
                     action = 'add'
                     added += 1
                 elif command == '-':
                     affects_old = True
                     action = 'del'
                     deleted += 1
                 elif command == ' ':
                     affects_old = affects_new = True
                     action = 'unmod'
                 else:
                     raise Exception('error parsing diff - unknown command in line %r at -%s+%s' % (line, old_line, new_line))
                 if not _newline_marker.match(line):
                     old_line += affects_old
                     new_line += affects_new
                     lines.append({
                         'old_lineno':   affects_old and old_line or '',
                         'new_lineno':   affects_new and new_line or '',
                         'action':       action,
                         'line':         line[1:],
                     })
                 line = next(diff_lines)
                 if _newline_marker.match(line):
                     # we need to append to lines, since this is not
                     # counted in the line specs of diff
                     lines.append({
                         'old_lineno':   '...',
                         'new_lineno':   '...',
                         'action':       'context',
                         'line':         line,
                     })
                     line = next(diff_lines)
             if old_line > old_end:
                 raise Exception('error parsing diff - more than %s "-" lines at -%s+%s' % (old_end, old_line, new_line))
             if new_line > new_end:
                 raise Exception('error parsing diff - more than %s "+" lines at -%s+%s' % (new_end, old_line, new_line))
     except StopIteration:
         pass
     if old_line != old_end or new_line != new_end:
         raise Exception('diff processing broken when old %s<>%s or new %s<>%s line %r' % (old_line, old_end, new_line, new_end, line))
     return chunks, added, deleted
 # Used for inline highlighter word split, must match the substitutions in _escaper
 _token_re = re.compile(r'()(&amp;|&lt;|&gt;|<u>\t</u>|<u class="cr"></u>| <i></i>|\W+?)')
 def _highlight_inline_diff(old, new):
     """
     Highlight simple add/remove in two lines given as info dicts. They are
     modified in place and given markup with <del>/<ins>.
     """
     assert old['action'] == 'del'
     assert new['action'] == 'add'
     oldwords = _token_re.split(old['line'])
     newwords = _token_re.split(new['line'])
     sequence = difflib.SequenceMatcher(None, oldwords, newwords)
     oldfragments, newfragments = [], []
     for tag, i1, i2, j1, j2 in sequence.get_opcodes():
         oldfrag = ''.join(oldwords[i1:i2])
         newfrag = ''.join(newwords[j1:j2])
         if tag != 'equal':
             if oldfrag:
                 oldfrag = '<del>%s</del>' % oldfrag

kallithea/lib/helpers.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/>.
 """
 Helper functions
 Consists of functions to typically be used within templates, but also
 available to Controllers. This module is available to both as 'h'.
 """
 import hashlib
 import json
 import logging
 import random
 import re
 import textwrap
 import urllib.parse
 from beaker.cache import cache_region
 from pygments import highlight as code_highlight
 from pygments.formatters.html import HtmlFormatter
 from tg.i18n import ugettext as _
 from webhelpers2.html import HTML, escape, literal
 from webhelpers2.html.tags import NotGiven, Option, Options, _input, _make_safe_id_component, checkbox, end_form
 from webhelpers2.html.tags import form as insecure_form
 from webhelpers2.html.tags import hidden, link_to, password, radio
 from webhelpers2.html.tags import select as webhelpers2_select
 from webhelpers2.html.tags import submit, text, textarea
 from webhelpers2.number import format_byte_size
 from webhelpers2.text import chop_at, truncate, wrap_paragraphs
 from kallithea.config.routing import url
 from kallithea.lib.annotate import annotate_highlight
 #==============================================================================
 # PERMS
 #==============================================================================
 from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel
 from kallithea.lib.markup_renderer import url_re
 from kallithea.lib.pygmentsutils import get_custom_lexer
 from kallithea.lib.utils2 import MENTIONS_REGEX, AttributeDict
 from kallithea.lib.utils2 import age as _age
-from kallithea.lib.utils2 import credentials_filter, safe_bytes, safe_int, safe_unicode, str2bool, time_to_datetime
+from kallithea.lib.utils2 import credentials_filter, safe_bytes, safe_int, safe_str, str2bool, time_to_datetime
 from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError
 #==============================================================================
 # SCM FILTERS available via h.
 #==============================================================================
 from kallithea.lib.vcs.utils import author_email, author_name
 log = logging.getLogger(__name__)
 def canonical_url(*args, **kargs):
     '''Like url(x, qualified=True), but returns url that not only is qualified
     but also canonical, as configured in canonical_url'''
     from kallithea import CONFIG
     try:
         parts = CONFIG.get('canonical_url', '').split('://', 1)
         kargs['host'] = parts[1]
         kargs['protocol'] = parts[0]
     except IndexError:
         kargs['qualified'] = True
     return url(*args, **kargs)
 def canonical_hostname():
     '''Return canonical hostname of system'''
     from kallithea import CONFIG
     try:
         parts = CONFIG.get('canonical_url', '').split('://', 1)
         return parts[1].split('/', 1)[0]
     except IndexError:
         parts = url('home', qualified=True).split('://', 1)
         return parts[1].split('/', 1)[0]
 def html_escape(s):
     """Return string with all html escaped.
     This is also safe for javascript in html but not necessarily correct.
     """
     return (s
         .replace('&', '&amp;')
         .replace(">", "&gt;")
         .replace("<", "&lt;")
         .replace('"', "&quot;")
         .replace("'", "&apos;") # Note: this is HTML5 not HTML4 and might not work in mails
+        )
 def js(value):
     """Convert Python value to the corresponding JavaScript representation.
     This is necessary to safely insert arbitrary values into HTML <script>
     sections e.g. using Mako template expression substitution.
     Note: Rather than using this function, it's preferable to avoid the
     insertion of values into HTML <script> sections altogether. Instead,
     data should (to the extent possible) be passed to JavaScript using
     data attributes or AJAX calls, eliminating the need for JS specific
     escaping.
     Note: This is not safe for use in attributes (e.g. onclick), because
     quotes are not escaped.
     Because the rules for parsing <script> varies between XHTML (where
     normal rules apply for any special characters) and HTML (where
     entities are not interpreted, but the literal string "</script>"
     is forbidden), the function ensures that the result never contains
     '&', '<' and '>', thus making it safe in both those contexts (but
     not in attributes).
     """
     return literal(
         ('(' + json.dumps(value) + ')')
         # In JSON, the following can only appear in string literals.
         .replace('&', r'\x26')
         .replace('<', r'\x3c')
         .replace('>', r'\x3e')
+    )
 def jshtml(val):
     """HTML escapes a string value, then converts the resulting string
     to its corresponding JavaScript representation (see `js`).
     This is used when a plain-text string (possibly containing special
     HTML characters) will be used by a script in an HTML context (e.g.
     element.innerHTML or jQuery's 'html' method).
     If in doubt, err on the side of using `jshtml` over `js`, since it's
     better to escape too much than too little.
     """
     return js(escape(val))
 def shorter(s, size=20, firstline=False, postfix='...'):
     """Truncate s to size, including the postfix string if truncating.
     If firstline, truncate at newline.
     """
     if firstline:
         s = s.split('\n', 1)[0].rstrip()
     if len(s) > size:
         return s[:size - len(postfix)] + postfix
     return s
 def reset(name, value, id=NotGiven, **attrs):
     """Create a reset button, similar to webhelpers2.html.tags.submit ."""
     return _input("reset", name, value, id, attrs)
 def select(name, selected_values, options, id=NotGiven, **attrs):
     """Convenient wrapper of webhelpers2 to let it accept options as a tuple list"""
     if isinstance(options, list):
         option_list = options
         # Handle old value,label lists ... where value also can be value,label lists
         options = Options()
         for x in option_list:
             if isinstance(x, tuple) and len(x) == 2:
                 value, label = x
             elif isinstance(x, str):
                 value = label = x
             else:
                 log.error('invalid select option %r', x)
                 raise
             if isinstance(value, list):
                 og = options.add_optgroup(label)
                 for x in value:
                     if isinstance(x, tuple) and len(x) == 2:
                         group_value, group_label = x
                     elif isinstance(x, str):
                         group_value = group_label = x
                     else:
                         log.error('invalid select option %r', x)
                         raise
                     og.add_option(group_label, group_value)
             else:
                 options.add_option(label, value)
     return webhelpers2_select(name, selected_values, options, id=id, **attrs)
 safeid = _make_safe_id_component
 def FID(raw_id, path):
     """
     Creates a unique ID for filenode based on it's hash of path and revision
     it's safe to use in urls
     :param raw_id:
     :param path:
     """
     return 'C-%s-%s' % (short_id(raw_id), hashlib.md5(safe_bytes(path)).hexdigest()[:12])
 class _FilesBreadCrumbs(object):
     def __call__(self, repo_name, rev, paths):
         url_l = [link_to(repo_name, url('files_home',
                                         repo_name=repo_name,
                                         revision=rev, f_path=''),
                          class_='ypjax-link')]
         paths_l = paths.split('/')
         for cnt, p in enumerate(paths_l):
             if p != '':
                 url_l.append(link_to(p,
                                      url('files_home',
                                          repo_name=repo_name,
                                          revision=rev,
                                          f_path='/'.join(paths_l[:cnt + 1])
                                          ),
                                      class_='ypjax-link'
+                                     )
+                             )
         return literal('/'.join(url_l))
 files_breadcrumbs = _FilesBreadCrumbs()
 class CodeHtmlFormatter(HtmlFormatter):
     """
     My code Html Formatter for source codes
     """
     def wrap(self, source, outfile):
         return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
     def _wrap_code(self, source):
         for cnt, it in enumerate(source):
             i, t = it
             t = '<span id="L%s">%s</span>' % (cnt + 1, t)
             yield i, t
     def _wrap_tablelinenos(self, inner):
         inner_lines = []
         lncount = 0
         for t, line in inner:
             if t:
                 lncount += 1
             inner_lines.append(line)
         fl = self.linenostart
         mw = len(str(lncount + fl - 1))
         sp = self.linenospecial
         st = self.linenostep
         la = self.lineanchors
         aln = self.anchorlinenos
         nocls = self.noclasses
         if sp:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if i % sp == 0:
                         if aln:
                             lines.append('<a href="#%s%d" class="special">%*d</a>' %
                                          (la, i, mw, i))
                         else:
                             lines.append('<span class="special">%*d</span>' % (mw, i))
                     else:
                         if aln:
                             lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                         else:
                             lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         else:
             lines = []
             for i in range(fl, fl + lncount):
                 if i % st == 0:
                     if aln:
                         lines.append('<a href="#%s%d">%*d</a>' % (la, i, mw, i))
                     else:
                         lines.append('%*d' % (mw, i))
                 else:
                     lines.append('')
             ls = '\n'.join(lines)
         # in case you wonder about the seemingly redundant <div> here: since the
         # content in the other cell also is wrapped in a div, some browsers in
         # some configurations seem to mess up the formatting...
         if nocls:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td><div class="linenodiv">'
                       '<pre>' + ls + '</pre></div></td>'
                       '<td id="hlcode" class="code">')
         else:
             yield 0, ('<table class="%stable">' % self.cssclass +
                       '<tr><td class="linenos"><div class="linenodiv">'
                       '<pre>' + ls + '</pre></div></td>'
                       '<td id="hlcode" class="code">')
         yield 0, ''.join(inner_lines)
         yield 0, '</td></tr></table>'
 _whitespace_re = re.compile(r'(\t)|( )(?=\n|</div>)')
 def _markup_whitespace(m):
     groups = m.groups()
     if groups[0]:
         return '<u>\t</u>'
     if groups[1]:
         return ' <i></i>'
 def markup_whitespace(s):
     return _whitespace_re.sub(_markup_whitespace, s)
 def pygmentize(filenode, **kwargs):
     """
     pygmentize function using pygments
     :param filenode:
     """
     lexer = get_custom_lexer(filenode.extension) or filenode.lexer
     return literal(markup_whitespace(
-        code_highlight(safe_unicode(filenode.content), lexer, CodeHtmlFormatter(**kwargs))))
+        code_highlight(safe_str(filenode.content), lexer, CodeHtmlFormatter(**kwargs))))
 def hsv_to_rgb(h, s, v):
     if s == 0.0:
         return v, v, v
     i = int(h * 6.0)  # XXX assume int() truncates!
     f = (h * 6.0) - i
     p = v * (1.0 - s)
     q = v * (1.0 - s * f)
     t = v * (1.0 - s * (1.0 - f))
     i = i % 6
     if i == 0:
         return v, t, p
     if i == 1:
         return q, v, p
     if i == 2:
         return p, v, t
     if i == 3:
         return p, q, v
     if i == 4:
         return t, p, v
     if i == 5:
         return v, p, q
 def gen_color(n=10000):
     """generator for getting n of evenly distributed colors using
     hsv color and golden ratio. It always return same order of colors
     :returns: RGB tuple
     """
     golden_ratio = 0.618033988749895
     h = 0.22717784590367374
     for _unused in range(n):
         h += golden_ratio
         h %= 1
         HSV_tuple = [h, 0.95, 0.95]
         RGB_tuple = hsv_to_rgb(*HSV_tuple)
         yield [str(int(x * 256)) for x in RGB_tuple]
 def pygmentize_annotation(repo_name, filenode, **kwargs):
     """
     pygmentize function for annotation
     :param filenode:
     """
     cgenerator = gen_color()
     color_dict = {}
     def get_color_string(cs):
         if cs in color_dict:
             col = color_dict[cs]
         else:
             col = color_dict[cs] = next(cgenerator)
         return "color: rgb(%s)! important;" % (', '.join(col))
     def url_func(changeset):
         author = escape(changeset.author)
         date = changeset.date
         message = escape(changeset.message)
         tooltip_html = ("<b>Author:</b> %s<br/>"
                         "<b>Date:</b> %s</b><br/>"
                         "<b>Message:</b> %s") % (author, date, message)
         lnk_format = show_id(changeset)
         uri = link_to(
                 lnk_format,
                 url('changeset_home', repo_name=repo_name,
                     revision=changeset.raw_id),
                 style=get_color_string(changeset.raw_id),
                 **{'data-toggle': 'popover',
                    'data-content': tooltip_html}
+              )
         uri += '\n'
         return uri
     return literal(markup_whitespace(annotate_highlight(filenode, url_func, **kwargs)))
 class _Message(object):
     """A message returned by ``pop_flash_messages()``.
     Converting the message to a string returns the message text. Instances
     also have the following attributes:
     * ``category``: the category specified when the message was created.
     * ``message``: the html-safe message text.
     """
     def __init__(self, category, message):
         self.category = category
         self.message = message
 def _session_flash_messages(append=None, clear=False):
     """Manage a message queue in tg.session: return the current message queue
     after appending the given message, and possibly clearing the queue."""
     key = 'flash'
     from tg import session
     if key in session:
         flash_messages = session[key]
     else:
         if append is None:  # common fast path - also used for clearing empty queue
             return []  # don't bother saving
         flash_messages = []
         session[key] = flash_messages
     if append is not None and append not in flash_messages:
         flash_messages.append(append)
     if clear:
         session.pop(key, None)
     session.save()
     return flash_messages
 def flash(message, category, logf=None):
     """
     Show a message to the user _and_ log it through the specified function
     category: notice (default), warning, error, success
     logf: a custom log function - such as log.debug
     logf defaults to log.info, unless category equals 'success', in which
     case logf defaults to log.debug.
     """
     assert category in ('error', 'success', 'warning'), category
     if hasattr(message, '__html__'):
         # render to HTML for storing in cookie
         safe_message = unicode(message)
     else:
         # Apply str - the message might be an exception with __str__
         # Escape, so we can trust the result without further escaping, without any risk of injection
         safe_message = html_escape(unicode(message))
     if logf is None:
         logf = log.info
         if category == 'success':
             logf = log.debug
     logf('Flash %s: %s', category, safe_message)
     _session_flash_messages(append=(category, safe_message))
 def pop_flash_messages():
     """Return all accumulated messages and delete them from the session.
     The return value is a list of ``Message`` objects.
     """
     return [_Message(category, message) for category, message in _session_flash_messages(clear=True)]
 age = lambda x, y=False: _age(x, y)
 capitalize = lambda x: x.capitalize()
 email = author_email
 short_id = lambda x: x[:12]
 hide_credentials = lambda x: ''.join(credentials_filter(x))
 def show_id(cs):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
     :param cs: changeset instance
     """
     from kallithea import CONFIG
     def_len = safe_int(CONFIG.get('show_sha_length', 12))
     show_rev = str2bool(CONFIG.get('show_revision_number', False))
     raw_id = cs.raw_id[:def_len]
     if show_rev:
         return 'r%s:%s' % (cs.revision, raw_id)
     else:
         return raw_id
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S")
     return ""
 def is_git(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
@@ @@ -1033,288 +1033,288 @@ _URLIFY_RE = re.compile(r''' @@
 # URL markup
 (?P<url>%s) |
 # @mention markup
 (?P<mention>%s) |
 # Changeset hash markup
 (?<!\w|[-_])
   (?P<hash>[0-9a-f]{12,40})
 (?!\w|[-_]) |
 # Markup of *bold text*
 (?:
   (?:^|(?<=\s))
   (?P<bold> [*] (?!\s) [^*\n]* (?<!\s) [*] )
   (?![*\w])
 ) |
 # "Stylize" markup
 \[see\ \=&gt;\ *(?P<seen>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |
 \[license\ \=&gt;\ *(?P<license>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |
 \[(?P<tagtype>requires|recommends|conflicts|base)\ \=&gt;\ *(?P<tagvalue>[a-zA-Z0-9\-\/]*)\] |
 \[(?:lang|language)\ \=&gt;\ *(?P<lang>[a-zA-Z\-\/\#\+]*)\] |
 \[(?P<tag>[a-z]+)\]
 ''' % (url_re.pattern, MENTIONS_REGEX.pattern),
     re.VERBOSE | re.MULTILINE | re.IGNORECASE)
 def urlify_text(s, repo_name=None, link_=None, truncate=None, stylize=False, truncatef=truncate):
     """
     Parses given text message and make literal html with markup.
     The text will be truncated to the specified length.
     Hashes are turned into changeset links to specified repository.
     URLs links to what they say.
     Issues are linked to given issue-server.
     If link_ is provided, all text not already linking somewhere will link there.
     """
     def _replace(match_obj):
         url = match_obj.group('url')
         if url is not None:
             return '<a href="%(url)s">%(url)s</a>' % {'url': url}
         mention = match_obj.group('mention')
         if mention is not None:
             return '<b>%s</b>' % mention
         hash_ = match_obj.group('hash')
         if hash_ is not None and repo_name is not None:
             from kallithea.config.routing import url  # doh, we need to re-import url to mock it later
             return '<a class="changeset_hash" href="%(url)s">%(hash)s</a>' % {
                  'url': url('changeset_home', repo_name=repo_name, revision=hash_),
                  'hash': hash_,
+                }
         bold = match_obj.group('bold')
         if bold is not None:
             return '<b>*%s*</b>' % _urlify(bold[1:-1])
         if stylize:
             seen = match_obj.group('seen')
             if seen:
                 return '<div class="label label-meta" data-tag="see">see =&gt; %s</div>' % seen
             license = match_obj.group('license')
             if license:
                 return '<div class="label label-meta" data-tag="license"><a href="http://www.opensource.org/licenses/%s">%s</a></div>' % (license, license)
             tagtype = match_obj.group('tagtype')
             if tagtype:
                 tagvalue = match_obj.group('tagvalue')
                 return '<div class="label label-meta" data-tag="%s">%s =&gt; <a href="/%s">%s</a></div>' % (tagtype, tagtype, tagvalue, tagvalue)
             lang = match_obj.group('lang')
             if lang:
                 return '<div class="label label-meta" data-tag="lang">%s</div>' % lang
             tag = match_obj.group('tag')
             if tag:
                 return '<div class="label label-meta" data-tag="%s">%s</div>' % (tag, tag)
         return match_obj.group(0)
     def _urlify(s):
         """
         Extract urls from text and make html links out of them
         """
         return _URLIFY_RE.sub(_replace, s)
     if truncate is None:
         s = s.rstrip()
     else:
         s = truncatef(s, truncate, whole_word=True)
     s = html_escape(s)
     s = _urlify(s)
     if repo_name is not None:
         s = urlify_issues(s, repo_name)
     if link_ is not None:
         # make href around everything that isn't a href already
         s = linkify_others(s, link_)
     s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')
     # Turn HTML5 into more valid HTML4 as required by some mail readers.
     # (This is not done in one step in html_escape, because character codes like
     # &#123; risk to be seen as an issue reference due to the presence of '#'.)
     s = s.replace("&apos;", "&#39;")
     return literal(s)
 def linkify_others(t, l):
     """Add a default link to html with links.
     HTML doesn't allow nesting of links, so the outer link must be broken up
     in pieces and give space for other links.
     """
     urls = re.compile(r'(\<a.*?\<\/a\>)',)
     links = []
     for e in urls.split(t):
         if e.strip() and not urls.match(e):
             links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
         else:
             links.append(e)
     return ''.join(links)
 # Global variable that will hold the actual urlify_issues function body.
 # Will be set on first use when the global configuration has been read.
 _urlify_issues_f = None
 def urlify_issues(newtext, repo_name):
     """Urlify issue references according to .ini configuration"""
     global _urlify_issues_f
     if _urlify_issues_f is None:
         from kallithea import CONFIG
         from kallithea.model.db import URL_SEP
         assert CONFIG['sqlalchemy.url'] # make sure config has been loaded
         # Build chain of urlify functions, starting with not doing any transformation
         tmp_urlify_issues_f = lambda s: s
         issue_pat_re = re.compile(r'issue_pat(.*)')
         for k in CONFIG:
             # Find all issue_pat* settings that also have corresponding server_link and prefix configuration
             m = issue_pat_re.match(k)
             if m is None:
                 continue
             suffix = m.group(1)
             issue_pat = CONFIG.get(k)
             issue_server_link = CONFIG.get('issue_server_link%s' % suffix)
             issue_sub = CONFIG.get('issue_sub%s' % suffix)
             if not issue_pat or not issue_server_link or issue_sub is None: # issue_sub can be empty but should be present
                 log.error('skipping incomplete issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
                 continue
             # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound
             try:
                 issue_re = re.compile(issue_pat)
             except re.error as e:
                 log.error('skipping invalid issue pattern %r: %r -> %r %r. Error: %s', suffix, issue_pat, issue_server_link, issue_sub, str(e))
                 continue
             log.debug('issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
             def issues_replace(match_obj,
                                issue_server_link=issue_server_link, issue_sub=issue_sub):
                 try:
                     issue_url = match_obj.expand(issue_server_link)
                 except (IndexError, re.error) as e:
                     log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                     issue_url = issue_server_link
                 issue_url = issue_url.replace('{repo}', repo_name)
                 issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])
                 # if issue_sub is empty use the matched issue reference verbatim
                 if not issue_sub:
                     issue_text = match_obj.group()
                 else:
                     try:
                         issue_text = match_obj.expand(issue_sub)
                     except (IndexError, re.error) as e:
                         log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                         issue_text = match_obj.group()
                 return (
                     '<a class="issue-tracker-link" href="%(url)s">'
                     '%(text)s'
                     '</a>'
                     ) % {
                      'url': issue_url,
                      'text': issue_text,
+                    }
             tmp_urlify_issues_f = (lambda s,
                                           issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f:
                                    issue_re.sub(issues_replace, chain_f(s)))
         # Set tmp function globally - atomically
         _urlify_issues_f = tmp_urlify_issues_f
     return _urlify_issues_f(newtext)
 def render_w_mentions(source, repo_name=None):
     """
     Render plain text with revision hashes and issue references urlified
     and with @mention highlighting.
     """
-    s = safe_unicode(source)
+    s = safe_str(source)
     s = urlify_text(s, repo_name=repo_name)
     return literal('<div class="formatted-fixed">%s</div>' % s)
 def short_ref(ref_type, ref_name):
     if ref_type == 'rev':
         return short_id(ref_name)
     return ref_name
 def link_to_ref(repo_name, ref_type, ref_name, rev=None):
     """
     Return full markup for a href to changeset_home for a changeset.
     If ref_type is branch it will link to changelog.
     ref_name is shortened if ref_type is 'rev'.
     if rev is specified show it too, explicitly linking to that revision.
     """
     txt = short_ref(ref_type, ref_name)
     if ref_type == 'branch':
         u = url('changelog_home', repo_name=repo_name, branch=ref_name)
     else:
         u = url('changeset_home', repo_name=repo_name, revision=ref_name)
     l = link_to(repo_name + '#' + txt, u)
     if rev and ref_type != 'rev':
         l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))
     return l
 def changeset_status(repo, revision):
     from kallithea.model.changeset_status import ChangesetStatusModel
     return ChangesetStatusModel().get_status(repo, revision)
 def changeset_status_lbl(changeset_status):
     from kallithea.model.db import ChangesetStatus
     return ChangesetStatus.get_status_lbl(changeset_status)
 def get_permission_name(key):
     from kallithea.model.db import Permission
     return dict(Permission.PERMS).get(key)
 def journal_filter_help():
     return _(textwrap.dedent('''
         Example filter terms:
             repository:vcs
             username:developer
             action:*push*
             ip:127.0.0.1
             date:20120101
             date:[20120101100000 TO 20120102]
         Generate wildcards using '*' character:
             "repository:vcs*" - search everything starting with 'vcs'
             "repository:*vcs*" - search for repository containing 'vcs'
         Optional AND / OR operators in queries
             "repository:vcs OR repository:test"
             "username:test AND repository:test*"
     '''))
 def not_mapped_error(repo_name):
     flash(_('%s repository is not mapped to db perhaps'
             ' it was created or renamed from the filesystem'
             ' please run the application again'
             ' in order to rescan repositories') % repo_name, category='error')
 def ip_range(ip_addr):
     from kallithea.model.db import UserIpMap
     s, e = UserIpMap._get_ip_range(ip_addr)
     return '%s - %s' % (s, e)
 session_csrf_secret_name = "_session_csrf_secret_token"
 def session_csrf_secret_token():
     """Return (and create) the current session's CSRF protection token."""
     from tg import session
     if not session_csrf_secret_name in session:
         session[session_csrf_secret_name] = str(random.getrandbits(128))
         session.save()
     return session[session_csrf_secret_name]
 def form(url, method="post", **attrs):
     """Like webhelpers.html.tags.form , but automatically adding
     session_csrf_secret_token for POST. The secret is thus never leaked in GET
     URLs.
     """
     form = insecure_form(url, method, **attrs)
     if method.lower() == 'get':
         return form
     return form + HTML.div(hidden(session_csrf_secret_name, session_csrf_secret_token()), style="display: none;")

kallithea/lib/indexers/daemon.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.lib.indexers.daemon
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 A daemon will read from task table and run tasks
 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 26, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import os
 import sys
 import traceback
 from os.path import dirname
 from shutil import rmtree
 from time import mktime
 from whoosh.index import create_in, exists_in, open_dir
 from whoosh.qparser import QueryParser
 from kallithea.config.conf import INDEX_EXTENSIONS, INDEX_FILENAMES
 from kallithea.lib.indexers import CHGSET_IDX_NAME, CHGSETS_SCHEMA, IDX_NAME, SCHEMA
-from kallithea.lib.utils2 import safe_unicode
+from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs.exceptions import ChangesetError, NodeDoesNotExistError, RepositoryError
 from kallithea.model.db import Repository
 from kallithea.model.scm import ScmModel
 # Add location of top level folder to sys.path
 project_path = dirname(dirname(dirname(dirname(os.path.realpath(__file__)))))
 sys.path.append(project_path)
 log = logging.getLogger('whoosh_indexer')
 class WhooshIndexingDaemon(object):
     """
     Daemon for atomic indexing jobs
     """
     def __init__(self, indexname=IDX_NAME, index_location=None,
                  repo_location=None, repo_list=None,
                  repo_update_list=None):
         self.indexname = indexname
         self.index_location = index_location
         if not index_location:
             raise Exception('You have to provide index location')
         self.repo_location = repo_location
         if not repo_location:
             raise Exception('You have to provide repositories location')
         self.repo_paths = ScmModel().repo_scan(self.repo_location)
         # filter repo list
         if repo_list:
             repo_list = set(repo_list)
             self.filtered_repo_paths = {}
             for repo_name, repo in self.repo_paths.items():
                 if repo_name in repo_list:
                     self.filtered_repo_paths[repo_name] = repo
             self.repo_paths = self.filtered_repo_paths
         # filter update repo list
         self.filtered_repo_update_paths = {}
         if repo_update_list:
             self.filtered_repo_update_paths = {}
             for repo_name, repo in self.repo_paths.items():
                 if repo_name in repo_update_list:
                     self.filtered_repo_update_paths[repo_name] = repo
             self.repo_paths = self.filtered_repo_update_paths
         self.initial = True
         if not os.path.isdir(self.index_location):
             os.makedirs(self.index_location)
             log.info('Cannot run incremental index since it does not '
                      'yet exist - running full build')
         elif not exists_in(self.index_location, IDX_NAME):
             log.info('Running full index build, as the file content '
                      'index does not exist')
         elif not exists_in(self.index_location, CHGSET_IDX_NAME):
             log.info('Running full index build, as the changeset '
                      'index does not exist')
         else:
             self.initial = False
     def _get_index_revision(self, repo):
         db_repo = Repository.get_by_repo_name(repo.name)
         landing_rev = 'tip'
         if db_repo:
             _rev_type, _rev = db_repo.landing_rev
             landing_rev = _rev
         return landing_rev
     def _get_index_changeset(self, repo, index_rev=None):
         if not index_rev:
             index_rev = self._get_index_revision(repo)
         cs = repo.get_changeset(index_rev)
         return cs
     def get_paths(self, repo):
         """
         recursive walk in root dir and return a set of all path in that dir
         based on repository walk function
         """
         index_paths_ = set()
         try:
             cs = self._get_index_changeset(repo)
             for _topnode, _dirs, files in cs.walk('/'):
                 for f in files:
                     index_paths_.add(os.path.join(repo.path, f.path))
         except RepositoryError:
             log.debug(traceback.format_exc())
             pass
         return index_paths_
     def get_node(self, repo, path, index_rev=None):
         """
         gets a filenode based on given full path.
         :param repo: scm repo instance
         :param path: full path including root location
         :return: FileNode
         """
         # FIXME: paths should be normalized ... or even better: don't include repo.path
         assert path.startswith(repo.path)
         assert path[len(repo.path)] in (os.path.sep, os.path.altsep)
         node_path = path[len(repo.path) + 1:]
         cs = self._get_index_changeset(repo, index_rev=index_rev)
         node = cs.get_node(node_path)
         return node
     def is_indexable_node(self, node):
         """
         Just index the content of chosen files, skipping binary files
         """
         return (node.extension in INDEX_EXTENSIONS or node.name in INDEX_FILENAMES) and \
                not node.is_binary
     def get_node_mtime(self, node):
         return mktime(node.last_changeset.date.timetuple())
     def add_doc(self, writer, path, repo, repo_name, index_rev=None):
         """
         Adding doc to writer this function itself fetches data from
         the instance of vcs backend
         """
         try:
             node = self.get_node(repo, path, index_rev)
         except (ChangesetError, NodeDoesNotExistError):
             log.debug("    >> %s - not found in %s %s", path, repo, index_rev)
             return 0, 0
         indexed = indexed_w_content = 0
         if self.is_indexable_node(node):
             bytes_content = node.content
             if b'\0' in bytes_content:
                 log.warning('    >> %s - no text content', path)
                 u_content = u''
             else:
                 log.debug('    >> %s', path)
-                u_content = safe_unicode(bytes_content)
+                u_content = safe_str(bytes_content)
                 indexed_w_content += 1
         else:
             log.debug('    >> %s - not indexable', path)
             # just index file name without it's content
             u_content = u''
             indexed += 1
         writer.add_document(
             fileid=path,
             owner=repo.contact,
             repository_rawname=repo_name,
             repository=repo_name,
             path=path,
             content=u_content,
             modtime=self.get_node_mtime(node),
             extension=node.extension
+        )
         return indexed, indexed_w_content
     def index_changesets(self, writer, repo_name, repo, start_rev=None):
         """
         Add all changeset in the vcs repo starting at start_rev
         to the index writer
         :param writer: the whoosh index writer to add to
         :param repo_name: name of the repository from whence the
           changeset originates including the repository group
         :param repo: the vcs repository instance to index changesets for,
           the presumption is the repo has changesets to index
         :param start_rev=None: the full sha id to start indexing from
           if start_rev is None then index from the first changeset in
           the repo
         """
         if start_rev is None:
             start_rev = repo[0].raw_id
         log.debug('Indexing changesets in %s, starting at rev %s',
                   repo_name, start_rev)
         indexed = 0
         cs_iter = repo.get_changesets(start=start_rev)
         total = len(cs_iter)
         for cs in cs_iter:
             indexed += 1
             log.debug('    >> %s %s/%s', cs, indexed, total)
             writer.add_document(
                 raw_id=cs.raw_id,
                 owner=repo.contact,
                 date=cs._timestamp,
                 repository_rawname=repo_name,
                 repository=repo_name,
                 author=cs.author,
                 message=cs.message,
                 last=cs.last,
                 added=u' '.join(node.path for node in cs.added).lower(),
                 removed=u' '.join(node.path for node in cs.removed).lower(),
                 changed=u' '.join(node.path for node in cs.changed).lower(),
                 parents=u' '.join(cs.raw_id for cs in cs.parents),
+            )
         return indexed
     def index_files(self, file_idx_writer, repo_name, repo):
         """
         Index files for given repo_name
         :param file_idx_writer: the whoosh index writer to add to
         :param repo_name: name of the repository we're indexing
         :param repo: instance of vcs repo
         """
         i_cnt = iwc_cnt = 0
         log.debug('Building file index for %s @revision:%s', repo_name,
                                                 self._get_index_revision(repo))
         index_rev = self._get_index_revision(repo)
         for idx_path in self.get_paths(repo):
             i, iwc = self.add_doc(file_idx_writer, idx_path, repo, repo_name, index_rev)
             i_cnt += i
             iwc_cnt += iwc
         log.debug('added %s files %s with content for repo %s',
                   i_cnt + iwc_cnt, iwc_cnt, repo.path)
         return i_cnt, iwc_cnt
     def update_changeset_index(self):
         idx = open_dir(self.index_location, indexname=CHGSET_IDX_NAME)
         with idx.searcher() as searcher:
             writer = idx.writer()
             writer_is_dirty = False
             try:
                 indexed_total = 0
                 repo_name = None
                 for repo_name, repo in sorted(self.repo_paths.items()):
                     log.debug('Updating changeset index for repo %s', repo_name)
                     # skip indexing if there aren't any revs in the repo
                     num_of_revs = len(repo)
                     if num_of_revs < 1:
                         continue
                     qp = QueryParser('repository', schema=CHGSETS_SCHEMA)
                     q = qp.parse(u"last:t AND %s" % repo_name)
                     results = searcher.search(q)
                     # default to scanning the entire repo
                     last_rev = 0
                     start_id = None
                     if len(results) > 0:
                         # assuming that there is only one result, if not this
                         # may require a full re-index.
                         start_id = results[0]['raw_id']
                         last_rev = repo.get_changeset(revision=start_id).revision
                     # there are new changesets to index or a new repo to index
                     if last_rev == 0 or num_of_revs > last_rev + 1:
                         # delete the docs in the index for the previous
                         # last changeset(s)
                         for hit in results:
                             q = qp.parse(u"last:t AND %s AND raw_id:%s" %
                                             (repo_name, hit['raw_id']))
                             writer.delete_by_query(q)
                         # index from the previous last changeset + all new ones
                         indexed_total += self.index_changesets(writer,
                                                 repo_name, repo, start_id)
                         writer_is_dirty = True
                 log.debug('indexed %s changesets for repo %s',
                           indexed_total, repo_name
+                )
             finally:
                 if writer_is_dirty:
                     log.debug('>> COMMITING CHANGES TO CHANGESET INDEX<<')
                     writer.commit(merge=True)
                     log.debug('>>> FINISHED REBUILDING CHANGESET INDEX <<<')
                 else:
                     log.debug('>> NOTHING TO COMMIT TO CHANGESET INDEX<<')
     def update_file_index(self):
         log.debug(u'STARTING INCREMENTAL INDEXING UPDATE FOR EXTENSIONS %s '
                   'AND REPOS %s', INDEX_EXTENSIONS, ' and '.join(self.repo_paths))
         idx = open_dir(self.index_location, indexname=self.indexname)
         # The set of all paths in the index
         indexed_paths = set()
         # The set of all paths we need to re-index
         to_index = set()
         writer = idx.writer()
         writer_is_dirty = False
         try:
             with idx.reader() as reader:
                 # Loop over the stored fields in the index
                 for fields in reader.all_stored_fields():
                     indexed_path = fields['path']
                     indexed_repo_path = fields['repository']
                     indexed_paths.add(indexed_path)
                     if indexed_repo_path not in self.filtered_repo_update_paths:
                         continue
                     repo = self.repo_paths[indexed_repo_path]
                     try:
                         node = self.get_node(repo, indexed_path)
                         # Check if this file was changed since it was indexed
                         indexed_time = fields['modtime']
                         mtime = self.get_node_mtime(node)
                         if mtime > indexed_time:
                             # The file has changed, delete it and add it to
                             # the list of files to reindex
                             log.debug(
                                 'adding to reindex list %s mtime: %s vs %s',
                                     indexed_path, mtime, indexed_time
+                            )
                             writer.delete_by_term('fileid', indexed_path)
                             writer_is_dirty = True
                             to_index.add(indexed_path)
                     except (ChangesetError, NodeDoesNotExistError):
                         # This file was deleted since it was indexed
                         log.debug('removing from index %s', indexed_path)
                         writer.delete_by_term('path', indexed_path)
                         writer_is_dirty = True
             # Loop over the files in the filesystem
             # Assume we have a function that gathers the filenames of the
             # documents to be indexed
             ri_cnt_total = 0  # indexed

kallithea/lib/markup_renderer.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.lib.markup_renderer
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Renderer for markup languages with ability to parse using rst or markdown
 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: Oct 27, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import re
 import traceback
 import bleach
 import markdown as markdown_mod
-from kallithea.lib.utils2 import MENTIONS_REGEX, safe_unicode
+from kallithea.lib.utils2 import MENTIONS_REGEX, safe_str
 log = logging.getLogger(__name__)
 url_re = re.compile(r'''\bhttps?://(?:[\da-zA-Z0-9@:.-]+)'''
                     r'''(?:[/a-zA-Z0-9_=@#~&+%.,:;?!*()-]*[/a-zA-Z0-9_=@#~])?''')
 class MarkupRenderer(object):
     RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES = ['include', 'meta', 'raw']
     MARKDOWN_PAT = re.compile(r'md|mkdn?|mdown|markdown', re.IGNORECASE)
     RST_PAT = re.compile(r're?st', re.IGNORECASE)
     PLAIN_PAT = re.compile(r'readme', re.IGNORECASE)
     @classmethod
     def _detect_renderer(cls, source, filename):
         """
         runs detection of what renderer should be used for generating html
         from a markup language
         filename can be also explicitly a renderer name
         """
         if cls.MARKDOWN_PAT.findall(filename):
             return cls.markdown
         elif cls.RST_PAT.findall(filename):
             return cls.rst
         elif cls.PLAIN_PAT.findall(filename):
             return cls.rst
         return cls.plain
     @classmethod
     def _flavored_markdown(cls, text):
         """
         Github style flavored markdown
         :param text:
         """
         from hashlib import md5
         # Extract pre blocks.
         extractions = {}
         def pre_extraction_callback(matchobj):
             digest = md5(matchobj.group(0)).hexdigest()
             extractions[digest] = matchobj.group(0)
             return "{gfm-extraction-%s}" % digest
         pattern = re.compile(r'<pre>.*?</pre>', re.MULTILINE | re.DOTALL)
         text = re.sub(pattern, pre_extraction_callback, text)
         # Prevent foo_bar_baz from ending up with an italic word in the middle.
         def italic_callback(matchobj):
             s = matchobj.group(0)
             if list(s).count('_') >= 2:
                 return s.replace('_', r'\_')
             return s
         text = re.sub(r'^(?! {4}|\t)\w+_\w+_\w[\w_]*', italic_callback, text)
         # In very clear cases, let newlines become <br /> tags.
         def newline_callback(matchobj):
             if len(matchobj.group(1)) == 1:
                 return matchobj.group(0).rstrip() + '  \n'
             else:
                 return matchobj.group(0)
         pattern = re.compile(r'^[\w\<][^\n]*(\n+)', re.MULTILINE)
         text = re.sub(pattern, newline_callback, text)
         # Insert pre block extractions.
         def pre_insert_callback(matchobj):
             return '\n\n' + extractions[matchobj.group(1)]
         text = re.sub(r'{gfm-extraction-([0-9a-f]{32})\}',
                       pre_insert_callback, text)
         return text
     @classmethod
     def render(cls, source, filename=None):
         """
         Renders a given filename using detected renderer
         it detects renderers based on file extension or mimetype.
         At last it will just do a simple html replacing new lines with <br/>
         >>> MarkupRenderer.render('''<img id="a" style="margin-top:-1000px;color:red" src="http://example.com/test.jpg">''', '.md')
         u'<p><img id="a" src="http://example.com/test.jpg" style="color: red;"></p>'
         >>> MarkupRenderer.render('''<img class="c d" src="file://localhost/test.jpg">''', 'b.mkd')
         u'<p><img class="c d"></p>'
         >>> MarkupRenderer.render('''<a href="foo">foo</a>''', 'c.mkdn')
         u'<p><a href="foo">foo</a></p>'
         >>> MarkupRenderer.render('''<script>alert(1)</script>''', 'd.mdown')
         u'&lt;script&gt;alert(1)&lt;/script&gt;'
         >>> MarkupRenderer.render('''<div onclick="alert(2)">yo</div>''', 'markdown')
         u'<div>yo</div>'
         >>> MarkupRenderer.render('''<a href="javascript:alert(3)">yo</a>''', 'md')
         u'<p><a>yo</a></p>'
         """
         renderer = cls._detect_renderer(source, filename)
         readme_data = renderer(source)
         # Allow most HTML, while preventing XSS issues:
         # no <script> tags, no onclick attributes, no javascript
         # "protocol", and also limit styling to prevent defacing.
         return bleach.clean(readme_data,
             tags=['a', 'abbr', 'b', 'blockquote', 'br', 'code', 'dd',
                   'div', 'dl', 'dt', 'em', 'h1', 'h2', 'h3', 'h4', 'h5',
                   'h6', 'hr', 'i', 'img', 'li', 'ol', 'p', 'pre', 'span',
                   'strong', 'sub', 'sup', 'table', 'tbody', 'td', 'th',
                   'thead', 'tr', 'ul'],
             attributes=['class', 'id', 'style', 'label', 'title', 'alt', 'href', 'src'],
             styles=['color'],
             protocols=['http', 'https', 'mailto'],
+            )
     @classmethod
     def plain(cls, source, universal_newline=True):
-        source = safe_unicode(source)
+        source = safe_str(source)
         if universal_newline:
             newline = '\n'
             source = newline.join(source.splitlines())
         def url_func(match_obj):
             url_full = match_obj.group(0)
             return '<a href="%(url)s">%(url)s</a>' % ({'url': url_full})
         source = url_re.sub(url_func, source)
         return '<br />' + source.replace("\n", '<br />')
     @classmethod
     def markdown(cls, source, safe=True, flavored=False):
         """
         Convert Markdown (possibly GitHub Flavored) to INSECURE HTML, possibly
         with "safe" fall-back to plaintext. Output from this method should be sanitized before use.
         >>> MarkupRenderer.markdown('''<img id="a" style="margin-top:-1000px;color:red" src="http://example.com/test.jpg">''')
         u'<p><img id="a" style="margin-top:-1000px;color:red" src="http://example.com/test.jpg"></p>'
         >>> MarkupRenderer.markdown('''<img class="c d" src="file://localhost/test.jpg">''')
         u'<p><img class="c d" src="file://localhost/test.jpg"></p>'
         >>> MarkupRenderer.markdown('''<a href="foo">foo</a>''')
         u'<p><a href="foo">foo</a></p>'
         >>> MarkupRenderer.markdown('''<script>alert(1)</script>''')
         u'<script>alert(1)</script>'
         >>> MarkupRenderer.markdown('''<div onclick="alert(2)">yo</div>''')
         u'<div onclick="alert(2)">yo</div>'
         >>> MarkupRenderer.markdown('''<a href="javascript:alert(3)">yo</a>''')
         u'<p><a href="javascript:alert(3)">yo</a></p>'
         >>> MarkupRenderer.markdown('''## Foo''')
         u'<h2>Foo</h2>'
         >>> print MarkupRenderer.markdown('''
         ...     #!/bin/bash
         ...     echo "hello"
         ... ''')
         <table class="code-highlighttable"><tr><td class="linenos"><div class="linenodiv"><pre>1
 </pre></div></td><td class="code"><div class="code-highlight"><pre><span></span><span class="ch">#!/bin/bash</span>
         <span class="nb">echo</span> <span class="s2">&quot;hello&quot;</span>
         </pre></div>
         </td></tr></table>
         """
-        source = safe_unicode(source)
+        source = safe_str(source)
         try:
             if flavored:
                 source = cls._flavored_markdown(source)
             return markdown_mod.markdown(
                 source,
                 extensions=['markdown.extensions.codehilite', 'markdown.extensions.extra'],
                 extension_configs={'markdown.extensions.codehilite': {'css_class': 'code-highlight'}})
         except Exception:
             log.error(traceback.format_exc())
             if safe:
                 log.debug('Falling back to render in plain mode')
                 return cls.plain(source)
             else:
                 raise
     @classmethod
     def rst(cls, source, safe=True):
-        source = safe_unicode(source)
+        source = safe_str(source)
         try:
             from docutils.core import publish_parts
             from docutils.parsers.rst import directives
             docutils_settings = dict([(alias, None) for alias in
                                 cls.RESTRUCTUREDTEXT_DISALLOWED_DIRECTIVES])
             docutils_settings.update({'input_encoding': 'unicode',
                                       'report_level': 4})
             for k, v in docutils_settings.items():
                 directives.register_directive(k, v)
             parts = publish_parts(source=source,
                                   writer_name="html4css1",
                                   settings_overrides=docutils_settings)
             return parts['html_title'] + parts["fragment"]
         except ImportError:
             log.warning('Install docutils to use this function')
             return cls.plain(source)
         except Exception:
             log.error(traceback.format_exc())
             if safe:
                 log.debug('Falling back to render in plain mode')
                 return cls.plain(source)
             else:
                 raise
     @classmethod
     def rst_with_mentions(cls, source):
         def wrapp(match_obj):
             uname = match_obj.groups()[0]
             return r'\ **@%(uname)s**\ ' % {'uname': uname}
         mention_hl = MENTIONS_REGEX.sub(wrapp, source).strip()
         return cls.rst(mention_hl)

kallithea/lib/middleware/permanent_repo_url.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.lib.middleware.permanent_repo_url
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 middleware to handle permanent repo URLs, replacing PATH_INFO '/_123/yada' with
 '/name/of/repo/yada' after looking 123 up in the database.
 """
 from kallithea.lib.utils import fix_repo_id_name
-from kallithea.lib.utils2 import safe_bytes, safe_unicode
+from kallithea.lib.utils2 import safe_bytes, safe_str
 class PermanentRepoUrl(object):
     def __init__(self, app, config):
         self.application = app
         self.config = config
     def __call__(self, environ, start_response):
         # Extract path_info as get_path_info does, but do it explicitly because
         # we also have to do the reverse operation when patching it back in
-        path_info = safe_unicode(environ['PATH_INFO'])
+        path_info = safe_str(environ['PATH_INFO'])
         if path_info.startswith('/'): # it must
             path_info = '/' + fix_repo_id_name(path_info[1:])
             environ['PATH_INFO'] = safe_bytes(path_info)
         return self.application(environ, start_response)

kallithea/lib/utils2.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.lib.utils2
 ~~~~~~~~~~~~~~~~~~~~
 Some simple helper functions.
 Note: all these functions should be independent of Kallithea classes, i.e.
 models, controllers, etc.  to prevent import cycles.
 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 5, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 from __future__ import print_function
 import binascii
 import datetime
 import json
 import os
 import pwd
 import re
 import time
 import urllib.parse
 import urlobject
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from webhelpers2.text import collapse, remove_formatting, strip_tags
-from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str, safe_unicode  # re-export
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 def str2bool(_str):
     """
     returns True/False value from given string, it tries to translate the
     string into boolean
     :param _str: string value to translate into boolean
     :rtype: boolean
     :returns: boolean from given string
     """
     if _str is None:
         return False
     if _str in (True, False):
         return _str
     _str = str(_str).strip().lower()
     return _str in ('t', 'true', 'y', 'yes', 'on', '1')
 def aslist(obj, sep=None, strip=True):
     """
     Returns given string separated by sep as list
     :param obj:
     :param sep:
     :param strip:
     """
     if isinstance(obj, (str)):
         lst = obj.split(sep)
         if strip:
             lst = [v.strip() for v in lst]
         return lst
     elif isinstance(obj, (list, tuple)):
         return obj
     elif obj is None:
         return []
     else:
         return [obj]
 def convert_line_endings(line, mode):
     """
     Converts a given line  "line end" according to given mode
     Available modes are::
 - Unix
 - Mac
 - DOS
     :param line: given line to convert
     :param mode: mode to convert to
     :rtype: str
     :return: converted line according to mode
     """
     if mode == 0:
         line = line.replace('\r\n', '\n')
         line = line.replace('\r', '\n')
     elif mode == 1:
         line = line.replace('\r\n', '\r')
         line = line.replace('\n', '\r')
     elif mode == 2:
         line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
     return line
 def detect_mode(line, default):
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
     :param line: str line
     :param default: default
     :rtype: int
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 def generate_api_key():
     """
     Generates a random (presumably unique) API key.
     This value is used in URLs and "Bearer" HTTP Authorization headers,
     which in practice means it should only contain URL-safe characters
     (RFC 3986):
         unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
     """
     # Hexadecimal certainly qualifies as URL-safe.
     return ascii_str(binascii.hexlify(os.urandom(20)))
 def safe_int(val, default=None):
     """
     Returns int() of val if val is not convertable to int use default
     instead
     :param val:
     :param default:
     """
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
     return val
 def remove_suffix(s, suffix):
     if s.endswith(suffix):
         s = s[:-1 * len(suffix)]
     return s
 def remove_prefix(s, prefix):
     if s.startswith(prefix):
         s = s[len(prefix):]
     return s
 def age(prevdate, show_short_version=False, now=None):
     """
     turns a datetime into an age string.
     If show_short_version is True, then it will generate a not so accurate but shorter string,
     example: 2days ago, instead of 2 days and 23 hours ago.
     :param prevdate: datetime object
     :param show_short_version: if it should approximate the date and return a shorter string
     :rtype: unicode
     :returns: unicode words describing age
     """
     now = now or datetime.datetime.now()
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     future = False
     if prevdate > now:
         now, prevdate = prevdate, now
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     from dateutil import relativedelta
     for part in order:
         d = relativedelta.relativedelta(now, prevdate)
         deltas[part] = getattr(d, part + 's')
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
         part = order[num]
         carry_part = order[num - 1]
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if prevdate.month == 2 and (prevdate.year % 4 == 0 and
             (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)
         ):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[prevdate.month - 1]
         deltas['month'] -= 1
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
     # In short version, we want nicer handling of ages of more than a year
     if show_short_version:
         if deltas['year'] == 1:
             # ages between 1 and 2 years: show as months
             deltas['month'] += 12
             deltas['year'] = 0
         if deltas['year'] >= 2:
             # ages 2+ years: round
             if deltas['month'] > 6:
                 deltas['year'] += 1
                 deltas['month'] = 0

kallithea/lib/vcs/backends/git/changeset.py

➞

Show inline comments

 import re
 from io import BytesIO
 from itertools import chain
 from subprocess import PIPE, Popen
 from dulwich import objects
 from dulwich.config import ConfigFile
 from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError, ChangesetError, ImproperArchiveTypeError, NodeDoesNotExistError, RepositoryError, VCSError
 from kallithea.lib.vcs.nodes import (
     AddedFileNodesGenerator, ChangedFileNodesGenerator, DirNode, FileNode, NodeKind, RemovedFileNodesGenerator, RootNode, SubModuleNode)
-from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, date_fromtimestamp, safe_int, safe_unicode
+from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, date_fromtimestamp, safe_int, safe_str
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 class GitChangeset(BaseChangeset):
     """
     Represents state of the repository at a revision.
     """
     def __init__(self, repository, revision):
         self._stat_modes = {}
         self.repository = repository
         try:
             commit = self.repository._repo[ascii_bytes(revision)]
             if isinstance(commit, objects.Tag):
                 revision = safe_str(commit.object[1])
                 commit = self.repository._repo.get_object(commit.object[1])
         except KeyError:
             raise RepositoryError("Cannot get object with id %s" % revision)
         self.raw_id = ascii_str(commit.id)
         self.short_id = self.raw_id[:12]
         self._commit = commit  # a Dulwich Commmit with .id
         self._tree_id = commit.tree
         self._committer_property = 'committer'
         self._author_property = 'author'
         self._date_property = 'commit_time'
         self._date_tz_property = 'commit_timezone'
         self.revision = repository.revisions.index(self.raw_id)
         self.nodes = {}
         self._paths = {}
     @LazyProperty
     def bookmarks(self):
         return ()
     @LazyProperty
     def message(self):
-        return safe_unicode(self._commit.message)
+        return safe_str(self._commit.message)
     @LazyProperty
     def committer(self):
-        return safe_unicode(getattr(self._commit, self._committer_property))
+        return safe_str(getattr(self._commit, self._committer_property))
     @LazyProperty
     def author(self):
-        return safe_unicode(getattr(self._commit, self._author_property))
+        return safe_str(getattr(self._commit, self._author_property))
     @LazyProperty
     def date(self):
         return date_fromtimestamp(getattr(self._commit, self._date_property),
                                   getattr(self._commit, self._date_tz_property))
     @LazyProperty
     def _timestamp(self):
         return getattr(self._commit, self._date_property)
     @LazyProperty
     def status(self):
         """
         Returns modified, added, removed, deleted files for current changeset
         """
         return self.changed, self.added, self.removed
     @LazyProperty
     def tags(self):
         _tags = []
         for tname, tsha in self.repository.tags.items():
             if tsha == self.raw_id:
                 _tags.append(tname)
         return _tags
     @LazyProperty
     def branch(self):
         # Note: This function will return one branch name for the changeset -
         # that might not make sense in Git where branches() is a better match
         # for the basic model
         heads = self.repository._heads(reverse=False)
         ref = heads.get(self._commit.id)
         if ref:
-            return safe_unicode(ref)
+            return safe_str(ref)
     @LazyProperty
     def branches(self):
         heads = self.repository._heads(reverse=True)
         return [b for b in heads if heads[b] == self._commit.id] # FIXME: Inefficient ... and returning None!
     def _fix_path(self, path):
         """
         Paths are stored without trailing slash so we need to get rid off it if
         needed.
         """
         if path.endswith('/'):
             path = path.rstrip('/')
         return path
     def _get_id_for_path(self, path):
         # FIXME: Please, spare a couple of minutes and make those codes cleaner;
         if path not in self._paths:
             path = path.strip('/')
             # set root tree
             tree = self.repository._repo[self._tree_id]
             if path == '':
                 self._paths[''] = tree.id
                 return tree.id
             splitted = path.split('/')
             dirs, name = splitted[:-1], splitted[-1]
             curdir = ''
             # initially extract things from root dir
             for item, stat, id in tree.items():
                 if curdir:
                     name = '/'.join((curdir, item))
                 else:
                     name = item
                 self._paths[name] = id
                 self._stat_modes[name] = stat
             for dir in dirs:
                 if curdir:
                     curdir = '/'.join((curdir, dir))
                 else:
                     curdir = dir
                 dir_id = None
                 for item, stat, id in tree.items():
                     if dir == item:
                         dir_id = id
                 if dir_id:
                     # Update tree
                     tree = self.repository._repo[dir_id]
                     if not isinstance(tree, objects.Tree):
                         raise ChangesetError('%s is not a directory' % curdir)
                 else:
                     raise ChangesetError('%s have not been found' % curdir)
                 # cache all items from the given traversed tree
                 for item, stat, id in tree.items():
                     if curdir:
                         name = '/'.join((curdir, item))
                     else:
                         name = item
                     self._paths[name] = id
                     self._stat_modes[name] = stat
             if path not in self._paths:
                 raise NodeDoesNotExistError("There is no file nor directory "
                     "at the given path '%s' at revision %s"
                     % (path, self.short_id))
         return self._paths[path]
     def _get_kind(self, path):
         obj = self.repository._repo[self._get_id_for_path(path)]
         if isinstance(obj, objects.Blob):
             return NodeKind.FILE
         elif isinstance(obj, objects.Tree):
             return NodeKind.DIR
     def _get_filectx(self, path):
         path = self._fix_path(path)
         if self._get_kind(path) != NodeKind.FILE:
             raise ChangesetError("File does not exist for revision %s at "
                 " '%s'" % (self.raw_id, path))
         return path
     def _get_file_nodes(self):
         return chain(*(t[2] for t in self.walk()))
     @LazyProperty
     def parents(self):
         """
         Returns list of parents changesets.
         """
         return [self.repository.get_changeset(ascii_str(parent_id))
                 for parent_id in self._commit.parents]
     @LazyProperty
     def children(self):
         """
         Returns list of children changesets.
         """
         rev_filter = settings.GIT_REV_FILTER
         so = self.repository.run_git_command(
             ['rev-list', rev_filter, '--children']
+        )
         return [
             self.repository.get_changeset(cs)
             for parts in (l.split(' ') for l in so.splitlines())
             if parts[0] == self.raw_id
             for cs in parts[1:]
+        ]
     def next(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 next_ = cs.revision + 1
                 next_rev = cs.repository.revisions[next_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(next_rev)
             if not branch or branch == cs.branch:
                 return cs
     def prev(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 prev_ = cs.revision - 1
                 if prev_ < 0:
                     raise IndexError
                 prev_rev = cs.repository.revisions[prev_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(prev_rev)
             if not branch or branch == cs.branch:
                 return cs
     def diff(self, ignore_whitespace=True, context=3):
         # Only used to feed diffstat
         rev1 = self.parents[0] if self.parents else self.repository.EMPTY_CHANGESET
         rev2 = self
         return b''.join(self.repository.get_diff(rev1, rev2,
                                     ignore_whitespace=ignore_whitespace,
                                     context=context))
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given ``path``.
         """
         # ensure path is traversed
         self._get_id_for_path(path)
         return self._stat_modes[path]
     def get_file_content(self, path):
         """
         Returns content of the file at given ``path``.
         """
         id = self._get_id_for_path(path)
         blob = self.repository._repo[id]
         return blob.as_pretty_string()
     def get_file_size(self, path):
         """
         Returns size of the file at given ``path``.
         """
         id = self._get_id_for_path(path)
         blob = self.repository._repo[id]
         return blob.raw_length()
     def get_file_changeset(self, path):
         """
         Returns last commit of the file at the given ``path``.
         """
         return self.get_file_history(path, limit=1)[0]
     def get_file_history(self, path, limit=None):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         TODO: This function now uses os underlying 'git' and 'grep' commands
         which is generally not good. Should be replaced with algorithm
         iterating commits.
         """

kallithea/lib/vcs/backends/git/inmemory.py

➞

Show inline comments

 import datetime
 import posixpath
 import stat
 import time
 from dulwich import objects
 from kallithea.lib.vcs.backends.base import BaseInMemoryChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.utils import ascii_str, safe_bytes
 class GitInMemoryChangeset(BaseInMemoryChangeset):
     def commit(self, message, author, parents=None, branch=None, date=None,
                **kwargs):
         """
         Performs in-memory commit (doesn't check workdir in any way) and
         returns newly created ``Changeset``. Updates repository's
         ``revisions``.
         :param message: message of the commit
         :param author: full username, i.e. "Joe Doe <joe.doe@example.com>"
         :param parents: single parent or sequence of parents from which commit
           would be derived
         :param date: ``datetime.datetime`` instance. Defaults to
           ``datetime.datetime.now()``.
         :param branch: branch name, as string. If none given, default backend's
           branch would be used.
         :raises ``CommitError``: if any error occurs while committing
         """
         self.check_integrity(parents)
         from .repository import GitRepository
         if branch is None:
             branch = GitRepository.DEFAULT_BRANCH_NAME
         repo = self.repository._repo
         object_store = repo.object_store
-        ENCODING = b"UTF-8"  # TODO: should probably be kept in sync with safe_unicode/safe_bytes and vcs/conf/settings.py DEFAULT_ENCODINGS
+        ENCODING = b"UTF-8"  # TODO: should probably be kept in sync with safe_str/safe_bytes and vcs/conf/settings.py DEFAULT_ENCODINGS
         # Create tree and populates it with blobs
         commit_tree = self.parents[0] and repo[self.parents[0]._commit.tree] or \
             objects.Tree()
         for node in self.added + self.changed:
             # Compute subdirs if needed
             dirpath, nodename = posixpath.split(node.path)
             dirnames = safe_bytes(dirpath).split(b'/') if dirpath else []
             parent = commit_tree
             ancestors = [('', parent)]
             # Tries to dig for the deepest existing tree
             while dirnames:
                 curdir = dirnames.pop(0)
                 try:
                     dir_id = parent[curdir][1]
                 except KeyError:
                     # put curdir back into dirnames and stops
                     dirnames.insert(0, curdir)
                     break
                 else:
                     # If found, updates parent
                     parent = self.repository._repo[dir_id]
                     ancestors.append((curdir, parent))
             # Now parent is deepest existing tree and we need to create subtrees
             # for dirnames (in reverse order) [this only applies for nodes from added]
             new_trees = []
             blob = objects.Blob.from_string(node.content)
             node_path = safe_bytes(node.name)
             if dirnames:
                 # If there are trees which should be created we need to build
                 # them now (in reverse order)
                 reversed_dirnames = list(reversed(dirnames))
                 curtree = objects.Tree()
                 curtree[node_path] = node.mode, blob.id
                 new_trees.append(curtree)
                 for dirname in reversed_dirnames[:-1]:
                     newtree = objects.Tree()
                     #newtree.add(stat.S_IFDIR, dirname, curtree.id)
                     newtree[dirname] = stat.S_IFDIR, curtree.id
                     new_trees.append(newtree)
                     curtree = newtree
                 parent[reversed_dirnames[-1]] = stat.S_IFDIR, curtree.id
             else:
                 parent.add(name=node_path, mode=node.mode, hexsha=blob.id)
             new_trees.append(parent)
             # Update ancestors
             for parent, tree, path in reversed([(a[1], b[1], b[0]) for a, b in
                 zip(ancestors, ancestors[1:])]
             ):
                 parent[path] = stat.S_IFDIR, tree.id
                 object_store.add_object(tree)
             object_store.add_object(blob)
             for tree in new_trees:
                 object_store.add_object(tree)
         for node in self.removed:
             paths = safe_bytes(node.path).split(b'/')
             tree = commit_tree
             trees = [tree]
             # Traverse deep into the forest...
             for path in paths:
                 try:
                     obj = self.repository._repo[tree[path][1]]
                     if isinstance(obj, objects.Tree):
                         trees.append(obj)
                         tree = obj
                 except KeyError:
                     break
             # Cut down the blob and all rotten trees on the way back...
             for path, tree in reversed(list(zip(paths, trees))):
                 del tree[path]
                 if tree:
                     # This tree still has elements - don't remove it or any
                     # of it's parents
                     break
         object_store.add_object(commit_tree)
         # Create commit
         commit = objects.Commit()
         commit.tree = commit_tree.id
         commit.parents = [p._commit.id for p in self.parents if p]
         commit.author = commit.committer = safe_bytes(author)
         commit.encoding = ENCODING
         commit.message = safe_bytes(message)
         # Compute date
         if date is None:
             date = time.time()
         elif isinstance(date, datetime.datetime):
             date = time.mktime(date.timetuple())
         author_time = kwargs.pop('author_time', date)
         commit.commit_time = int(date)
         commit.author_time = int(author_time)
         tz = time.timezone
         author_tz = kwargs.pop('author_timezone', tz)
         commit.commit_timezone = tz
         commit.author_timezone = author_tz
         object_store.add_object(commit)
         # Update vcs repository object & recreate dulwich repo
         ref = b'refs/heads/%s' % safe_bytes(branch)
         repo.refs[ref] = commit.id
         self.repository.revisions.append(ascii_str(commit.id))
         # invalidate parsed refs after commit
         self.repository._parsed_refs = self.repository._get_parsed_refs()
         tip = self.repository.get_changeset()
         self.reset()
         return tip
     def _get_missing_trees(self, path, root_tree):
         """
         Creates missing ``Tree`` objects for the given path.
         :param path: path given as a string. It may be a path to a file node
           (i.e. ``foo/bar/baz.txt``) or directory path - in that case it must
           end with slash (i.e. ``foo/bar/``).
         :param root_tree: ``dulwich.objects.Tree`` object from which we start
           traversing (should be commit's root tree)
         """
         dirpath = posixpath.split(path)[0]
         dirs = dirpath.split('/')
         if not dirs or dirs == ['']:
             return []
         def get_tree_for_dir(tree, dirname):
             for name, mode, id in tree.items():
                 if name == dirname:
                     obj = self.repository._repo[id]
                     if isinstance(obj, objects.Tree):
                         return obj
                     else:
                         raise RepositoryError("Cannot create directory %s "
                             "at tree %s as path is occupied and is not a "
                             "Tree" % (dirname, tree))
             return None
         trees = []
         parent = root_tree
         for dirname in dirs:
             tree = get_tree_for_dir(parent, dirname)
             if tree is None:
                 tree = objects.Tree()
                 parent.add(stat.S_IFDIR, dirname, tree.id)
                 parent = tree
             # Always append tree
             trees.append(tree)
         return trees

kallithea/lib/vcs/backends/git/repository.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     vcs.backends.git.repository
     ~~~~~~~~~~~~~~~~~~~~~~~~~~~
     Git repository implementation.
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import errno
 import logging
 import os
 import re
 import time
 import urllib.error
 import urllib.parse
 import urllib.request
 from collections import OrderedDict
 import mercurial.url  # import httpbasicauthhandler, httpdigestauthhandler
 import mercurial.util  # import url as hg_url
 from dulwich.config import ConfigFile
 from dulwich.objects import Tag
 from dulwich.repo import NotGitRepository, Repo
 from kallithea.lib.vcs import subprocessio
 from kallithea.lib.vcs.backends.base import BaseRepository, CollectionGenerator
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import (
     BranchDoesNotExistError, ChangesetDoesNotExistError, EmptyRepositoryError, RepositoryError, TagAlreadyExistError, TagDoesNotExistError)
-from kallithea.lib.vcs.utils import ascii_str, date_fromtimestamp, makedate, safe_bytes, safe_unicode
+from kallithea.lib.vcs.utils import ascii_str, date_fromtimestamp, makedate, safe_bytes, safe_str
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.utils.paths import abspath, get_user_home
 from .changeset import GitChangeset
 from .inmemory import GitInMemoryChangeset
 from .workdir import GitWorkdir
 SHA_PATTERN = re.compile(r'^([0-9a-fA-F]{12}|[0-9a-fA-F]{40})$')
 log = logging.getLogger(__name__)
 class GitRepository(BaseRepository):
     """
     Git repository backend.
     """
     DEFAULT_BRANCH_NAME = 'master'
     scm = 'git'
     def __init__(self, repo_path, create=False, src_url=None,
                  update_after_clone=False, bare=False):
         self.path = abspath(repo_path)
         self.repo = self._get_repo(create, src_url, update_after_clone, bare)
         self.bare = self.repo.bare
     @property
     def _config_files(self):
         return [
             self.bare and abspath(self.path, 'config')
                       or abspath(self.path, '.git', 'config'),
              abspath(get_user_home(), '.gitconfig'),
+         ]
     @property
     def _repo(self):
         return self.repo
     @property
     def head(self):
         try:
             return self._repo.head()
         except KeyError:
             return None
     @property
     def _empty(self):
         """
         Checks if repository is empty ie. without any changesets
         """
         try:
             self.revisions[0]
         except (KeyError, IndexError):
             return True
         return False
     @LazyProperty
     def revisions(self):
         """
         Returns list of revisions' ids, in ascending order.  Being lazy
         attribute allows external tools to inject shas from cache.
         """
         return self._get_all_revisions()
     @classmethod
     def _run_git_command(cls, cmd, cwd=None):
         """
         Runs given ``cmd`` as git command and returns output bytes in a tuple
         (stdout, stderr) ... or raise RepositoryError.
         :param cmd: git command to be executed
         :param cwd: passed directly to subprocess
         """
         # need to clean fix GIT_DIR !
         gitenv = dict(os.environ)
         gitenv.pop('GIT_DIR', None)
         gitenv['GIT_CONFIG_NOGLOBAL'] = '1'
         assert isinstance(cmd, list), cmd
         cmd = [settings.GIT_EXECUTABLE_PATH, '-c', 'core.quotepath=false'] + cmd
         try:
             p = subprocessio.SubprocessIOChunker(cmd, cwd=cwd, env=gitenv, shell=False)
         except (EnvironmentError, OSError) as err:
             # output from the failing process is in str(EnvironmentError)
             msg = ("Couldn't run git command %s.\n"
                    "Subprocess failed with '%s': %s\n" %
                    (cmd, type(err).__name__, err)
             ).strip()
             log.error(msg)
             raise RepositoryError(msg)
         try:
             stdout = b''.join(p.output)
             stderr = b''.join(p.error)
         finally:
             p.close()
         # TODO: introduce option to make commands fail if they have any stderr output?
         if stderr:
             log.debug('stderr from %s:\n%s', cmd, stderr)
         else:
             log.debug('stderr from %s: None', cmd)
         return stdout, stderr
     def run_git_command(self, cmd):
         """
         Runs given ``cmd`` as git command with cwd set to current repo.
         Returns stdout as unicode str ... or raise RepositoryError.
         """
         cwd = None
         if os.path.isdir(self.path):
             cwd = self.path
         stdout, _stderr = self._run_git_command(cmd, cwd=cwd)
-        return safe_unicode(stdout)
+        return safe_str(stdout)
     @classmethod
     def _check_url(cls, url):
         """
         Function will check given url and try to verify if it's a valid
         link. Sometimes it may happened that git will issue basic
         auth request that can cause whole API to hang when used from python
         or other external calls.
         On failures it'll raise urllib2.HTTPError, exception is also thrown
         when the return code is non 200
         """
         # check first if it's not an local url
         if os.path.isdir(url) or url.startswith('file:'):
             return True
         if url.startswith('git://'):
             return True
         if '+' in url[:url.find('://')]:
             url = url[url.find('+') + 1:]
         handlers = []
         url_obj = mercurial.util.url(safe_bytes(url))
         test_uri, authinfo = url_obj.authinfo()
         if not test_uri.endswith('info/refs'):
             test_uri = test_uri.rstrip('/') + '/info/refs'
         url_obj.passwd = b'*****'
         cleaned_uri = str(url_obj)
         if authinfo:
             # create a password manager
             passmgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
             passmgr.add_password(*authinfo)
             handlers.extend((mercurial.url.httpbasicauthhandler(passmgr),
                              mercurial.url.httpdigestauthhandler(passmgr)))
         o = urllib.request.build_opener(*handlers)
         o.addheaders = [('User-Agent', 'git/1.7.8.0')]  # fake some git
         req = urllib.request.Request(
             "%s?%s" % (
                 test_uri,
                 urllib.parse.urlencode({"service": 'git-upload-pack'})
             ))
         try:
             resp = o.open(req)
             if resp.code != 200:
                 raise Exception('Return Code is not 200')
         except Exception as e:
             # means it cannot be cloned
             raise urllib.error.URLError("[%s] org_exc: %s" % (cleaned_uri, e))
         # now detect if it's proper git repo
         gitdata = resp.read()
         if 'service=git-upload-pack' not in gitdata:
             raise urllib.error.URLError(
                 "url [%s] does not look like an git" % cleaned_uri)
         return True
     def _get_repo(self, create, src_url=None, update_after_clone=False,
                   bare=False):
         if create and os.path.exists(self.path):
             raise RepositoryError("Location already exist")
         if src_url and not create:
             raise RepositoryError("Create should be set to True if src_url is "
                                   "given (clone operation creates repository)")
         try:
             if create and src_url:
                 GitRepository._check_url(src_url)
                 self.clone(src_url, update_after_clone, bare)
                 return Repo(self.path)
             elif create:
                 os.makedirs(self.path)
                 if bare:
                     return Repo.init_bare(self.path)
                 else:
                     return Repo.init(self.path)
             else:
                 return Repo(self.path)
         except (NotGitRepository, OSError) as err:
             raise RepositoryError(err)
     def _get_all_revisions(self):
         # we must check if this repo is not empty, since later command
         # fails if it is. And it's cheaper to ask than throw the subprocess
         # errors
         try:
             self._repo.head()
         except KeyError:
             return []
         rev_filter = settings.GIT_REV_FILTER
         cmd = ['rev-list', rev_filter, '--reverse', '--date-order']
         try:
             so = self.run_git_command(cmd)
         except RepositoryError:
             # Can be raised for empty repositories
             return []
         return so.splitlines()
     def _get_all_revisions2(self):
         # alternate implementation using dulwich
         includes = [ascii_str(sha) for key, (sha, type_) in self._parsed_refs.items()
                     if type_ != b'T']
         return [c.commit.id for c in self._repo.get_walker(include=includes)]
     def _get_revision(self, revision):
         """
         Given any revision identifier, returns a 40 char string with revision hash.
         """
         if self._empty:
             raise EmptyRepositoryError("There are no changesets yet")
         if revision in (None, '', 'tip', 'HEAD', 'head', -1):
             revision = -1
         if isinstance(revision, int):
             try:
                 return self.revisions[revision]
             except IndexError:
                 msg = "Revision %r does not exist for %s" % (revision, self.name)
                 raise ChangesetDoesNotExistError(msg)
         if isinstance(revision, (str, unicode)):
             if revision.isdigit() and (len(revision) < 12 or len(revision) == revision.count('0')):
                 try:
                     return self.revisions[int(revision)]
                 except IndexError:
                     msg = "Revision %r does not exist for %s" % (revision, self)
                     raise ChangesetDoesNotExistError(msg)
             # get by branch/tag name
             _ref_revision = self._parsed_refs.get(safe_bytes(revision))
             if _ref_revision:  # and _ref_revision[1] in [b'H', b'RH', b'T']:
                 return ascii_str(_ref_revision[0])
             if revision in self.revisions:
                 return revision
             # maybe it's a tag ? we don't have them in self.revisions
             if revision in self.tags.values():
                 return revision
             if SHA_PATTERN.match(revision):
                 msg = "Revision %r does not exist for %s" % (revision, self.name)
                 raise ChangesetDoesNotExistError(msg)
         raise ChangesetDoesNotExistError("Given revision %r not recognized" % revision)
     def get_ref_revision(self, ref_type, ref_name):
         """
         Returns ``GitChangeset`` object representing repository's
         changeset at the given ``revision``.
         """
         return self._get_revision(ref_name)
     def _get_archives(self, archive_name='tip'):
         for i in [('zip', '.zip'), ('gz', '.tar.gz'), ('bz2', '.tar.bz2')]:
             yield {"type": i[0], "extension": i[1], "node": archive_name}
     def _get_url(self, url):
         """
         Returns normalized url. If schema is not given, would fall to
         filesystem (``file:///``) schema.
         """
         if url != 'default' and '://' not in url:
             url = ':///'.join(('file', url))
         return url
     @LazyProperty
     def name(self):
         return os.path.basename(self.path)
     @LazyProperty
     def last_change(self):
         """
         Returns last change made on this repository as datetime object
         """
         return date_fromtimestamp(self._get_mtime(), makedate()[1])
     def _get_mtime(self):
         try:
             return time.mktime(self.get_changeset().date.timetuple())
         except RepositoryError:
             idx_loc = '' if self.bare else '.git'
             # fallback to filesystem
             in_path = os.path.join(self.path, idx_loc, "index")
             he_path = os.path.join(self.path, idx_loc, "HEAD")
             if os.path.exists(in_path):
                 return os.stat(in_path).st_mtime
             else:
                 return os.stat(he_path).st_mtime
     @LazyProperty
     def description(self):
-        return safe_unicode(self._repo.get_description() or b'unknown')
+        return safe_str(self._repo.get_description() or b'unknown')
     @LazyProperty
     def contact(self):
         undefined_contact = u'Unknown'
         return undefined_contact
     @property
     def branches(self):
         if not self.revisions:
             return {}
         sortkey = lambda ctx: ctx[0]
         _branches = [(key, ascii_str(sha))
                      for key, (sha, type_) in self._parsed_refs.items() if type_ == b'H']
         return OrderedDict(sorted(_branches, key=sortkey, reverse=False))
     @LazyProperty
     def closed_branches(self):
         return {}
     @LazyProperty
     def tags(self):
         return self._get_tags()
     def _get_tags(self):
         if not self.revisions:
             return {}
         sortkey = lambda ctx: ctx[0]
         _tags = [(key, ascii_str(sha))
                  for key, (sha, type_) in self._parsed_refs.items() if type_ == b'T']
         return OrderedDict(sorted(_tags, key=sortkey, reverse=True))
     def tag(self, name, user, revision=None, message=None, date=None,
             **kwargs):
         """
         Creates and returns a tag for the given ``revision``.
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param revision: changeset id for which new tag would be created
         :param message: message of the tag's commit
         :param date: date of tag's commit
         :raises TagAlreadyExistError: if tag with same name already exists
         """
         if name in self.tags:
             raise TagAlreadyExistError("Tag %s already exists" % name)
         changeset = self.get_changeset(revision)
         message = message or "Added tag %s for commit %s" % (name,
             changeset.raw_id)
         self._repo.refs[b"refs/tags/%s" % safe_bytes(name)] = changeset._commit.id
         self._parsed_refs = self._get_parsed_refs()
         self.tags = self._get_tags()
         return changeset
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
         :param name: name of the tag to be removed
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param message: message of the tag's removal commit
         :param date: date of tag's removal commit
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         if name not in self.tags:
             raise TagDoesNotExistError("Tag %s does not exist" % name)
         # self._repo.refs is a DiskRefsContainer, and .path gives the full absolute path of '.git'
         tagpath = os.path.join(self._repo.refs.path, 'refs', 'tags', name)
         try:
             os.remove(tagpath)
             self._parsed_refs = self._get_parsed_refs()
             self.tags = self._get_tags()
         except OSError as e:
             raise RepositoryError(e.strerror)
     @LazyProperty
     def bookmarks(self):
         """
         Gets bookmarks for this repository
         """
         return {}
     @LazyProperty
     def _parsed_refs(self):
         return self._get_parsed_refs()
     def _get_parsed_refs(self):
         """Return refs as a dict, like:
         { b'v0.2.0': [b'599ba911aa24d2981225f3966eb659dfae9e9f30', b'T'] }
         """
         _repo = self._repo
         refs = _repo.get_refs()
         keys = [(b'refs/heads/', b'H'),
                 (b'refs/remotes/origin/', b'RH'),
                 (b'refs/tags/', b'T')]
         _refs = {}
         for ref, sha in refs.items():
             for k, type_ in keys:
                 if ref.startswith(k):
                     _key = ref[len(k):]
                     if type_ == b'T':
                         obj = _repo.get_object(sha)
                         if isinstance(obj, Tag):
                             sha = _repo.get_object(sha).object[1]
                     _refs[_key] = [sha, type_]
                     break
         return _refs
     def _heads(self, reverse=False):
         refs = self._repo.get_refs()
         heads = {}
         for key, val in refs.items():
             for ref_key in [b'refs/heads/', b'refs/remotes/origin/']:
                 if key.startswith(ref_key):
                     n = key[len(ref_key):]
                     if n not in [b'HEAD']:
                         heads[n] = val
         return heads if reverse else dict((y, x) for x, y in heads.items())
     def get_changeset(self, revision=None):
         """
         Returns ``GitChangeset`` object representing commit from git repository
         at the given revision or head (most recent commit) if None given.
         """
         if isinstance(revision, GitChangeset):
             return revision
         return GitChangeset(repository=self, revision=self._get_revision(revision))
     def get_changesets(self, start=None, end=None, start_date=None,
            end_date=None, branch_name=None, reverse=False, max_revisions=None):
         """
         Returns iterator of ``GitChangeset`` objects from start to end (both
         are inclusive), in ascending date order (unless ``reverse`` is set).
         :param start: changeset ID, as str; first returned changeset
         :param end: changeset ID, as str; last returned changeset
         :param start_date: if specified, changesets with commit date less than
           ``start_date`` would be filtered out from returned set
         :param end_date: if specified, changesets with commit date greater than
           ``end_date`` would be filtered out from returned set
         :param branch_name: if specified, changesets not reachable from given
           branch would be filtered out from returned set
         :param reverse: if ``True``, returned generator would be reversed
           (meaning that returned changesets would have descending date order)
         :raise BranchDoesNotExistError: If given ``branch_name`` does not
             exist.
         :raise ChangesetDoesNotExistError: If changeset for given ``start`` or
           ``end`` could not be found.
         """
         if branch_name and branch_name not in self.branches:
             raise BranchDoesNotExistError("Branch '%s' not found"
                                           % branch_name)
         # actually we should check now if it's not an empty repo to not spaw
         # subprocess commands
         if self._empty:
             raise EmptyRepositoryError("There are no changesets yet")
         # %H at format means (full) commit hash, initial hashes are retrieved
         # in ascending date order
         cmd = ['log', '--date-order', '--reverse', '--pretty=format:%H']
         if max_revisions:
             cmd += ['--max-count=%s' % max_revisions]
         if start_date:
             cmd += ['--since', start_date.strftime('%m/%d/%y %H:%M:%S')]
         if end_date:
             cmd += ['--until', end_date.strftime('%m/%d/%y %H:%M:%S')]
         if branch_name:
             cmd.append(branch_name)
         else:
             cmd.append(settings.GIT_REV_FILTER)
         revs = self.run_git_command(cmd).splitlines()
         start_pos = 0
         end_pos = len(revs)
         if start:
             _start = self._get_revision(start)
             try:
                 start_pos = revs.index(_start)
             except ValueError:
                 pass
         if end is not None:
             _end = self._get_revision(end)
             try:
                 end_pos = revs.index(_end)

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

0 comments (0 inline, 0 general)