Changeset - 39a59e6915bb
[Not reviewed]
default
0 2 0
Mads Kiilerich - 9 years ago 2016-09-25 17:21:07
madski@unity3d.com
helpers: refactor and optimize urlify_issues

Avoid parsing the configuration and compiling regexps every time a string is
processed.

Instead, the configuration is parsed once and turned into a function that apply
the compiled regexps in a chain.

A next iteration can perhaps run them in parallel and integrate with the
general urlify machinery.
2 files changed with 56 insertions and 50 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/helpers.py
Show inline comments
 
@@ -587,912 +587,914 @@ def action_parser(user_log, feed=False, 
 
                if rev.op and rev.ref_name:
 
                    if rev.op == 'delete_branch':
 
                        lbl = _('Deleted branch: %s') % rev.ref_name
 
                    elif rev.op == 'tag':
 
                        lbl = _('Created tag: %s') % rev.ref_name
 
                    else:
 
                        lbl = 'Unknown operation %s' % rev.op
 
                else:
 
                    lazy_cs = True
 
                    lbl = rev.short_id[:8]
 
                    url_ = url('changeset_home', repo_name=repo_name,
 
                               revision=rev.raw_id)
 
            else:
 
                # changeset cannot be found - it might have been stripped or removed
 
                lbl = rev[:12]
 
                title_ = _('Changeset not found')
 
            if parse_cs:
 
                return link_to(lbl, url_, title=title_, class_='tooltip')
 
            return link_to(lbl, url_, raw_id=rev.raw_id, repo_name=repo_name,
 
                           class_='lazy-cs' if lazy_cs else '')
 

	
 
        def _get_op(rev_txt):
 
            _op = None
 
            _name = rev_txt
 
            if len(rev_txt.split('=>')) == 2:
 
                _op, _name = rev_txt.split('=>')
 
            return _op, _name
 

	
 
        revs = []
 
        if len(filter(lambda v: v != '', revs_ids)) > 0:
 
            repo = None
 
            for rev in revs_ids[:revs_top_limit]:
 
                _op, _name = _get_op(rev)
 

	
 
                # we want parsed changesets, or new log store format is bad
 
                if parse_cs:
 
                    try:
 
                        if repo is None:
 
                            repo = user_log.repository.scm_instance
 
                        _rev = repo.get_changeset(rev)
 
                        revs.append(_rev)
 
                    except ChangesetDoesNotExistError:
 
                        log.error('cannot find revision %s in this repo', rev)
 
                        revs.append(rev)
 
                else:
 
                    _rev = AttributeDict({
 
                        'short_id': rev[:12],
 
                        'raw_id': rev,
 
                        'message': '',
 
                        'op': _op,
 
                        'ref_name': _name
 
                    })
 
                    revs.append(_rev)
 
        cs_links = [" " + ', '.join(
 
            [lnk(rev, repo_name) for rev in revs[:revs_limit]]
 
        )]
 
        _op1, _name1 = _get_op(revs_ids[0])
 
        _op2, _name2 = _get_op(revs_ids[-1])
 

	
 
        _rev = '%s...%s' % (_name1, _name2)
 

	
 
        compare_view = (
 
            ' <div class="compare_view tooltip" title="%s">'
 
            '<a href="%s">%s</a> </div>' % (
 
                _('Show all combined changesets %s->%s') % (
 
                    revs_ids[0][:12], revs_ids[-1][:12]
 
                ),
 
                url('changeset_home', repo_name=repo_name,
 
                    revision=_rev
 
                ),
 
                _('Compare view')
 
            )
 
        )
 

	
 
        # if we have exactly one more than normally displayed
 
        # just display it, takes less space than displaying
 
        # "and 1 more revisions"
 
        if len(revs_ids) == revs_limit + 1:
 
            cs_links.append(", " + lnk(revs[revs_limit], repo_name))
 

	
 
        # hidden-by-default ones
 
        if len(revs_ids) > revs_limit + 1:
 
            uniq_id = revs_ids[0]
 
            html_tmpl = (
 
                '<span> %s <a class="show_more" id="_%s" '
 
                'href="#more">%s</a> %s</span>'
 
            )
 
            if not feed:
 
                cs_links.append(html_tmpl % (
 
                      _('and'),
 
                      uniq_id, _('%s more') % (len(revs_ids) - revs_limit),
 
                      _('revisions')
 
                    )
 
                )
 

	
 
            if not feed:
 
                html_tmpl = '<span id="%s" style="display:none">, %s </span>'
 
            else:
 
                html_tmpl = '<span id="%s"> %s </span>'
 

	
 
            morelinks = ', '.join(
 
              [lnk(rev, repo_name) for rev in revs[revs_limit:]]
 
            )
 

	
 
            if len(revs_ids) > revs_top_limit:
 
                morelinks += ', ...'
 

	
 
            cs_links.append(html_tmpl % (uniq_id, morelinks))
 
        if len(revs) > 1:
 
            cs_links.append(compare_view)
 
        return ''.join(cs_links)
 

	
 
    def get_fork_name():
 
        repo_name = action_params
 
        url_ = url('summary_home', repo_name=repo_name)
 
        return _('Fork name %s') % link_to(action_params, url_)
 

	
 
    def get_user_name():
 
        user_name = action_params
 
        return user_name
 

	
 
    def get_users_group():
 
        group_name = action_params
 
        return group_name
 

	
 
    def get_pull_request():
 
        pull_request_id = action_params
 
        nice_id = PullRequest.make_nice_id(pull_request_id)
 

	
 
        deleted = user_log.repository is None
 
        if deleted:
 
            repo_name = user_log.repository_name
 
        else:
 
            repo_name = user_log.repository.repo_name
 

	
 
        return link_to(_('Pull request %s') % nice_id,
 
                    url('pullrequest_show', repo_name=repo_name,
 
                    pull_request_id=pull_request_id))
 

	
 
    def get_archive_name():
 
        archive_name = action_params
 
        return archive_name
 

	
 
    # action : translated str, callback(extractor), icon
 
    action_map = {
 
    'user_deleted_repo':           (_('[deleted] repository'),
 
                                    None, 'icon-trashcan'),
 
    'user_created_repo':           (_('[created] repository'),
 
                                    None, 'icon-plus'),
 
    'user_created_fork':           (_('[created] repository as fork'),
 
                                    None, 'icon-fork'),
 
    'user_forked_repo':            (_('[forked] repository'),
 
                                    get_fork_name, 'icon-fork'),
 
    'user_updated_repo':           (_('[updated] repository'),
 
                                    None, 'icon-pencil'),
 
    'user_downloaded_archive':      (_('[downloaded] archive from repository'),
 
                                    get_archive_name, 'icon-download-cloud'),
 
    'admin_deleted_repo':          (_('[delete] repository'),
 
                                    None, 'icon-trashcan'),
 
    'admin_created_repo':          (_('[created] repository'),
 
                                    None, 'icon-plus'),
 
    'admin_forked_repo':           (_('[forked] repository'),
 
                                    None, 'icon-fork'),
 
    'admin_updated_repo':          (_('[updated] repository'),
 
                                    None, 'icon-pencil'),
 
    'admin_created_user':          (_('[created] user'),
 
                                    get_user_name, 'icon-user'),
 
    'admin_updated_user':          (_('[updated] user'),
 
                                    get_user_name, 'icon-user'),
 
    'admin_created_users_group':   (_('[created] user group'),
 
                                    get_users_group, 'icon-pencil'),
 
    'admin_updated_users_group':   (_('[updated] user group'),
 
                                    get_users_group, 'icon-pencil'),
 
    'user_commented_revision':     (_('[commented] on revision in repository'),
 
                                    get_cs_links, 'icon-comment'),
 
    'user_commented_pull_request': (_('[commented] on pull request for'),
 
                                    get_pull_request, 'icon-comment'),
 
    'user_closed_pull_request':    (_('[closed] pull request for'),
 
                                    get_pull_request, 'icon-ok'),
 
    'push':                        (_('[pushed] into'),
 
                                    get_cs_links, 'icon-move-up'),
 
    'push_local':                  (_('[committed via Kallithea] into repository'),
 
                                    get_cs_links, 'icon-pencil'),
 
    'push_remote':                 (_('[pulled from remote] into repository'),
 
                                    get_cs_links, 'icon-move-up'),
 
    'pull':                        (_('[pulled] from'),
 
                                    None, 'icon-move-down'),
 
    'started_following_repo':      (_('[started following] repository'),
 
                                    None, 'icon-heart'),
 
    'stopped_following_repo':      (_('[stopped following] repository'),
 
                                    None, 'icon-heart-empty'),
 
    }
 

	
 
    action_str = action_map.get(action, action)
 
    if feed:
 
        action = action_str[0].replace('[', '').replace(']', '')
 
    else:
 
        action = action_str[0] \
 
            .replace('[', '<span class="journal_highlight">') \
 
            .replace(']', '</span>')
 

	
 
    action_params_func = lambda: ""
 

	
 
    if callable(action_str[1]):
 
        action_params_func = action_str[1]
 

	
 
    def action_parser_icon():
 
        action = user_log.action
 
        action_params = None
 
        x = action.split(':')
 

	
 
        if len(x) > 1:
 
            action, action_params = x
 

	
 
        tmpl = """<i class="%s" alt="%s"></i>"""
 
        ico = action_map.get(action, ['', '', ''])[2]
 
        return literal(tmpl % (ico, action))
 

	
 
    # returned callbacks we need to call to get
 
    return [lambda: literal(action), action_params_func, action_parser_icon]
 

	
 

	
 

	
 
#==============================================================================
 
# PERMS
 
#==============================================================================
 
from kallithea.lib.auth import HasPermissionAny, \
 
    HasRepoPermissionAny, HasRepoGroupPermissionAny
 

	
 

	
 
#==============================================================================
 
# GRAVATAR URL
 
#==============================================================================
 
def gravatar_div(email_address, cls='', size=30, **div_attributes):
 
    """Return an html literal with a div around a gravatar if they are enabled.
 
    Extra keyword parameters starting with 'div_' will get the prefix removed
 
    and be used as attributes on the div. The default class is 'gravatar'.
 
    """
 
    from pylons import tmpl_context as c
 
    if not c.visual.use_gravatar:
 
        return ''
 
    if 'div_class' not in div_attributes:
 
        div_attributes['div_class'] = "gravatar"
 
    attributes = []
 
    for k, v in sorted(div_attributes.items()):
 
        assert k.startswith('div_'), k
 
        attributes.append(' %s="%s"' % (k[4:], escape(v)))
 
    return literal("""<div%s>%s</div>""" %
 
                   (''.join(attributes),
 
                    gravatar(email_address, cls=cls, size=size)))
 

	
 
def gravatar(email_address, cls='', size=30):
 
    """return html element of the gravatar
 

	
 
    This method will return an <img> with the resolution double the size (for
 
    retina screens) of the image. If the url returned from gravatar_url is
 
    empty then we fallback to using an icon.
 

	
 
    """
 
    from pylons import tmpl_context as c
 
    if not c.visual.use_gravatar:
 
        return ''
 

	
 
    src = gravatar_url(email_address, size * 2)
 

	
 
    if src:
 
        # here it makes sense to use style="width: ..." (instead of, say, a
 
        # stylesheet) because we using this to generate a high-res (retina) size
 
        html = ('<img alt="" class="{cls}" style="width: {size}px; height: {size}px" src="{src}"/>'
 
            .format(cls=cls, size=size, src=src))
 

	
 
    else:
 
        # if src is empty then there was no gravatar, so we use a font icon
 
        html = ("""<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""
 
            .format(cls=cls, size=size, src=src))
 

	
 
    return literal(html)
 

	
 
def gravatar_url(email_address, size=30, default=''):
 
    # doh, we need to re-import those to mock it later
 
    from kallithea.config.routing import url
 
    from pylons import tmpl_context as c
 
    if not c.visual.use_gravatar:
 
        return ""
 

	
 
    _def = 'anonymous@kallithea-scm.org'  # default gravatar
 
    email_address = email_address or _def
 

	
 
    if email_address == _def:
 
        return default
 

	
 
    parsed_url = urlparse.urlparse(url.current(qualified=True))
 
    url = (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL ) \
 
               .replace('{email}', email_address) \
 
               .replace('{md5email}', hashlib.md5(safe_str(email_address).lower()).hexdigest()) \
 
               .replace('{netloc}', parsed_url.netloc) \
 
               .replace('{scheme}', parsed_url.scheme) \
 
               .replace('{size}', safe_str(size))
 
    return url
 

	
 
class Page(_Page):
 
    """
 
    Custom pager to match rendering style with YUI paginator
 
    """
 

	
 
    def __init__(self, *args, **kwargs):
 
        kwargs.setdefault('url', url.current)
 
        _Page.__init__(self, *args, **kwargs)
 

	
 
    def _get_pos(self, cur_page, max_page, items):
 
        edge = (items / 2) + 1
 
        if (cur_page <= edge):
 
            radius = max(items / 2, items - cur_page)
 
        elif (max_page - cur_page) < edge:
 
            radius = (items - 1) - (max_page - cur_page)
 
        else:
 
            radius = items / 2
 

	
 
        left = max(1, (cur_page - (radius)))
 
        right = min(max_page, cur_page + (radius))
 
        return left, cur_page, right
 

	
 
    def _range(self, regexp_match):
 
        """
 
        Return range of linked pages (e.g. '1 2 [3] 4 5 6 7 8').
 

	
 
        Arguments:
 

	
 
        regexp_match
 
            A "re" (regular expressions) match object containing the
 
            radius of linked pages around the current page in
 
            regexp_match.group(1) as a string
 

	
 
        This function is supposed to be called as a callable in
 
        re.sub.
 

	
 
        """
 
        radius = int(regexp_match.group(1))
 

	
 
        # Compute the first and last page number within the radius
 
        # e.g. '1 .. 5 6 [7] 8 9 .. 12'
 
        # -> leftmost_page  = 5
 
        # -> rightmost_page = 9
 
        leftmost_page, _cur, rightmost_page = self._get_pos(self.page,
 
                                                            self.last_page,
 
                                                            (radius * 2) + 1)
 
        nav_items = []
 

	
 
        # Create a link to the first page (unless we are on the first page
 
        # or there would be no need to insert '..' spacers)
 
        if self.page != self.first_page and self.first_page < leftmost_page:
 
            nav_items.append(self._pagerlink(self.first_page, self.first_page))
 

	
 
        # Insert dots if there are pages between the first page
 
        # and the currently displayed page range
 
        if leftmost_page - self.first_page > 1:
 
            # Wrap in a SPAN tag if nolink_attr is set
 
            text_ = '..'
 
            if self.dotdot_attr:
 
                text_ = HTML.span(c=text_, **self.dotdot_attr)
 
            nav_items.append(text_)
 

	
 
        for thispage in xrange(leftmost_page, rightmost_page + 1):
 
            # Highlight the current page number and do not use a link
 
            text_ = str(thispage)
 
            if thispage == self.page:
 
                # Wrap in a SPAN tag if nolink_attr is set
 
                if self.curpage_attr:
 
                    text_ = HTML.span(c=text_, **self.curpage_attr)
 
                nav_items.append(text_)
 
            # Otherwise create just a link to that page
 
            else:
 
                nav_items.append(self._pagerlink(thispage, text_))
 

	
 
        # Insert dots if there are pages between the displayed
 
        # page numbers and the end of the page range
 
        if self.last_page - rightmost_page > 1:
 
            text_ = '..'
 
            # Wrap in a SPAN tag if nolink_attr is set
 
            if self.dotdot_attr:
 
                text_ = HTML.span(c=text_, **self.dotdot_attr)
 
            nav_items.append(text_)
 

	
 
        # Create a link to the very last page (unless we are on the last
 
        # page or there would be no need to insert '..' spacers)
 
        if self.page != self.last_page and rightmost_page < self.last_page:
 
            nav_items.append(self._pagerlink(self.last_page, self.last_page))
 

	
 
        #_page_link = url.current()
 
        #nav_items.append(literal('<link rel="prerender" href="%s?page=%s">' % (_page_link, str(int(self.page)+1))))
 
        #nav_items.append(literal('<link rel="prefetch" href="%s?page=%s">' % (_page_link, str(int(self.page)+1))))
 
        return self.separator.join(nav_items)
 

	
 
    def pager(self, format='~2~', page_param='page', partial_param='partial',
 
        show_if_single_page=False, separator=' ', onclick=None,
 
        symbol_first='<<', symbol_last='>>',
 
        symbol_previous='<', symbol_next='>',
 
        link_attr=None,
 
        curpage_attr=None,
 
        dotdot_attr=None, **kwargs):
 
        self.curpage_attr = curpage_attr or {'class': 'pager_curpage'}
 
        self.separator = separator
 
        self.pager_kwargs = kwargs
 
        self.page_param = page_param
 
        self.partial_param = partial_param
 
        self.onclick = onclick
 
        self.link_attr = link_attr or {'class': 'pager_link', 'rel': 'prerender'}
 
        self.dotdot_attr = dotdot_attr or {'class': 'pager_dotdot'}
 

	
 
        # Don't show navigator if there is no more than one page
 
        if self.page_count == 0 or (self.page_count == 1 and not show_if_single_page):
 
            return ''
 

	
 
        from string import Template
 
        # Replace ~...~ in token format by range of pages
 
        result = re.sub(r'~(\d+)~', self._range, format)
 

	
 
        # Interpolate '%' variables
 
        result = Template(result).safe_substitute({
 
            'first_page': self.first_page,
 
            'last_page': self.last_page,
 
            'page': self.page,
 
            'page_count': self.page_count,
 
            'items_per_page': self.items_per_page,
 
            'first_item': self.first_item,
 
            'last_item': self.last_item,
 
            'item_count': self.item_count,
 
            'link_first': self.page > self.first_page and \
 
                    self._pagerlink(self.first_page, symbol_first) or '',
 
            'link_last': self.page < self.last_page and \
 
                    self._pagerlink(self.last_page, symbol_last) or '',
 
            'link_previous': self.previous_page and \
 
                    self._pagerlink(self.previous_page, symbol_previous) \
 
                    or HTML.span(symbol_previous, class_="yui-pg-previous"),
 
            'link_next': self.next_page and \
 
                    self._pagerlink(self.next_page, symbol_next) \
 
                    or HTML.span(symbol_next, class_="yui-pg-next")
 
        })
 

	
 
        return literal(result)
 

	
 

	
 
#==============================================================================
 
# REPO PAGER, PAGER FOR REPOSITORY
 
#==============================================================================
 
class RepoPage(Page):
 

	
 
    def __init__(self, collection, page=1, items_per_page=20,
 
                 item_count=None, **kwargs):
 

	
 
        """Create a "RepoPage" instance. special pager for paging
 
        repository
 
        """
 
        # TODO: call baseclass __init__
 
        self._url_generator = kwargs.pop('url', url.current)
 

	
 
        # Safe the kwargs class-wide so they can be used in the pager() method
 
        self.kwargs = kwargs
 

	
 
        # Save a reference to the collection
 
        self.original_collection = collection
 

	
 
        self.collection = collection
 

	
 
        # The self.page is the number of the current page.
 
        # The first page has the number 1!
 
        try:
 
            self.page = int(page)  # make it int() if we get it as a string
 
        except (ValueError, TypeError):
 
            self.page = 1
 

	
 
        self.items_per_page = items_per_page
 

	
 
        # Unless the user tells us how many items the collections has
 
        # we calculate that ourselves.
 
        if item_count is not None:
 
            self.item_count = item_count
 
        else:
 
            self.item_count = len(self.collection)
 

	
 
        # Compute the number of the first and last available page
 
        if self.item_count > 0:
 
            self.first_page = 1
 
            self.page_count = int(math.ceil(float(self.item_count) /
 
                                            self.items_per_page))
 
            self.last_page = self.first_page + self.page_count - 1
 

	
 
            # Make sure that the requested page number is the range of
 
            # valid pages
 
            if self.page > self.last_page:
 
                self.page = self.last_page
 
            elif self.page < self.first_page:
 
                self.page = self.first_page
 

	
 
            # Note: the number of items on this page can be less than
 
            #       items_per_page if the last page is not full
 
            self.first_item = max(0, (self.item_count) - (self.page *
 
                                                          items_per_page))
 
            self.last_item = ((self.item_count - 1) - items_per_page *
 
                              (self.page - 1))
 

	
 
            self.items = list(self.collection[self.first_item:self.last_item + 1])
 

	
 
            # Links to previous and next page
 
            if self.page > self.first_page:
 
                self.previous_page = self.page - 1
 
            else:
 
                self.previous_page = None
 

	
 
            if self.page < self.last_page:
 
                self.next_page = self.page + 1
 
            else:
 
                self.next_page = None
 

	
 
        # No items available
 
        else:
 
            self.first_page = None
 
            self.page_count = 0
 
            self.last_page = None
 
            self.first_item = None
 
            self.last_item = None
 
            self.previous_page = None
 
            self.next_page = None
 
            self.items = []
 

	
 
        # This is a subclass of the 'list' type. Initialise the list now.
 
        list.__init__(self, reversed(self.items))
 

	
 

	
 
def changed_tooltip(nodes):
 
    """
 
    Generates a html string for changed nodes in changeset page.
 
    It limits the output to 30 entries
 

	
 
    :param nodes: LazyNodesGenerator
 
    """
 
    if nodes:
 
        pref = ': <br/> '
 
        suf = ''
 
        if len(nodes) > 30:
 
            suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)
 
        return literal(pref + '<br/> '.join([safe_unicode(x.path)
 
                                             for x in nodes[:30]]) + suf)
 
    else:
 
        return ': ' + _('No files')
 

	
 

	
 
def repo_link(groups_and_repos):
 
    """
 
    Makes a breadcrumbs link to repo within a group
 
    joins &raquo; on each group to create a fancy link
 

	
 
    ex::
 
        group >> subgroup >> repo
 

	
 
    :param groups_and_repos:
 
    :param last_url:
 
    """
 
    groups, just_name, repo_name = groups_and_repos
 
    last_url = url('summary_home', repo_name=repo_name)
 
    last_link = link_to(just_name, last_url)
 

	
 
    def make_link(group):
 
        return link_to(group.name,
 
                       url('repos_group_home', group_name=group.group_name))
 
    return literal(' &raquo; '.join(map(make_link, groups) + ['<span>%s</span>' % last_link]))
 

	
 

	
 
def fancy_file_stats(stats):
 
    """
 
    Displays a fancy two colored bar for number of added/deleted
 
    lines of code on file
 

	
 
    :param stats: two element list of added/deleted lines of code
 
    """
 
    from kallithea.lib.diffs import NEW_FILENODE, DEL_FILENODE, \
 
        MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE
 

	
 
    def cgen(l_type, a_v, d_v):
 
        mapping = {'tr': 'top-right-rounded-corner-mid',
 
                   'tl': 'top-left-rounded-corner-mid',
 
                   'br': 'bottom-right-rounded-corner-mid',
 
                   'bl': 'bottom-left-rounded-corner-mid'}
 
        map_getter = lambda x: mapping[x]
 

	
 
        if l_type == 'a' and d_v:
 
            #case when added and deleted are present
 
            return ' '.join(map(map_getter, ['tl', 'bl']))
 

	
 
        if l_type == 'a' and not d_v:
 
            return ' '.join(map(map_getter, ['tr', 'br', 'tl', 'bl']))
 

	
 
        if l_type == 'd' and a_v:
 
            return ' '.join(map(map_getter, ['tr', 'br']))
 

	
 
        if l_type == 'd' and not a_v:
 
            return ' '.join(map(map_getter, ['tr', 'br', 'tl', 'bl']))
 

	
 
    a, d = stats['added'], stats['deleted']
 
    width = 100
 

	
 
    if stats['binary']:
 
        #binary mode
 
        lbl = ''
 
        bin_op = 1
 

	
 
        if BIN_FILENODE in stats['ops']:
 
            lbl = 'bin+'
 

	
 
        if NEW_FILENODE in stats['ops']:
 
            lbl += _('new file')
 
            bin_op = NEW_FILENODE
 
        elif MOD_FILENODE in stats['ops']:
 
            lbl += _('mod')
 
            bin_op = MOD_FILENODE
 
        elif DEL_FILENODE in stats['ops']:
 
            lbl += _('del')
 
            bin_op = DEL_FILENODE
 
        elif RENAMED_FILENODE in stats['ops']:
 
            lbl += _('rename')
 
            bin_op = RENAMED_FILENODE
 

	
 
        #chmod can go with other operations
 
        if CHMOD_FILENODE in stats['ops']:
 
            _org_lbl = _('chmod')
 
            lbl += _org_lbl if lbl.endswith('+') else '+%s' % _org_lbl
 

	
 
        #import ipdb;ipdb.set_trace()
 
        b_d = '<div class="bin bin%s %s" style="width:100%%">%s</div>' % (bin_op, cgen('a', a_v='', d_v=0), lbl)
 
        b_a = '<div class="bin bin1" style="width:0%"></div>'
 
        return literal('<div style="width:%spx">%s%s</div>' % (width, b_a, b_d))
 

	
 
    t = stats['added'] + stats['deleted']
 
    unit = float(width) / (t or 1)
 

	
 
    # needs > 9% of width to be visible or 0 to be hidden
 
    a_p = max(9, unit * a) if a > 0 else 0
 
    d_p = max(9, unit * d) if d > 0 else 0
 
    p_sum = a_p + d_p
 

	
 
    if p_sum > width:
 
        #adjust the percentage to be == 100% since we adjusted to 9
 
        if a_p > d_p:
 
            a_p = a_p - (p_sum - width)
 
        else:
 
            d_p = d_p - (p_sum - width)
 

	
 
    a_v = a if a > 0 else ''
 
    d_v = d if d > 0 else ''
 

	
 
    d_a = '<div class="added %s" style="width:%s%%">%s</div>' % (
 
        cgen('a', a_v, d_v), a_p, a_v
 
    )
 
    d_d = '<div class="deleted %s" style="width:%s%%">%s</div>' % (
 
        cgen('d', a_v, d_v), d_p, d_v
 
    )
 
    return literal('<div style="width:%spx">%s%s</div>' % (width, d_a, d_d))
 

	
 

	
 
_URLIFY_RE = re.compile(r'''
 
# URL markup
 
(?P<url>%s) |
 
# @mention markup
 
(?P<mention>%s) |
 
# Changeset hash markup
 
(?<!\w|[-_])
 
  (?P<hash>[0-9a-f]{12,40})
 
(?!\w|[-_]) |
 
# Markup of *bold text*
 
(?:
 
  (?:^|(?<=\s))
 
  (?P<bold> [*] (?!\s) [^*\n]* (?<!\s) [*] )
 
  (?![*\w])
 
) |
 
# "Stylize" markup
 
\[see\ \=&gt;\ *(?P<seen>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |
 
\[license\ \=&gt;\ *(?P<license>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |
 
\[(?P<tagtype>requires|recommends|conflicts|base)\ \=&gt;\ *(?P<tagvalue>[a-zA-Z0-9\-\/]*)\] |
 
\[(?:lang|language)\ \=&gt;\ *(?P<lang>[a-zA-Z\-\/\#\+]*)\] |
 
\[(?P<tag>[a-z]+)\]
 
''' % (url_re.pattern, MENTIONS_REGEX.pattern),
 
    re.VERBOSE | re.MULTILINE | re.IGNORECASE)
 

	
 

	
 

	
 
def urlify_text(s, repo_name=None, link_=None, truncate=None, stylize=False, truncatef=truncate):
 
    """
 
    Parses given text message and make literal html with markup.
 
    The text will be truncated to the specified length.
 
    Hashes are turned into changeset links to specified repository.
 
    URLs links to what they say.
 
    Issues are linked to given issue-server.
 
    If link_ is provided, all text not already linking somewhere will link there.
 
    """
 

	
 
    def _replace(match_obj):
 
        url = match_obj.group('url')
 
        if url is not None:
 
            return '<a href="%(url)s">%(url)s</a>' % {'url': url}
 
        mention = match_obj.group('mention')
 
        if mention is not None:
 
            return '<b>%s</b>' % mention
 
        hash_ = match_obj.group('hash')
 
        if hash_ is not None and repo_name is not None:
 
            from kallithea.config.routing import url  # doh, we need to re-import url to mock it later
 
            return '<a class="revision-link" href="%(url)s">%(hash)s</a>' % {
 
                 'url': url('changeset_home', repo_name=repo_name, revision=hash_),
 
                 'hash': hash_,
 
                }
 
        bold = match_obj.group('bold')
 
        if bold is not None:
 
            return '<b>*%s*</b>' % _urlify(bold[1:-1])
 
        if stylize:
 
            seen = match_obj.group('seen')
 
            if seen:
 
                return '<div class="metatag" tag="see">see =&gt; %s</div>' % seen
 
            license = match_obj.group('license')
 
            if license:
 
                return '<div class="metatag" tag="license"><a href="http:\/\/www.opensource.org/licenses/%s">%s</a></div>' % (license, license)
 
            tagtype = match_obj.group('tagtype')
 
            if tagtype:
 
                tagvalue = match_obj.group('tagvalue')
 
                return '<div class="metatag" tag="%s">%s =&gt; <a href="/%s">%s</a></div>' % (tagtype, tagtype, tagvalue, tagvalue)
 
            lang = match_obj.group('lang')
 
            if lang:
 
                return '<div class="metatag" tag="lang">%s</div>' % lang
 
            tag = match_obj.group('tag')
 
            if tag:
 
                return '<div class="metatag" tag="%s">%s</div>' % (tag, tag)
 
        return match_obj.group(0)
 

	
 
    def _urlify(s):
 
        """
 
        Extract urls from text and make html links out of them
 
        """
 
        return _URLIFY_RE.sub(_replace, s)
 

	
 
    if truncate is None:
 
        s = s.rstrip()
 
    else:
 
        s = truncatef(s, truncate, whole_word=True)
 
    s = html_escape(s)
 
    s = _urlify(s)
 
    if repo_name is not None:
 
        s = urlify_issues(s, repo_name)
 
    if link_ is not None:
 
        # make href around everything that isn't a href already
 
        s = linkify_others(s, link_)
 
    s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')
 
    return literal(s)
 

	
 

	
 
def linkify_others(t, l):
 
    """Add a default link to html with links.
 
    HTML doesn't allow nesting of links, so the outer link must be broken up
 
    in pieces and give space for other links.
 
    """
 
    urls = re.compile(r'(\<a.*?\<\/a\>)',)
 
    links = []
 
    for e in urls.split(t):
 
        if e.strip() and not urls.match(e):
 
            links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
 
        else:
 
            links.append(e)
 

	
 
    return ''.join(links)
 

	
 

	
 
def _urlify_issues_replace_f(repo_name, ISSUE_SERVER_LNK, ISSUE_PREFIX):
 
    def urlify_issues_replace(match_obj):
 
        pref = ''
 
        if match_obj.group().startswith(' '):
 
            pref = ' '
 

	
 
        issue_id = ''.join(match_obj.groups())
 
        issue_url = ISSUE_SERVER_LNK.replace('{id}', issue_id)
 
        issue_url = issue_url.replace('{repo}', repo_name)
 
        issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])
 

	
 
        return (
 
            '%(pref)s<a class="%(cls)s" href="%(url)s">'
 
            '%(issue-prefix)s%(id-repr)s'
 
            '</a>'
 
            ) % {
 
             'pref': pref,
 
             'cls': 'issue-tracker-link',
 
             'url': issue_url,
 
             'id-repr': issue_id,
 
             'issue-prefix': ISSUE_PREFIX,
 
             'serv': ISSUE_SERVER_LNK,
 
            }
 
    return urlify_issues_replace
 
# Global variable that will hold the actual urlify_issues function body.
 
# Will be set on first use when the global configuration has been read.
 
_urlify_issues_f = None
 

	
 

	
 
def urlify_issues(newtext, repo_name):
 
    from kallithea import CONFIG as conf
 
    """Urlify issue references according to .ini configuration"""
 
    global _urlify_issues_f
 
    if _urlify_issues_f is None:
 
        from kallithea import CONFIG
 
        assert CONFIG['sqlalchemy.url'] # make sure config has been loaded
 

	
 
        # Build chain of urlify functions, starting with not doing any transformation
 
        tmp_urlify_issues_f = lambda s: s
 

	
 
    # allow multiple issue servers to be used
 
    valid_indices = [
 
        x.group(1)
 
        for x in map(lambda x: re.match(r'issue_pat(.*)', x), conf.keys())
 
        if x and 'issue_server_link%s' % x.group(1) in conf
 
        and 'issue_prefix%s' % x.group(1) in conf
 
    ]
 

	
 
    if valid_indices:
 
        log.debug('found issue server suffixes `%s` during valuation of: %s',
 
                  ','.join(valid_indices), newtext)
 
        issue_pat_re = re.compile(r'issue_pat(.*)')
 
        for k in CONFIG.keys():
 
            # Find all issue_pat* settings that also have corresponding server_link and prefix configuration
 
            m = issue_pat_re.match(k)
 
            if m is None:
 
                continue
 
            suffix = m.group(1)
 
            issue_pat = CONFIG.get(k)
 
            issue_server_link = CONFIG.get('issue_server_link%s' % suffix)
 
            issue_prefix = CONFIG.get('issue_prefix%s' % suffix)
 
            if issue_pat and issue_server_link and issue_prefix:
 
                log.debug('issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_prefix)
 
            else:
 
                log.error('skipping incomplete issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_prefix)
 
                continue
 

	
 
    for pattern_index in valid_indices:
 
        ISSUE_PATTERN = conf.get('issue_pat%s' % pattern_index)
 
        ISSUE_SERVER_LNK = conf.get('issue_server_link%s' % pattern_index)
 
        ISSUE_PREFIX = conf.get('issue_prefix%s' % pattern_index)
 
            # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound
 
            issue_re = re.compile(issue_pat)
 
            def issues_replace(match_obj,
 
                               issue_server_link=issue_server_link, issue_prefix=issue_prefix):
 
                leadingspace = ' ' if match_obj.group().startswith(' ') else ''
 
                issue_id = ''.join(match_obj.groups())
 
                issue_url = issue_server_link.replace('{id}', issue_id)
 
                issue_url = issue_url.replace('{repo}', repo_name)
 
                issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])
 
                return (
 
                    '%(leadingspace)s<a class="issue-tracker-link" href="%(url)s">'
 
                    '%(issue-prefix)s%(id-repr)s'
 
                    '</a>'
 
                    ) % {
 
                     'leadingspace': leadingspace,
 
                     'url': issue_url,
 
                     'id-repr': issue_id,
 
                     'issue-prefix': issue_prefix,
 
                     'serv': issue_server_link,
 
                    }
 
            tmp_urlify_issues_f = (lambda s,
 
                                          issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f:
 
                                   issue_re.sub(issues_replace, chain_f(s)))
 

	
 
        log.debug('pattern suffix `%s` PAT:%s SERVER_LINK:%s PREFIX:%s',
 
                  pattern_index, ISSUE_PATTERN, ISSUE_SERVER_LNK,
 
                  ISSUE_PREFIX)
 

	
 
        URL_PAT = re.compile(ISSUE_PATTERN)
 
        # Set tmp function globally - atomically
 
        _urlify_issues_f = tmp_urlify_issues_f
 

	
 
        urlify_issues_replace = _urlify_issues_replace_f(repo_name, ISSUE_SERVER_LNK, ISSUE_PREFIX)
 
        newtext = URL_PAT.sub(urlify_issues_replace, newtext)
 
        log.debug('processed prefix:`%s` => %s', pattern_index, newtext)
 

	
 
    return newtext
 
    return _urlify_issues_f(newtext)
 

	
 

	
 
def render_w_mentions(source, repo_name=None):
 
    """
 
    Render plain text with revision hashes and issue references urlified
 
    and with @mention highlighting.
 
    """
 
    s = safe_unicode(source)
 
    s = urlify_text(s, repo_name=repo_name)
 
    return literal('<div class="formatted-fixed">%s</div>' % s)
 

	
 

	
 
def short_ref(ref_type, ref_name):
 
    if ref_type == 'rev':
 
        return short_id(ref_name)
 
    return ref_name
 

	
 
def link_to_ref(repo_name, ref_type, ref_name, rev=None):
 
    """
 
    Return full markup for a href to changeset_home for a changeset.
 
    If ref_type is branch it will link to changelog.
 
    ref_name is shortened if ref_type is 'rev'.
 
    if rev is specified show it too, explicitly linking to that revision.
 
    """
 
    txt = short_ref(ref_type, ref_name)
 
    if ref_type == 'branch':
 
        u = url('changelog_home', repo_name=repo_name, branch=ref_name)
 
    else:
 
        u = url('changeset_home', repo_name=repo_name, revision=ref_name)
 
    l = link_to(repo_name + '#' + txt, u)
 
    if rev and ref_type != 'rev':
 
        l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))
 
    return l
 

	
 
def changeset_status(repo, revision):
 
    return ChangesetStatusModel().get_status(repo, revision)
 

	
 

	
 
def changeset_status_lbl(changeset_status):
 
    return ChangesetStatus.get_status_lbl(changeset_status)
 

	
 

	
 
def get_permission_name(key):
 
    return dict(Permission.PERMS).get(key)
 

	
 

	
 
def journal_filter_help():
 
    return _(textwrap.dedent('''
 
        Example filter terms:
 
            repository:vcs
 
            username:developer
 
            action:*push*
 
            ip:127.0.0.1
 
            date:20120101
 
            date:[20120101100000 TO 20120102]
 

	
 
        Generate wildcards using '*' character:
 
            "repository:vcs*" - search everything starting with 'vcs'
 
            "repository:*vcs*" - search for repository containing 'vcs'
 

	
 
        Optional AND / OR operators in queries
 
            "repository:vcs OR repository:test"
 
            "username:test AND repository:test*"
 
    '''))
 

	
 

	
 
def not_mapped_error(repo_name):
 
    flash(_('%s repository is not mapped to db perhaps'
 
            ' it was created or renamed from the filesystem'
 
            ' please run the application again'
 
            ' in order to rescan repositories') % repo_name, category='error')
 

	
 

	
 
def ip_range(ip_addr):
 
    from kallithea.model.db import UserIpMap
 
    s, e = UserIpMap._get_ip_range(ip_addr)
 
    return '%s - %s' % (s, e)
 

	
 

	
 
def form(url, method="post", **attrs):
 
    """Like webhelpers.html.tags.form but automatically using secure_form with
 
    authentication_token for POST. authentication_token is thus never leaked
 
    in the URL."""
 
    if method.lower() == 'get':
 
        return insecure_form(url, method=method, **attrs)
 
    # webhelpers will turn everything but GET into POST
 
    return secure_form(url, method=method, **attrs)
kallithea/tests/functional/test_files.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import os
 
import posixpath
 
import mimetypes
 
from kallithea.tests.base import *
 
from kallithea.model.db import Repository
 
from kallithea.model.meta import Session
 
from kallithea.tests.fixture import Fixture
 

	
 
fixture = Fixture()
 

	
 
ARCHIVE_SPECS = {
 
    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),
 
    '.tar.gz': ('application/x-gzip', 'tgz', ''),
 
    '.zip': ('application/zip', 'zip', ''),
 
}
 

	
 
HG_NODE_HISTORY = fixture.load_resource('hg_node_history_response.json')
 
GIT_NODE_HISTORY = fixture.load_resource('git_node_history_response.json')
 

	
 

	
 
def _set_downloads(repo_name, set_to):
 
    repo = Repository.get_by_repo_name(repo_name)
 
    repo.enable_downloads = set_to
 
    Session().commit()
 

	
 

	
 
class TestFilesController(TestController):
 

	
 
    def test_index(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='/'))
 
        # Test response...
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/vcs"><i class="icon-folder-open"></i><span>vcs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.gitignore"><i class="icon-doc"></i><span>.gitignore</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgignore"><i class="icon-doc"></i><span>.hgignore</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.hgtags"><i class="icon-doc"></i><span>.hgtags</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/.travis.yml"><i class="icon-doc"></i><span>.travis.yml</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/MANIFEST.in"><i class="icon-doc"></i><span>MANIFEST.in</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/run_test_and_report.sh"><i class="icon-doc"></i><span>run_test_and_report.sh</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.cfg"><i class="icon-doc"></i><span>setup.cfg</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/setup.py"><i class="icon-doc"></i><span>setup.py</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/test_and_report.sh"><i class="icon-doc"></i><span>test_and_report.sh</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/96507bd11ecc815ebc6270fdf6db110928c09c1e/tox.ini"><i class="icon-doc"></i><span>tox.ini</span></a>' % HG_REPO)
 

	
 
    def test_index_revision(self):
 
        self.log_user()
 

	
 
        response = self.app.get(
 
            url(controller='files', action='index',
 
                repo_name=HG_REPO,
 
                revision='7ba66bec8d6dbba14a2155be32408c435c5f4492',
 
                f_path='/')
 
        )
 

	
 
        #Test response...
 

	
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/docs"><i class="icon-folder-open"></i><span>docs</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-dir ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/tests"><i class="icon-folder-open"></i><span>tests</span></a>' % HG_REPO)
 
        response.mustcontain('<a class="browser-file ypjax-link" href="/%s/files/7ba66bec8d6dbba14a2155be32408c435c5f4492/README.rst"><i class="icon-doc"></i><span>README.rst</span></a>' % HG_REPO)
 
        response.mustcontain('1.1 KiB')
 

	
 
    def test_index_different_branch(self):
 
        self.log_user()
 

	
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='97e8b885c04894463c51898e14387d80c30ed1ee',
 
                                    f_path='/'))
 

	
 
        response.mustcontain("""<option selected="selected" value="97e8b885c04894463c51898e14387d80c30ed1ee">git at 97e8b885c048</option>""")
 

	
 
    def test_index_paging(self):
 
        self.log_user()
 

	
 
        for r in [(73, 'a066b25d5df7016b45a41b7e2a78c33b57adc235'),
 
                  (92, 'cc66b61b8455b264a7a8a2d8ddc80fcfc58c221e'),
 
                  (109, '75feb4c33e81186c87eac740cee2447330288412'),
 
                  (1, '3d8f361e72ab303da48d799ff1ac40d5ac37c67e'),
 
                  (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:
 

	
 
            response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision=r[1],
 
                                    f_path='/'))
 

	
 
            response.mustcontain("""@ r%s:%s""" % (r[0], r[1][:12]))
 

	
 
    def test_file_source(self):
 
        # Force the global cache to be populated now when we know the right .ini has been loaded.
 
        # (Without this, the test would fail.)
 
        import kallithea.lib.helpers
 
        kallithea.lib.helpers._urlify_issues_f = None
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='8911406ad776fdd3d0b9932a2e89677e57405a48',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        response.mustcontain("""<div class="commit">Partially implemented <a class="issue-tracker-link" href="https://issues.example.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.<br/>"""
 
"""In addition some other __str__ are unicode as well<br/>"""
 
"""Added test for unicode<br/>"""
 
"""Improved test to clone into uniq repository.<br/>"""
 
"""removed extra unicode conversion in diff.</div>
 
""")
 

	
 
        response.mustcontain("""<option selected="selected" value="8911406ad776fdd3d0b9932a2e89677e57405a48">default at 8911406ad776</option>""")
 

	
 
    def test_file_source_history(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py'),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 
        assert response.body == HG_NODE_HISTORY
 

	
 
    def test_file_source_history_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py'),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'},)
 
        assert response.body == GIT_NODE_HISTORY
 

	
 
    def test_file_annotation(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 

	
 
        response.mustcontain("""r356:25213a5fbb04""")
 

	
 
    def test_file_annotation_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='index',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain("""r345:c994f0de03b2""")
 

	
 
    def test_file_annotation_history(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 

	
 
        assert response.body == HG_NODE_HISTORY
 

	
 
    def test_file_annotation_history_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='history',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True),
 
                                extra_environ={'HTTP_X_PARTIAL_XHR': '1'})
 

	
 
        assert response.body == GIT_NODE_HISTORY
 

	
 
    def test_file_authors(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='authors',
 
                                    repo_name=HG_REPO,
 
                                    revision='tip',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain('Marcin Kuzminski')
 
        response.mustcontain('Lukasz Balcerzak')
 

	
 
    def test_file_authors_git(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='authors',
 
                                    repo_name=GIT_REPO,
 
                                    revision='master',
 
                                    f_path='vcs/nodes.py',
 
                                    annotate=True))
 
        response.mustcontain('Marcin Kuzminski')
 
        response.mustcontain('Lukasz Balcerzak')
 

	
 
    def test_archival(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for arch_ext, info in ARCHIVE_SPECS.items():
 
            short = '27cd5cce30c9%s' % arch_ext
 
            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
 
            filename = '%s-%s' % (HG_REPO, short)
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 

	
 
            assert response.status == '200 OK'
 
            heads = [
 
                ('Pragma', 'no-cache'),
 
                ('Cache-Control', 'no-cache'),
 
                ('Content-Disposition', 'attachment; filename=%s' % filename),
 
                ('Content-Type', '%s; charset=utf-8' % info[0]),
 
            ]
 
            assert response.response._headers.items() == heads
 

	
 
    def test_archival_wrong_ext(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for arch_ext in ['tar', 'rar', 'x', '..ax', '.zipz']:
 
            fname = '27cd5cce30c96924232dffcd24178a07ffeb5dfc%s' % arch_ext
 

	
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 
            response.mustcontain('Unknown archive type')
 

	
 
    def test_archival_wrong_revision(self):
 
        self.log_user()
 
        _set_downloads(HG_REPO, set_to=True)
 
        for rev in ['00x000000', 'tar', 'wrong', '@##$@$42413232', '232dffcd']:
 
            fname = '%s.zip' % rev
 

	
 
            response = self.app.get(url(controller='files',
 
                                        action='archivefile',
 
                                        repo_name=HG_REPO,
 
                                        fname=fname))
 
            response.mustcontain('Unknown revision')
 

	
 
    #==========================================================================
 
    # RAW FILE
 
    #==========================================================================
 
    def test_raw_file_ok(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        assert response.content_disposition == "attachment; filename=nodes.py"
 
        assert response.content_type == mimetypes.guess_type("nodes.py")[0]
 

	
 
    def test_raw_file_wrong_cs(self):
 
        self.log_user()
 
        rev = u'ERRORce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/nodes.py'
 

	
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = """Such revision does not exist for this repository"""
 
        response.mustcontain(msg)
 

	
 
    def test_raw_file_wrong_f_path(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/ERRORnodes.py'
 
        response = self.app.get(url(controller='files', action='rawfile',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
 
        response.mustcontain(msg)
 

	
 
    #==========================================================================
 
    # RAW RESPONSE - PLAIN
 
    #==========================================================================
 
    def test_raw_ok(self):
 
        self.log_user()
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision='27cd5cce30c96924232dffcd24178a07ffeb5dfc',
 
                                    f_path='vcs/nodes.py'))
 

	
 
        assert response.content_type == "text/plain"
 

	
 
    def test_raw_wrong_cs(self):
 
        self.log_user()
 
        rev = u'ERRORcce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/nodes.py'
 

	
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 

	
 
        msg = """Such revision does not exist for this repository"""
 
        response.mustcontain(msg)
 

	
 
    def test_raw_wrong_f_path(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        f_path = 'vcs/ERRORnodes.py'
 
        response = self.app.get(url(controller='files', action='raw',
 
                                    repo_name=HG_REPO,
 
                                    revision=rev,
 
                                    f_path=f_path), status=404)
 
        msg = "There is no file nor directory at the given path: &#39;%s&#39; at revision %s" % (f_path, rev[:12])
 
        response.mustcontain(msg)
 

	
 
    def test_ajaxed_files_list(self):
 
        self.log_user()
 
        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'
 
        response = self.app.get(
 
            url('files_nodelist_home', repo_name=HG_REPO, f_path='/',
 
                revision=rev),
 
            extra_environ={'HTTP_X_PARTIAL_XHR': '1'},
 
        )
 
        response.mustcontain("vcs/web/simplevcs/views/repository.py")
 

	
 
    # Hg - ADD FILE
 
    def test_add_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'))
 

	
 
    def test_add_file_into_hg_missing_content(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': '',
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No content')
 

	
 
    def test_add_file_into_hg_missing_filename(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No filename')
 

	
 
    @parametrize('location,filename', [
 
        ('/abs', 'foo'),
 
        ('../rel', 'foo'),
 
        ('file/../foo', 'foo'),
 
    ])
 
    def test_add_file_into_hg_bad_filenames(self, location, filename):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
 

	
 
    @parametrize('cnt,location,filename', [
 
        (1, '', 'foo.txt'),
 
        (2, 'dir', 'foo.rst'),
 
        (3, 'rel/dir', 'foo.bar'),
 
    ])
 
    def test_add_file_into_hg(self, cnt, location, filename):
 
        self.log_user()
 
        repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='hg')
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - add file
 
    def test_add_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'))
 

	
 
    def test_add_file_into_git_missing_content(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                     'content': '',
 
                                     '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        self.checkSessionFlash(response, 'No content')
 

	
 
    def test_add_file_into_git_missing_filename(self):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'No filename')
 

	
 
    @parametrize('location,filename', [
 
        ('/abs', 'foo'),
 
        ('../rel', 'foo'),
 
        ('file/../foo', 'foo'),
 
    ])
 
    def test_add_file_into_git_bad_filenames(self, location, filename):
 
        self.log_user()
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 

	
 
        self.checkSessionFlash(response, 'Location must be relative path and must not contain .. in path')
 

	
 
    @parametrize('cnt,location,filename', [
 
        (1, '', 'foo.txt'),
 
        (2, 'dir', 'foo.rst'),
 
        (3, 'rel/dir', 'foo.bar'),
 
    ])
 
    def test_add_file_into_git(self, cnt, location, filename):
 
        self.log_user()
 
        repo = fixture.create_repo(u'commit-test-%s' % cnt, repo_type='git')
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "foo",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Hg - EDIT
 
    def test_edit_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_edit_home',
 
                                      repo_name=HG_REPO,
 
                                      revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_edit_file_view_not_on_branch_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only edit files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_edit_file_view_commit_changes_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'content': "def py():\n print 'hello world'\n",
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - edit
 
    def test_edit_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_edit_home',
 
                                      repo_name=GIT_REPO,
 
                                      revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_edit_file_view_not_on_branch_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only edit files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_edit_file_view_commit_changes_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-edit-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_edit_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'content': "def py():\n print 'hello world'\n",
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Hg - delete
 
    def test_delete_file_view_hg(self):
 
        self.log_user()
 
        response = self.app.get(url('files_delete_home',
 
                                     repo_name=HG_REPO,
 
                                     revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_delete_file_view_not_on_branch_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only delete files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_delete_file_view_commit_changes_hg(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='hg')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                                   'Successfully deleted file %s' % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    # Git - delete
 
    def test_delete_file_view_git(self):
 
        self.log_user()
 
        response = self.app.get(url('files_delete_home',
 
                                     repo_name=HG_REPO,
 
                                     revision='tip', f_path='vcs/nodes.py'))
 

	
 
    def test_delete_file_view_not_on_branch_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip', f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.get(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision='tip', f_path=posixpath.join(location, filename)),
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                'You can only delete files with revision being a valid branch')
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
 

	
 
    def test_delete_file_view_commit_changes_git(self):
 
        self.log_user()
 
        repo = fixture.create_repo(u'test-delete-repo', repo_type='git')
 

	
 
        ## add file
 
        location = 'vcs'
 
        filename = 'nodes.py'
 
        response = self.app.post(url('files_add_home',
 
                                      repo_name=repo.repo_name,
 
                                      revision='tip',
 
                                      f_path='/'),
 
                                 params={
 
                                    'content': "def py():\n print 'hello'\n",
 
                                    'filename': filename,
 
                                    'location': location,
 
                                    '_authentication_token': self.authentication_token(),
 
                                 },
 
                                 status=302)
 
        response.follow()
 
        try:
 
            self.checkSessionFlash(response, 'Successfully committed to %s'
 
                                   % posixpath.join(location, filename))
 
            response = self.app.post(url('files_delete_home',
 
                                          repo_name=repo.repo_name,
 
                                          revision=repo.scm_instance.DEFAULT_BRANCH_NAME,
 
                                          f_path=posixpath.join(location, filename)),
 
                                     params={
 
                                        'message': 'i committed',
 
                                        '_authentication_token': self.authentication_token(),
 
                                     },
 
                                    status=302)
 
            self.checkSessionFlash(response,
 
                                   'Successfully deleted file %s' % posixpath.join(location, filename))
 
        finally:
 
            fixture.destroy_repo(repo.repo_name)
0 comments (0 inline, 0 general)