kallithea Changeset - 3dbb625d5f9c

Changeset - 3dbb625d5f9c

Parent rev.

Child rev.

[Not reviewed]

default

0 13 0

Mads Kiilerich (mads) - 8 years ago 2018-02-12 02:38:02
mads@kiilerich.com

Grafted from: c1860c537e45

vcs: introduce 'branches' attribute on changesets, making it possible for Git to show multiple branches for a changeset

Mercurial changesets will always have have exactly one branch (which might be
"default"). The VCS data model was the same.

Git allows for a changeset to have 0 or more branches ... and possibly one of
them as active. The right data model is thus to have an enumerable of branches.

We thus add a 'branches' attribute and use it where applicable.

The existing 'branch' attribute used some heuristics to decide which branch use
as "the" branch ... and in some places code (and tests) rely on that. We thus
keep that old method, knowing that some of its uses probably should move to
'branches'.

The code for retrieving Git branches is based on work by Dominik Ruf.

13 files changed with 48 insertions and 21 deletions:

kallithea/controllers/feed.py

kallithea/controllers/files.py

kallithea/controllers/pullrequests.py

kallithea/lib/vcs/backends/base.py

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

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

kallithea/templates/changelog/changelog_table.html

kallithea/templates/changeset/changeset.html

kallithea/templates/changeset/changeset_range.html

kallithea/tests/vcs/test_branches.py

kallithea/tests/vcs/test_changesets.py

kallithea/tests/vcs/test_git.py

kallithea/tests/vcs/test_hg.py

0 comments (0 inline, 0 general)

kallithea/controllers/feed.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.feed
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 Feed controller for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 23, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from tg import response, tmpl_context as c
 from tg.i18n import ugettext as _
 from beaker.cache import cache_region, region_invalidate
 from webhelpers.feedgenerator import Atom1Feed, Rss201rev2Feed
 from kallithea import CONFIG
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import LoginRequired, HasRepoPermissionLevelDecorator
 from kallithea.lib.base import BaseRepoController
 from kallithea.lib.diffs import DiffProcessor
 from kallithea.model.db import CacheInvalidation
 from kallithea.lib.utils2 import safe_int, str2bool, safe_unicode
 log = logging.getLogger(__name__)
 language = 'en-us'
 ttl = "5"
 class FeedController(BaseRepoController):
     @LoginRequired(api_access=True, allow_default_user=True)
     @HasRepoPermissionLevelDecorator('read')
     def _before(self, *args, **kwargs):
         super(FeedController, self)._before(*args, **kwargs)
     def _get_title(self, cs):
         return h.shorter(cs.message, 160)
     def __get_desc(self, cs):
         desc_msg = [(_('%s committed on %s')
                      % (h.person(cs.author), h.fmt_date(cs.date))) + '<br/>']
         # branches, tags, bookmarks
         if cs.branch:
             desc_msg.append('branch: %s<br/>' % cs.branch)
         for branch in cs.branches:
             desc_msg.append('branch: %s<br/>' % branch)
         for book in cs.bookmarks:
             desc_msg.append('bookmark: %s<br/>' % book)
         for tag in cs.tags:
             desc_msg.append('tag: %s<br/>' % tag)
         changes = []
         diff_limit = safe_int(CONFIG.get('rss_cut_off_limit', 32 * 1024))
         raw_diff = cs.diff()
         diff_processor = DiffProcessor(raw_diff,
                                        diff_limit=diff_limit,
                                        inline_diff=False)
         for st in diff_processor.parsed:
             st.update({'added': st['stats']['added'],
                        'removed': st['stats']['deleted']})
             changes.append('\n %(operation)s %(filename)s '
                            '(%(added)s lines added, %(removed)s lines removed)'
                             % st)
         if diff_processor.limited_diff:
             changes = changes + ['\n ' +
                                  _('Changeset was too big and was cut off...')]
         # rev link
         _url = h.canonical_url('changeset_home', repo_name=c.db_repo.repo_name,
                    revision=cs.raw_id)
         desc_msg.append('changeset: <a href="%s">%s</a>' % (_url, cs.raw_id[:8]))
         desc_msg.append('<pre>')
         desc_msg.append(h.urlify_text(cs.message))
         desc_msg.append('\n')
         desc_msg.extend(changes)
         if str2bool(CONFIG.get('rss_include_diff', False)):
             desc_msg.append('\n\n')
             desc_msg.append(raw_diff)
         desc_msg.append('</pre>')
         return map(safe_unicode, desc_msg)
     def atom(self, repo_name):
         """Produce an atom-1.0 feed via feedgenerator module"""
         @cache_region('long_term', '_get_feed_from_cache')
         def _get_feed_from_cache(key, kind):
             feed = Atom1Feed(
                 title=_('%s %s feed') % (c.site_name, repo_name),
                 link=h.canonical_url('summary_home', repo_name=repo_name),
                 description=_('Changes on %s repository') % repo_name,
                 language=language,
                 ttl=ttl
+            )
             rss_items_per_page = safe_int(CONFIG.get('rss_items_per_page', 20))
             for cs in reversed(list(c.db_repo_scm_instance[-rss_items_per_page:])):
                 feed.add_item(title=self._get_title(cs),
                               link=h.canonical_url('changeset_home', repo_name=repo_name,
                                        revision=cs.raw_id),
                               author_name=cs.author,
                               description=''.join(self.__get_desc(cs)),
                               pubdate=cs.date,
+                              )
             response.content_type = feed.mime_type
             return feed.writeString('utf-8')
         kind = 'ATOM'
         valid = CacheInvalidation.test_and_set_valid(repo_name, kind)
         if not valid:
             region_invalidate(_get_feed_from_cache, None, '_get_feed_from_cache', repo_name, kind)
         return _get_feed_from_cache(repo_name, kind)
     def rss(self, repo_name):
         """Produce an rss2 feed via feedgenerator module"""
         @cache_region('long_term', '_get_feed_from_cache')
         def _get_feed_from_cache(key, kind):
             feed = Rss201rev2Feed(
                 title=_('%s %s feed') % (c.site_name, repo_name),
                 link=h.canonical_url('summary_home', repo_name=repo_name),
                 description=_('Changes on %s repository') % repo_name,
                 language=language,
                 ttl=ttl
+            )
             rss_items_per_page = safe_int(CONFIG.get('rss_items_per_page', 20))
             for cs in reversed(list(c.db_repo_scm_instance[-rss_items_per_page:])):
                 feed.add_item(title=self._get_title(cs),
                               link=h.canonical_url('changeset_home', repo_name=repo_name,
                                        revision=cs.raw_id),
                               author_name=cs.author,
                               description=''.join(self.__get_desc(cs)),
                               pubdate=cs.date,
+                             )
             response.content_type = feed.mime_type
             return feed.writeString('utf-8')
         kind = 'RSS'
         valid = CacheInvalidation.test_and_set_valid(repo_name, kind)
         if not valid:
             region_invalidate(_get_feed_from_cache, None, '_get_feed_from_cache', repo_name, kind)
         return _get_feed_from_cache(repo_name, kind)

kallithea/controllers/files.py

➞

Show inline comments

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

kallithea/controllers/pullrequests.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.pullrequests
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 pull requests controller for Kallithea for initializing pull requests
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: May 7, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from tg import request, tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound, HTTPForbidden, HTTPBadRequest
 from kallithea.config.routing import url
 from kallithea.lib import helpers as h
 from kallithea.lib import diffs
 from kallithea.lib.auth import LoginRequired, HasRepoPermissionLevelDecorator
 from kallithea.lib.base import BaseRepoController, render, jsonify
 from kallithea.lib.page import Page
 from kallithea.lib.utils import action_logger
 from kallithea.lib.vcs.exceptions import EmptyRepositoryError, ChangesetDoesNotExistError
 from kallithea.lib.vcs.utils import safe_str
 from kallithea.lib.vcs.utils.hgcompat import unionrepo
 from kallithea.model.db import PullRequest, ChangesetStatus, ChangesetComment, \
     PullRequestReviewer, Repository, User
 from kallithea.model.pull_request import CreatePullRequestAction, CreatePullRequestIterationAction, PullRequestModel
 from kallithea.model.meta import Session
 from kallithea.model.repo import RepoModel
 from kallithea.model.comment import ChangesetCommentsModel
 from kallithea.model.changeset_status import ChangesetStatusModel
 from kallithea.model.forms import PullRequestForm, PullRequestPostForm
 from kallithea.lib.utils2 import safe_int
 from kallithea.controllers.changeset import _ignorews_url, _context_url, \
     create_comment
 from kallithea.controllers.compare import CompareController
 from kallithea.lib.graphmod import graph_data
 log = logging.getLogger(__name__)
 def _get_reviewer(user_id):
     """Look up user by ID and validate it as a potential reviewer."""
     try:
         user = User.get(int(user_id))
     except ValueError:
         user = None
     if user is None or user.is_default_user:
         h.flash(_('Invalid reviewer "%s" specified') % user_id, category='error')
         raise HTTPBadRequest()
     return user
 class PullrequestsController(BaseRepoController):
     def _get_repo_refs(self, repo, rev=None, branch=None, branch_rev=None):
         """return a structure with repo's interesting changesets, suitable for
         the selectors in pullrequest.html
         rev: a revision that must be in the list somehow and selected by default
         branch: a branch that must be in the list and selected by default - even if closed
         branch_rev: a revision of which peers should be preferred and available."""
         # list named branches that has been merged to this named branch - it should probably merge back
         peers = []
         if rev:
             rev = safe_str(rev)
         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):
                 abranch = repo.get_changeset(i).branch
                 if abranch not in peerbranches:
                     n = 'branch:%s:%s' % (abranch, repo.get_changeset(abranch).raw_id)
                     peers.append((n, abranch))
                     peerbranches.add(abranch)
                 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 _get_is_allowed_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(h.literal(_('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):
         repo_model = RepoModel()
         c.users_array = repo_model.get_users_js()
         c.pull_request = PullRequest.get_or_404(pull_request_id)
         c.allowed_to_change_status = self._get_is_allowed_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:
                 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')
         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
                             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))
                         c.update_msg = _('The following additional changes are available on %s:') % c.cs_branch_name
                     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)
         status = request.POST.get('changeset_status')
         close_pr = request.POST.get('save_close')
         delete = request.POST.get('save_delete')
         f_path = request.POST.get('f_path')
         line_no = request.POST.get('line')
         if (status or close_pr or delete) and (f_path or line_no):
             # status votes and closing is only possible in general comments
             raise HTTPBadRequest()
         allowed_to_change_status = self._get_is_allowed_change_status(pull_request)
         if not allowed_to_change_status:
             if status or close_pr:
                 h.flash(_('No permission to change pull request status'), 'error')
                 raise HTTPForbidden()
         if delete == "delete":
             if (pull_request.owner_id == request.authuser.user_id or
                 h.HasPermissionAny('hg.admin')() or
                 h.HasRepoPermissionLevel('admin')(pull_request.org_repo.repo_name) or
                 h.HasRepoPermissionLevel('admin')(pull_request.other_repo.repo_name)
                 ) and not pull_request.is_closed():
                 PullRequestModel().delete(pull_request)
                 Session().commit()
                 h.flash(_('Successfully deleted pull request %s') % pull_request_id,
                         category='success')
                 return {
                    'location': url('my_pullrequests'), # or repo pr list?
+                }
                 raise HTTPFound(location=url('my_pullrequests')) # or repo pr list?
             raise HTTPForbidden()
         text = request.POST.get('text', '').strip()
         comment = create_comment(
             text,
             status,
             pull_request_id=pull_request_id,
             f_path=f_path,
             line_no=line_no,
             closing_pr=close_pr,
+        )
         action_logger(request.authuser,
                       'user_commented_pull_request:%s' % pull_request_id,
                       c.db_repo, request.ip_addr)
         if status:
             ChangesetStatusModel().set_status(
                 c.db_repo.repo_id,
                 status,
                 request.authuser.user_id,
                 comment,
                 pull_request=pull_request_id
+            )
         if close_pr:
             PullRequestModel().close_pull_request(pull_request_id)
             action_logger(request.authuser,
                           'user_closed_pull_request:%s' % pull_request_id,
                           c.db_repo, request.ip_addr)
         Session().commit()
         if not request.environ.get('HTTP_X_PARTIAL_XHR'):
             raise HTTPFound(location=pull_request.url())
         data = {
            'target_id': h.safeid(h.safe_unicode(request.POST.get('f_path'))),
+        }
         if comment is not None:
             c.comment = comment
             data.update(comment.get_dict())
             data.update({'rendered_text':
                          render('changeset/changeset_comment_block.html')})
         return data
     @LoginRequired()
     @HasRepoPermissionLevelDecorator('read')
     @jsonify
     def delete_comment(self, repo_name, comment_id):
         co = ChangesetComment.get(comment_id)
         if co.pull_request.is_closed():
             # don't allow deleting comments on closed pull request
             raise HTTPForbidden()
         owner = co.author_id == request.authuser.user_id
         repo_admin = h.HasRepoPermissionLevel('admin')(c.repo_name)
         if h.HasPermissionAny('hg.admin')() or repo_admin or owner:
             ChangesetCommentsModel().delete(comment=co)
             Session().commit()
             return True
         else:
             raise HTTPForbidden()

kallithea/lib/vcs/backends/base.py

➞

Show inline comments

