kallithea Changeset - 95ca00cd722f

Changeset - 95ca00cd722f

Parent rev.

Child rev.

[Not reviewed]

stable

0 5 0

Mads Kiilerich - 6 years ago 2020-02-03 15:46:22
mads@kiilerich.com

Grafted from: 7e5b9fa298df

cleanup: minor correctness fixes

"Trivial" potential problems spotted with pytype.

5 files changed with 9 insertions and 5 deletions:

kallithea/controllers/admin/repo_groups.py

kallithea/controllers/pullrequests.py

kallithea/lib/diffs.py

kallithea/lib/utils2.py

kallithea/lib/vcs/nodes.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/repo_groups.py

➞

Show inline comments

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

kallithea/controllers/pullrequests.py

➞

Show inline comments

@@ @@ -89,557 +89,558 @@ class PullrequestsController(BaseRepoCon @@
         if branch:
             branch = safe_str(branch)
         if branch_rev:
             branch_rev = safe_str(branch_rev)
             # a revset not restricting to merge() would be better
             # (especially because it would get the branch point)
             # ... but is currently too expensive
             # including branches of children could be nice too
             peerbranches = set()
             for i in repo._repo.revs(
                 "sort(parents(branch(id(%s)) and merge()) - branch(id(%s)), -rev)",
                 branch_rev, branch_rev
             ):
                 for abranch in repo.get_changeset(i).branches:
                     if abranch not in peerbranches:
                         n = 'branch:%s:%s' % (abranch, repo.get_changeset(abranch).raw_id)
                         peers.append((n, abranch))
                         peerbranches.add(abranch)
         selected = None
         tiprev = repo.tags.get('tip')
         tipbranch = None
         branches = []
         for abranch, branchrev in repo.branches.iteritems():
             n = 'branch:%s:%s' % (abranch, branchrev)
             desc = abranch
             if branchrev == tiprev:
                 tipbranch = abranch
                 desc = '%s (current tip)' % desc
             branches.append((n, desc))
             if rev == branchrev:
                 selected = n
             if branch == abranch:
                 if not rev:
                     selected = n
                 branch = None
         if branch:  # branch not in list - it is probably closed
             branchrev = repo.closed_branches.get(branch)
             if branchrev:
                 n = 'branch:%s:%s' % (branch, branchrev)
                 branches.append((n, _('%s (closed)') % branch))
                 selected = n
                 branch = None
             if branch:
                 log.debug('branch %r not found in %s', branch, repo)
         bookmarks = []
         for bookmark, bookmarkrev in repo.bookmarks.iteritems():
             n = 'book:%s:%s' % (bookmark, bookmarkrev)
             bookmarks.append((n, bookmark))
             if rev == bookmarkrev:
                 selected = n
         tags = []
         for tag, tagrev in repo.tags.iteritems():
             if tag == 'tip':
                 continue
             n = 'tag:%s:%s' % (tag, tagrev)
             tags.append((n, tag))
             # note: even if rev == tagrev, don't select the static tag - it must be chosen explicitly
         # prio 1: rev was selected as existing entry above
         # prio 2: create special entry for rev; rev _must_ be used
         specials = []
         if rev and selected is None:
             selected = 'rev:%s:%s' % (rev, rev)
             specials = [(selected, '%s: %s' % (_("Changeset"), rev[:12]))]
         # prio 3: most recent peer branch
         if peers and not selected:
             selected = peers[0][0]
         # prio 4: tip revision
         if not selected:
             if h.is_hg(repo):
                 if tipbranch:
                     selected = 'branch:%s:%s' % (tipbranch, tiprev)
                 else:
                     selected = 'tag:null:' + repo.EMPTY_CHANGESET
                     tags.append((selected, 'null'))
             else:
                 if 'master' in repo.branches:
                     selected = 'branch:master:%s' % repo.branches['master']
                 else:
                     k, v = repo.branches.items()[0]
                     selected = 'branch:%s:%s' % (k, v)
         groups = [(specials, _("Special")),
                   (peers, _("Peer branches")),
                   (bookmarks, _("Bookmarks")),
                   (branches, _("Branches")),
                   (tags, _("Tags")),
+                  ]
         return [g for g in groups if g[0]], selected
     def _is_allowed_to_change_status(self, pull_request):
         if pull_request.is_closed():
             return False
         owner = request.authuser.user_id == pull_request.owner_id
         reviewer = PullRequestReviewer.query() \
             .filter(PullRequestReviewer.pull_request == pull_request) \
             .filter(PullRequestReviewer.user_id == request.authuser.user_id) \
             .count() != 0
         return request.authuser.admin or owner or reviewer
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def show_all(self, repo_name):
         c.from_ = request.GET.get('from_') or ''
         c.closed = request.GET.get('closed') or ''
         p = safe_int(request.GET.get('page'), 1)
         q = PullRequest.query(include_closed=c.closed, sorted=True)
         if c.from_:
             q = q.filter_by(org_repo=c.db_repo)
         else:
             q = q.filter_by(other_repo=c.db_repo)
         c.pull_requests = q.all()
         c.pullrequests_pager = Page(c.pull_requests, page=p, items_per_page=100)
         return render('/pullrequests/pullrequest_show_all.html')
     @LoginRequired()
     def show_my(self):
         c.closed = request.GET.get('closed') or ''
         c.my_pull_requests = PullRequest.query(
             include_closed=c.closed,
             sorted=True,
         ).filter_by(owner_id=request.authuser.user_id).all()
         c.participate_in_pull_requests = []
         c.participate_in_pull_requests_todo = []
         done_status = set([ChangesetStatus.STATUS_APPROVED, ChangesetStatus.STATUS_REJECTED])
         for pr in PullRequest.query(
             include_closed=c.closed,
             reviewer_id=request.authuser.user_id,
             sorted=True,
         ):
             status = pr.user_review_status(request.authuser.user_id) # very inefficient!!!
             if status in done_status:
                 c.participate_in_pull_requests.append(pr)
             else:
                 c.participate_in_pull_requests_todo.append(pr)
         return render('/pullrequests/pullrequest_show_my.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def index(self):
         org_repo = c.db_repo
         org_scm_instance = org_repo.scm_instance
         try:
             org_scm_instance.get_changeset()
         except EmptyRepositoryError as e:
             h.flash(_('There are no changesets yet'),
                     category='warning')
             raise HTTPFound(location=url('summary_home', repo_name=org_repo.repo_name))
         org_rev = request.GET.get('rev_end')
         # rev_start is not directly useful - its parent could however be used
         # as default for other and thus give a simple compare view
         rev_start = request.GET.get('rev_start')
         other_rev = None
         if rev_start:
             starters = org_repo.get_changeset(rev_start).parents
             if starters:
                 other_rev = starters[0].raw_id
             else:
                 other_rev = org_repo.scm_instance.EMPTY_CHANGESET
         branch = request.GET.get('branch')
         c.cs_repos = [(org_repo.repo_name, org_repo.repo_name)]
         c.default_cs_repo = org_repo.repo_name
         c.cs_refs, c.default_cs_ref = self._get_repo_refs(org_scm_instance, rev=org_rev, branch=branch)
         default_cs_ref_type, default_cs_branch, default_cs_rev = c.default_cs_ref.split(':')
         if default_cs_ref_type != 'branch':
             default_cs_branch = org_repo.get_changeset(default_cs_rev).branch
         # add org repo to other so we can open pull request against peer branches on itself
         c.a_repos = [(org_repo.repo_name, '%s (self)' % org_repo.repo_name)]
         if org_repo.parent:
             # add parent of this fork also and select it.
             # use the same branch on destination as on source, if available.
             c.a_repos.append((org_repo.parent.repo_name, '%s (parent)' % org_repo.parent.repo_name))
             c.a_repo = org_repo.parent
             c.a_refs, c.default_a_ref = self._get_repo_refs(
                     org_repo.parent.scm_instance, branch=default_cs_branch, rev=other_rev)
         else:
             c.a_repo = org_repo
             c.a_refs, c.default_a_ref = self._get_repo_refs(org_scm_instance, rev=other_rev)
         # gather forks and add to this list ... even though it is rare to
         # request forks to pull from their parent
         for fork in org_repo.forks:
             c.a_repos.append((fork.repo_name, fork.repo_name))
         return render('/pullrequests/pullrequest.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def repo_info(self, repo_name):
         repo = c.db_repo
         refs, selected_ref = self._get_repo_refs(repo.scm_instance)
         return {
             'description': repo.description.split('\n', 1)[0],
             'selected_ref': selected_ref,
             'refs': refs,
+            }
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def create(self, repo_name):
         repo = c.db_repo
         try:
             _form = PullRequestForm(repo.repo_id)().to_python(request.POST)
         except formencode.Invalid as errors:
             log.error(traceback.format_exc())
             log.error(str(errors))
             msg = _('Error creating pull request: %s') % errors.msg
             h.flash(msg, 'error')
             raise HTTPBadRequest
         # heads up: org and other might seem backward here ...
         org_ref = _form['org_ref'] # will have merge_rev as rev but symbolic name
         org_repo = Repository.guess_instance(_form['org_repo'])
         other_ref = _form['other_ref'] # will have symbolic name and head revision
         other_repo = Repository.guess_instance(_form['other_repo'])
         reviewers = []
         title = _form['pullrequest_title']
         description = _form['pullrequest_desc'].strip()
         owner = User.get(request.authuser.user_id)
         try:
             cmd = CreatePullRequestAction(org_repo, other_repo, org_ref, other_ref, title, description, owner, reviewers)
         except CreatePullRequestAction.ValidationError as e:
             h.flash(str(e), category='error', logf=log.error)
             raise HTTPNotFound
         try:
             pull_request = cmd.execute()
             Session().commit()
         except Exception:
             h.flash(_('Error occurred while creating pull request'),
                     category='error')
             log.error(traceback.format_exc())
             raise HTTPFound(location=url('pullrequest_home', repo_name=repo_name))
         h.flash(_('Successfully opened new pull request'),
                 category='success')
         raise HTTPFound(location=pull_request.url())
     def create_new_iteration(self, old_pull_request, new_rev, title, description, reviewers):
         owner = User.get(request.authuser.user_id)
         new_org_rev = self._get_ref_rev(old_pull_request.org_repo, 'rev', new_rev)
         new_other_rev = self._get_ref_rev(old_pull_request.other_repo, old_pull_request.other_ref_parts[0], old_pull_request.other_ref_parts[1])
         try:
             cmd = CreatePullRequestIterationAction(old_pull_request, new_org_rev, new_other_rev, title, description, owner, reviewers)
         except CreatePullRequestAction.ValidationError as e:
             h.flash(str(e), category='error', logf=log.error)
             raise HTTPNotFound
         try:
             pull_request = cmd.execute()
             Session().commit()
         except Exception:
             h.flash(_('Error occurred while creating pull request'),
                     category='error')
             log.error(traceback.format_exc())
             raise HTTPFound(location=old_pull_request.url())
         h.flash(_('New pull request iteration created'),
                 category='success')
         raise HTTPFound(location=pull_request.url())
     # pullrequest_post for PR editing
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     def post(self, repo_name, pull_request_id):
         pull_request = PullRequest.get_or_404(pull_request_id)
         if pull_request.is_closed():
             raise HTTPForbidden()
         assert pull_request.other_repo.repo_name == repo_name
         # only owner or admin can update it
         owner = pull_request.owner_id == request.authuser.user_id
         repo_admin = h.HasRepoPermissionLevel('admin')(c.repo_name)
         if not (h.HasPermissionAny('hg.admin')() or repo_admin or owner):
             raise HTTPForbidden()
         _form = PullRequestPostForm()().to_python(request.POST)
         cur_reviewers = set(pull_request.get_reviewer_users())
         new_reviewers = set(_get_reviewer(s) for s in _form['review_members'])
         old_reviewers = set(_get_reviewer(s) for s in _form['org_review_members'])
         other_added = cur_reviewers - old_reviewers
         other_removed = old_reviewers - cur_reviewers
         if other_added:
             h.flash(_('Meanwhile, the following reviewers have been added: %s') %
                     (', '.join(u.username for u in other_added)),
                     category='warning')
         if other_removed:
             h.flash(_('Meanwhile, the following reviewers have been removed: %s') %
                     (', '.join(u.username for u in other_removed)),
                     category='warning')
         if _form['updaterev']:
             return self.create_new_iteration(pull_request,
                                       _form['updaterev'],
                                       _form['pullrequest_title'],
                                       _form['pullrequest_desc'],
                                       new_reviewers)
         added_reviewers = new_reviewers - old_reviewers - cur_reviewers
         removed_reviewers = (old_reviewers - new_reviewers) & cur_reviewers
         old_description = pull_request.description
         pull_request.title = _form['pullrequest_title']
         pull_request.description = _form['pullrequest_desc'].strip() or _('No description')
         pull_request.owner = User.get_by_username(_form['owner'])
         user = User.get(request.authuser.user_id)
         PullRequestModel().mention_from_description(user, pull_request, old_description)
         PullRequestModel().add_reviewers(user, pull_request, added_reviewers)
         PullRequestModel().remove_reviewers(user, pull_request, removed_reviewers)
         Session().commit()
         h.flash(_('Pull request updated'), category='success')
         raise HTTPFound(location=pull_request.url())
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def delete(self, repo_name, pull_request_id):
         pull_request = PullRequest.get_or_404(pull_request_id)
         # only owner can delete it !
         if pull_request.owner_id == request.authuser.user_id:
             PullRequestModel().delete(pull_request)
             Session().commit()
             h.flash(_('Successfully deleted pull request'),
                     category='success')
             raise HTTPFound(location=url('my_pullrequests'))
         raise HTTPForbidden()
     @LoginRequired(allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def show(self, repo_name, pull_request_id, extra=None):
         c.pull_request = PullRequest.get_or_404(pull_request_id)
         c.allowed_to_change_status = self._is_allowed_to_change_status(c.pull_request)
         cc_model = ChangesetCommentsModel()
         cs_model = ChangesetStatusModel()
         # pull_requests repo_name we opened it against
         # ie. other_repo must match
         if repo_name != c.pull_request.other_repo.repo_name:
             raise HTTPNotFound
         # load compare data into template context
         c.cs_repo = c.pull_request.org_repo
         (c.cs_ref_type,
          c.cs_ref_name,
          c.cs_rev) = c.pull_request.org_ref.split(':')
         c.a_repo = c.pull_request.other_repo
         (c.a_ref_type,
          c.a_ref_name,
          c.a_rev) = c.pull_request.other_ref.split(':') # a_rev is ancestor
         org_scm_instance = c.cs_repo.scm_instance # property with expensive cache invalidation check!!!
         try:
             c.cs_ranges = []
             for x in c.pull_request.revisions:
             try:
                 c.cs_ranges.append(org_scm_instance.get_changeset(x))
         except ChangesetDoesNotExistError:
             c.cs_ranges = []
             h.flash(_('Revision %s not found in %s') % (x, c.cs_repo.repo_name),
                 'error')
                 break
         c.cs_ranges_org = None # not stored and not important and moving target - could be calculated ...
         revs = [ctx.revision for ctx in reversed(c.cs_ranges)]
         c.jsdata = graph_data(org_scm_instance, revs)
         c.is_range = False
         try:
             if c.a_ref_type == 'rev': # this looks like a free range where target is ancestor
                 cs_a = org_scm_instance.get_changeset(c.a_rev)
                 root_parents = c.cs_ranges[0].parents
                 c.is_range = cs_a in root_parents
                 #c.merge_root = len(root_parents) > 1 # a range starting with a merge might deserve a warning
         except ChangesetDoesNotExistError: # probably because c.a_rev not found
             pass
         except IndexError: # probably because c.cs_ranges is empty, probably because revisions are missing
             pass
         avail_revs = set()
         avail_show = []
         c.cs_branch_name = c.cs_ref_name
         c.a_branch_name = None
         other_scm_instance = c.a_repo.scm_instance
         c.update_msg = ""
         c.update_msg_other = ""
         try:
             if not c.cs_ranges:
                 c.update_msg = _('Error: changesets not found when displaying pull request from %s.') % c.cs_rev
             elif org_scm_instance.alias == 'hg' and c.a_ref_name != 'ancestor':
                 if c.cs_ref_type != 'branch':
                     c.cs_branch_name = org_scm_instance.get_changeset(c.cs_ref_name).branch # use ref_type ?
                 c.a_branch_name = c.a_ref_name
                 if c.a_ref_type != 'branch':
                     try:
                         c.a_branch_name = other_scm_instance.get_changeset(c.a_ref_name).branch # use ref_type ?
                     except EmptyRepositoryError:
                         c.a_branch_name = 'null' # not a branch name ... but close enough
                 # candidates: descendants of old head that are on the right branch
                 #             and not are the old head itself ...
                 #             and nothing at all if old head is a descendant of target ref name
                 if not c.is_range and other_scm_instance._repo.revs('present(%s)::&%s', c.cs_ranges[-1].raw_id, c.a_branch_name):
                     c.update_msg = _('This pull request has already been merged to %s.') % c.a_branch_name
                 elif c.pull_request.is_closed():
                     c.update_msg = _('This pull request has been closed and can not be updated.')
                 else: # look for descendants of PR head on source branch in org repo
                     avail_revs = org_scm_instance._repo.revs('%s:: & branch(%s)',
                                                              revs[0], c.cs_branch_name)
                     if len(avail_revs) > 1: # more than just revs[0]
                         # also show changesets that not are descendants but would be merged in
                         targethead = other_scm_instance.get_changeset(c.a_branch_name).raw_id
                         if org_scm_instance.path != other_scm_instance.path:
                             # Note: org_scm_instance.path must come first so all
                             # valid revision numbers are 100% org_scm compatible
                             # - both for avail_revs and for revset results
                             try:
                                 hgrepo = unionrepo.makeunionrepository(org_scm_instance.baseui,
                                                                        org_scm_instance.path,
                                                                        other_scm_instance.path)
                             except AttributeError: # makeunionrepository was introduced in Mercurial 4.8 23f2299e9e53
                                 hgrepo = unionrepo.unionrepository(org_scm_instance.baseui,
                                                                    org_scm_instance.path,
                                                                    other_scm_instance.path)
                         else:
                             hgrepo = org_scm_instance._repo
                         show = set(hgrepo.revs('::%ld & !::parents(%s) & !::%s',
                                                avail_revs, revs[0], targethead))
                         if show:
                             c.update_msg = _('The following additional changes are available on %s:') % c.cs_branch_name
                         else:
                             c.update_msg = _('No additional changesets found for iterating on this pull request.')
                     else:
                         show = set()
                         avail_revs = set() # drop revs[0]
                         c.update_msg = _('No additional changesets found for iterating on this pull request.')
                     # TODO: handle branch heads that not are tip-most
                     brevs = org_scm_instance._repo.revs('%s - %ld - %s', c.cs_branch_name, avail_revs, revs[0])
                     if brevs:
                         # also show changesets that are on branch but neither ancestors nor descendants
                         show.update(org_scm_instance._repo.revs('::%ld - ::%ld - ::%s', brevs, avail_revs, c.a_branch_name))
                         show.add(revs[0]) # make sure graph shows this so we can see how they relate
                         c.update_msg_other = _('Note: Branch %s has another head: %s.') % (c.cs_branch_name,
                             h.short_id(org_scm_instance.get_changeset((max(brevs))).raw_id))
                     avail_show = sorted(show, reverse=True)
             elif org_scm_instance.alias == 'git':
                 c.cs_repo.scm_instance.get_changeset(c.cs_rev) # check it exists - raise ChangesetDoesNotExistError if not
                 c.update_msg = _("Git pull requests don't support iterating yet.")
         except ChangesetDoesNotExistError:
             c.update_msg = _('Error: some changesets not found when displaying pull request from %s.') % c.cs_rev
         c.avail_revs = avail_revs
         c.avail_cs = [org_scm_instance.get_changeset(r) for r in avail_show]
         c.avail_jsdata = graph_data(org_scm_instance, avail_show)
         raw_ids = [x.raw_id for x in c.cs_ranges]
         c.cs_comments = c.cs_repo.get_comments(raw_ids)
         c.cs_statuses = c.cs_repo.statuses(raw_ids)
         ignore_whitespace = request.GET.get('ignorews') == '1'
         line_context = safe_int(request.GET.get('context'), 3)
         c.ignorews_url = _ignorews_url
         c.context_url = _context_url
         fulldiff = request.GET.get('fulldiff')
         diff_limit = None if fulldiff else self.cut_off_limit
         # we swap org/other ref since we run a simple diff on one repo
         log.debug('running diff between %s and %s in %s',
                   c.a_rev, c.cs_rev, org_scm_instance.path)
         try:
             raw_diff = diffs.get_diff(org_scm_instance, rev1=safe_str(c.a_rev), rev2=safe_str(c.cs_rev),
                                       ignore_whitespace=ignore_whitespace, context=line_context)
         except ChangesetDoesNotExistError:
             raw_diff = _("The diff can't be shown - the PR revisions could not be found.")
         diff_processor = diffs.DiffProcessor(raw_diff or '', diff_limit=diff_limit)
         c.limited_diff = diff_processor.limited_diff
         c.file_diff_data = []
         c.lines_added = 0
         c.lines_deleted = 0
         for f in diff_processor.parsed:
             st = f['stats']
             c.lines_added += st['added']
             c.lines_deleted += st['deleted']
             filename = f['filename']
             fid = h.FID('', filename)
             html_diff = diffs.as_html(enable_comments=True, parsed_lines=[f])
             c.file_diff_data.append((fid, None, f['operation'], f['old_filename'], filename, html_diff, st))
         # inline comments
         c.inline_cnt = 0
         c.inline_comments = cc_model.get_inline_comments(
                                 c.db_repo.repo_id,
                                 pull_request=pull_request_id)
         # count inline comments
         for __, lines in c.inline_comments:
             for comments in lines.values():
                 c.inline_cnt += len(comments)
         # comments
         c.comments = cc_model.get_comments(c.db_repo.repo_id, pull_request=pull_request_id)
         # (badly named) pull-request status calculation based on reviewer votes
         (c.pull_request_reviewers,
          c.pull_request_pending_reviewers,
          c.current_voting_result,
          ) = cs_model.calculate_pull_request_result(c.pull_request)
         c.changeset_statuses = ChangesetStatus.STATUSES
         c.is_ajax_preview = False
         c.ancestors = None # [c.a_rev] ... but that is shown in an other way
         return render('/pullrequests/pullrequest_show.html')
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def comment(self, repo_name, pull_request_id):
         pull_request = PullRequest.get_or_404(pull_request_id)
         allowed_to_change_status = self._is_allowed_to_change_status(pull_request)
         return create_cs_pr_comment(repo_name, pull_request=pull_request,
                 allowed_to_change_status=allowed_to_change_status)
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         return delete_cs_pr_comment(repo_name, comment_id)

kallithea/lib/diffs.py

➞

Show inline comments

@@ @@ -176,503 +176,503 @@ def wrap_to_table(html): @@
                 <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 = filter(lambda o: isinstance(o, SubModuleNode),
                             [filenode_new, filenode_old])
         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
     return cs1, cs2, a_path, html_diff, stats, op
 def get_gitdiff(filenode_old, filenode_new, ignore_whitespace=True, context=3):
     """
     Returns git style diff between given ``filenode_old`` and ``filenode_new``.
     """
     # make sure we pass in default context
     context = context or 3
     submodules = filter(lambda o: isinstance(o, SubModuleNode),
                         [filenode_new, filenode_old])
     if submodules:
         return ''
     for filenode in (filenode_old, filenode_new):
         if not isinstance(filenode, FileNode):
             raise VCSError("Given object should be FileNode object, not %s"
                 % filenode.__class__)
     repo = filenode_new.changeset.repository
     old_raw_id = getattr(filenode_old.changeset, 'raw_id', repo.EMPTY_CHANGESET)
     new_raw_id = getattr(filenode_new.changeset, 'raw_id', repo.EMPTY_CHANGESET)
     vcs_gitdiff = get_diff(repo, old_raw_id, new_raw_id, filenode_new.path,
                            ignore_whitespace, context)
     return vcs_gitdiff
 def get_diff(scm_instance, rev1, rev2, path=None, ignore_whitespace=False, context=3):
     """
     A thin wrapper around vcs lib get_diff.
     """
     try:
         return scm_instance.get_diff(rev1, rev2, path=path,
                                      ignore_whitespace=ignore_whitespace, context=context)
     except MemoryError:
         h.flash('MemoryError: Diff is too big', category='error')
         return ''
 NEW_FILENODE = 1
 DEL_FILENODE = 2
 MOD_FILENODE = 3
 RENAMED_FILENODE = 4
 COPIED_FILENODE = 5
 CHMOD_FILENODE = 6
 BIN_FILENODE = 7
 class DiffProcessor(object):
     """
     Give it a unified or git diff and it returns a list of the files that were
     mentioned in the diff together with a dict of meta information that
     can be used to render it in a HTML template.
     """
     _diff_git_re = re.compile('^diff --git', re.MULTILINE)
     def __init__(self, diff, vcs='hg', diff_limit=None, inline_diff=True):
         """
         :param diff:   a text in diff format
         :param vcs: type of version control hg or git
         :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, basestring):
             raise Exception('Diff must be a basestring got %s instead' % type(diff))
         self._diff = 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, buffer(self._diff, start, end - start))
             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'].iteritems()
                   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 = lineiter.next()
                     while True:
                         # find a first del line
                         while peekline['action'] != 'del':
                             peekline = lineiter.next()
                         delline = peekline
                         peekline = lineiter.next()
                         # if not followed by add, eat all following del lines
                         if peekline['action'] != 'add':
                             while peekline['action'] == 'del':
                                 peekline = lineiter.next()
                             continue
                         # found an add - make sure it is the only one
                         addline = peekline
                         try:
                             peekline = lineiter.next()
                         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))
 _git_header_re = re.compile(r"""
     ^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(r"""
     ^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)
 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 not rest.startswith('@') and not rest.startswith('literal ') and not rest.startswith('delta '):
         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(r'.*\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:
         chunks = []
         line = diff_lines.next()
         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 = diff_lines.next()
             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 = diff_lines.next()
                 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 = diff_lines.next()
             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
             if newfrag:
                 newfrag = '<ins>%s</ins>' % newfrag
         oldfragments.append(oldfrag)
         newfragments.append(newfrag)
     old['line'] = "".join(oldfragments)
     new['line'] = "".join(newfragments)

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 os
 import pwd
 import re
 import time
 import urllib
 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.compat import json
 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, (basestring)):
         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
     """
     from string import replace
     if mode == 0:
         line = replace(line, '\r\n', '\n')
         line = replace(line, '\r', '\n')
     elif mode == 1:
         line = replace(line, '\r\n', '\r')
         line = replace(line, '\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 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 safe_unicode(str_, from_encoding=None):
     """
     safe unicode function. Does few trick to turn str_ into unicode
     In case of UnicodeDecode error we try to return it with encoding detected
     by chardet library if it fails fallback to unicode with errors replaced
     :param str_: string to decode
     :rtype: unicode
     :returns: unicode object
     """
     if isinstance(str_, unicode):
         return str_
     if not from_encoding:
         import kallithea
         DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
                                                         'utf-8'), sep=',')
         from_encoding = DEFAULT_ENCODINGS
     if not isinstance(from_encoding, (list, tuple)):
         from_encoding = [from_encoding]
     try:
         return unicode(str_)
     except UnicodeDecodeError:
         pass
     for enc in from_encoding:
         try:
             return unicode(str_, enc)
         except UnicodeDecodeError:
             pass
     try:
         import chardet
         encoding = chardet.detect(str_)['encoding']
         if encoding is None:
             raise Exception()
         return str_.decode(encoding)
     except (ImportError, UnicodeDecodeError, Exception):
         return unicode(str_, from_encoding[0], 'replace')
 def safe_str(unicode_, to_encoding=None):
     """
     safe str function. Does few trick to turn unicode_ into string
     In case of UnicodeEncodeError we try to return it with encoding detected
     by chardet library if it fails fallback to string with errors replaced
     :param unicode_: unicode to encode
     :rtype: str
     :returns: str object
     """
     # if it's not basestr cast to str
     if not isinstance(unicode_, basestring):
         return str(unicode_)
     if isinstance(unicode_, str):
         return unicode_
     if not to_encoding:
         import kallithea
         DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
                                                         'utf-8'), sep=',')
         to_encoding = DEFAULT_ENCODINGS
     if not isinstance(to_encoding, (list, tuple)):
         to_encoding = [to_encoding]
     for enc in to_encoding:
         try:
             return unicode_.encode(enc)
         except UnicodeEncodeError:
             pass
     try:
         import chardet
         encoding = chardet.detect(unicode_)['encoding']
         if encoding is None:
             raise UnicodeEncodeError()
         return unicode_.encode(encoding)
     except (ImportError, UnicodeEncodeError):
         return unicode_.encode(to_encoding[0], 'replace')
 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
     # Format the result
     fmt_funcs = {
         'year': lambda d: ungettext(u'%d year', '%d years', d) % d,
         'month': lambda d: ungettext(u'%d month', '%d months', d) % d,
         'day': lambda d: ungettext(u'%d day', '%d days', d) % d,
         'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext(u'%d second', '%d seconds', d) % d,
+    }
     for i, part in enumerate(order):
         value = deltas[part]
         if value == 0:
             continue
         if i < 5:
             sub_part = order[i + 1]
             sub_value = deltas[sub_part]
         else:
             sub_value = 0
         if sub_value == 0 or show_short_version:
             if future:
                 return _('in %s') % fmt_funcs[part](value)
             else:
                 return _('%s ago') % fmt_funcs[part](value)
         if future:
             return _('in %s and %s') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
         else:
             return _('%s and %s ago') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
     return _('just now')
 def uri_filter(uri):
     """
     Removes user:password from given url string
     :param uri:
     :rtype: unicode
     :returns: filtered list of strings
     """
     if not uri:
-        return ''
+        return []
     proto = ''
     for pat in ('https://', 'http://', 'git://'):
         if uri.startswith(pat):
             uri = uri[len(pat):]
             proto = pat
             break
     # remove passwords and username
     uri = uri[uri.find('@') + 1:]
     # get the port
     cred_pos = uri.find(':')
     if cred_pos == -1:
         host, port = uri, None
     else:
         host, port = uri[:cred_pos], uri[cred_pos + 1:]
     return filter(None, [proto, host, port])
 def credentials_filter(uri):
     """
     Returns a url with removed credentials
     :param uri:
     """
     uri = uri_filter(uri)
     # check if we have port
     if len(uri) > 2 and uri[2]:
         uri[2] = ':' + uri[2]
     return ''.join(uri)
 def get_clone_url(clone_uri_tmpl, prefix_url, repo_name, repo_id, username=None):
     parsed_url = urlobject.URLObject(prefix_url)
     prefix = safe_unicode(urllib.unquote(parsed_url.path.rstrip('/')))
     try:
         system_user = pwd.getpwuid(os.getuid()).pw_name
     except Exception: # TODO: support all systems - especially Windows
         system_user = 'kallithea' # hardcoded default value ...
     args = {
         'scheme': parsed_url.scheme,
         'user': safe_unicode(urllib.quote(safe_str(username or ''))),
         'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")
         'prefix': prefix, # undocumented, empty or starting with /
         'repo': repo_name,
         'repoid': str(repo_id),
         'system_user': safe_unicode(system_user),
         'hostname': parsed_url.hostname,
+    }
     url = re.sub('{([^{}]+)}', lambda m: args.get(m.group(1), m.group(0)), clone_uri_tmpl)
     # remove leading @ sign if it's present. Case of empty user
     url_obj = urlobject.URLObject(url)
     if not url_obj.username:
         url_obj = url_obj.with_username(None)
     return safe_unicode(url_obj)
 def get_changeset_safe(repo, rev):
     """
     Safe version of get_changeset if this changeset doesn't exists for a
     repo it returns a Dummy one instead
     :param repo:
     :param rev:
     """
     from kallithea.lib.vcs.backends.base import BaseRepository
     from kallithea.lib.vcs.exceptions import RepositoryError
     from kallithea.lib.vcs.backends.base import EmptyChangeset
     if not isinstance(repo, BaseRepository):
         raise Exception('You must pass an Repository '
                         'object as first argument got %s' % type(repo))
     try:
         cs = repo.get_changeset(rev)
     except (RepositoryError, LookupError):
         cs = EmptyChangeset(requested_revision=rev)
     return cs
 def datetime_to_time(dt):
     if dt:
         return time.mktime(dt.timetuple())
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, basestring):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
 # Must match regexp in kallithea/public/js/base.js MentionsAutoComplete()
 # Check char before @ - it must not look like we are in an email addresses.
 # Matching is greedy so we don't have to look beyond the end.
 MENTIONS_REGEX = re.compile(r'(?:^|(?<=[^a-zA-Z0-9]))@([a-zA-Z0-9][-_.a-zA-Z0-9]*[a-zA-Z0-9])')
 def extract_mentioned_usernames(text):
     r"""
     Returns list of (possible) usernames @mentioned in given text.
     >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
     ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']
     """
     return MENTIONS_REGEX.findall(text)
 def extract_mentioned_users(text):
     """ Returns set of actual database Users @mentioned in given text. """
     from kallithea.model.db import User
     result = set()
     for name in extract_mentioned_usernames(text):
         user = User.get_by_username(name, case_insensitive=True)
         if user is not None and not user.is_default_user:
             result.add(user)
     return result
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__
 def obfuscate_url_pw(engine):
     from sqlalchemy.engine import url as sa_url
     from sqlalchemy.exc import ArgumentError
     try:
         _url = sa_url.make_url(engine or '')
     except ArgumentError:
         return engine
     if _url.password:
         _url.password = 'XXXXX'
     return str(_url)
 class HookEnvironmentError(Exception): pass
 def get_hook_environment():
     """
     Get hook context by deserializing the global KALLITHEA_EXTRAS environment
     variable.
     Called early in Git out-of-process hooks to get .ini config path so the
     basic environment can be configured properly. Also used in all hooks to get
     information about the action that triggered it.
     """
     try:
         kallithea_extras = os.environ['KALLITHEA_EXTRAS']
     except KeyError:
         raise HookEnvironmentError("Environment variable KALLITHEA_EXTRAS not found")
     extras = json.loads(kallithea_extras)
     for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:
     try:
         for k in ['username', 'repository', 'scm', 'action', 'ip', 'config']:
             extras[k]
     except KeyError:
         raise HookEnvironmentError('Missing key %s in KALLITHEA_EXTRAS %s' % (k, extras))
     return AttributeDict(extras)
 def set_hook_environment(username, ip_addr, repo_name, repo_alias, action=None):
     """Prepare global context for running hooks by serializing data in the
     global KALLITHEA_EXTRAS environment variable.
     Most importantly, this allow Git hooks to do proper logging and updating of
     caches after pushes.
     Must always be called before anything with hooks are invoked.
     """
     from kallithea import CONFIG
     extras = {
         'ip': ip_addr, # used in log_push/pull_action action_logger
         'username': username,
         'action': action or 'push_local', # used in log_push_action_raw_ids action_logger
         'repository': repo_name,
         'scm': repo_alias, # used to pick hack in log_push_action_raw_ids
         'config': CONFIG['__file__'], # used by git hook to read config
+    }
     os.environ['KALLITHEA_EXTRAS'] = json.dumps(extras)
 def get_current_authuser():
     """
     Gets kallithea user from threadlocal tmpl_context variable if it's
     defined, else returns None.
     """
     from tg import tmpl_context
     if hasattr(tmpl_context, 'authuser'):
         return tmpl_context.authuser
     return None
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute. Example::
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid)
             # calls
     """
     def __init__(self, attr_name):
         self.attr_name = attr_name
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
     def __call__(self):
         return self
 # alias
 OAttr = OptionalAttr
 class Optional(object):
     """
     Defines an optional parameter::
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
     is equivalent of::
         param = Optional.extract(param)
     """
     def __init__(self, type_):
         self.type_ = type_
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
     def __call__(self):
         return self.getval()
     def getval(self):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             # use params name
             return self.type_.attr_name
         return self.type_
     @classmethod
     def extract(cls, val):
         """
         Extracts value from Optional() instance
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             return val.getval()
         return val
 def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):
     return _cleanstringsub('_', safe_str(s)).rstrip('_')
 def recursive_replace(str_, replace=' '):
     """
     Recursive replace of given sign to just one instance
     :param str_: given string
     :param replace: char to find and replace multiple instances
     Examples::
     >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')
     'Mighty-Mighty-Bo-sstones'
     """
     if str_.find(replace * 2) == -1:
         return str_
     else:
         str_ = str_.replace(replace * 2, replace)
         return recursive_replace(str_, replace)
 def repo_name_slug(value):
     """
     Return slug of name of repository
     This function is called on each creation/modification
     of repository to prevent bad names in repo
     """
     slug = remove_formatting(value)
     slug = strip_tags(slug)
     for c in r"""`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:
         slug = slug.replace(c, '-')
     slug = recursive_replace(slug, '-')
     slug = collapse(slug, '-')
     return slug
 def ask_ok(prompt, retries=4, complaint='Yes or no please!'):
     while True:
         ok = raw_input(prompt)
         if ok in ('y', 'ye', 'yes'):
             return True
         if ok in ('n', 'no', 'nop', 'nope'):
             return False
         retries = retries - 1
         if retries < 0:
             raise IOError
         print(complaint)

kallithea/lib/vcs/nodes.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 """
     vcs.nodes
     ~~~~~~~~~
     Module holding everything related to vcs nodes.
     :created_on: Apr 8, 2010
     :copyright: (c) 2010-2011 by Marcin Kuzminski, Lukasz Balcerzak.
 """
 import mimetypes
 import posixpath
 import stat
 from kallithea.lib.vcs.backends.base import EmptyChangeset
 from kallithea.lib.vcs.exceptions import NodeError, RemovedFileNodeError
 from kallithea.lib.vcs.utils import safe_str, safe_unicode
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 class NodeKind:
     SUBMODULE = -1
     DIR = 1
     FILE = 2
 class NodeState:
     ADDED = u'added'
     CHANGED = u'changed'
     NOT_CHANGED = u'not changed'
     REMOVED = u'removed'
 class NodeGeneratorBase(object):
     """
     Base class for removed added and changed filenodes, it's a lazy generator
     class that will create filenodes only on iteration or call
     The len method doesn't need to create filenodes at all
     """
     def __init__(self, current_paths, cs):
         self.cs = cs
         self.current_paths = current_paths
     def __call__(self):
         return [n for n in self]
     def __getslice__(self, i, j):
         for p in self.current_paths[i:j]:
             yield self.cs.get_node(p)
     def __len__(self):
         return len(self.current_paths)
     def __iter__(self):
         for p in self.current_paths:
             yield self.cs.get_node(p)
 class AddedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding Added files for current changeset
     """
     pass
 class ChangedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding Changed files for current changeset
     """
     pass
 class RemovedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding removed files for current changeset
     """
     def __iter__(self):
         for p in self.current_paths:
             yield RemovedFileNode(path=p)
     def __getslice__(self, i, j):
         for p in self.current_paths[i:j]:
             yield RemovedFileNode(path=p)
 class Node(object):
     """
     Simplest class representing file or directory on repository.  SCM backends
     should use ``FileNode`` and ``DirNode`` subclasses rather than ``Node``
     directly.
     Node's ``path`` cannot start with slash as we operate on *relative* paths
     only. Moreover, every single node is identified by the ``path`` attribute,
     so it cannot end with slash, too. Otherwise, path could lead to mistakes.
     """
     def __init__(self, path, kind):
         if path.startswith('/'):
             raise NodeError("Cannot initialize Node objects with slash at "
                             "the beginning as only relative paths are supported")
         self.path = safe_str(path.rstrip('/'))  # we store paths as str
         if path == '' and kind != NodeKind.DIR:
             raise NodeError("Only DirNode and its subclasses may be "
                             "initialized with empty path")
         self.kind = kind
         #self.dirs, self.files = [], []
         if self.is_root() and not self.is_dir():
             raise NodeError("Root node cannot be FILE kind")
     @LazyProperty
     def parent(self):
         parent_path = self.get_parent_path()
         if parent_path:
             if self.changeset:
                 return self.changeset.get_node(parent_path)
             return DirNode(parent_path)
         return None
     @LazyProperty
     def unicode_path(self):
         return safe_unicode(self.path)
     @LazyProperty
     def name(self):
         """
         Returns name of the node so if its path
         then only last part is returned.
         """
         return safe_unicode(self.path.rstrip('/').split('/')[-1])
     def _get_kind(self):
         return self._kind
     def _set_kind(self, kind):
         if hasattr(self, '_kind'):
             raise NodeError("Cannot change node's kind")
         else:
             self._kind = kind
             # Post setter check (path's trailing slash)
             if self.path.endswith('/'):
                 raise NodeError("Node's path cannot end with slash")
     kind = property(_get_kind, _set_kind)
     def __cmp__(self, other):
         """
         Comparator using name of the node, needed for quick list sorting.
         """
         kind_cmp = cmp(self.kind, other.kind)
         if kind_cmp:
             return kind_cmp
         return cmp(self.name, other.name)
     def __eq__(self, other):
         for attr in ['name', 'path', 'kind']:
             if getattr(self, attr) != getattr(other, attr):
                 return False
         if self.is_file():
             if self.content != other.content:
                 return False
         else:
             # For DirNode's check without entering each dir
             self_nodes_paths = list(sorted(n.path for n in self.nodes))
             other_nodes_paths = list(sorted(n.path for n in self.nodes))
             if self_nodes_paths != other_nodes_paths:
                 return False
         return True
     def __nq__(self, other):
         return not self.__eq__(other)
     def __repr__(self):
         return '<%s %r>' % (self.__class__.__name__, self.path)
     def __str__(self):
         return self.__repr__()
     def __unicode__(self):
         return self.name
     def get_parent_path(self):
         """
         Returns node's parent path or empty string if node is root.
         """
         if self.is_root():
             return ''
         return posixpath.dirname(self.path.rstrip('/')) + '/'
     def is_file(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.FILE``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.FILE
     def is_dir(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.DIR``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.DIR
     def is_root(self):
         """
         Returns ``True`` if node is a root node and ``False`` otherwise.
         """
         return self.kind == NodeKind.DIR and self.path == ''
     def is_submodule(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.SUBMODULE``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.SUBMODULE
     @LazyProperty
     def added(self):
         return self.state is NodeState.ADDED
     @LazyProperty
     def changed(self):
         return self.state is NodeState.CHANGED
     @LazyProperty
     def not_changed(self):
         return self.state is NodeState.NOT_CHANGED
     @LazyProperty
     def removed(self):
         return self.state is NodeState.REMOVED
 class FileNode(Node):
     """
     Class representing file nodes.
     :attribute: path: path to the node, relative to repository's root
     :attribute: content: if given arbitrary sets content of the file
     :attribute: changeset: if given, first time content is accessed, callback
     :attribute: mode: octal stat mode for a node. Default is 0100644.
     """
     def __init__(self, path, content=None, changeset=None, mode=None):
         """
         Only one of ``content`` and ``changeset`` may be given. Passing both
         would raise ``NodeError`` exception.
         :param path: relative path to the node
         :param content: content may be passed to constructor
         :param changeset: if given, will use it to lazily fetch content
         :param mode: octal representation of ST_MODE (i.e. 0100644)
         """
         if content and changeset:
             raise NodeError("Cannot use both content and changeset")
         super(FileNode, self).__init__(path, kind=NodeKind.FILE)
         self.changeset = changeset
         self._content = content
         self._mode = mode or 0100644
     @LazyProperty
     def mode(self):
         """
         Returns lazily mode of the FileNode. If ``changeset`` is not set, would
         use value given at initialization or 0100644 (default).
         """
         if self.changeset:
             mode = self.changeset.get_file_mode(self.path)
         else:
             mode = self._mode
         return mode
     def _get_content(self):
         if self.changeset:
             content = self.changeset.get_file_content(self.path)
         else:
             content = self._content
         return content
     @property
     def content(self):
         """
         Returns lazily content of the FileNode. If possible, would try to
         decode content from UTF-8.
         """
         content = self._get_content()
         if bool(content and '\0' in content):
             return content
         return safe_unicode(content)
     @LazyProperty
     def size(self):
         if self.changeset:
             return self.changeset.get_file_size(self.path)
         raise NodeError("Cannot retrieve size of the file without related "
             "changeset attribute")
     @LazyProperty
     def message(self):
         if self.changeset:
             return self.last_changeset.message
         raise NodeError("Cannot retrieve message of the file without related "
             "changeset attribute")
     @LazyProperty
     def last_changeset(self):
         if self.changeset:
             return self.changeset.get_file_changeset(self.path)
         raise NodeError("Cannot retrieve last changeset of the file without "
             "related changeset attribute")
     def get_mimetype(self):
         """
         Mimetype is calculated based on the file's content.
         """
         mtype, encoding = mimetypes.guess_type(self.name)
         if mtype is None:
             if self.is_binary:
                 mtype = 'application/octet-stream'
                 encoding = None
             else:
                 mtype = 'text/plain'
                 encoding = None
                 # try with pygments
                 from pygments import lexers
                 try:
                     from pygments import lexers
                     mt = lexers.get_lexer_for_filename(self.name).mimetypes
                 except lexers.ClassNotFound:
                     mt = None
                 if mt:
                     mtype = mt[0]
         return mtype, encoding
     @LazyProperty
     def mimetype(self):
         """
         Wrapper around full mimetype info. It returns only type of fetched
         mimetype without the encoding part. use get_mimetype function to fetch
         full set of (type,encoding)
         """
         return self.get_mimetype()[0]
     @LazyProperty
     def mimetype_main(self):
         return self.mimetype.split('/')[0]
     @LazyProperty
     def lexer(self):
         """
         Returns pygment's lexer class. Would try to guess lexer taking file's
         content, name and mimetype.
         """
         from pygments import lexers
         try:
             lexer = lexers.guess_lexer_for_filename(self.name, self.content, stripnl=False)
         except lexers.ClassNotFound:
             lexer = lexers.TextLexer(stripnl=False)
         # returns first alias
         return lexer
     @LazyProperty
     def lexer_alias(self):
         """
         Returns first alias of the lexer guessed for this file.
         """
         return self.lexer.aliases[0]
     @LazyProperty
     def history(self):
         """
         Returns a list of changeset for this file in which the file was changed
         """
         if self.changeset is None:
             raise NodeError('Unable to get changeset for this FileNode')
         return self.changeset.get_file_history(self.path)
     @LazyProperty
     def annotate(self):
         """
         Returns a list of three element tuples with lineno,changeset and line
         """
         if self.changeset is None:
             raise NodeError('Unable to get changeset for this FileNode')
         return self.changeset.get_file_annotate(self.path)
     @LazyProperty
     def state(self):
         if not self.changeset:
             raise NodeError("Cannot check state of the node if it's not "
                 "linked with changeset")
         elif self.path in (node.path for node in self.changeset.added):
             return NodeState.ADDED
         elif self.path in (node.path for node in self.changeset.changed):
             return NodeState.CHANGED
         else:
             return NodeState.NOT_CHANGED
     @property
     def is_binary(self):
         """
         Returns True if file has binary content.
         """
         _bin = '\0' in self._get_content()
         return _bin
     def is_browser_compatible_image(self):
         return self.mimetype in [
             "image/gif",
             "image/jpeg",
             "image/png",
             "image/bmp"
+        ]
     @LazyProperty
     def extension(self):
         """Returns filenode extension"""
         return self.name.split('.')[-1]
     @property
     def is_executable(self):
         """
         Returns ``True`` if file has executable flag turned on.
         """
         return bool(self.mode & stat.S_IXUSR)
     def __repr__(self):
         return '<%s %r @ %s>' % (self.__class__.__name__, self.path,
                                  getattr(self.changeset, 'short_id', ''))
 class RemovedFileNode(FileNode):
     """
     Dummy FileNode class - trying to access any public attribute except path,
     name, kind or state (or methods/attributes checking those two) would raise
     RemovedFileNodeError.
     """
     ALLOWED_ATTRIBUTES = [
         'name', 'path', 'state', 'is_root', 'is_file', 'is_dir', 'kind',
         'added', 'changed', 'not_changed', 'removed'
+    ]
     def __init__(self, path):
         """
         :param path: relative path to the node
         """
         super(RemovedFileNode, self).__init__(path=path)
     def __getattribute__(self, attr):
         if attr.startswith('_') or attr in RemovedFileNode.ALLOWED_ATTRIBUTES:
             return super(RemovedFileNode, self).__getattribute__(attr)
         raise RemovedFileNodeError("Cannot access attribute %s on "
             "RemovedFileNode" % attr)
     @LazyProperty
     def state(self):
         return NodeState.REMOVED
 class DirNode(Node):
     """
     DirNode stores list of files and directories within this node.
     Nodes may be used standalone but within repository context they
     lazily fetch data within same repository's changeset.
     """
     def __init__(self, path, nodes=(), changeset=None):
         """
         Only one of ``nodes`` and ``changeset`` may be given. Passing both
         would raise ``NodeError`` exception.
         :param path: relative path to the node
         :param nodes: content may be passed to constructor
         :param changeset: if given, will use it to lazily fetch content
         :param size: always 0 for ``DirNode``
         """
         if nodes and changeset:
             raise NodeError("Cannot use both nodes and changeset")
         super(DirNode, self).__init__(path, NodeKind.DIR)
         self.changeset = changeset
         self._nodes = nodes
     @LazyProperty
     def content(self):
         raise NodeError("%s represents a dir and has no ``content`` attribute"
             % self)
     @LazyProperty
     def nodes(self):
         if self.changeset:
             nodes = self.changeset.get_nodes(self.path)
         else:
             nodes = self._nodes
         self._nodes_dict = dict((node.path, node) for node in nodes)
         return sorted(nodes)
     @LazyProperty
     def files(self):
         return sorted((node for node in self.nodes if node.is_file()))
     @LazyProperty
     def dirs(self):
         return sorted((node for node in self.nodes if node.is_dir()))
     def __iter__(self):
         for node in self.nodes:
             yield node
     def get_node(self, path):
         """
         Returns node from within this particular ``DirNode``, so it is now
         allowed to fetch, i.e. node located at 'docs/api/index.rst' from node
         'docs'. In order to access deeper nodes one must fetch nodes between
         them first - this would work::
            docs = root.get_node('docs')
            docs.get_node('api').get_node('index.rst')
         :param: path - relative to the current node
         .. note::
            To access lazily (as in example above) node have to be initialized
            with related changeset object - without it node is out of
            context and may know nothing about anything else than nearest
            (located at same level) nodes.
         """
         try:
             path = path.rstrip('/')
             if path == '':
                 raise NodeError("Cannot retrieve node without path")
             self.nodes  # access nodes first in order to set _nodes_dict
             paths = path.split('/')
             if len(paths) == 1:
                 if not self.is_root():
                     path = '/'.join((self.path, paths[0]))
                 else:
                     path = paths[0]
                 return self._nodes_dict[path]
             elif len(paths) > 1:
                 if self.changeset is None:
                     raise NodeError("Cannot access deeper "
                                     "nodes without changeset")
                 else:
                     path1, path2 = paths[0], '/'.join(paths[1:])
                     return self.get_node(path1).get_node(path2)
             else:
                 raise KeyError
         except KeyError:
             raise NodeError("Node does not exist at %s" % path)
     @LazyProperty
     def state(self):
         raise NodeError("Cannot access state of DirNode")
     @LazyProperty
     def size(self):
         size = 0
         for root, dirs, files in self.changeset.walk(self.path):
             for f in files:
                 size += f.size
         return size
     def __repr__(self):
         return '<%s %r @ %s>' % (self.__class__.__name__, self.path,
                                  getattr(self.changeset, 'short_id', ''))
 class RootNode(DirNode):
     """
     DirNode being the root node of the repository.
     """
     def __init__(self, nodes=(), changeset=None):
         super(RootNode, self).__init__(path='', nodes=nodes,
             changeset=changeset)
     def __repr__(self):
         return '<%s>' % self.__class__.__name__
 class SubModuleNode(Node):
     """
     represents a SubModule of Git or SubRepo of Mercurial
     """
     is_binary = False
     size = 0
     def __init__(self, name, url, changeset=None, alias=None):
         self.path = name
         self.kind = NodeKind.SUBMODULE
         self.alias = alias
         # we have to use emptyChangeset here since this can point to svn/git/hg
         # submodules we cannot get from repository
         self.changeset = EmptyChangeset(str(changeset), alias=alias)
         self.url = url
     def __repr__(self):
         return '<%s %r @ %s>' % (self.__class__.__name__, self.path,
                                  getattr(self.changeset, 'short_id', ''))
     @LazyProperty
     def name(self):
         """
         Returns name of the node so if its path
         then only last part is returned.
         """
         org = safe_unicode(self.path.rstrip('/').split('/')[-1])
         return u'%s @ %s' % (org, self.changeset.short_id)

0 comments (0 inline, 0 general)