@@ @@ -273,804 +273,809 @@ class BaseRepository(object): @@
         created changeset.
         :raises ``NothingChangedError``: if no changes has been made
         """
         raise NotImplementedError
     def get_state(self):
         """
         Returns dictionary with ``added``, ``changed`` and ``removed`` lists
         containing ``FileNode`` objects.
         """
         raise NotImplementedError
     def get_config_value(self, section, name, config_file=None):
         """
         Returns configuration value for a given [``section``] and ``name``.
         :param section: Section we want to retrieve value from
         :param name: Name of configuration we want to retrieve
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     def get_user_name(self, config_file=None):
         """
         Returns user's name from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     def get_user_email(self, config_file=None):
         """
         Returns user's email from global configuration file.
         :param config_file: A path to file which should be used to retrieve
           configuration from (might also be a list of file paths)
         """
         raise NotImplementedError
     # =========== #
     # WORKDIR API #
     # =========== #
     @LazyProperty
     def workdir(self):
         """
         Returns ``Workdir`` instance for this repository.
         """
         raise NotImplementedError
 class BaseChangeset(object):
     """
     Each backend should implement it's changeset representation.
     **Attributes**
         ``repository``
             repository object within which changeset exists
         ``id``
             may be ``raw_id`` or i.e. for mercurial's tip just ``tip``
         ``raw_id``
             raw changeset representation (i.e. full 40 length sha for git
             backend)
         ``short_id``
             shortened (if apply) version of ``raw_id``; it would be simple
             shortcut for ``raw_id[:12]`` for git/mercurial backends or same
             as ``raw_id`` for subversion
         ``revision``
             revision number as integer
         ``files``
             list of ``FileNode`` (``Node`` with NodeKind.FILE) objects
         ``dirs``
             list of ``DirNode`` (``Node`` with NodeKind.DIR) objects
         ``nodes``
             combined list of ``Node`` objects
         ``author``
             author of the changeset, as unicode
         ``message``
             message of the changeset, as unicode
         ``parents``
             list of parent changesets
         ``last``
             ``True`` if this is last changeset in repository, ``False``
             otherwise; trying to access this attribute while there is no
             changesets would raise ``EmptyRepositoryError``
     """
     def __str__(self):
         return '<%s at %s:%s>' % (self.__class__.__name__, self.revision,
             self.short_id)
     def __repr__(self):
         return self.__str__()
     def __unicode__(self):
         return u'%s:%s' % (self.revision, self.short_id)
     def __eq__(self, other):
         return self.raw_id == other.raw_id
     def __json__(self, with_file_list=False):
         if with_file_list:
             return dict(
                 short_id=self.short_id,
                 raw_id=self.raw_id,
                 revision=self.revision,
                 message=self.message,
                 date=self.date,
                 author=self.author,
                 added=[safe_unicode(el.path) for el in self.added],
                 changed=[safe_unicode(el.path) for el in self.changed],
                 removed=[safe_unicode(el.path) for el in self.removed],
+            )
         else:
             return dict(
                 short_id=self.short_id,
                 raw_id=self.raw_id,
                 revision=self.revision,
                 message=self.message,
                 date=self.date,
                 author=self.author,
+            )
     @LazyProperty
     def last(self):
         if self.repository is None:
             raise ChangesetError("Cannot check if it's most recent revision")
         return self.raw_id == self.repository.revisions[-1]
     @LazyProperty
     def parents(self):
         """
         Returns list of parents changesets.
         """
         raise NotImplementedError
     @LazyProperty
     def children(self):
         """
         Returns list of children changesets.
         """
         raise NotImplementedError
     @LazyProperty
     def id(self):
         """
         Returns string identifying this changeset.
         """
         raise NotImplementedError
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this changeset.
         """
         raise NotImplementedError
     @LazyProperty
     def short_id(self):
         """
         Returns shortened version of ``raw_id`` attribute, as string,
         identifying this changeset, useful for web representation.
         """
         raise NotImplementedError
     @LazyProperty
     def revision(self):
         """
         Returns integer identifying this changeset.
         """
         raise NotImplementedError
     @LazyProperty
     def committer(self):
         """
         Returns Committer for given commit
         """
         raise NotImplementedError
     @LazyProperty
     def committer_name(self):
         """
         Returns Author name for given commit
         """
         return author_name(self.committer)
     @LazyProperty
     def committer_email(self):
         """
         Returns Author email address for given commit
         """
         return author_email(self.committer)
     @LazyProperty
     def author(self):
         """
         Returns Author for given commit
         """
         raise NotImplementedError
     @LazyProperty
     def author_name(self):
         """
         Returns Author name for given commit
         """
         return author_name(self.author)
     @LazyProperty
     def author_email(self):
         """
         Returns Author email address for given commit
         """
         return author_email(self.author)
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_content(self, path):
         """
         Returns content of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_size(self, path):
         """
         Returns size of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_changeset(self, path):
         """
         Returns last commit of the file at the given ``path``.
         """
         raise NotImplementedError
     def get_file_history(self, path):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         """
         raise NotImplementedError
     def get_nodes(self, path):
         """
         Returns combined ``DirNode`` and ``FileNode`` objects list representing
         state of changeset at the given ``path``.
         :raises ``ChangesetError``: if node at the given ``path`` is not
           instance of ``DirNode``
         """
         raise NotImplementedError
     def get_node(self, path):
         """
         Returns ``Node`` object from the given ``path``.
         :raises ``NodeDoesNotExistError``: if there is no node at the given
           ``path``
         """
         raise NotImplementedError
     def fill_archive(self, stream=None, kind='tgz', prefix=None):
         """
         Fills up given stream.
         :param stream: file like object.
         :param kind: one of following: ``zip``, ``tar``, ``tgz``
             or ``tbz2``. Default: ``tgz``.
         :param prefix: name of root directory in archive.
             Default is repository name and changeset's raw_id joined with dash.
             repo-tip.<kind>
         """
         raise NotImplementedError
     def get_chunked_archive(self, **kwargs):
         """
         Returns iterable archive. Tiny wrapper around ``fill_archive`` method.
         :param chunk_size: extra parameter which controls size of returned
             chunks. Default:8k.
         """
         chunk_size = kwargs.pop('chunk_size', 8192)
         stream = kwargs.get('stream')
         self.fill_archive(**kwargs)
         while True:
             data = stream.read(chunk_size)
             if not data:
                 break
             yield data
     @LazyProperty
     def root(self):
         """
         Returns ``RootNode`` object for this changeset.
         """
         return self.get_node('')
     def next(self, branch=None):
         """
         Returns next changeset from current, if branch is gives it will return
         next changeset belonging to this branch
         :param branch: show changesets within the given named branch
         """
         raise NotImplementedError
     def prev(self, branch=None):
         """
         Returns previous changeset from current, if branch is gives it will
         return previous changeset belonging to this branch
         :param branch: show changesets within the given named branch
         """
         raise NotImplementedError
     @LazyProperty
     def added(self):
         """
         Returns list of added ``FileNode`` objects.
         """
         raise NotImplementedError
     @LazyProperty
     def changed(self):
         """
         Returns list of modified ``FileNode`` objects.
         """
         raise NotImplementedError
     @LazyProperty
     def removed(self):
         """
         Returns list of removed ``FileNode`` objects.
         """
         raise NotImplementedError
     @LazyProperty
     def size(self):
         """
         Returns total number of bytes from contents of all filenodes.
         """
         return sum((node.size for node in self.get_filenodes_generator()))
     def walk(self, topurl=''):
         """
         Similar to os.walk method. Instead of filesystem it walks through
         changeset starting at given ``topurl``.  Returns generator of tuples
         (topnode, dirnodes, filenodes).
         """
         topnode = self.get_node(topurl)
         yield (topnode, topnode.dirs, topnode.files)
         for dirnode in topnode.dirs:
             for tup in self.walk(dirnode.path):
                 yield tup
     def get_filenodes_generator(self):
         """
         Returns generator that yields *all* file nodes.
         """
         for topnode, dirs, files in self.walk():
             for node in files:
                 yield node
     def as_dict(self):
         """
         Returns dictionary with changeset's attributes and their values.
         """
         data = get_dict_for_attrs(self, ['id', 'raw_id', 'short_id',
             'revision', 'date', 'message'])
         data['author'] = {'name': self.author_name, 'email': self.author_email}
         data['added'] = [safe_unicode(node.path) for node in self.added]
         data['changed'] = [safe_unicode(node.path) for node in self.changed]
         data['removed'] = [safe_unicode(node.path) for node in self.removed]
         return data
     @LazyProperty
     def closesbranch(self):
         return False
     @LazyProperty
     def obsolete(self):
         return False
     @LazyProperty
     def bumped(self):
         return False
     @LazyProperty
     def divergent(self):
         return False
     @LazyProperty
     def extinct(self):
         return False
     @LazyProperty
     def unstable(self):
         return False
     @LazyProperty
     def phase(self):
         return ''
 class BaseWorkdir(object):
     """
     Working directory representation of single repository.
     :attribute: repository: repository object of working directory
     """
     def __init__(self, repository):
         self.repository = repository
     def get_branch(self):
         """
         Returns name of current branch.
         """
         raise NotImplementedError
     def get_changeset(self):
         """
         Returns current changeset.
         """
         raise NotImplementedError
     def get_added(self):
         """
         Returns list of ``FileNode`` objects marked as *new* in working
         directory.
         """
         raise NotImplementedError
     def get_changed(self):
         """
         Returns list of ``FileNode`` objects *changed* in working directory.
         """
         raise NotImplementedError
     def get_removed(self):
         """
         Returns list of ``RemovedFileNode`` objects marked as *removed* in
         working directory.
         """
         raise NotImplementedError
     def get_untracked(self):
         """
         Returns list of ``FileNode`` objects which are present within working
         directory however are not tracked by repository.
         """
         raise NotImplementedError
     def get_status(self):
         """
         Returns dict with ``added``, ``changed``, ``removed`` and ``untracked``
         lists.
         """
         raise NotImplementedError
     def commit(self, message, **kwargs):
         """
         Commits local (from working directory) changes and returns newly
         created
         ``Changeset``. Updates repository's ``revisions`` list.
         :raises ``CommitError``: if any error occurs while committing
         """
         raise NotImplementedError
     def update(self, revision=None):
         """
         Fetches content of the given revision and populates it within working
         directory.
         """
         raise NotImplementedError
     def checkout_branch(self, branch=None):
         """
         Checks out ``branch`` or the backend's default branch.
         Raises ``BranchDoesNotExistError`` if the branch does not exist.
         """
         raise NotImplementedError
 class BaseInMemoryChangeset(object):
     """
     Represents differences between repository's state (most recent head) and
     changes made *in place*.
     **Attributes**
         ``repository``
             repository object for this in-memory-changeset
         ``added``
             list of ``FileNode`` objects marked as *added*
         ``changed``
             list of ``FileNode`` objects marked as *changed*
         ``removed``
             list of ``FileNode`` or ``RemovedFileNode`` objects marked to be
             *removed*
         ``parents``
             list of ``Changeset`` representing parents of in-memory changeset.
             Should always be 2-element sequence.
     """
     def __init__(self, repository):
         self.repository = repository
         self.added = []
         self.changed = []
         self.removed = []
         self.parents = []
     def add(self, *filenodes):
         """
         Marks given ``FileNode`` objects as *to be committed*.
         :raises ``NodeAlreadyExistsError``: if node with same path exists at
           latest changeset
         :raises ``NodeAlreadyAddedError``: if node with same path is already
           marked as *added*
         """
         # Check if not already marked as *added* first
         for node in filenodes:
             if node.path in (n.path for n in self.added):
                 raise NodeAlreadyAddedError("Such FileNode %s is already "
                     "marked for addition" % node.path)
         for node in filenodes:
             self.added.append(node)
     def change(self, *filenodes):
         """
         Marks given ``FileNode`` objects to be *changed* in next commit.
         :raises ``EmptyRepositoryError``: if there are no changesets yet
         :raises ``NodeAlreadyExistsError``: if node with same path is already
           marked to be *changed*
         :raises ``NodeAlreadyRemovedError``: if node with same path is already
           marked to be *removed*
         :raises ``NodeDoesNotExistError``: if node doesn't exist in latest
           changeset
         :raises ``NodeNotChangedError``: if node hasn't really be changed
         """
         for node in filenodes:
             if node.path in (n.path for n in self.removed):
                 raise NodeAlreadyRemovedError("Node at %s is already marked "
                     "as removed" % node.path)
         try:
             self.repository.get_changeset()
         except EmptyRepositoryError:
             raise EmptyRepositoryError("Nothing to change - try to *add* new "
                 "nodes rather than changing them")
         for node in filenodes:
             if node.path in (n.path for n in self.changed):
                 raise NodeAlreadyChangedError("Node at '%s' is already "
                     "marked as changed" % node.path)
             self.changed.append(node)
     def remove(self, *filenodes):
         """
         Marks given ``FileNode`` (or ``RemovedFileNode``) objects to be
         *removed* in next commit.
         :raises ``NodeAlreadyRemovedError``: if node has been already marked to
           be *removed*
         :raises ``NodeAlreadyChangedError``: if node has been already marked to
           be *changed*
         """
         for node in filenodes:
             if node.path in (n.path for n in self.removed):
                 raise NodeAlreadyRemovedError("Node is already marked to "
                     "for removal at %s" % node.path)
             if node.path in (n.path for n in self.changed):
                 raise NodeAlreadyChangedError("Node is already marked to "
                     "be changed at %s" % node.path)
             # We only mark node as *removed* - real removal is done by
             # commit method
             self.removed.append(node)
     def reset(self):
         """
         Resets this instance to initial state (cleans ``added``, ``changed``
         and ``removed`` lists).
         """
         self.added = []
         self.changed = []
         self.removed = []
         self.parents = []
     def get_ipaths(self):
         """
         Returns generator of paths from nodes marked as added, changed or
         removed.
         """
         for node in itertools.chain(self.added, self.changed, self.removed):
             yield node.path
     def get_paths(self):
         """
         Returns list of paths from nodes marked as added, changed or removed.
         """
         return list(self.get_ipaths())
     def check_integrity(self, parents=None):
         """
         Checks in-memory changeset's integrity. Also, sets parents if not
         already set.
         :raises CommitError: if any error occurs (i.e.
           ``NodeDoesNotExistError``).
         """
         if not self.parents:
             parents = parents or []
             if len(parents) == 0:
                 try:
                     parents = [self.repository.get_changeset(), None]
                 except EmptyRepositoryError:
                     parents = [None, None]
             elif len(parents) == 1:
                 parents += [None]
             self.parents = parents
         # Local parents, only if not None
         parents = [p for p in self.parents if p]
         # Check nodes marked as added
         for p in parents:
             for node in self.added:
                 try:
                     p.get_node(node.path)
                 except NodeDoesNotExistError:
                     pass
                 else:
                     raise NodeAlreadyExistsError("Node at %s already exists "
                         "at %s" % (node.path, p))
         # Check nodes marked as changed
         missing = set(self.changed)
         not_changed = set(self.changed)
         if self.changed and not parents:
             raise NodeDoesNotExistError(str(self.changed[0].path))
         for p in parents:
             for node in self.changed:
                 try:
                     old = p.get_node(node.path)
                     missing.remove(node)
                     # if content actually changed, remove node from unchanged
                     if old.content != node.content:
                         not_changed.remove(node)
                 except NodeDoesNotExistError:
                     pass
         if self.changed and missing:
             raise NodeDoesNotExistError("Node at %s is missing "
                 "(parents: %s)" % (node.path, parents))
         if self.changed and not_changed:
             raise NodeNotChangedError("Node at %s wasn't actually changed "
                 "since parents' changesets: %s" % (not_changed.pop().path,
                     parents)
+            )
         # Check nodes marked as removed
         if self.removed and not parents:
             raise NodeDoesNotExistError("Cannot remove node at %s as there "
                 "were no parents specified" % self.removed[0].path)
         really_removed = set()
         for p in parents:
             for node in self.removed:
                 try:
                     p.get_node(node.path)
                     really_removed.add(node)
                 except ChangesetError:
                     pass
         not_removed = set(self.removed) - really_removed
         if not_removed:
             raise NodeDoesNotExistError("Cannot remove node at %s from "
                 "following parents: %s" % (not_removed[0], parents))
     def commit(self, message, author, parents=None, branch=None, date=None,
             **kwargs):
         """
         Performs in-memory commit (doesn't check workdir in any way) and
         returns newly created ``Changeset``. Updates repository's
         ``revisions``.
         .. note::
             While overriding this method each backend's should call
             ``self.check_integrity(parents)`` in the first place.
         :param message: message of the commit
         :param author: full username, i.e. "Joe Doe <joe.doe@example.com>"
         :param parents: single parent or sequence of parents from which commit
           would be derived
         :param date: ``datetime.datetime`` instance. Defaults to
           ``datetime.datetime.now()``.
         :param branch: branch name, as string. If none given, default backend's
           branch would be used.
         :raises ``CommitError``: if any error occurs while committing
         """
         raise NotImplementedError
 class EmptyChangeset(BaseChangeset):
     """
     An dummy empty changeset. It's possible to pass hash when creating
     an EmptyChangeset
     """
     def __init__(self, cs='0' * 40, repo=None, requested_revision=None,
                  alias=None, revision=-1, message='', author='', date=None):
         self._empty_cs = cs
         self.revision = revision
         self.message = message
         self.author = author
         self.date = date or datetime.datetime.fromtimestamp(0)
         self.repository = repo
         self.requested_revision = requested_revision
         self.alias = alias
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifying this changeset, useful for web
         representation.
         """
         return self._empty_cs
     @LazyProperty
     def branch(self):
         from kallithea.lib.vcs.backends import get_backend
         return get_backend(self.alias).DEFAULT_BRANCH_NAME
     @LazyProperty
     def branches(self):
         from kallithea.lib.vcs.backends import get_backend
         return [get_backend(self.alias).DEFAULT_BRANCH_NAME]
     @LazyProperty
     def short_id(self):
         return self.raw_id[:12]
     def get_file_changeset(self, path):
         return self
     def get_file_content(self, path):
         return u''
     def get_file_size(self, path):
         return 0
 class CollectionGenerator(object):
     def __init__(self, repo, revs):
         self.repo = repo
         self.revs = revs
     def __len__(self):
         return len(self.revs)
     def __iter__(self):
         for rev in self.revs:
             yield self.repo.get_changeset(rev)
     def __getitem__(self, what):
         """Return either a single element by index, or a sliced collection."""
         if isinstance(what, slice):
             return CollectionGenerator(self.repo, self.revs[what])
         else:
             # single item
             return self.repo.get_changeset(self.revs[what])
     def __repr__(self):
         return '<CollectionGenerator[len:%s]>' % (len(self))

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

➞

Show inline comments

 import re
 from itertools import chain
 from dulwich import objects
 from subprocess import Popen, PIPE
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset
 from kallithea.lib.vcs.exceptions import (
     RepositoryError, ChangesetError, NodeDoesNotExistError, VCSError,
     ChangesetDoesNotExistError, ImproperArchiveTypeError
+)
 from kallithea.lib.vcs.nodes import (
     FileNode, DirNode, NodeKind, RootNode, SubModuleNode,
     ChangedFileNodesGenerator, AddedFileNodesGenerator, RemovedFileNodesGenerator
+)
 from kallithea.lib.vcs.utils import (
     safe_unicode, safe_str, safe_int, date_fromtimestamp
+)
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 class GitChangeset(BaseChangeset):
     """
     Represents state of the repository at single revision.
     """
     def __init__(self, repository, revision):
         self._stat_modes = {}
         self.repository = repository
         revision = safe_str(revision)
         try:
             commit = self.repository._repo[revision]
             if isinstance(commit, objects.Tag):
                 revision = safe_str(commit.object[1])
                 commit = self.repository._repo.get_object(commit.object[1])
         except KeyError:
             raise RepositoryError("Cannot get object with id %s" % revision)
         self.raw_id = revision
         self.id = self.raw_id
         self.short_id = self.raw_id[:12]
         self._commit = commit
         self._tree_id = commit.tree
         self._committer_property = 'committer'
         self._author_property = 'author'
         self._date_property = 'commit_time'
         self._date_tz_property = 'commit_timezone'
         self.revision = repository.revisions.index(revision)
         self.nodes = {}
         self._paths = {}
     @LazyProperty
     def bookmarks(self):
         return ()
     @LazyProperty
     def message(self):
         return safe_unicode(self._commit.message)
     @LazyProperty
     def committer(self):
         return safe_unicode(getattr(self._commit, self._committer_property))
     @LazyProperty
     def author(self):
         return safe_unicode(getattr(self._commit, self._author_property))
     @LazyProperty
     def date(self):
         return date_fromtimestamp(getattr(self._commit, self._date_property),
                                   getattr(self._commit, self._date_tz_property))
     @LazyProperty
     def _timestamp(self):
         return getattr(self._commit, self._date_property)
     @LazyProperty
     def status(self):
         """
         Returns modified, added, removed, deleted files for current changeset
         """
         return self.changed, self.added, self.removed
     @LazyProperty
     def tags(self):
         _tags = []
         for tname, tsha in self.repository.tags.iteritems():
             if tsha == self.raw_id:
                 _tags.append(tname)
         return _tags
     @LazyProperty
     def branch(self):
         # Note: This function will return one branch name for the changeset -
         # that might not make sense in Git where branches() is a better match
         # for the basic model
         heads = self.repository._heads(reverse=False)
         ref = heads.get(self.raw_id)
         if ref:
             return safe_unicode(ref)
     @LazyProperty
     def branches(self):
         heads = self.repository._heads(reverse=True)
         return [b for b in heads if heads[b] == self.raw_id] # FIXME: Inefficient ... and returning None!
     def _fix_path(self, path):
         """
         Paths are stored without trailing slash so we need to get rid off it if
         needed.
         """
         if path.endswith('/'):
             path = path.rstrip('/')
         return path
     def _get_id_for_path(self, path):
         path = safe_str(path)
         # FIXME: Please, spare a couple of minutes and make those codes cleaner;
         if path not in self._paths:
             path = path.strip('/')
             # set root tree
             tree = self.repository._repo[self._tree_id]
             if path == '':
                 self._paths[''] = tree.id
                 return tree.id
             splitted = path.split('/')
             dirs, name = splitted[:-1], splitted[-1]
             curdir = ''
             # initially extract things from root dir
             for item, stat, id in tree.iteritems():
                 if curdir:
                     name = '/'.join((curdir, item))
                 else:
                     name = item
                 self._paths[name] = id
                 self._stat_modes[name] = stat
             for dir in dirs:
                 if curdir:
                     curdir = '/'.join((curdir, dir))
                 else:
                     curdir = dir
                 dir_id = None
                 for item, stat, id in tree.iteritems():
                     if dir == item:
                         dir_id = id
                 if dir_id:
                     # Update tree
                     tree = self.repository._repo[dir_id]
                     if not isinstance(tree, objects.Tree):
                         raise ChangesetError('%s is not a directory' % curdir)
                 else:
                     raise ChangesetError('%s have not been found' % curdir)
                 # cache all items from the given traversed tree
                 for item, stat, id in tree.iteritems():
                     if curdir:
                         name = '/'.join((curdir, item))
                     else:
                         name = item
                     self._paths[name] = id
                     self._stat_modes[name] = stat
             if path not in self._paths:
                 raise NodeDoesNotExistError("There is no file nor directory "
                     "at the given path '%s' at revision %s"
                     % (path, safe_str(self.short_id)))
         return self._paths[path]
     def _get_kind(self, path):
         obj = self.repository._repo[self._get_id_for_path(path)]
         if isinstance(obj, objects.Blob):
             return NodeKind.FILE
         elif isinstance(obj, objects.Tree):
             return NodeKind.DIR
     def _get_filectx(self, path):
         path = self._fix_path(path)
         if self._get_kind(path) != NodeKind.FILE:
             raise ChangesetError("File does not exist for revision %s at "
                 " '%s'" % (self.raw_id, path))
         return path
     def _get_file_nodes(self):
         return chain(*(t[2] for t in self.walk()))
     @LazyProperty
     def parents(self):
         """
         Returns list of parents changesets.
         """
         return [self.repository.get_changeset(parent)
                 for parent in self._commit.parents]
     @LazyProperty
     def children(self):
         """
         Returns list of children changesets.
         """
         rev_filter = settings.GIT_REV_FILTER
         so, se = self.repository.run_git_command(
             ['rev-list', rev_filter, '--children']
+        )
         children = []
         pat = re.compile(r'^%s' % self.raw_id)
         for l in so.splitlines():
             if pat.match(l):
                 childs = l.split(' ')[1:]
                 children.extend(childs)
         return [self.repository.get_changeset(cs) for cs in children]
     def next(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 next_ = cs.revision + 1
                 next_rev = cs.repository.revisions[next_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(next_rev)
             if not branch or branch == cs.branch:
                 return cs
     def prev(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 prev_ = cs.revision - 1
                 if prev_ < 0:
                     raise IndexError
                 prev_rev = cs.repository.revisions[prev_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(prev_rev)
             if not branch or branch == cs.branch:
                 return cs
     def diff(self, ignore_whitespace=True, context=3):
         rev1 = self.parents[0] if self.parents else self.repository.EMPTY_CHANGESET
         rev2 = self
         return ''.join(self.repository.get_diff(rev1, rev2,
                                     ignore_whitespace=ignore_whitespace,
                                     context=context))
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given ``path``.
         """
         # ensure path is traversed
         path = safe_str(path)
         self._get_id_for_path(path)
         return self._stat_modes[path]
     def get_file_content(self, path):
         """
         Returns content of the file at given ``path``.
         """
         id = self._get_id_for_path(path)
         blob = self.repository._repo[id]
         return blob.as_pretty_string()
     def get_file_size(self, path):
         """
         Returns size of the file at given ``path``.
         """
         id = self._get_id_for_path(path)
         blob = self.repository._repo[id]
         return blob.raw_length()
     def get_file_changeset(self, path):
         """
         Returns last commit of the file at the given ``path``.
         """
         return self.get_file_history(path, limit=1)[0]
     def get_file_history(self, path, limit=None):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         TODO: This function now uses os underlying 'git' and 'grep' commands
         which is generally not good. Should be replaced with algorithm
         iterating commits.
         """
         self._get_filectx(path)
         cs_id = safe_str(self.id)
         f_path = safe_str(path)
         if limit is not None:
             cmd = ['log', '-n', str(safe_int(limit, 0)),
                    '--pretty=format:%H', '-s', cs_id, '--', f_path]
         else:
             cmd = ['log',
                    '--pretty=format:%H', '-s', cs_id, '--', f_path]
         so, se = self.repository.run_git_command(cmd)
         ids = re.findall(r'[0-9a-fA-F]{40}', so)
         return [self.repository.get_changeset(sha) for sha in ids]
     def get_file_history_2(self, path):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         """
         self._get_filectx(path)
         from dulwich.walk import Walker
         include = [self.id]
         walker = Walker(self.repository._repo.object_store, include,
                         paths=[path], max_entries=1)
         return [self.repository.get_changeset(sha)
                 for sha in (x.commit.id for x in walker)]
     def get_file_annotate(self, path):
         """
         Returns a generator of four element tuples with
             lineno, sha, changeset lazy loader and line
         TODO: This function now uses os underlying 'git' command which is
         generally not good. Should be replaced with algorithm iterating
         commits.
         """
         cmd = ['blame', '-l', '--root', '-r', self.id, '--', path]
         # -l     ==> outputs long shas (and we need all 40 characters)
         # --root ==> doesn't put '^' character for boundaries
         # -r sha ==> blames for the given revision
         so, se = self.repository.run_git_command(cmd)
         for i, blame_line in enumerate(so.split('\n')[:-1]):
             ln_no = i + 1
             sha, line = re.split(r' ', blame_line, 1)
             yield (ln_no, sha, lambda: self.repository.get_changeset(sha), line)
     def fill_archive(self, stream=None, kind='tgz', prefix=None,
                      subrepos=False):
         """
         Fills up given stream.
         :param stream: file like object.
         :param kind: one of following: ``zip``, ``tgz`` or ``tbz2``.
             Default: ``tgz``.
         :param prefix: name of root directory in archive.
             Default is repository name and changeset's raw_id joined with dash
             (``repo-tip.<KIND>``).
         :param subrepos: include subrepos in this archive.
         :raise ImproperArchiveTypeError: If given kind is wrong.
         :raise VcsError: If given stream is None
         """
         allowed_kinds = settings.ARCHIVE_SPECS.keys()
         if kind not in allowed_kinds:
             raise ImproperArchiveTypeError('Archive kind not supported use one'
                 'of %s', allowed_kinds)
         if prefix is None:
             prefix = '%s-%s' % (self.repository.name, self.short_id)
         elif prefix.startswith('/'):
             raise VCSError("Prefix cannot start with leading slash")
         elif prefix.strip() == '':
             raise VCSError("Prefix cannot be empty")
         if kind == 'zip':
             frmt = 'zip'
         else:
             frmt = 'tar'
         _git_path = settings.GIT_EXECUTABLE_PATH
         cmd = '%s archive --format=%s --prefix=%s/ %s' % (_git_path,
                                                 frmt, prefix, self.raw_id)
         if kind == 'tgz':
             cmd += ' | gzip -9'
         elif kind == 'tbz2':
             cmd += ' | bzip2 -9'
         if stream is None:
             raise VCSError('You need to pass in a valid stream for filling'
                            ' with archival data')
         popen = Popen(cmd, stdout=PIPE, stderr=PIPE, shell=True,
                       cwd=self.repository.path)
         buffer_size = 1024 * 8
         chunk = popen.stdout.read(buffer_size)
         while chunk:
             stream.write(chunk)
             chunk = popen.stdout.read(buffer_size)
         # Make sure all descriptors would be read
         popen.communicate()
     def get_nodes(self, path):
         if self._get_kind(path) != NodeKind.DIR:
             raise ChangesetError("Directory does not exist for revision %s at "
                 " '%s'" % (self.revision, path))
         path = self._fix_path(path)
         id = self._get_id_for_path(path)
         tree = self.repository._repo[id]
         dirnodes = []
         filenodes = []
         als = self.repository.alias
         for name, stat, id in tree.iteritems():
             if objects.S_ISGITLINK(stat):
                 dirnodes.append(SubModuleNode(name, url=None, changeset=id,
                                               alias=als))
                 continue
             obj = self.repository._repo.get_object(id)
             if path != '':
                 obj_path = '/'.join((path, name))
             else:
                 obj_path = name
             if obj_path not in self._stat_modes:
                 self._stat_modes[obj_path] = stat
             if isinstance(obj, objects.Tree):
                 dirnodes.append(DirNode(obj_path, changeset=self))
             elif isinstance(obj, objects.Blob):
                 filenodes.append(FileNode(obj_path, changeset=self, mode=stat))
             else:
                 raise ChangesetError("Requested object should be Tree "
                                      "or Blob, is %r" % type(obj))
         nodes = dirnodes + filenodes
         for node in nodes:
             if node.path not in self.nodes:
                 self.nodes[node.path] = node
         nodes.sort()
         return nodes
     def get_node(self, path):
         if isinstance(path, unicode):
             path = path.encode('utf-8')
         path = self._fix_path(path)
         if path not in self.nodes:
             try:
                 id_ = self._get_id_for_path(path)
             except ChangesetError:
                 raise NodeDoesNotExistError("Cannot find one of parents' "
                     "directories for a given path: %s" % path)
             _GL = lambda m: m and objects.S_ISGITLINK(m)
             if _GL(self._stat_modes.get(path)):
                 node = SubModuleNode(path, url=None, changeset=id_,
                                      alias=self.repository.alias)
             else:
                 obj = self.repository._repo.get_object(id_)
                 if isinstance(obj, objects.Tree):
                     if path == '':
                         node = RootNode(changeset=self)
                     else:
                         node = DirNode(path, changeset=self)
                     node._tree = obj
                 elif isinstance(obj, objects.Blob):
                     node = FileNode(path, changeset=self)
                     node._blob = obj
                 else:
                     raise NodeDoesNotExistError("There is no file nor directory "
                         "at the given path '%s' at revision %s"
                         % (path, self.short_id))
             # cache node
             self.nodes[path] = node
         return self.nodes[path]
     @LazyProperty
     def affected_files(self):
         """
         Gets a fast accessible file changes for given changeset
         """
         added, modified, deleted = self._changes_cache
         return list(added.union(modified).union(deleted))
     @LazyProperty
     def _diff_name_status(self):
         output = []
         for parent in self.parents:
             cmd = ['diff', '--name-status', parent.raw_id, self.raw_id,
                    '--encoding=utf8']
             so, se = self.repository.run_git_command(cmd)
             output.append(so.strip())
         return '\n'.join(output)
     @LazyProperty
     def _changes_cache(self):
         added = set()
         modified = set()
         deleted = set()
         _r = self.repository._repo
         parents = self.parents
         if not self.parents:
             parents = [EmptyChangeset()]
         for parent in parents:
             if isinstance(parent, EmptyChangeset):
                 oid = None
             else:
                 oid = _r[parent.raw_id].tree
             changes = _r.object_store.tree_changes(oid, _r[self.raw_id].tree)
             for (oldpath, newpath), (_, _), (_, _) in changes:
                 if newpath and oldpath:
                     modified.add(newpath)
                 elif newpath and not oldpath:
                     added.add(newpath)
                 elif not newpath and oldpath:
                     deleted.add(oldpath)
         return added, modified, deleted
     def _get_paths_for_status(self, status):
         """
         Returns sorted list of paths for given ``status``.
         :param status: one of: *added*, *modified* or *deleted*
         """
         added, modified, deleted = self._changes_cache
         return sorted({
             'added': list(added),
             'modified': list(modified),
             'deleted': list(deleted)}[status]
+        )
     @LazyProperty
     def added(self):
         """
         Returns list of added ``FileNode`` objects.
         """
         if not self.parents:
             return list(self._get_file_nodes())
         return AddedFileNodesGenerator([n for n in
                                 self._get_paths_for_status('added')], self)
     @LazyProperty
     def changed(self):
         """
         Returns list of modified ``FileNode`` objects.
         """
         if not self.parents:
             return []
         return ChangedFileNodesGenerator([n for n in
                                 self._get_paths_for_status('modified')], self)
     @LazyProperty
     def removed(self):
         """
         Returns list of removed ``FileNode`` objects.
         """
         if not self.parents:
             return []
         return RemovedFileNodesGenerator([n for n in
                                 self._get_paths_for_status('deleted')], self)
     extra = {}

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

➞

Show inline comments

 import os
 import posixpath
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.backends.base import BaseChangeset
 from kallithea.lib.vcs.exceptions import (
     ChangesetDoesNotExistError, ChangesetError, ImproperArchiveTypeError,
     NodeDoesNotExistError, VCSError
+)
 from kallithea.lib.vcs.nodes import (
     AddedFileNodesGenerator, ChangedFileNodesGenerator, DirNode, FileNode,
     NodeKind, RemovedFileNodesGenerator, RootNode, SubModuleNode
+)
 from kallithea.lib.vcs.utils import safe_str, safe_unicode, date_fromtimestamp
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.vcs.utils.paths import get_dirs_for_path
 from kallithea.lib.vcs.utils.hgcompat import archival, hex
 from mercurial import obsolete
 class MercurialChangeset(BaseChangeset):
     """
     Represents state of the repository at the single revision.
     """
     def __init__(self, repository, revision):
         self.repository = repository
         assert isinstance(revision, basestring), repr(revision)
         self.raw_id = revision
         self._ctx = repository._repo[revision]
         self.revision = self._ctx._rev
         self.nodes = {}
     @LazyProperty
     def tags(self):
         return map(safe_unicode, self._ctx.tags())
     @LazyProperty
     def branch(self):
         return safe_unicode(self._ctx.branch())
     @LazyProperty
     def branches(self):
         return [safe_unicode(self._ctx.branch())]
     @LazyProperty
     def closesbranch(self):
         return self._ctx.closesbranch()
     @LazyProperty
     def obsolete(self):
         return self._ctx.obsolete()
     @LazyProperty
     def bumped(self):
         return self._ctx.bumped()
     @LazyProperty
     def divergent(self):
         return self._ctx.divergent()
     @LazyProperty
     def extinct(self):
         return self._ctx.extinct()
     @LazyProperty
     def unstable(self):
         return self._ctx.unstable()
     @LazyProperty
     def phase(self):
         if(self._ctx.phase() == 1):
             return 'Draft'
         elif(self._ctx.phase() == 2):
             return 'Secret'
         else:
             return ''
     @LazyProperty
     def successors(self):
         successors = obsolete.successorssets(self._ctx._repo, self._ctx.node())
         if successors:
             # flatten the list here handles both divergent (len > 1)
             # and the usual case (len = 1)
             successors = [hex(n)[:12] for sub in successors for n in sub if n != self._ctx.node()]
         return successors
     @LazyProperty
     def precursors(self):
         precursors = set()
         nm = self._ctx._repo.changelog.nodemap
         for p in self._ctx._repo.obsstore.precursors.get(self._ctx.node(), ()):
             pr = nm.get(p[0])
             if pr is not None:
                 precursors.add(hex(p[0])[:12])
         return precursors
     @LazyProperty
     def bookmarks(self):
         return map(safe_unicode, self._ctx.bookmarks())
     @LazyProperty
     def message(self):
         return safe_unicode(self._ctx.description())
     @LazyProperty
     def committer(self):
         return safe_unicode(self.author)
     @LazyProperty
     def author(self):
         return safe_unicode(self._ctx.user())
     @LazyProperty
     def date(self):
         return date_fromtimestamp(*self._ctx.date())
     @LazyProperty
     def _timestamp(self):
         return self._ctx.date()[0]
     @LazyProperty
     def status(self):
         """
         Returns modified, added, removed, deleted files for current changeset
         """
         return self.repository._repo.status(self._ctx.p1().node(),
                                             self._ctx.node())
     @LazyProperty
     def _file_paths(self):
         return list(self._ctx)
     @LazyProperty
     def _dir_paths(self):
         p = list(set(get_dirs_for_path(*self._file_paths)))
         p.insert(0, '')
         return p
     @LazyProperty
     def _paths(self):
         return self._dir_paths + self._file_paths
     @LazyProperty
     def id(self):
         if self.last:
             return u'tip'
         return self.short_id
     @LazyProperty
     def short_id(self):
         return self.raw_id[:12]
     @LazyProperty
     def parents(self):
         """
         Returns list of parents changesets.
         """
         return [self.repository.get_changeset(parent.rev())
                 for parent in self._ctx.parents() if parent.rev() >= 0]
     @LazyProperty
     def children(self):
         """
         Returns list of children changesets.
         """
         return [self.repository.get_changeset(child.rev())
                 for child in self._ctx.children() if child.rev() >= 0]
     def next(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 next_ = cs.repository.revisions.index(cs.raw_id) + 1
                 next_rev = cs.repository.revisions[next_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(next_rev)
             if not branch or branch == cs.branch:
                 return cs
     def prev(self, branch=None):
         if branch and self.branch != branch:
             raise VCSError('Branch option used on changeset not belonging '
                            'to that branch')
         cs = self
         while True:
             try:
                 prev_ = cs.repository.revisions.index(cs.raw_id) - 1
                 if prev_ < 0:
                     raise IndexError
                 prev_rev = cs.repository.revisions[prev_]
             except IndexError:
                 raise ChangesetDoesNotExistError
             cs = cs.repository.get_changeset(prev_rev)
             if not branch or branch == cs.branch:
                 return cs
     def diff(self, ignore_whitespace=True, context=3):
         return ''.join(self._ctx.diff(git=True,
                                       ignore_whitespace=ignore_whitespace,
                                       context=context))
     def _fix_path(self, path):
         """
         Paths are stored without trailing slash so we need to get rid off it if
         needed. Also mercurial keeps filenodes as str so we need to decode
         from unicode to str
         """
         if path.endswith('/'):
             path = path.rstrip('/')
         return safe_str(path)
     def _get_kind(self, path):
         path = self._fix_path(path)
         if path in self._file_paths:
             return NodeKind.FILE
         elif path in self._dir_paths:
             return NodeKind.DIR
         else:
             raise ChangesetError("Node does not exist at the given path '%s'"
                 % (path))
     def _get_filectx(self, path):
         path = self._fix_path(path)
         if self._get_kind(path) != NodeKind.FILE:
             raise ChangesetError("File does not exist for revision %s at "
                 " '%s'" % (self.raw_id, path))
         return self._ctx.filectx(path)
     def _extract_submodules(self):
         """
         returns a dictionary with submodule information from substate file
         of hg repository
         """
         return self._ctx.substate
     def get_file_mode(self, path):
         """
         Returns stat mode of the file at the given ``path``.
         """
         fctx = self._get_filectx(path)
         if 'x' in fctx.flags():
             return 0100755
         else:
             return 0100644
     def get_file_content(self, path):
         """
         Returns content of the file at given ``path``.
         """
         fctx = self._get_filectx(path)
         return fctx.data()
     def get_file_size(self, path):
         """
         Returns size of the file at given ``path``.
         """
         fctx = self._get_filectx(path)
         return fctx.size()
     def get_file_changeset(self, path):
         """
         Returns last commit of the file at the given ``path``.
         """
         return self.get_file_history(path, limit=1)[0]
     def get_file_history(self, path, limit=None):
         """
         Returns history of file as reversed list of ``Changeset`` objects for
         which file at given ``path`` has been modified.
         """
         fctx = self._get_filectx(path)
         hist = []
         cnt = 0
         for cs in reversed([x for x in fctx.filelog()]):
             cnt += 1
             hist.append(hex(fctx.filectx(cs).node()))
             if limit is not None and cnt == limit:
                 break
         return [self.repository.get_changeset(node) for node in hist]
     def get_file_annotate(self, path):
         """
         Returns a generator of four element tuples with
             lineno, sha, changeset lazy loader and line
         """
         fctx = self._get_filectx(path)
         for i, (aline, l) in enumerate(fctx.annotate(linenumber=False)):
             ln_no = i + 1
             try:
                 fctx = aline.fctx
             except AttributeError: # aline.fctx was introduced in Mercurial 4.4
                 fctx = aline[0]
             sha = hex(fctx.node())
             yield (ln_no, sha, lambda: self.repository.get_changeset(sha), l)
     def fill_archive(self, stream=None, kind='tgz', prefix=None,
                      subrepos=False):
         """
         Fills up given stream.
         :param stream: file like object.
         :param kind: one of following: ``zip``, ``tgz`` or ``tbz2``.
             Default: ``tgz``.
         :param prefix: name of root directory in archive.
             Default is repository name and changeset's raw_id joined with dash
             (``repo-tip.<KIND>``).
         :param subrepos: include subrepos in this archive.
         :raise ImproperArchiveTypeError: If given kind is wrong.
         :raise VcsError: If given stream is None
         """
         allowed_kinds = settings.ARCHIVE_SPECS.keys()
         if kind not in allowed_kinds:
             raise ImproperArchiveTypeError('Archive kind not supported use one'
                 'of %s', allowed_kinds)
         if stream is None:
             raise VCSError('You need to pass in a valid stream for filling'
                            ' with archival data')
         if prefix is None:
             prefix = '%s-%s' % (self.repository.name, self.short_id)
         elif prefix.startswith('/'):
             raise VCSError("Prefix cannot start with leading slash")
         elif prefix.strip() == '':
             raise VCSError("Prefix cannot be empty")
         archival.archive(self.repository._repo, stream, self.raw_id,
                          kind, prefix=prefix, subrepos=subrepos)
     def get_nodes(self, path):
         """
         Returns combined ``DirNode`` and ``FileNode`` objects list representing
         state of changeset at the given ``path``. If node at the given ``path``
         is not instance of ``DirNode``, ChangesetError would be raised.
         """
         if self._get_kind(path) != NodeKind.DIR:
             raise ChangesetError("Directory does not exist for revision %s at "
                 " '%s'" % (self.revision, path))
         path = self._fix_path(path)
         filenodes = [FileNode(f, changeset=self) for f in self._file_paths
             if os.path.dirname(f) == path]
         dirs = path == '' and '' or [d for d in self._dir_paths
             if d and posixpath.dirname(d) == path]
         dirnodes = [DirNode(d, changeset=self) for d in dirs
             if os.path.dirname(d) == path]
         als = self.repository.alias
         for k, vals in self._extract_submodules().iteritems():
             #vals = url,rev,type
             loc = vals[0]
             cs = vals[1]
             dirnodes.append(SubModuleNode(k, url=loc, changeset=cs,
                                           alias=als))
         nodes = dirnodes + filenodes
         # cache nodes
         for node in nodes:
             self.nodes[node.path] = node
         nodes.sort()
         return nodes
     def get_node(self, path):
         """
         Returns ``Node`` object from the given ``path``. If there is no node at
         the given ``path``, ``ChangesetError`` would be raised.
         """
         path = self._fix_path(path)
         if path not in self.nodes:
             if path in self._file_paths:
                 node = FileNode(path, changeset=self)
             elif path in self._dir_paths or path in self._dir_paths:
                 if path == '':
                     node = RootNode(changeset=self)
                 else:
                     node = DirNode(path, changeset=self)
             else:
                 raise NodeDoesNotExistError("There is no file nor directory "
                     "at the given path: '%s' at revision %s"
                     % (path, self.short_id))
             # cache node
             self.nodes[path] = node
         return self.nodes[path]
     @LazyProperty
     def affected_files(self):
         """
         Gets a fast accessible file changes for given changeset
         """
         return self._ctx.files()
     @property
     def added(self):
         """
         Returns list of added ``FileNode`` objects.
         """
         return AddedFileNodesGenerator([n for n in self.status[1]], self)
     @property
     def changed(self):
         """
         Returns list of modified ``FileNode`` objects.
         """
         return ChangedFileNodesGenerator([n for n in self.status[0]], self)
     @property
     def removed(self):
         """
         Returns list of removed ``FileNode`` objects.
         """
         return RemovedFileNodesGenerator([n for n in self.status[2]], self)
     @LazyProperty
     def extra(self):
         return self._ctx.extra()

kallithea/templates/changelog/changelog_table.html

➞

Show inline comments

 ## Render changelog table with id 'changesets' with the range of changesets,
 ## statuses, and comments.
 ## Optionally, pass a js snippet to run whenever a table resize is triggered.
 <%def name="changelog(repo_name, cs_range, cs_statuses, cs_comments, show_checkbox=False, show_branch=True, show_index=False, resize_js='')">
     <% num_cs = len(cs_range) %>
     <table class="table" id="changesets">
     <tbody>
       %for cnt,cs in enumerate(cs_range):
       <tr id="chg_${cnt+1}" class="${'mergerow' if len(cs.parents) > 1 else ''}">
         %if show_checkbox:
         <td class="checkbox-column">
           ${h.checkbox(cs.raw_id,class_="changeset_range")}
         </td>
         %endif
         %if show_index:
         <td class="changeset-logical-index">
           <%
               index = num_cs - cnt
               if index == 1:
                   title = _('First (oldest) changeset in this list')
               elif index == num_cs:
                   title = _('Last (most recent) changeset in this list')
               else:
                   title = _('Position in this list of changesets')
           %>
           <span data-toggle="tooltip" title="${title}">
             ${index}
           </span>
         </td>
         %endif
         <td class="status">
           %if cs_statuses.get(cs.raw_id):
             %if cs_statuses.get(cs.raw_id)[2]:
               <a data-toggle="tooltip"
                   title="${_('Changeset status: %s by %s\nClick to open associated pull request %s') % (cs_statuses.get(cs.raw_id)[1], cs_statuses.get(cs.raw_id)[5].username, cs_statuses.get(cs.raw_id)[4])}"
                   href="${h.url('pullrequest_show',repo_name=cs_statuses.get(cs.raw_id)[3],pull_request_id=cs_statuses.get(cs.raw_id)[2])}">
                 <i class="icon-circle changeset-status-${cs_statuses.get(cs.raw_id)[0]}"></i>
               </a>
             %else:
               <a data-toggle="tooltip"
                   title="${_('Changeset status: %s by %s') % (cs_statuses.get(cs.raw_id)[1], cs_statuses.get(cs.raw_id)[5].username)}"
                   href="${cs_comments[cs.raw_id][0].url()}">
                 <i class="icon-circle changeset-status-${cs_statuses.get(cs.raw_id)[0]}"></i>
               </a>
             %endif
           %endif
         </td>
         <td class="author" data-toggle="tooltip" title="${cs.author}">
           ${h.gravatar(h.email_or_none(cs.author), size=16)}
           <span class="user">${h.person(cs.author)}</span>
         </td>
         <td class="hash">
           ${h.link_to(h.show_id(cs),h.url('changeset_home',repo_name=repo_name,revision=cs.raw_id), class_='changeset_hash')}
         </td>
         <td class="date">
           <div data-toggle="tooltip" title="${h.fmt_date(cs.date)}">${h.age(cs.date,True)}</div>
         </td>
         <% message_lines = cs.message.splitlines() %>
         %if len(message_lines) > 1:
         <td class="expand_commit" title="${_('Expand commit message')}">
           <i class="icon-align-left"></i>
         </td>
         %else:
         <td class="expand_commit"></td>
         %endif
         <td class="mid">
           <div class="log-container">
             <div class="message">
               <div class="message-firstline">${h.urlify_text(message_lines[0], c.repo_name,h.url('changeset_home',repo_name=repo_name,revision=cs.raw_id))}</div>
               %if len(message_lines) > 1:
               <div class="message-full hidden">${h.urlify_text(cs.message, repo_name)}</div>
               %endif
             </div>
             <div class="extra-container">
               %if cs_comments.get(cs.raw_id):
                 <a class="comments-container comments-cnt" href="${cs_comments[cs.raw_id][0].url()}" data-toggle="tooltip" title="${_('%s comments') % len(cs_comments[cs.raw_id])}">${len(cs_comments[cs.raw_id])}<i class="icon-comment-discussion"></i>
                 </a>
               %endif
               %for book in cs.bookmarks:
                 <span class="label label-bookmark" title="${_('Bookmark %s') % book}">${h.link_to(book,h.url('changeset_home',repo_name=repo_name,revision=cs.raw_id))}</span>
               %endfor
               %for tag in cs.tags:
                 <span class="label label-tag" title="${_('Tag %s') % tag}">${h.link_to(tag,h.url('changeset_home',repo_name=repo_name,revision=cs.raw_id))}</span>
               %endfor
               %if cs.bumped:
                 <span class="label label-bumped" title="Bumped">Bumped</span>
               %endif
               %if cs.divergent:
                 <span class="label label-divergent" title="Divergent">Divergent</span>
               %endif
               %if cs.extinct:
                 <span class="label label-extinct" title="Extinct">Extinct</span>
               %endif
               %if cs.unstable:
                 <span class="label label-unstable" title="Unstable">Unstable</span>
               %endif
               %if cs.phase:
                 <span class="label label-phase" title="Phase">${cs.phase}</span>
               %endif
               %if show_branch and cs.branch:
                 <span class="label label-branch" title="${_('Branch %s' % cs.branch)}">${h.link_to(cs.branch,h.url('changelog_home',repo_name=repo_name,branch=cs.branch))}</span>
               %if show_branch:
                 %for branch in cs.branches:
                   <span class="label label-branch" title="${_('Branch %s' % branch)}">${h.link_to(branch,h.url('changelog_home',repo_name=repo_name,branch=branch))}</span>
                 %endfor
               %endif
             </div>
           </div>
         </td>
       </tr>
       %endfor
     </tbody>
     </table>
 <script type="text/javascript">
   $(document).ready(function() {
     $('#changesets .expand_commit').on('click',function(e){
       $(this).next('.mid').find('.message > div').toggleClass('hidden');
       ${resize_js};
     });
   });
 </script>
 </%def>

kallithea/templates/changeset/changeset.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%namespace name="comment" file="/changeset/changeset_file_comment.html"/>
 <%block name="title">
     ${_('%s Changeset') % c.repo_name} - ${h.show_id(c.changeset)}
 </%block>
 <%def name="breadcrumbs_links()">
     ${_('Changeset')} - <span class='changeset_hash'>${h.show_id(c.changeset)}</span>
 </%def>
 <%block name="header_menu">
     ${self.menu('repositories')}
 </%block>
 <%def name="main()">
 ${self.repo_context_bar('changelog', c.changeset.raw_id)}
 <div class="panel panel-primary">
   <div class="panel-heading clearfix">
     ${self.breadcrumbs()}
   </div>
   <script>
     var _USERS_AC_DATA = ${h.js(c.users_array)};
     AJAX_COMMENT_URL = ${h.js(url('changeset_comment',repo_name=c.repo_name,revision=c.changeset.raw_id))};
     AJAX_COMMENT_DELETE_URL = ${h.js(url('changeset_comment_delete',repo_name=c.repo_name,comment_id='__COMMENT_ID__'))};
   </script>
   <div class="panel-body">
     <div class="panel panel-default">
         <div class="panel-heading clearfix">
             ${self.parent_child_navigation()}
                 <div class="pull-left" title="${_('Changeset status')}">
                     %if c.statuses:
                         <i class="icon-circle changeset-status-${c.statuses[0]}"></i>
                         [${h.changeset_status_lbl(c.statuses[0])}]
                     %endif
                 </div>
                 <div class="diff-actions pull-left">
                   <a href="${h.url('changeset_raw_home',repo_name=c.repo_name,revision=c.changeset.raw_id)}"
                      data-toggle="tooltip"
                      title="${_('Raw diff')}"><i class="icon-diff"></i></a>
                   <a href="${h.url('changeset_patch_home',repo_name=c.repo_name,revision=c.changeset.raw_id)}"
                      data-toggle="tooltip"
                      title="${_('Patch diff')}"><i class="icon-file-powerpoint"></i></a>
                   <a href="${h.url('changeset_download_home',repo_name=c.repo_name,revision=c.changeset.raw_id,diff='download')}"
                      data-toggle="tooltip"
                      title="${_('Download diff')}"><i class="icon-floppy"></i></a>
                   ${c.ignorews_url(request.GET)}
                   ${c.context_url(request.GET)}
                 </div>
         </div>
         <div class="panel-body">
             <div class="form-group changeset_content_header clearfix">
                 <div class="pull-right">
                     <span class="logtags">
                         %if len(c.changeset.parents)>1:
                         <span class="label label-merge">${_('Merge')}</span>
                         %endif
                         %for book in c.changeset.bookmarks:
                         <span class="label label-bookmark" title="${_('Bookmark %s') % book}">${h.link_to(book,h.url('changeset_home',repo_name=c.repo_name,revision=c.changeset.raw_id))}</span>
                         %endfor
                         %for tag in c.changeset.tags:
                          <span class="label label-tag"  title="${_('Tag %s') % tag}">${h.link_to(tag,h.url('changeset_home',repo_name=c.repo_name,revision=c.changeset.raw_id))}</span>
                         %endfor
                         %if c.changeset.branch:
                          <span class="label label-branch" title="${_('Branch %s') % c.changeset.branch}">${h.link_to(c.changeset.branch,h.url('changelog_home',repo_name=c.repo_name,branch=c.changeset.branch))}</span>
                         %endif
                         %for branch in c.changeset.branches:
                           <span class="label label-branch" title="${_('Branch %s') % branch}">${h.link_to(branch,h.url('changelog_home',repo_name=c.repo_name,branch=branch))}</span>
                         %endfor
                     </span>
                     <div class="changes">
                         % if (len(c.changeset.affected_files) <= c.affected_files_cut_off) or c.fulldiff:
                          <span class="label deleted" title="${_('Removed')}">${len(c.changeset.removed)}</span>
                          <span class="label changed" title="${_('Changed')}">${len(c.changeset.changed)}</span>
                          <span class="label added" title="${_('Added')}">${len(c.changeset.added)}</span>
                         % else:
                          <span class="label deleted" title="${_('Affected %s files') % len(c.changeset.affected_files)}">!</span>
                          <span class="label changed" title="${_('Affected %s files') % len(c.changeset.affected_files)}">!</span>
                          <span class="label added"   title="${_('Affected %s files') % len(c.changeset.affected_files)}">!</span>
                         % endif
                     </div>
                 </div>
                 <div class="pull-left">
                      <div class="author">
                          ${h.gravatar_div(h.email_or_none(c.changeset.author), size=20)}
                          <span><b>${h.person(c.changeset.author,'full_name_and_username')}</b> - ${h.age(c.changeset.date,True)} ${h.fmt_date(c.changeset.date)}</span><br/>
                          <span>${h.email_or_none(c.changeset.author)}</span><br/>
                      </div>
                      <% rev = c.changeset.extra.get('source') %>
                      %if rev:
                      <div>
                        ${_('Grafted from:')} ${h.link_to(h.short_id(rev),h.url('changeset_home',repo_name=c.repo_name,revision=rev), class_="changeset_hash")}
                      </div>
                      %endif
                      <% rev = c.changeset.extra.get('transplant_source', '').encode('hex') %>
                      %if rev:
                      <div>
                        ${_('Transplanted from:')} ${h.link_to(h.short_id(rev),h.url('changeset_home',repo_name=c.repo_name,revision=rev), class_="changeset_hash")}
                      </div>
                      %endif
                      % if hasattr(c.changeset, 'successors') and c.changeset.successors:
                      <div class='successors'>
                        <span class='successors_header'>${_('Replaced by:')} </span>
                        % for i, s in enumerate(c.changeset.successors):
                            <%
                            comma = ""
                            if i != len(c.changeset.successors)-1:
                              comma = ", "
                            %>
                          <a class='successors_hash' href="${h.url('changeset_home',repo_name=c.repo_name, revision=s)}">${s}</a>${comma}
                        % endfor
                      </div>
                      % endif
                      % if hasattr(c.changeset, 'precursors') and c.changeset.precursors:
                      <div class='precursors'>
                        <span class='precursors_header'>${_('Preceded by:')} </span>
                        % for i, s in enumerate(c.changeset.precursors):
                            <%
                            comma = ""
                            if i != len(c.changeset.precursors)-1:
                              comma = ", "
                            %>
                            <a class="precursors_hash" href="${h.url('changeset_home',repo_name=c.repo_name, revision=s)}">${s}</a>${comma}
                        % endfor
                      </div>
                      % endif
                 </div>
             </div>
             <div class="form-group">${h.urlify_text(c.changeset.message, c.repo_name)}</div>
             <div>
               <% a_rev, cs_rev, file_diff_data = c.changes[c.changeset.raw_id] %>
               % if c.limited_diff:
                   ${ungettext('%s file changed', '%s files changed', len(file_diff_data)) % len(file_diff_data)}:
               % else:
                   ${ungettext('%s file changed with %s insertions and %s deletions', '%s files changed with %s insertions and %s deletions', len(file_diff_data)) % (len(file_diff_data), c.lines_added, c.lines_deleted)}:
               %endif
               </div>
               <div class="cs_files">
                 %for fid, url_fid, op, a_path, path, diff, stats in file_diff_data:
                     <div class="cs_${op} clearfix">
                       <span class="node">
                           <i class="icon-diff-${op}"></i>${h.link_to(h.safe_unicode(path), '#%s' % fid)}
                       </span>
                       <div class="changes">${h.fancy_file_stats(stats)}</div>
                     </div>
                 %endfor
                 %if c.limited_diff:
                   <h5>${_('Changeset was too big and was cut off...')} <a href="${h.url.current(fulldiff=1, **request.GET.mixed())}">${_('Show full diff anyway')}</a></h5>
                 %endif
             </div>
             <div class="comments-number">
                 ${comment.comment_count(c.inline_cnt, len(c.comments))}
             </div>
         </div>
     </div>
     ## diff block
     <div class="commentable-diff">
     <%namespace name="diff_block" file="/changeset/diff_block.html"/>
     ${diff_block.diff_block_js()}
     <% a_rev, cs_rev, file_diff_data = c.changes[c.changeset.raw_id] %>
     ${diff_block.diff_block(c.repo_name, 'rev', a_rev, a_rev,
                             c.repo_name, 'rev', cs_rev, cs_rev, file_diff_data)}
     % if c.limited_diff:
       <h4>${_('Changeset was too big and was cut off...')} <a href="${h.url.current(fulldiff=1, **request.GET.mixed())}">${_('Show full diff anyway')}</a></h4>
     % endif
     </div>
     ## template for inline comment form
     ${comment.comment_inline_form()}
     ## render comments and inlines
     ${comment.generate_comments()}
     ## main comment form and it status
     ${comment.comments()}
     </div>
     ## FORM FOR MAKING JS ACTION AS CHANGESET COMMENTS
     <script type="text/javascript">
       $(document).ready(function(){
           $('.code-difftable').on('click', '.add-bubble', function(e){
               show_comment_form($(this));
           });
           move_comments($(".comments .comments-list-chunk"));
           // hack: re-navigate to target after JS is done ... if a target is set and setting href thus won't reload
           if (window.location.hash != "") {
               window.location.href = window.location.href;
+          }
       });
     </script>
   </div>
 </%def>

kallithea/templates/changeset/changeset_range.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%block name="title">
     ${_('%s Changesets') % c.repo_name} - ${h.show_id(c.cs_ranges[0])} &gt; ${h.show_id(c.cs_ranges[-1])}
 </%block>
 <%def name="breadcrumbs_links()">
     ${_('Changesets')} -
     ${h.link_to(h.show_id(c.cs_ranges[0]),h.url('changeset_home',repo_name=c.repo_name,revision=c.cs_ranges[0].raw_id))}
     <i class="icon-right"></i>
     ${h.link_to(h.show_id(c.cs_ranges[-1]),h.url('changeset_home',repo_name=c.repo_name,revision=c.cs_ranges[-1].raw_id))}
 </%def>
 <%block name="header_menu">
     ${self.menu('repositories')}
 </%block>
 <%def name="main()">
 ${self.repo_context_bar('changelog')}
 <div class="panel panel-primary">
     <div class="panel-heading clearfix">
         <div class="pull-left">
             ${self.breadcrumbs()}
         </div>
         <div class="pull-right">
             <a href="${h.url('compare_url',repo_name=c.repo_name,org_ref_type='rev',org_ref_name=getattr(c.cs_ranges[0].parents[0] if c.cs_ranges[0].parents else h.EmptyChangeset(),'raw_id'),other_ref_type='rev',other_ref_name=c.cs_ranges[-1].raw_id)}" class="btn btn-default btn-sm"><i class="icon-git-compare"></i>Compare Revisions</a>
         </div>
     </div>
     <div class="panel-body">
         <div>
                 <table class="table compare_view_commits">
                 %for cnt,cs in enumerate(c.cs_ranges):
                   <tr>
                     %if c.visual.use_gravatar:
                     <td>${h.gravatar_div(h.email_or_none(cs.author), size=14)}</td>
                     %endif
                     <td>${h.link_to(h.short_id(cs.raw_id),h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}</td>
                     <td class="author">${h.person(cs.author)}</td>
                     <td><span data-toggle="tooltip" title="${h.age(cs.date)}">${cs.date}</span></td>
                     <td>
                       %if c.statuses:
                         <i class="icon-circle changeset-status-${c.statuses[cnt]}" title="${_('Changeset status: %s') % h.changeset_status_lbl(c.statuses[cnt])}"></i>
                       %endif
                     </td>
                     <td><div class="message">${h.urlify_text(h.wrap_paragraphs(cs.message),c.repo_name)}</div></td>
                   </tr>
                 %endfor
                 </table>
                 <h4>${_('Files affected')}</h4>
                 <div class="cs_files">
                     %for cs in c.cs_ranges:
                         <h6>${h.link_to(h.show_id(cs),h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}</h6>
                         <% a_rev, cs_rev, file_diff_data = c.changes[cs.raw_id] %>
                         %for fid, url_fid, op, a_path, path, diff, stats in file_diff_data:
                             <div class="cs_${op} clearfix">
                                 <span class="node">
                                     <i class="icon-diff-${op}"></i>
                                     ${h.link_to(h.safe_unicode(path), '#%s' % fid)}
                                 </span>
                                 <div class="changes">${h.fancy_file_stats(stats)}</div>
                             </div>
                         %endfor
                     %endfor
                 </div>
         </div>
     </div>
     <%namespace name="comment" file="/changeset/changeset_file_comment.html"/>
     <%namespace name="diff_block" file="/changeset/diff_block.html"/>
     ${diff_block.diff_block_js()}
     %for cs in c.cs_ranges:
         <div class="panel-body">
           ## diff block
           <div class="h3">
           ${h.gravatar_div(h.email_or_none(cs.author), size=20)}
           <a data-toggle="tooltip" title="${cs.message}" href="${h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id)}">${h.show_id(cs)}</a>
             <div class="right">
               <span class="logtags">
                 %if len(cs.parents)>1:
                 <span class="label label-merge">${_('Merge')}</span>
                 %endif
                 %if h.is_hg(c.db_repo_scm_instance):
                   %for book in cs.bookmarks:
                   <span class="label label-bookmark" title="${_('Bookmark %s') % book}">
                      ${h.link_to(book,h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}
                   </span>
                   %endfor
                 %endif
                 %for tag in cs.tags:
                     <span class="label label-tag" title="${_('Tag %s') % tag}">
                     ${h.link_to(tag,h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}</span>
                 %endfor
                 %if cs.branch:
                 <span class="label label-branch" title="${_('Branch %s') % cs.branch}">
                    ${h.link_to(cs.branch,h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}
                 </span>
                 %endif
                 %for branch in cs.branches:
                   <span class="label label-branch" title="${_('Branch %s') % branch}">
                     ${h.link_to(branch,h.url('changeset_home',repo_name=c.cs_repo.repo_name,revision=cs.raw_id))}
                   </span>
                 %endfor
               </span>
             </div>
           </div>
           <% a_rev, cs_rev, file_diff_data = c.changes[cs.raw_id] %>
           ${diff_block.diff_block(c.repo_name, 'rev', a_rev, a_rev,
                                   c.repo_name, 'rev', cs_rev, cs_rev, file_diff_data)}
         </div>
     %endfor
 </div>
 </%def>

kallithea/tests/vcs/test_branches.py

➞

Show inline comments

 import datetime
 from kallithea.lib import vcs
 from kallithea.lib.vcs.nodes import FileNode
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import SCM_TESTS
 class BranchesTestCaseMixin(_BackendTestMixin):
     @classmethod
     def _get_commits(cls):
         commits = [
+            {
                 'message': 'Initial commit',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar', content='Foobar'),
                     FileNode('foobar2', content='Foobar II'),
                     FileNode('foo/bar/baz', content='baz here!'),
                 ],
             },
+            {
                 'message': 'Changes...',
                 'author': 'Jane Doe <jane.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 21),
                 'added': [
                     FileNode('some/new.txt', content='news...'),
                 ],
                 'changed': [
                     FileNode('foobar', 'Foobar I'),
                 ],
                 'removed': [],
             },
+        ]
         return commits
     def test_simple(self):
         tip = self.repo.get_changeset()
         assert tip.date == datetime.datetime(2010, 1, 1, 21)
     def test_new_branch(self):
         # This check must not be removed to ensure the 'branches' LazyProperty
         # gets hit *before* the new 'foobar' branch got created:
         assert 'foobar' not in self.repo.branches
         self.imc.add(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
+        )
         assert 'foobar' in self.repo.branches
         assert foobar_tip.branch == 'foobar'
         assert foobar_tip.branches == ['foobar']
     def test_new_head(self):
         tip = self.repo.get_changeset()
         self.imc.add(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
             parents=[tip],
+        )
         self.imc.change(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\nand more...\n'))
         newtip = self.imc.commit(
             message=u'At default branch',
             author=u'joe',
             branch=foobar_tip.branch,
             parents=[foobar_tip],
+        )
         newest_tip = self.imc.commit(
             message=u'Merged with %s' % foobar_tip.raw_id,
             author=u'joe',
             branch=self.backend_class.DEFAULT_BRANCH_NAME,
             parents=[newtip, foobar_tip],
+        )
         assert newest_tip.branch == self.backend_class.DEFAULT_BRANCH_NAME
         assert newest_tip.branches == [self.backend_class.DEFAULT_BRANCH_NAME]
     def test_branch_with_slash_in_name(self):
         self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))
         self.imc.commit(u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         assert 'issue/123' in self.repo.branches
     def test_branch_with_slash_in_name_and_similar_without(self):
         self.imc.add(vcs.nodes.FileNode('extrafile', content='Some data\n'))
         self.imc.commit(u'Branch with a slash!', author=u'joe',
             branch='issue/123')
         self.imc.add(vcs.nodes.FileNode('extrafile II', content='Some data\n'))
         self.imc.commit(u'Branch without a slash...', author=u'joe',
             branch='123')
         assert 'issue/123' in self.repo.branches
         assert '123' in self.repo.branches
 # For each backend create test case class
 for alias in SCM_TESTS:
     attrs = {
         'backend_alias': alias,
+    }
     cls_name = ''.join(('test %s branches' % alias).title().split())
     globals()[cls_name] = type(cls_name, (BranchesTestCaseMixin,), attrs)

kallithea/tests/vcs/test_changesets.py

➞

Show inline comments

 # encoding: utf8
 import time
 import datetime
 import pytest
 from kallithea.lib import vcs
 from kallithea.lib.vcs.backends.base import BaseChangeset
 from kallithea.lib.vcs.nodes import (
     FileNode, AddedFileNodesGenerator,
     ChangedFileNodesGenerator, RemovedFileNodesGenerator
+)
 from kallithea.lib.vcs.exceptions import (
     BranchDoesNotExistError, ChangesetDoesNotExistError,
     RepositoryError, EmptyRepositoryError
+)
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import SCM_TESTS, get_new_dir
 class TestBaseChangeset(object):
     def test_as_dict(self):
         changeset = BaseChangeset()
         changeset.id = 'ID'
         changeset.raw_id = 'RAW_ID'
         changeset.short_id = 'SHORT_ID'
         changeset.revision = 1009
         changeset.date = datetime.datetime(2011, 1, 30, 1, 45)
         changeset.message = 'Message of a commit'
         changeset.author = 'Joe Doe <joe.doe@example.com>'
         changeset.added = [FileNode('foo/bar/baz'), FileNode(u'foobar'), FileNode(u'blåbærgrød')]
         changeset.changed = []
         changeset.removed = []
         assert changeset.as_dict() == {
             'id': 'ID',
             'raw_id': 'RAW_ID',
             'short_id': 'SHORT_ID',
             'revision': 1009,
             'date': datetime.datetime(2011, 1, 30, 1, 45),
             'message': 'Message of a commit',
             'author': {
                 'name': 'Joe Doe',
                 'email': 'joe.doe@example.com',
             },
             'added': ['foo/bar/baz', 'foobar', u'bl\xe5b\xe6rgr\xf8d'],
             'changed': [],
             'removed': [],
+        }
 class _ChangesetsWithCommitsTestCaseixin(_BackendTestMixin):
     recreate_repo_per_test = True
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': 'Commit %d' % x,
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
+            }
     def test_new_branch(self):
         self.imc.add(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
+        )
         assert 'foobar' in self.repo.branches
         assert foobar_tip.branch == 'foobar'
         assert foobar_tip.branches == ['foobar']
         # 'foobar' should be the only branch that contains the new commit
         branch_tips = self.repo.branches.values()
         assert branch_tips.count(str(foobar_tip.raw_id)) == 1
     def test_new_head_in_default_branch(self):
         tip = self.repo.get_changeset()
         self.imc.add(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\n'))
         foobar_tip = self.imc.commit(
             message=u'New branch: foobar',
             author=u'joe',
             branch='foobar',
             parents=[tip],
+        )
         self.imc.change(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\nand more...\n'))
         newtip = self.imc.commit(
             message=u'At default branch',
             author=u'joe',
             branch=foobar_tip.branch,
             parents=[foobar_tip],
+        )
         newest_tip = self.imc.commit(
             message=u'Merged with %s' % foobar_tip.raw_id,
             author=u'joe',
             branch=self.backend_class.DEFAULT_BRANCH_NAME,
             parents=[newtip, foobar_tip],
+        )
         assert newest_tip.branch == self.backend_class.DEFAULT_BRANCH_NAME
         assert newest_tip.branches == [self.backend_class.DEFAULT_BRANCH_NAME]
     def test_get_changesets_respects_branch_name(self):
         tip = self.repo.get_changeset()
         self.imc.add(vcs.nodes.FileNode('docs/index.txt',
             content='Documentation\n'))
         doc_changeset = self.imc.commit(
             message=u'New branch: docs',
             author=u'joe',
             branch='docs',
+        )
         self.imc.add(vcs.nodes.FileNode('newfile', content=''))
         self.imc.commit(
             message=u'Back in default branch',
             author=u'joe',
             parents=[tip],
+        )
         default_branch_changesets = self.repo.get_changesets(
             branch_name=self.repo.DEFAULT_BRANCH_NAME)
         assert doc_changeset not in default_branch_changesets
     def test_get_changeset_by_branch(self):
         for branch, sha in self.repo.branches.iteritems():
             assert sha == self.repo.get_changeset(branch).raw_id
     def test_get_changeset_by_tag(self):
         for tag, sha in self.repo.tags.iteritems():
             assert sha == self.repo.get_changeset(tag).raw_id
     def test_get_changeset_parents(self):
         for test_rev in [1, 2, 3]:
             sha = self.repo.get_changeset(test_rev-1)
             assert [sha] == self.repo.get_changeset(test_rev).parents
     def test_get_changeset_children(self):
         for test_rev in [1, 2, 3]:
             sha = self.repo.get_changeset(test_rev+1)
             assert [sha] == self.repo.get_changeset(test_rev).children
 class _ChangesetsTestCaseMixin(_BackendTestMixin):
     recreate_repo_per_test = False
     @classmethod
     def _get_commits(cls):
         start_date = datetime.datetime(2010, 1, 1, 20)
         for x in xrange(5):
             yield {
                 'message': u'Commit %d' % x,
                 'author': u'Joe Doe <joe.doe@example.com>',
                 'date': start_date + datetime.timedelta(hours=12 * x),
                 'added': [
                     FileNode('file_%d.txt' % x, content='Foobar %d' % x),
                 ],
+            }
     def test_simple(self):
         tip = self.repo.get_changeset()
         assert tip.date == datetime.datetime(2010, 1, 3, 20)
     def test_get_changesets_is_ordered_by_date(self):
         changesets = list(self.repo.get_changesets())
         ordered_by_date = sorted(changesets,
             key=lambda cs: cs.date)
         assert changesets == ordered_by_date
     def test_get_changesets_respects_start(self):
         second_id = self.repo.revisions[1]
         changesets = list(self.repo.get_changesets(start=second_id))
         assert len(changesets) == 4
     def test_get_changesets_numerical_id_respects_start(self):
         second_id = 1
         changesets = list(self.repo.get_changesets(start=second_id))
         assert len(changesets) == 4
     def test_get_changesets_includes_start_changeset(self):
         second_id = self.repo.revisions[1]
         changesets = list(self.repo.get_changesets(start=second_id))
         assert changesets[0].raw_id == second_id
     def test_get_changesets_respects_end(self):
         second_id = self.repo.revisions[1]
         changesets = list(self.repo.get_changesets(end=second_id))
         assert changesets[-1].raw_id == second_id
         assert len(changesets) == 2
     def test_get_changesets_numerical_id_respects_end(self):
         second_id = 1
         changesets = list(self.repo.get_changesets(end=second_id))
         assert changesets.index(changesets[-1]) == second_id
         assert len(changesets) == 2
     def test_get_changesets_respects_both_start_and_end(self):
         second_id = self.repo.revisions[1]
         third_id = self.repo.revisions[2]
         changesets = list(self.repo.get_changesets(start=second_id,
             end=third_id))
         assert len(changesets) == 2
     def test_get_changesets_numerical_id_respects_both_start_and_end(self):
         changesets = list(self.repo.get_changesets(start=2, end=3))
         assert len(changesets) == 2
     def test_get_changesets_on_empty_repo_raises_EmptyRepository_error(self):
         Backend = self.get_backend()
         repo_path = get_new_dir(str(time.time()))
         repo = Backend(repo_path, create=True)
         with pytest.raises(EmptyRepositoryError):
             list(repo.get_changesets(start='foobar'))
     def test_get_changesets_includes_end_changeset(self):
         second_id = self.repo.revisions[1]
         changesets = list(self.repo.get_changesets(end=second_id))
         assert changesets[-1].raw_id == second_id
     def test_get_changesets_respects_start_date(self):
         start_date = datetime.datetime(2010, 2, 1)
         for cs in self.repo.get_changesets(start_date=start_date):
             assert cs.date >= start_date
     def test_get_changesets_respects_end_date(self):
         start_date = datetime.datetime(2010, 1, 1)
         end_date = datetime.datetime(2010, 2, 1)
         for cs in self.repo.get_changesets(start_date=start_date,
                                            end_date=end_date):
             assert cs.date >= start_date
             assert cs.date <= end_date
     def test_get_changesets_respects_start_date_and_end_date(self):
         end_date = datetime.datetime(2010, 2, 1)
         for cs in self.repo.get_changesets(end_date=end_date):
             assert cs.date <= end_date
     def test_get_changesets_respects_reverse(self):
         changesets_id_list = [cs.raw_id for cs in
             self.repo.get_changesets(reverse=True)]
         assert changesets_id_list == list(reversed(self.repo.revisions))
     def test_get_filenodes_generator(self):
         tip = self.repo.get_changeset()
         filepaths = [node.path for node in tip.get_filenodes_generator()]
         assert filepaths == ['file_%d.txt' % x for x in xrange(5)]
     def test_size(self):
         tip = self.repo.get_changeset()
         size = 5 * len('Foobar N') # Size of 5 files
         assert tip.size == size
     def test_author(self):
         tip = self.repo.get_changeset()
         assert tip.author == u'Joe Doe <joe.doe@example.com>'
     def test_author_name(self):
         tip = self.repo.get_changeset()
         assert tip.author_name == u'Joe Doe'
     def test_author_email(self):
         tip = self.repo.get_changeset()
         assert tip.author_email == u'joe.doe@example.com'
     def test_get_changesets_raise_changesetdoesnotexist_for_wrong_start(self):
         with pytest.raises(ChangesetDoesNotExistError):
             list(self.repo.get_changesets(start='foobar'))
     def test_get_changesets_raise_changesetdoesnotexist_for_wrong_end(self):
         with pytest.raises(ChangesetDoesNotExistError):
             list(self.repo.get_changesets(end='foobar'))
     def test_get_changesets_raise_branchdoesnotexist_for_wrong_branch_name(self):
         with pytest.raises(BranchDoesNotExistError):
             list(self.repo.get_changesets(branch_name='foobar'))
     def test_get_changesets_raise_repositoryerror_for_wrong_start_end(self):
         start = self.repo.revisions[-1]
         end = self.repo.revisions[0]
         with pytest.raises(RepositoryError):
             list(self.repo.get_changesets(start=start, end=end))
     def test_get_changesets_numerical_id_reversed(self):
         with pytest.raises(RepositoryError):
             [x for x in self.repo.get_changesets(start=3, end=2)]
     def test_get_changesets_numerical_id_respects_both_start_and_end_last(self):
         with pytest.raises(RepositoryError):
             last = len(self.repo.revisions)
             list(self.repo.get_changesets(start=last-1, end=last-2))
     def test_get_changesets_numerical_id_last_zero_error(self):
         with pytest.raises(RepositoryError):
             last = len(self.repo.revisions)
             list(self.repo.get_changesets(start=last-1, end=0))
 class _ChangesetsChangesTestCaseMixin(_BackendTestMixin):
     recreate_repo_per_test = False
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': u'Initial',
                 'author': u'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foo/bar', content='foo'),
                     FileNode('foo/bał', content='foo'),
                     FileNode('foobar', content='foo'),
                     FileNode('qwe', content='foo'),
                 ],
             },
+            {
                 'message': u'Massive changes',
                 'author': u'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [FileNode('fallout', content='War never changes')],
                 'changed': [
                     FileNode('foo/bar', content='baz'),
                     FileNode('foobar', content='baz'),
                 ],
                 'removed': [FileNode('qwe')],
             },
+        ]
     def test_initial_commit(self):
         changeset = self.repo.get_changeset(0)
         assert sorted(list(changeset.added)) == sorted([
             changeset.get_node('foo/bar'),
             changeset.get_node('foo/bał'),
             changeset.get_node('foobar'),
             changeset.get_node('qwe'),
         ])
         assert list(changeset.changed) == []
         assert list(changeset.removed) == []
         assert u'foo/ba\u0142' in changeset.as_dict()['added']
         assert u'foo/ba\u0142' in changeset.__json__(with_file_list=True)['added']
     def test_head_added(self):
         changeset = self.repo.get_changeset()
         assert isinstance(changeset.added, AddedFileNodesGenerator)
         assert list(changeset.added) == [
             changeset.get_node('fallout'),
+        ]
         assert isinstance(changeset.changed, ChangedFileNodesGenerator)
         assert list(changeset.changed) == [
             changeset.get_node('foo/bar'),
             changeset.get_node('foobar'),
+        ]
         assert isinstance(changeset.removed, RemovedFileNodesGenerator)
         assert len(changeset.removed) == 1
         assert list(changeset.removed)[0].path == 'qwe'
     def test_get_filemode(self):
         changeset = self.repo.get_changeset()
         assert 33188 == changeset.get_file_mode('foo/bar')
     def test_get_filemode_non_ascii(self):
         changeset = self.repo.get_changeset()
         assert 33188 == changeset.get_file_mode('foo/bał')
         assert 33188 == changeset.get_file_mode(u'foo/bał')
 # For each backend create test case class
 for alias in SCM_TESTS:
     attrs = {
         'backend_alias': alias,
+    }
     # tests with additional commits
     cls_name = 'Test' + alias.title() + 'ChangesetsWithCommits'
     globals()[cls_name] = type(cls_name, (_ChangesetsWithCommitsTestCaseixin,), attrs)
     # tests without additional commits
     cls_name = 'Test' + alias.title() + 'Changesets'
     globals()[cls_name] = type(cls_name, (_ChangesetsTestCaseMixin,), attrs)
     # tests changes
     cls_name = 'Test' + alias.title() + 'ChangesetsChanges'
     globals()[cls_name] = type(cls_name, (_ChangesetsChangesTestCaseMixin,), attrs)

kallithea/tests/vcs/test_git.py

➞

Show inline comments

 import os
 import sys
 import mock
 import datetime
 import urllib2
 import pytest
 from kallithea.lib.vcs.backends.git import GitRepository, GitChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import NodeKind, FileNode, DirNode, NodeState
 from kallithea.model.scm import ScmModel
 from kallithea.tests.vcs.base import _BackendTestMixin
 from kallithea.tests.vcs.conf import TEST_GIT_REPO, TEST_GIT_REPO_CLONE, TESTS_TMP_PATH, get_new_dir
 class TestGitRepository(object):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_GIT_REPO_CLONE):
             pytest.fail('Cannot test git clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_GIT_REPO_CLONE)
     def setup_method(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_wrong_repo_path(self):
         wrong_repo_path = os.path.join(TESTS_TMP_PATH, 'errorrepo')
         with pytest.raises(RepositoryError):
             GitRepository(wrong_repo_path)
     def test_git_cmd_injection(self):
         repo_inject_path = TEST_GIT_REPO + '; echo "Cake";'
         with pytest.raises(urllib2.URLError):
             # Should fail because URL will contain the parts after ; too
             GitRepository(get_new_dir('injection-repo'), src_url=repo_inject_path, update_after_clone=True, create=True)
         with pytest.raises(RepositoryError):
             # Should fail on direct clone call, which as of this writing does not happen outside of class
             clone_fail_repo = GitRepository(get_new_dir('injection-repo'), create=True)
             clone_fail_repo.clone(repo_inject_path, update_after_clone=True,)
         # Verify correct quoting of evil characters that should work on posix file systems
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             # and some versions of the git client don't like ` and '
             tricky_path = get_new_dir("tricky-path-repo-$")
         else:
             tricky_path = get_new_dir("tricky-path-repo-$'\"`")
         successfully_cloned = GitRepository(tricky_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         assert not successfully_cloned._repo.bare
         if sys.platform == 'win32':
             # windows does not allow '"' in dir names
             # and some versions of the git client don't like ` and '
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$")
         else:
             tricky_path_2 = get_new_dir("tricky-path-2-repo-$'\"`")
         successfully_cloned2 = GitRepository(tricky_path_2, src_url=tricky_path, bare=True, create=True)
         # Repo should have been created and thus used correct quoting for clone
         assert successfully_cloned2._repo.bare
         # Should pass because URL has been properly quoted
         successfully_cloned.pull(tricky_path_2)
         successfully_cloned2.fetch(tricky_path)
     def test_repo_create_with_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         repo = GitRepository(repo_path, src_url=None, bare=True, create=True)
         # Repo should have been created
         assert repo._repo.bare
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = GitRepository(TEST_GIT_REPO)
         repo_clone = GitRepository(TEST_GIT_REPO_CLONE,
             src_url=TEST_GIT_REPO, create=True, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # Checking hashes of changesets should be enough
         for changeset in repo.get_changesets():
             raw_id = changeset.raw_id
             assert raw_id == repo_clone.get_changeset(raw_id).raw_id
     def test_repo_clone_with_spaces_in_path(self):
         repo_path = get_new_dir("path with spaces")
         successfully_cloned = GitRepository(repo_path, src_url=TEST_GIT_REPO, update_after_clone=True, create=True)
         # Repo should have been created
         assert not successfully_cloned._repo.bare
         successfully_cloned.pull(TEST_GIT_REPO)
         self.repo.fetch(repo_path)
     def test_repo_clone_without_create(self):
         with pytest.raises(RepositoryError):
             GitRepository(TEST_GIT_REPO_CLONE + '_wo_create', src_url=TEST_GIT_REPO)
     def test_repo_clone_with_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_with_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # check if current workdir was updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         assert os.path.isfile(fpath) == True, 'Repo was cloned and updated but file %s could not be found' % fpath
     def test_repo_clone_without_update(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_without_update'
         repo_clone = GitRepository(clone_path,
             create=True, src_url=TEST_GIT_REPO, update_after_clone=False)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # check if current workdir was *NOT* updated
         fpath = os.path.join(clone_path, 'MANIFEST.in')
         # Make sure it's not bare repo
         assert not repo_clone._repo.bare
         assert os.path.isfile(fpath) == False, 'Repo was cloned and updated but file %s was found' % fpath
     def test_repo_clone_into_bare_repo(self):
         repo = GitRepository(TEST_GIT_REPO)
         clone_path = TEST_GIT_REPO_CLONE + '_bare.git'
         repo_clone = GitRepository(clone_path, create=True,
             src_url=repo.path, bare=True)
         assert repo_clone._repo.bare
     def test_create_repo_is_not_bare_by_default(self):
         repo = GitRepository(get_new_dir('not-bare-by-default'), create=True)
         assert not repo._repo.bare
     def test_create_bare_repo(self):
         repo = GitRepository(get_new_dir('bare-repo'), create=True, bare=True)
         assert repo._repo.bare
     def test_revisions(self):
         # there are 112 revisions (by now)
         # so we can assume they would be available from now on
         subset = set([
             'c1214f7e79e02fc37156ff215cd71275450cffc3',
             '38b5fe81f109cb111f549bfe9bb6b267e10bc557',
             'fa6600f6848800641328adbf7811fd2372c02ab2',
             '102607b09cdd60e2793929c4f90478be29f85a17',
             '49d3fd156b6f7db46313fac355dca1a0b94a0017',
             '2d1028c054665b962fa3d307adfc923ddd528038',
             'd7e0d30fbcae12c90680eb095a4f5f02505ce501',
             'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
             '8430a588b43b5d6da365400117c89400326e7992',
             'd955cd312c17b02143c04fa1099a352b04368118',
             'f67b87e5c629c2ee0ba58f85197e423ff28d735b',
             'add63e382e4aabc9e1afdc4bdc24506c269b7618',
             'f298fe1189f1b69779a4423f40b48edf92a703fc',
             'bd9b619eb41994cac43d67cf4ccc8399c1125808',
             '6e125e7c890379446e98980d8ed60fba87d0f6d1',
             'd4a54db9f745dfeba6933bf5b1e79e15d0af20bd',
             '0b05e4ed56c802098dfc813cbe779b2f49e92500',
             '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
             '45223f8f114c64bf4d6f853e3c35a369a6305520',
             'ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
             'f5ea29fc42ef67a2a5a7aecff10e1566699acd68',
             '27d48942240f5b91dfda77accd2caac94708cc7d',
             '622f0eb0bafd619d2560c26f80f09e3b0b0d78af',
             'e686b958768ee96af8029fe19c6050b1a8dd3b2b'])
         assert subset.issubset(set(self.repo.revisions))
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             revs = list(self.repo[sfrom:sto])
             assert len(revs) == size
             assert revs[0] == self.repo.get_changeset(sfrom)
             assert revs[-1] == self.repo.get_changeset(sto - 1)
     def test_branches(self):
         # TODO: Need more tests here
         # Removed (those are 'remotes' branches for cloned repo)
         #assert 'master' in self.repo.branches
         #assert 'gittree' in self.repo.branches
         #assert 'web-branch' in self.repo.branches
         for name, id in self.repo.branches.items():
             assert isinstance(self.repo.get_changeset(id), GitChangeset)
     def test_tags(self):
         # TODO: Need more tests here
         assert 'v0.1.1' in self.repo.tags
         assert 'v0.1.2' in self.repo.tags
         for name, id in self.repo.tags.items():
             assert isinstance(self.repo.get_changeset(id), GitChangeset)
     def _test_single_changeset_cache(self, revision):
         chset = self.repo.get_changeset(revision)
         assert revision in self.repo.changesets
         assert chset is self.repo.changesets[revision]
     def test_initial_changeset(self):
         id = self.repo.revisions[0]
         init_chset = self.repo.get_changeset(id)
         assert init_chset.message == 'initial import\n'
         assert init_chset.author == 'Marcin Kuzminski <marcin@python-blog.com>'
         for path in ('vcs/__init__.py',
                      'vcs/backends/BaseRepository.py',
                      'vcs/backends/__init__.py'):
             assert isinstance(init_chset.get_node(path), FileNode)
         for path in ('', 'vcs', 'vcs/backends'):
             assert isinstance(init_chset.get_node(path), DirNode)
         with pytest.raises(NodeDoesNotExistError):
             init_chset.get_node(path='foobar')
         node = init_chset.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
     def test_not_existing_changeset(self):
         with pytest.raises(RepositoryError):
             self.repo.get_changeset('f' * 40)
     def test_changeset10(self):
         chset10 = self.repo.get_changeset(self.repo.revisions[9])
         readme = """===
 VCS
 ===
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.content == readme
 class TestGitChangeset(object):
     def setup_method(self):
         self.repo = GitRepository(TEST_GIT_REPO)
     def test_default_changeset(self):
         tip = self.repo.get_changeset()
         assert tip == self.repo.get_changeset(None)
         assert tip == self.repo.get_changeset('tip')
     def test_root_node(self):
         tip = self.repo.get_changeset()
         assert tip.root is tip.get_node('')
     def test_lazy_fetch(self):
         """
         Test if changeset's nodes expands and are cached as we walk through
         the revision. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         assert commit_id in self.repo.revisions
         chset = self.repo.get_changeset(commit_id)
         assert len(chset.nodes) == 0
         root = chset.root
         assert len(chset.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates chset.nodes
         assert len(chset.nodes) == 9
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(chset.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates chset.nodes
         assert len(chset.nodes) == 17
         assert docs is chset.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is chset.get_node('docs')
     def test_nodes_with_changeset(self):
         commit_id = '2a13f185e4525f9d4b59882791a2d397b90d5ddc'
         chset = self.repo.get_changeset(commit_id)
         root = chset.root
         docs = root.get_node('docs')
         assert docs is chset.get_node('docs')
         api = docs.get_node('api')
         assert api is chset.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is chset.get_node('docs/api/index.rst')
         assert index is chset.get_node('docs') \
                              .get_node('api') \
                              .get_node('index.rst')
     def test_branch_and_tags(self):
         # Those tests seem to show wrong results:
         # in Git, only heads have a branch - most changesets don't
         rev0 = self.repo.revisions[0]
         chset0 = self.repo.get_changeset(rev0)
         assert chset0.branch is None # should be 'master'?
         assert chset0.branches == [] # should be 'master'?
         assert chset0.tags == []
         rev10 = self.repo.revisions[10]
         chset10 = self.repo.get_changeset(rev10)
         assert chset10.branch is None # should be 'master'?
         assert chset10.branches == [] # should be 'master'?
         assert chset10.tags == []
         rev44 = self.repo.revisions[44]
         chset44 = self.repo.get_changeset(rev44)
         assert chset44.branch is None # should be 'web-branch'?
         assert chset44.branches == [] # should be 'web-branch'?
         tip = self.repo.get_changeset('tip')
         assert 'tip' not in tip.tags # it should be?
         assert not tip.tags # how it is!
     def _test_slices(self, limit, offset):
         count = self.repo.count()
         changesets = self.repo.get_changesets(limit=limit, offset=offset)
         idx = 0
         for changeset in changesets:
             rev = offset + idx
             idx += 1
             rev_id = self.repo.revisions[rev]
             if idx > limit:
                 pytest.fail("Exceeded limit already (getting revision %s, "
                     "there are %s total revisions, offset=%s, limit=%s)"
                     % (rev_id, count, offset, limit))
             assert changeset == self.repo.get_changeset(rev_id)
         result = list(self.repo.get_changesets(limit=limit, offset=offset))
         start = offset
         end = limit and offset + limit or None
         sliced = list(self.repo[start:end])
         pytest.failUnlessEqual(result, sliced,
             msg="Comparison failed for limit=%s, offset=%s"
             "(get_changeset returned: %s and sliced: %s"
             % (limit, offset, result, sliced))
     def _test_file_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert node.is_file()
         assert node.size == size
     def test_file_size(self):
         to_check = (
             ('c1214f7e79e02fc37156ff215cd71275450cffc3',
                 'vcs/backends/BaseRepository.py', 502),
             ('d7e0d30fbcae12c90680eb095a4f5f02505ce501',
                 'vcs/backends/hg.py', 854),
             ('6e125e7c890379446e98980d8ed60fba87d0f6d1',
                 'setup.py', 1068),
             ('d955cd312c17b02143c04fa1099a352b04368118',
                 'vcs/backends/base.py', 2921),
             ('ca1eb7957a54bce53b12d1a51b13452f95bc7c7e',
                 'vcs/backends/base.py', 3936),
             ('f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def _test_dir_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert node.size == size
     def test_dir_size(self):
         to_check = (
             ('5f2c6ee195929b0be80749243c18121c9864a3b3', '/', 674076),
             ('7ab37bc680b4aa72c34d07b230c866c28e9fc204', '/', 674049),
             ('6892503fb8f2a552cef5f4d4cc2cdbd13ae1cd2f', '/', 671830),
+        )
         for revision, path, size in to_check:
             self._test_dir_size(revision, path, size)
     def test_repo_size(self):
         assert self.repo.size == 674076
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [
                 '54386793436c938cff89326944d4c2702340037d',
                 '51d254f0ecf5df2ce50c0b115741f4cf13985dab',
                 '998ed409c795fec2012b1c0ca054d99888b22090',
                 '5e0eb4c47f56564395f76333f319d26c79e2fb09',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '7cb3fd1b6d8c20ba89e2264f1c8baebc8a52d36e',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '191caa5b2c81ed17c0794bf7bb9958f4dcb0b87e',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/nodes.py': [
                 '33fa3223355104431402a888fa77a4e9956feb3e',
                 'fa014c12c26d10ba682fadb78f2a11c24c8118e1',
                 'e686b958768ee96af8029fe19c6050b1a8dd3b2b',
                 'ab5721ca0a081f26bf43d9051e615af2cc99952f',
                 'c877b68d18e792a66b7f4c529ea02c8f80801542',
                 '4313566d2e417cb382948f8d9d7c765330356054',
                 '6c2303a793671e807d1cfc70134c9ca0767d98c2',
                 '54386793436c938cff89326944d4c2702340037d',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
                 '1c6b3677b37ea064cb4b51714d8f7498f93f4b2b',
                 '2d03ca750a44440fb5ea8b751176d1f36f8e8f46',
                 '2a08b128c206db48c2f0b8f70df060e6db0ae4f8',
                 '30c26513ff1eb8e5ce0e1c6b477ee5dc50e2f34b',
                 'ac71e9503c2ca95542839af0ce7b64011b72ea7c',
                 '12669288fd13adba2a9b7dd5b870cc23ffab92d2',
                 '5a0c84f3e6fe3473e4c8427199d5a6fc71a9b382',
                 '12f2f5e2b38e6ff3fbdb5d722efed9aa72ecb0d5',
                 '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                 'f50f42baeed5af6518ef4b0cb2f1423f3851a941',
                 'd7e390a45f6aa96f04f5e7f583ad4f867431aa25',
                 'f15c21f97864b4f071cddfbf2750ec2e23859414',
                 'e906ef056cf539a4e4e5fc8003eaf7cf14dd8ade',
                 'ea2b108b48aa8f8c9c4a941f66c1a03315ca1c3b',
                 '84dec09632a4458f79f50ddbbd155506c460b4f9',
                 '0115510b70c7229dbc5dc49036b32e7d91d23acd',
                 '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                 '3bf1c5868e570e39569d094f922d33ced2fa3b2b',
                 'b8d04012574729d2c29886e53b1a43ef16dd00a1',
                 '6970b057cffe4aab0a792aa634c89f4bebf01441',
                 'dd80b0f6cf5052f17cc738c2951c4f2070200d7f',
                 'ff7ca51e58c505fec0dd2491de52c622bb7a806b',
             ],
             'vcs/backends/git.py': [
                 '4cf116ad5a457530381135e2f4c453e68a1b0105',
                 '9a751d84d8e9408e736329767387f41b36935153',
                 'cb681fb539c3faaedbcdf5ca71ca413425c18f01',
                 '428f81bb652bcba8d631bce926e8834ff49bdcc6',
                 '180ab15aebf26f98f714d8c68715e0f05fa6e1c7',
                 '2b8e07312a2e89e92b90426ab97f349f4bce2a3a',
                 '50e08c506174d8645a4bb517dd122ac946a0f3bf',
                 '54000345d2e78b03a99d561399e8e548de3f3203',
             ],
+        }
         for path, revs in files.items():
             node = self.repo.get_changeset(revs[0]).get_node(path)
             node_revs = [chset.raw_id for chset in node.history]
             assert set(revs).issubset(set(node_revs)), "We assumed that %s is subset of revisions for which file %s " \
                 "has been changed, and history of that node returned: %s" \
                 % (revs, path, node_revs)
     def test_file_annotate(self):
         files = {
             'vcs/backends/__init__.py': {
                 'c1214f7e79e02fc37156ff215cd71275450cffc3': {
                     'lines_no': 1,
                     'changesets': [
                         'c1214f7e79e02fc37156ff215cd71275450cffc3',
                     ],
                 },
                 '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647': {
                     'lines_no': 21,
                     'changesets': [
                         '49d3fd156b6f7db46313fac355dca1a0b94a0017',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
                 'e29b67bd158580fc90fc5e9111240b90e6e86064': {
                     'lines_no': 32,
                     'changesets': [
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '5eab1222a7cd4bfcbabc218ca6d04276d4e27378',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '54000345d2e78b03a99d561399e8e548de3f3203',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '2a13f185e4525f9d4b59882791a2d397b90d5ddc',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '78c3f0c23b7ee935ec276acb8b8212444c33c396',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '992f38217b979d0b0987d0bae3cc26dac85d9b19',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                         '16fba1ae9334d79b66d7afed2c2dfbfa2ae53647',
                     ],
                 },
             },
+        }
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_1
                 l2 = files[fname][rev]['changesets']
                 assert l1 == l2, "The lists of revision for %s@rev %s" \
                     "from annotation list should match each other, " \
                     "got \n%s \nvs \n%s " % (fname, rev, l1, l2)
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         node = self.repo \
             .get_changeset('e6ea6d16e2f26250124a1f4b4fe37a912f9d86a0') \
             .get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('33fa3223355104431402a888fa77a4e9956feb3e') \
             .get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
         node = self.repo \
             .get_changeset('e29b67bd158580fc90fc5e9111240b90e6e86064') \
             .get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(
             'fa6600f6848800641328adbf7811fd2372c02ab2')
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             chset.get_node(path)
         # but it would be one of ``removed`` (changeset's attribute)
         assert path in [rf.path for rf in chset.removed]
         chset = self.repo.get_changeset(
             '54386793436c938cff89326944d4c2702340037d')
         changed = ['setup.py', 'tests/test_nodes.py', 'vcs/backends/hg.py',
             'vcs/nodes.py']
         assert set(changed) == set([f.path for f in chset.changed])
     def test_commit_message_is_unicode(self):
         for cs in self.repo:
             assert type(cs.message) == unicode
     def test_changeset_author_is_unicode(self):
         for cs in self.repo:
             assert type(cs.author) == unicode
     def test_repo_files_content_is_unicode(self):
         changeset = self.repo.get_changeset()
         for node in changeset.get_node('/'):
             if node.is_file():
                 assert type(node.content) == unicode
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         tip = self.repo.get_changeset()
         with pytest.raises(VCSError):
             tip.get_node(path)
     def test_author_email(self):
         assert 'marcin@python-blog.com' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_email
         assert 'lukasz.balcerzak@python-center.pl' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_email
         assert '' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_email
     def test_author_username(self):
         assert 'Marcin Kuzminski' == self.repo.get_changeset('c1214f7e79e02fc37156ff215cd71275450cffc3').author_name
         assert 'Lukasz Balcerzak' == self.repo.get_changeset('ff7ca51e58c505fec0dd2491de52c622bb7a806b').author_name
         assert 'marcink none@none' == self.repo.get_changeset('8430a588b43b5d6da365400117c89400326e7992').author_name
 class TestGitSpecific():
     def test_error_is_raised_for_added_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with pytest.raises(VCSError):
             changeset.added
     def test_error_is_raised_for_changed_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with pytest.raises(VCSError):
             changeset.added
     def test_error_is_raised_for_removed_if_diff_name_status_is_wrong(self):
         repo = mock.MagicMock()
         changeset = GitChangeset(repo, 'foobar')
         changeset._diff_name_status = 'foobar'
         with pytest.raises(VCSError):
             changeset.added
 class TestGitSpecificWithRepo(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('foobar/static/js/admin/base.js', content='base'),
                     FileNode('foobar/static/admin', content='admin',
                         mode=0120000), # this is a link
                     FileNode('foo', content='foo'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('foo2', content='foo2'),
                 ],
             },
+        ]
     def test_paths_slow_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar').get_node('static').get_node('js').get_node('admin').get_node('base.js').content == 'base'
     def test_paths_fast_traversing(self):
         cs = self.repo.get_changeset()
         assert cs.get_node('foobar/static/js/admin/base.js').content == 'base'
     def test_workdir_get_branch(self):
         self.repo.run_git_command(['checkout', '-b', 'production'])
         # Regression test: one of following would fail if we don't check
         # .git/HEAD file
         self.repo.run_git_command(['checkout', 'production'])
         assert self.repo.workdir.get_branch() == 'production'
         self.repo.run_git_command(['checkout', 'master'])
         assert self.repo.workdir.get_branch() == 'master'
     def test_get_diff_runs_git_command_with_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1)])
     def test_get_diff_runs_git_command_with_str_hashes(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(self.repo.EMPTY_CHANGESET, 1)
         self.repo.run_git_command.assert_called_once_with(
             ['show', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(1)])
     def test_get_diff_runs_git_command_with_path_if_its_given(self):
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo')
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U3', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
     def test_get_diff_does_not_sanitize_valid_context(self):
         almost_overflowed_long_int = 2**31-1
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo', context=almost_overflowed_long_int)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U' + str(almost_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
     def test_get_diff_sanitizes_overflowing_context(self):
         overflowed_long_int = 2**31
         sanitized_overflowed_long_int = overflowed_long_int-1
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo', context=overflowed_long_int)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U' + str(sanitized_overflowed_long_int), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
     def test_get_diff_does_not_sanitize_zero_context(self):
         zero_context = 0
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo', context=zero_context)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U' + str(zero_context), '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
     def test_get_diff_sanitizes_negative_context(self):
         negative_context = -10
         self.repo.run_git_command = mock.Mock(return_value=['', ''])
         self.repo.get_diff(0, 1, 'foo', context=negative_context)
         self.repo.run_git_command.assert_called_once_with(
             ['diff', '-U0', '--full-index', '--binary', '-p', '-M', '--abbrev=40',
              self.repo._get_revision(0), self.repo._get_revision(1), '--', 'foo'])
 class TestGitRegression(_BackendTestMixin):
     backend_alias = 'git'
     @classmethod
     def _get_commits(cls):
         return [
+            {
                 'message': 'Initial',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 20),
                 'added': [
                     FileNode('bot/__init__.py', content='base'),
                     FileNode('bot/templates/404.html', content='base'),
                     FileNode('bot/templates/500.html', content='base'),
                 ],
             },
+            {
                 'message': 'Second',
                 'author': 'Joe Doe <joe.doe@example.com>',
                 'date': datetime.datetime(2010, 1, 1, 22),
                 'added': [
                     FileNode('bot/build/migrations/1.py', content='foo2'),
                     FileNode('bot/build/migrations/2.py', content='foo2'),
                     FileNode('bot/build/static/templates/f.html', content='foo2'),
                     FileNode('bot/build/static/templates/f1.html', content='foo2'),
                     FileNode('bot/build/templates/err.html', content='foo2'),
                     FileNode('bot/build/templates/err2.html', content='foo2'),
                 ],
             },
+        ]
     def test_similar_paths(self):
         cs = self.repo.get_changeset()
         paths = lambda *n: [x.path for x in n]
         assert paths(*cs.get_nodes('bot')) == ['bot/build', 'bot/templates', 'bot/__init__.py']
         assert paths(*cs.get_nodes('bot/build')) == ['bot/build/migrations', 'bot/build/static', 'bot/build/templates']
         assert paths(*cs.get_nodes('bot/build/static')) == ['bot/build/static/templates']
         # this get_nodes below causes troubles !
         assert paths(*cs.get_nodes('bot/build/static/templates')) == ['bot/build/static/templates/f.html', 'bot/build/static/templates/f1.html']
         assert paths(*cs.get_nodes('bot/build/templates')) == ['bot/build/templates/err.html', 'bot/build/templates/err2.html']
         assert paths(*cs.get_nodes('bot/templates/')) == ['bot/templates/404.html', 'bot/templates/500.html']
 class TestGitHooks(object):
     """
     Tests related to hook functionality of Git repositories.
     """
     def setup_method(self):
         # For each run we want a fresh repo.
         self.repo_directory = get_new_dir("githookrepo")
         self.repo = GitRepository(self.repo_directory, create=True)
         # Create a dictionary where keys are hook names, and values are paths to
         # them. Deduplicates code in tests a bit.
         self.hook_directory = self.repo.get_hook_location()
         self.kallithea_hooks = dict((h, os.path.join(self.hook_directory, h)) for h in ("pre-receive", "post-receive"))
     def test_hooks_created_if_missing(self):
         """
         Tests if hooks are installed in repository if they are missing.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             if os.path.exists(hook_path):
                 os.remove(hook_path)
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             assert os.path.exists(hook_path)
     def test_kallithea_hooks_updated(self):
         """
         Tests if hooks are updated if they are Kallithea hooks already.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("KALLITHEA_HOOK_VER=0.0.0\nJUST_BOGUS")
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 assert "JUST_BOGUS" not in f.read()
     def test_custom_hooks_untouched(self):
         """
         Tests if hooks are left untouched if they are not Kallithea hooks.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(repo=self.repo)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 assert "CUSTOM_HOOK" in f.read()
     def test_custom_hooks_forced_update(self):
         """
         Tests if hooks are forcefully updated even though they are custom hooks.
         """
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path, "w") as f:
                 f.write("#!/bin/bash\n#CUSTOM_HOOK")
         ScmModel().install_git_hooks(repo=self.repo, force_create=True)
         for hook, hook_path in self.kallithea_hooks.iteritems():
             with open(hook_path) as f:
                 assert "KALLITHEA_HOOK_VER" in f.read()

kallithea/tests/vcs/test_hg.py

➞

Show inline comments

 import os
 import pytest
 import mock
 from kallithea.lib.utils2 import safe_str
 from kallithea.lib.vcs.backends.hg import MercurialRepository, MercurialChangeset
 from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError
 from kallithea.lib.vcs.nodes import NodeKind, NodeState
 from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, \
     TEST_HG_REPO_PULL, TESTS_TMP_PATH
 class TestMercurialRepository(object):
     def __check_for_existing_repo(self):
         if os.path.exists(TEST_HG_REPO_CLONE):
             pytest.fail('Cannot test mercurial clone repo as location %s already '
                       'exists. You should manually remove it first.'
                       % TEST_HG_REPO_CLONE)
     def setup_method(self):
         self.repo = MercurialRepository(safe_str(TEST_HG_REPO))
     def test_wrong_repo_path(self):
         wrong_repo_path = os.path.join(TESTS_TMP_PATH, 'errorrepo')
         with pytest.raises(RepositoryError):
             MercurialRepository(wrong_repo_path)
     def test_unicode_path_repo(self):
         with pytest.raises(VCSError):
             MercurialRepository(u'iShouldFail')
     def test_repo_clone(self):
         self.__check_for_existing_repo()
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE,
             src_url=TEST_HG_REPO, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # Checking hashes of changesets should be enough
         for changeset in repo.get_changesets():
             raw_id = changeset.raw_id
             assert raw_id == repo_clone.get_changeset(raw_id).raw_id
     def test_repo_clone_with_update(self):
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_w_update',
             src_url=TEST_HG_REPO, update_after_clone=True)
         assert len(repo.revisions) == len(repo_clone.revisions)
         # check if current workdir was updated
         assert os.path.isfile(
             os.path.join(
                 TEST_HG_REPO_CLONE + '_w_update', 'MANIFEST.in'
+            )
+        )
     def test_repo_clone_without_update(self):
         repo = MercurialRepository(safe_str(TEST_HG_REPO))
         repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_wo_update',
             src_url=TEST_HG_REPO, update_after_clone=False)
         assert len(repo.revisions) == len(repo_clone.revisions)
         assert not os.path.isfile(
             os.path.join(
                 TEST_HG_REPO_CLONE + '_wo_update', 'MANIFEST.in'
+            )
+        )
     def test_pull(self):
         if os.path.exists(TEST_HG_REPO_PULL):
             pytest.fail('Cannot test mercurial pull command as location %s '
                       'already exists. You should manually remove it first'
                       % TEST_HG_REPO_PULL)
         repo_new = MercurialRepository(TEST_HG_REPO_PULL, create=True)
         assert len(self.repo.revisions) > len(repo_new.revisions)
         repo_new.pull(self.repo.path)
         repo_new = MercurialRepository(TEST_HG_REPO_PULL)
         assert len(self.repo.revisions) == len(repo_new.revisions)
     def test_revisions(self):
         # there are 21 revisions at bitbucket now
         # so we can assume they would be available from now on
         subset = set(['b986218ba1c9b0d6a259fac9b050b1724ed8e545',
                  '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
                  '6cba7170863a2411822803fa77a0a264f1310b35',
                  '56349e29c2af3ac913b28bde9a2c6154436e615b',
                  '2dda4e345facb0ccff1a191052dd1606dba6781d',
                  '6fff84722075f1607a30f436523403845f84cd9e',
                  '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',
                  '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
                  'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',
                  'be90031137367893f1c406e0a8683010fd115b79',
                  'db8e58be770518cbb2b1cdfa69146e47cd481481',
                  '84478366594b424af694a6c784cb991a16b87c21',
                  '17f8e105dddb9f339600389c6dc7175d395a535c',
                  '20a662e756499bde3095ffc9bc0643d1def2d0eb',
                  '2e319b85e70a707bba0beff866d9f9de032aa4f9',
                  '786facd2c61deb9cf91e9534735124fb8fc11842',
                  '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',
                  'aa6a0de05b7612707db567078e130a6cd114a9a7',
                  'eada5a770da98ab0dd7325e29d00e0714f228d09'
                 ])
         assert subset.issubset(set(self.repo.revisions))
         # check if we have the proper order of revisions
         org = ['b986218ba1c9b0d6a259fac9b050b1724ed8e545',
                 '3d8f361e72ab303da48d799ff1ac40d5ac37c67e',
                 '6cba7170863a2411822803fa77a0a264f1310b35',
                 '56349e29c2af3ac913b28bde9a2c6154436e615b',
                 '2dda4e345facb0ccff1a191052dd1606dba6781d',
                 '6fff84722075f1607a30f436523403845f84cd9e',
                 '7d4bc8ec6be56c0f10425afb40b6fc315a4c25e7',
                 '3803844fdbd3b711175fc3da9bdacfcd6d29a6fb',
                 'dc5d2c0661b61928834a785d3e64a3f80d3aad9c',
                 'be90031137367893f1c406e0a8683010fd115b79',
                 'db8e58be770518cbb2b1cdfa69146e47cd481481',
                 '84478366594b424af694a6c784cb991a16b87c21',
                 '17f8e105dddb9f339600389c6dc7175d395a535c',
                 '20a662e756499bde3095ffc9bc0643d1def2d0eb',
                 '2e319b85e70a707bba0beff866d9f9de032aa4f9',
                 '786facd2c61deb9cf91e9534735124fb8fc11842',
                 '94593d2128d38210a2fcd1aabff6dda0d6d9edf8',
                 'aa6a0de05b7612707db567078e130a6cd114a9a7',
                 'eada5a770da98ab0dd7325e29d00e0714f228d09',
                 '2c1885c735575ca478bf9e17b0029dca68824458',
                 'd9bcd465040bf869799b09ad732c04e0eea99fe9',
                 '469e9c847fe1f6f7a697b8b25b4bc5b48780c1a7',
                 '4fb8326d78e5120da2c7468dcf7098997be385da',
                 '62b4a097164940bd66030c4db51687f3ec035eed',
                 '536c1a19428381cfea92ac44985304f6a8049569',
                 '965e8ab3c44b070cdaa5bf727ddef0ada980ecc4',
                 '9bb326a04ae5d98d437dece54be04f830cf1edd9',
                 'f8940bcb890a98c4702319fbe36db75ea309b475',
                 'ff5ab059786ebc7411e559a2cc309dfae3625a3b',
                 '6b6ad5f82ad5bb6190037671bd254bd4e1f4bf08',
                 'ee87846a61c12153b51543bf860e1026c6d3dcba', ]
         assert org == self.repo.revisions[:31]
     def test_iter_slice(self):
         sliced = list(self.repo[:10])
         itered = list(self.repo)[:10]
         assert sliced == itered
     def test_slicing(self):
         # 4 1 5 10 95
         for sfrom, sto, size in [(0, 4, 4), (1, 2, 1), (10, 15, 5),
                                  (10, 20, 10), (5, 100, 95)]:
             revs = list(self.repo[sfrom:sto])
             assert len(revs) == size
             assert revs[0] == self.repo.get_changeset(sfrom)
             assert revs[-1] == self.repo.get_changeset(sto - 1)
     def test_branches(self):
         # TODO: Need more tests here
         # active branches
         assert 'default' in self.repo.branches
         assert 'stable' in self.repo.branches
         # closed
         assert 'git' in self.repo._get_branches(closed=True)
         assert 'web' in self.repo._get_branches(closed=True)
         for name, id in self.repo.branches.items():
             assert isinstance(self.repo.get_changeset(id), MercurialChangeset)
     def test_tip_in_tags(self):
         # tip is always a tag
         assert 'tip' in self.repo.tags
     def test_tip_changeset_in_tags(self):
         tip = self.repo.get_changeset()
         assert self.repo.tags['tip'] == tip.raw_id
     def test_initial_changeset(self):
         init_chset = self.repo.get_changeset(0)
         assert init_chset.message == 'initial import'
         assert init_chset.author == 'Marcin Kuzminski <marcin@python-blog.com>'
         assert sorted(init_chset._file_paths) == sorted([
             'vcs/__init__.py',
             'vcs/backends/BaseRepository.py',
             'vcs/backends/__init__.py',
         ])
         assert sorted(init_chset._dir_paths) == sorted(['', 'vcs', 'vcs/backends'])
         with pytest.raises(NodeDoesNotExistError):
             init_chset.get_node(path='foobar')
         node = init_chset.get_node('vcs/')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.DIR
         node = init_chset.get_node('vcs/__init__.py')
         assert hasattr(node, 'kind')
         assert node.kind == NodeKind.FILE
     def test_not_existing_changeset(self):
         # rawid
         with pytest.raises(RepositoryError):
             self.repo.get_changeset('abcd' * 10)
         # shortid
         with pytest.raises(RepositoryError):
             self.repo.get_changeset('erro' * 4)
         # numeric
         with pytest.raises(RepositoryError):
             self.repo.get_changeset(self.repo.count() + 1)
         # Small chance we ever get to this one
         revision = pow(2, 30)
         with pytest.raises(RepositoryError):
             self.repo.get_changeset(revision)
     def test_changeset10(self):
         chset10 = self.repo.get_changeset(10)
         readme = """===
 VCS
 ===
 Various Version Control System management abstraction layer for Python.
 Introduction
 ------------
 TODO: To be written...
 """
         node = chset10.get_node('README.rst')
         assert node.kind == NodeKind.FILE
         assert node.content == readme
     @mock.patch('kallithea.lib.vcs.backends.hg.repository.diffopts')
     def test_get_diff_does_not_sanitize_zero_context(self, mock_diffopts):
         zero_context = 0
         self.repo.get_diff(0, 1, 'foo', context=zero_context)
         mock_diffopts.assert_called_once_with(git=True, showfunc=True, ignorews=False, context=zero_context)
     @mock.patch('kallithea.lib.vcs.backends.hg.repository.diffopts')
     def test_get_diff_sanitizes_negative_context(self, mock_diffopts):
         negative_context = -10
         zero_context = 0
         self.repo.get_diff(0, 1, 'foo', context=negative_context)
         mock_diffopts.assert_called_once_with(git=True, showfunc=True, ignorews=False, context=zero_context)
 class TestMercurialChangeset(object):
     def setup_method(self):
         self.repo = MercurialRepository(safe_str(TEST_HG_REPO))
     def _test_equality(self, changeset):
         revision = changeset.revision
         assert changeset == self.repo.get_changeset(revision)
     def test_equality(self):
         revs = [0, 10, 20]
         changesets = [self.repo.get_changeset(rev) for rev in revs]
         for changeset in changesets:
             self._test_equality(changeset)
     def test_default_changeset(self):
         tip = self.repo.get_changeset('tip')
         assert tip == self.repo.get_changeset()
         assert tip == self.repo.get_changeset(revision=None)
         assert tip == list(self.repo[-1:])[0]
     def test_root_node(self):
         tip = self.repo.get_changeset('tip')
         assert tip.root is tip.get_node('')
     def test_lazy_fetch(self):
         """
         Test if changeset's nodes expands and are cached as we walk through
         the revision. This test is somewhat hard to write as order of tests
         is a key here. Written by running command after command in a shell.
         """
         chset = self.repo.get_changeset(45)
         assert len(chset.nodes) == 0
         root = chset.root
         assert len(chset.nodes) == 1
         assert len(root.nodes) == 8
         # accessing root.nodes updates chset.nodes
         assert len(chset.nodes) == 9
         docs = root.get_node('docs')
         # we haven't yet accessed anything new as docs dir was already cached
         assert len(chset.nodes) == 9
         assert len(docs.nodes) == 8
         # accessing docs.nodes updates chset.nodes
         assert len(chset.nodes) == 17
         assert docs is chset.get_node('docs')
         assert docs is root.nodes[0]
         assert docs is root.dirs[0]
         assert docs is chset.get_node('docs')
     def test_nodes_with_changeset(self):
         chset = self.repo.get_changeset(45)
         root = chset.root
         docs = root.get_node('docs')
         assert docs is chset.get_node('docs')
         api = docs.get_node('api')
         assert api is chset.get_node('docs/api')
         index = api.get_node('index.rst')
         assert index is chset.get_node('docs/api/index.rst')
         assert index is chset.get_node('docs').get_node('api').get_node('index.rst')
     def test_branch_and_tags(self):
         chset0 = self.repo.get_changeset(0)
         assert chset0.branch == 'default'
         assert chset0.branches == ['default']
         assert chset0.tags == []
         chset10 = self.repo.get_changeset(10)
         assert chset10.branch == 'default'
         assert chset10.branches == ['default']
         assert chset10.tags == []
         chset44 = self.repo.get_changeset(44)
         assert chset44.branch == 'web'
         assert chset44.branches == ['web']
         tip = self.repo.get_changeset('tip')
         assert 'tip' in tip.tags
     def _test_file_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert node.is_file()
         assert node.size == size
     def test_file_size(self):
         to_check = (
             (10, 'setup.py', 1068),
             (20, 'setup.py', 1106),
             (60, 'setup.py', 1074),
             (10, 'vcs/backends/base.py', 2921),
             (20, 'vcs/backends/base.py', 3936),
             (60, 'vcs/backends/base.py', 6189),
+        )
         for revision, path, size in to_check:
             self._test_file_size(revision, path, size)
     def _test_dir_size(self, revision, path, size):
         node = self.repo.get_changeset(revision).get_node(path)
         assert not node.is_file()
         assert node.size == size
     def test_dir_size(self):
         to_check = (
             ('96507bd11ecc', '/', 682421),
             ('a53d9201d4bc', '/', 682410),
             ('90243de06161', '/', 682006),
+        )
         for revision, path, size in to_check:
             self._test_dir_size(revision, path, size)
     def test_repo_size(self):
         assert self.repo.size == 682421
     def test_file_history(self):
         # we can only check if those revisions are present in the history
         # as we cannot update this test every time file is changed
         files = {
             'setup.py': [7, 18, 45, 46, 47, 69, 77],
             'vcs/nodes.py': [7, 8, 24, 26, 30, 45, 47, 49, 56, 57, 58, 59, 60,
 , 73, 76],
             'vcs/backends/hg.py': [4, 5, 6, 11, 12, 13, 14, 15, 16, 21, 22, 23,
 , 27, 28, 30, 31, 33, 35, 36, 37, 38, 39, 40, 41, 44, 45, 47,
 , 49, 53, 54, 55, 58, 60, 61, 67, 68, 69, 70, 73, 77, 78, 79,
 ],
+        }
         for path, revs in files.items():
             tip = self.repo.get_changeset(revs[-1])
             node = tip.get_node(path)
             node_revs = [chset.revision for chset in node.history]
             assert set(revs).issubset(set(node_revs)), \
                 "We assumed that %s is subset of revisions for which file %s " \
                 "has been changed, and history of that node returned: %s" \
                 % (revs, path, node_revs)
     def test_file_annotate(self):
         files = {
                  'vcs/backends/__init__.py':
                   {89: {'lines_no': 31,
                         'changesets': [32, 32, 61, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]},
 : {'lines_no': 1,
                         'changesets': [4]},
 : {'lines_no': 31,
                         'changesets': [32, 32, 45, 32, 32, 37, 32, 32, 32, 44,
 , 37, 37, 37, 45, 37, 44, 37, 37, 37,
 , 32, 32, 32, 37, 32, 37, 37, 32,
 , 32]}},
                  'vcs/exceptions.py':
                  {89: {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18,
                        'changesets': [16, 16, 16, 16, 16, 16, 16, 16, 16, 16,
 , 16, 17, 16, 16, 18, 18, 18]},
 : {'lines_no': 18, 'changesets': [16, 16, 16, 16, 16, 16,
 , 16, 16, 16, 16, 16,
 , 16, 16, 18, 18, 18]}},
                  'MANIFEST.in': {89: {'lines_no': 5,
                                       'changesets': [7, 7, 7, 71, 71]},
 : {'lines_no': 3,
                                       'changesets': [7, 7, 7]},
 : {'lines_no': 3,
                                      'changesets': [7, 7, 7]}}}
         for fname, revision_dict in files.items():
             for rev, data in revision_dict.items():
                 cs = self.repo.get_changeset(rev)
                 l1_1 = [x[1] for x in cs.get_file_annotate(fname)]
                 l1_2 = [x[2]().raw_id for x in cs.get_file_annotate(fname)]
                 assert l1_1 == l1_2
                 l1 = l1_2 = [x[2]().revision for x in cs.get_file_annotate(fname)]
                 l2 = files[fname][rev]['changesets']
                 assert l1 == l2, "The lists of revision for %s@rev%s" \
                     "from annotation list should match each other," \
                     "got \n%s \nvs \n%s " % (fname, rev, l1, l2)
     def test_changeset_state(self):
         """
         Tests which files have been added/changed/removed at particular revision
         """
         # rev 46ad32a4f974:
         # hg st --rev 46ad32a4f974
         #    changed: 13
         #    added:   20
         #    removed: 1
         changed = set(['.hgignore'
             , 'README.rst' , 'docs/conf.py' , 'docs/index.rst' , 'setup.py'
             , 'tests/test_hg.py' , 'tests/test_nodes.py' , 'vcs/__init__.py'
             , 'vcs/backends/__init__.py' , 'vcs/backends/base.py'
             , 'vcs/backends/hg.py' , 'vcs/nodes.py' , 'vcs/utils/__init__.py'])
         added = set(['docs/api/backends/hg.rst'
             , 'docs/api/backends/index.rst' , 'docs/api/index.rst'
             , 'docs/api/nodes.rst' , 'docs/api/web/index.rst'
             , 'docs/api/web/simplevcs.rst' , 'docs/installation.rst'
             , 'docs/quickstart.rst' , 'setup.cfg' , 'vcs/utils/baseui_config.py'
             , 'vcs/utils/web.py' , 'vcs/web/__init__.py' , 'vcs/web/exceptions.py'
             , 'vcs/web/simplevcs/__init__.py' , 'vcs/web/simplevcs/exceptions.py'
             , 'vcs/web/simplevcs/middleware.py' , 'vcs/web/simplevcs/models.py'
             , 'vcs/web/simplevcs/settings.py' , 'vcs/web/simplevcs/utils.py'
             , 'vcs/web/simplevcs/views.py'])
         removed = set(['docs/api.rst'])
         chset64 = self.repo.get_changeset('46ad32a4f974')
         assert set((node.path for node in chset64.added)) == added
         assert set((node.path for node in chset64.changed)) == changed
         assert set((node.path for node in chset64.removed)) == removed
         # rev b090f22d27d6:
         # hg st --rev b090f22d27d6
         #    changed: 13
         #    added:   20
         #    removed: 1
         chset88 = self.repo.get_changeset('b090f22d27d6')
         assert set((node.path for node in chset88.added)) == set()
         assert set((node.path for node in chset88.changed)) == set(['.hgignore'])
         assert set((node.path for node in chset88.removed)) == set()
         # 85:
         #    added:   2 ['vcs/utils/diffs.py', 'vcs/web/simplevcs/views/diffs.py']
         #    changed: 4 ['vcs/web/simplevcs/models.py', ...]
         #    removed: 1 ['vcs/utils/web.py']
         chset85 = self.repo.get_changeset(85)
         assert set((node.path for node in chset85.added)) == set([
             'vcs/utils/diffs.py',
             'vcs/web/simplevcs/views/diffs.py'
         ])
         assert set((node.path for node in chset85.changed)) == set([
             'vcs/web/simplevcs/models.py',
             'vcs/web/simplevcs/utils.py',
             'vcs/web/simplevcs/views/__init__.py',
             'vcs/web/simplevcs/views/repository.py',
         ])
         assert set((node.path for node in chset85.removed)) == set([
             'vcs/utils/web.py'
         ])
     def test_files_state(self):
         """
         Tests state of FileNodes.
         """
         chset = self.repo.get_changeset(85)
         node = chset.get_node('vcs/utils/diffs.py')
         assert node.state, NodeState.ADDED
         assert node.added
         assert not node.changed
         assert not node.not_changed
         assert not node.removed
         chset = self.repo.get_changeset(88)
         node = chset.get_node('.hgignore')
         assert node.state, NodeState.CHANGED
         assert not node.added
         assert node.changed
         assert not node.not_changed
         assert not node.removed
         chset = self.repo.get_changeset(85)
         node = chset.get_node('setup.py')
         assert node.state, NodeState.NOT_CHANGED
         assert not node.added
         assert not node.changed
         assert node.not_changed
         assert not node.removed
         # If node has REMOVED state then trying to fetch it would raise
         # ChangesetError exception
         chset = self.repo.get_changeset(2)
         path = 'vcs/backends/BaseRepository.py'
         with pytest.raises(NodeDoesNotExistError):
             chset.get_node(path)
         # but it would be one of ``removed`` (changeset's attribute)
         assert path in [rf.path for rf in chset.removed]
     def test_commit_message_is_unicode(self):
         for cm in self.repo:
             assert type(cm.message) == unicode
     def test_changeset_author_is_unicode(self):
         for cm in self.repo:
             assert type(cm.author) == unicode
     def test_repo_files_content_is_unicode(self):
         test_changeset = self.repo.get_changeset(100)
         for node in test_changeset.get_node('/'):
             if node.is_file():
                 assert type(node.content) == unicode
     def test_wrong_path(self):
         # There is 'setup.py' in the root dir but not there:
         path = 'foo/bar/setup.py'
         with pytest.raises(VCSError):
             self.repo.get_changeset().get_node(path)
     def test_archival_file(self):
         # TODO:
         pass
     def test_archival_as_generator(self):
         # TODO:
         pass
     def test_archival_wrong_kind(self):
         tip = self.repo.get_changeset()
         with pytest.raises(VCSError):
             tip.fill_archive(kind='error')
     def test_archival_empty_prefix(self):
         # TODO:
         pass
     def test_author_email(self):
         assert 'marcin@python-blog.com' == self.repo.get_changeset('b986218ba1c9').author_email
         assert 'lukasz.balcerzak@python-center.pl' == self.repo.get_changeset('3803844fdbd3').author_email
         assert '' == self.repo.get_changeset('84478366594b').author_email
     def test_author_username(self):
         assert 'Marcin Kuzminski' == self.repo.get_changeset('b986218ba1c9').author_name
         assert 'Lukasz Balcerzak' == self.repo.get_changeset('3803844fdbd3').author_name
         assert 'marcink' == self.repo.get_changeset('84478366594b').author_name

0 comments (0 inline, 0 general)