Changeset - d21c762f5c4b
[Not reviewed]
beta
0 2 0
Marcin Kuzminski - 13 years ago 2013-02-19 20:35:25
marcin@python-works.com
revision extraction function shouldn't be so eager, just extract commits that are
surrounded with spaces, or they are at beginning of the whole string
- wrote tests for that
- used parametrized for other tests
2 files changed with 63 insertions and 18 deletions:
0 comments (0 inline, 0 general)
rhodecode/lib/helpers.py
Show inline comments
 
@@ -615,561 +615,566 @@ def action_parser(user_log, feed=False, 
 
                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
 
        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') % pull_request_id,
 
                    url('pullrequest_show', repo_name=repo_name,
 
                    pull_request_id=pull_request_id))
 

	
 
    # action : translated str, callback(extractor), icon
 
    action_map = {
 
    'user_deleted_repo':           (_('[deleted] repository'),
 
                                    None, 'database_delete.png'),
 
    'user_created_repo':           (_('[created] repository'),
 
                                    None, 'database_add.png'),
 
    'user_created_fork':           (_('[created] repository as fork'),
 
                                    None, 'arrow_divide.png'),
 
    'user_forked_repo':            (_('[forked] repository'),
 
                                    get_fork_name, 'arrow_divide.png'),
 
    'user_updated_repo':           (_('[updated] repository'),
 
                                    None, 'database_edit.png'),
 
    'admin_deleted_repo':          (_('[delete] repository'),
 
                                    None, 'database_delete.png'),
 
    'admin_created_repo':          (_('[created] repository'),
 
                                    None, 'database_add.png'),
 
    'admin_forked_repo':           (_('[forked] repository'),
 
                                    None, 'arrow_divide.png'),
 
    'admin_updated_repo':          (_('[updated] repository'),
 
                                    None, 'database_edit.png'),
 
    'admin_created_user':          (_('[created] user'),
 
                                    get_user_name, 'user_add.png'),
 
    'admin_updated_user':          (_('[updated] user'),
 
                                    get_user_name, 'user_edit.png'),
 
    'admin_created_users_group':   (_('[created] users group'),
 
                                    get_users_group, 'group_add.png'),
 
    'admin_updated_users_group':   (_('[updated] users group'),
 
                                    get_users_group, 'group_edit.png'),
 
    'user_commented_revision':     (_('[commented] on revision in repository'),
 
                                    get_cs_links, 'comment_add.png'),
 
    'user_commented_pull_request': (_('[commented] on pull request for'),
 
                                    get_pull_request, 'comment_add.png'),
 
    'user_closed_pull_request':    (_('[closed] pull request for'),
 
                                    get_pull_request, 'tick.png'),
 
    'push':                        (_('[pushed] into'),
 
                                    get_cs_links, 'script_add.png'),
 
    'push_local':                  (_('[committed via RhodeCode] into repository'),
 
                                    get_cs_links, 'script_edit.png'),
 
    'push_remote':                 (_('[pulled from remote] into repository'),
 
                                    get_cs_links, 'connect.png'),
 
    'pull':                        (_('[pulled] from'),
 
                                    None, 'down_16.png'),
 
    'started_following_repo':      (_('[started following] repository'),
 
                                    None, 'heart_add.png'),
 
    'stopped_following_repo':      (_('[stopped following] repository'),
 
                                    None, 'heart_delete.png'),
 
    }
 

	
 
    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 = """<img src="%s%s" alt="%s"/>"""
 
        ico = action_map.get(action, ['', '', ''])[2]
 
        return literal(tmpl % ((url('/images/icons/')), ico, action))
 

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

	
 

	
 

	
 
#==============================================================================
 
# PERMS
 
#==============================================================================
 
from rhodecode.lib.auth import HasPermissionAny, HasPermissionAll, \
 
HasRepoPermissionAny, HasRepoPermissionAll, HasReposGroupPermissionAll, \
 
HasReposGroupPermissionAny
 

	
 

	
 
#==============================================================================
 
# GRAVATAR URL
 
#==============================================================================
 

	
 
def gravatar_url(email_address, size=30):
 
    from pylons import url  # doh, we need to re-import url to mock it later
 
    _def = 'anonymous@rhodecode.org'
 
    use_gravatar = str2bool(config['app_conf'].get('use_gravatar'))
 
    email_address = email_address or _def
 
    if (not use_gravatar or not email_address or email_address == _def):
 
        f = lambda a, l: min(l, key=lambda x: abs(x - a))
 
        return url("/images/user%s.png" % f(size, [14, 16, 20, 24, 30]))
 

	
 
    if use_gravatar and config['app_conf'].get('alternative_gravatar_url'):
 
        tmpl = config['app_conf'].get('alternative_gravatar_url', '')
 
        parsed_url = urlparse.urlparse(url.current(qualified=True))
 
        tmpl = tmpl.replace('{email}', email_address)\
 
                   .replace('{md5email}', hashlib.md5(email_address.lower()).hexdigest()) \
 
                   .replace('{netloc}', parsed_url.netloc)\
 
                   .replace('{scheme}', parsed_url.scheme)\
 
                   .replace('{size}', str(size))
 
        return tmpl
 

	
 
    ssl_enabled = 'https' == request.environ.get('wsgi.url_scheme')
 
    default = 'identicon'
 
    baseurl_nossl = "http://www.gravatar.com/avatar/"
 
    baseurl_ssl = "https://secure.gravatar.com/avatar/"
 
    baseurl = baseurl_ssl if ssl_enabled else baseurl_nossl
 

	
 
    if isinstance(email_address, unicode):
 
        #hashlib crashes on unicode items
 
        email_address = safe_str(email_address)
 
    # construct the url
 
    gravatar_url = baseurl + hashlib.md5(email_address.lower()).hexdigest() + "?"
 
    gravatar_url += urllib.urlencode({'d': default, 's': str(size)})
 

	
 
    return gravatar_url
 

	
 

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

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

	
 
        """Create a "RepoPage" instance. special pager for paging
 
        repository
 
        """
 
        self._url_generator = url
 

	
 
        # 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, last_url=None):
 
    """
 
    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, repo_name = groups_and_repos
 
    last_link = link_to(repo_name, last_url) if last_url else repo_name
 

	
 
    if not groups:
 
        if last_url:
 
            return last_link
 
        return repo_name
 
    else:
 
        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) + [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
 
    """
 
    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[0], stats[1]
 
    width = 100
 

	
 
    if a == 'b':
 
        #binary mode
 
        b_d = '<div class="bin%s %s" style="width:100%%">%s</div>' % (d, cgen('a', a_v='', d_v=0), 'bin')
 
        b_a = '<div class="bin1" style="width:0%%">%s</div>' % ('bin')
 
        return literal('<div style="width:%spx">%s%s</div>' % (width, b_a, b_d))
 

	
 
    t = stats[0] + stats[1]
 
    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))
 

	
 

	
 
def urlify_text(text_):
 

	
 
    url_pat = re.compile(r'''(http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]'''
 
                         '''|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+)''')
 

	
 
    def url_func(match_obj):
 
        url_full = match_obj.groups()[0]
 
        return '<a href="%(url)s">%(url)s</a>' % ({'url': url_full})
 

	
 
    return literal(url_pat.sub(url_func, text_))
 

	
 

	
 
def urlify_changesets(text_, repository):
 
    """
 
    Extract revision ids from changeset and make link from them
 

	
 
    :param text_:
 
    :param repository:
 
    """
 

	
 
    URL_PAT = re.compile(r'([0-9a-fA-F]{12,})')
 
    from pylons import url  # doh, we need to re-import url to mock it later
 
    URL_PAT = re.compile(r'(?:^|\s)([0-9a-fA-F]{12,40})(?:$|\s)')
 

	
 
    def url_func(match_obj):
 
        rev = match_obj.groups()[0]
 
        pref = ''
 
        suf = ''
 
        if match_obj.group().startswith(' '):
 
            pref = ' '
 
        if match_obj.group().endswith(' '):
 
            suf = ' '
 
        tmpl = (
 
        '%(pref)s<a class="%(cls)s" href="%(url)s">'
 
        '%(rev)s'
 
        '</a>'
 
        '%(suf)s'
 
        )
 
        return tmpl % {
 
         'pref': pref,
 
         'cls': 'revision-link',
 
         'url': url('changeset_home', repo_name=repository, revision=rev),
 
         'rev': rev,
 
         'suf': suf
 
        }
 

	
 
    newtext = URL_PAT.sub(url_func, text_)
 

	
 
    return newtext
 

	
 

	
 
def urlify_commit(text_, repository=None, link_=None):
 
    """
 
    Parses given text message and makes proper links.
 
    issues are linked to given issue-server, and rest is a changeset link
 
    if link_ is given, in other case it's a plain text
 

	
 
    :param text_:
 
    :param repository:
 
    :param link_: changeset link
 
    """
 
    import traceback
 

	
 
    def escaper(string):
 
        return string.replace('<', '&lt;').replace('>', '&gt;')
 

	
 
    def linkify_others(t, l):
 
        urls = re.compile(r'(\<a.*?\<\/a\>)',)
 
        links = []
 
        for e in urls.split(t):
 
            if not urls.match(e):
 
                links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
 
            else:
 
                links.append(e)
 

	
 
        return ''.join(links)
 

	
 
    # urlify changesets - extrac revisions and make link out of them
 
    newtext = urlify_changesets(escaper(text_), repository)
 

	
 
    try:
 
        from rhodecode import CONFIG
 
        conf = CONFIG
 

	
 
        # 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
 
        ]
 

	
 
        log.debug('found issue server suffixes `%s` during valuation of: %s'
 
                  % (','.join(valid_indices), newtext))
 

	
 
        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)
 

	
 
            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(r'%s' % ISSUE_PATTERN)
 

	
 
            def url_func(match_obj):
 
                pref = ''
 
                if match_obj.group().startswith(' '):
 
                    pref = ' '
 

	
 
                issue_id = ''.join(match_obj.groups())
 
                tmpl = (
 
                '%(pref)s<a class="%(cls)s" href="%(url)s">'
 
                '%(issue-prefix)s%(id-repr)s'
 
                '</a>'
 
                )
 
                url = ISSUE_SERVER_LNK.replace('{id}', issue_id)
 
                if repository:
 
                    url = url.replace('{repo}', repository)
 
                    repo_name = repository.split(URL_SEP)[-1]
 
                    url = url.replace('{repo_name}', repo_name)
 

	
 
                return tmpl % {
 
                     'pref': pref,
 
                     'cls': 'issue-tracker-link',
 
                     'url': url,
 
                     'id-repr': issue_id,
 
                     'issue-prefix': ISSUE_PREFIX,
 
                     'serv': ISSUE_SERVER_LNK,
 
                }
 
            newtext = URL_PAT.sub(url_func, newtext)
 
            log.debug('processed prefix:`%s` => %s' % (pattern_index, newtext))
 

	
 
        # if we actually did something above
 
        if link_:
 
            # wrap not links into final link => link_
 
            newtext = linkify_others(newtext, link_)
 
    except:
 
        log.error(traceback.format_exc())
 
        pass
 

	
 
    return literal(newtext)
 

	
 

	
 
def rst(source):
 
    return literal('<div class="rst-block">%s</div>' %
 
                   MarkupRenderer.rst(source))
 

	
 

	
 
def rst_w_mentions(source):
 
    """
 
    Wrapped rst renderer with @mention highlighting
 

	
 
    :param source:
 
    """
 
    return literal('<div class="rst-block">%s</div>' %
 
                   MarkupRenderer.rst_with_mentions(source))
 

	
 

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

	
 

	
 
def changeset_status_lbl(changeset_status):
 
    return dict(ChangesetStatus.STATUSES).get(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:marcin
 
            action:*push*
 
            ip:127.0.0.1
 
            date:20120101
 
            date:[20120101100000 TO 20120102]
 

	
 
        Generate wildcards using '*' character:
 
            "repositroy: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 rhodecode.model.db import UserIpMap
 
    s, e = UserIpMap._get_ip_range(ip_addr)
 
    return '%s - %s' % (s, e)
rhodecode/tests/test_libs.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
"""
 
    rhodecode.tests.test_libs
 
    ~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 

	
 
    Package for testing various lib/helper functions in rhodecode
 

	
 
    :created_on: Jun 9, 2011
 
    :copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com>
 
    :license: GPLv3, see COPYING for more details.
 
"""
 
# 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/>.
 
from __future__ import with_statement
 
import unittest
 
import datetime
 
import hashlib
 
import mock
 
from rhodecode.tests import *
 

	
 
proto = 'http'
 
TEST_URLS = [
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
 
     '%s://127.0.0.1:8080' % proto),
 
    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
 
     '%s://domain.org' % proto),
 
    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',
 
                                                '8080'],
 
     '%s://domain.org:8080' % proto),
 
]
 

	
 
proto = 'https'
 
TEST_URLS += [
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
 
     '%s://127.0.0.1:8080' % proto),
 
    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],
 
     '%s://domain.org' % proto),
 
    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',
 
                                                '8080'],
 
     '%s://domain.org:8080' % proto),
 
]
 

	
 

	
 
class TestLibs(unittest.TestCase):
 

	
 
    def test_uri_filter(self):
 
    @parameterized.expand(TEST_URLS)
 
    def test_uri_filter(self, test_url, expected, expected_creds):
 
        from rhodecode.lib.utils2 import uri_filter
 
        self.assertEqual(uri_filter(test_url), expected)
 

	
 
        for url in TEST_URLS:
 
            self.assertEqual(uri_filter(url[0]), url[1])
 

	
 
    def test_credentials_filter(self):
 
    @parameterized.expand(TEST_URLS)
 
    def test_credentials_filter(self, test_url, expected, expected_creds):
 
        from rhodecode.lib.utils2 import credentials_filter
 

	
 
        for url in TEST_URLS:
 
            self.assertEqual(credentials_filter(url[0]), url[2])
 
        self.assertEqual(credentials_filter(test_url), expected_creds)
 

	
 
    def test_str2bool(self):
 
        from rhodecode.lib.utils2 import str2bool
 
        test_cases = [
 
            ('t', True),
 
    @parameterized.expand([('t', True),
 
            ('true', True),
 
            ('y', True),
 
            ('yes', True),
 
            ('on', True),
 
            ('1', True),
 
            ('Y', True),
 
            ('yeS', True),
 
            ('Y', True),
 
            ('TRUE', True),
 
            ('T', True),
 
            ('False', False),
 
            ('F', False),
 
            ('FALSE', False),
 
            ('0', False),
 
            ('-1', False),
 
            ('', False), ]
 

	
 
        for case in test_cases:
 
            self.assertEqual(str2bool(case[0]), case[1])
 
                           ('', False)
 
    ])
 
    def test_str2bool(self, str_bool, expected):
 
        from rhodecode.lib.utils2 import str2bool
 
        self.assertEqual(str2bool(str_bool), expected)
 

	
 
    def test_mention_extractor(self):
 
        from rhodecode.lib.utils2 import extract_mentioned_users
 
        sample = (
 
            "@first hi there @marcink here's my email marcin@email.com "
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
 
            "@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl "
 
            "@marian.user just do it @marco-polo and next extract @marco_polo "
 
            "user.dot  hej ! not-needed maril@domain.org"
 
        )
 

	
 
        s = sorted([
 
        'first', 'marcink', 'lukaszb', 'one_more22', 'MARCIN', 'maRCiN', 'john',
 
        'marian.user', 'marco-polo', 'marco_polo'
 
        ], key=lambda k: k.lower())
 
        self.assertEqual(s, extract_mentioned_users(sample))
 

	
 
    def test_age(self):
 
        from rhodecode.lib.utils2 import age
 
        from dateutil import relativedelta
 
        n = datetime.datetime.now()
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 

	
 
        self.assertEqual(age(n), u'just now')
 
        self.assertEqual(age(n + delt(seconds=-1)), u'1 second ago')
 
        self.assertEqual(age(n + delt(seconds=-60 * 2)), u'2 minutes ago')
 
        self.assertEqual(age(n + delt(hours=-1)), u'1 hour ago')
 
        self.assertEqual(age(n + delt(hours=-24)), u'1 day ago')
 
        self.assertEqual(age(n + delt(hours=-24 * 5)), u'5 days ago')
 
        self.assertEqual(age(n + delt(months=-1)), u'1 month ago')
 
        self.assertEqual(age(n + delt(months=-1, days=-2)), u'1 month and 2 days ago')
 
        self.assertEqual(age(n + delt(years=-1, months=-1)), u'1 year and 1 month ago')
 

	
 
    def test_age_in_future(self):
 
        from rhodecode.lib.utils2 import age
 
        from dateutil import relativedelta
 
        n = datetime.datetime.now()
 
        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 

	
 
        self.assertEqual(age(n), u'just now')
 
        self.assertEqual(age(n + delt(seconds=1)), u'in 1 second')
 
        self.assertEqual(age(n + delt(seconds=60 * 2)), u'in 2 minutes')
 
        self.assertEqual(age(n + delt(hours=1)), u'in 1 hour')
 
        self.assertEqual(age(n + delt(hours=24)), u'in 1 day')
 
        self.assertEqual(age(n + delt(hours=24 * 5)), u'in 5 days')
 
        self.assertEqual(age(n + delt(months=1)), u'in 1 month')
 
        self.assertEqual(age(n + delt(months=1, days=1)), u'in 1 month and 1 day')
 
        self.assertEqual(age(n + delt(years=1, months=1)), u'in 1 year and 1 month')
 

	
 
    def test_tag_exctrator(self):
 
        sample = (
 
            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
 
            "[requires] [stale] [see<>=>] [see => http://url.com]"
 
            "[requires => url] [lang => python] [just a tag]"
 
            "[,d] [ => ULR ] [obsolete] [desc]]"
 
        )
 
        from rhodecode.lib.helpers import desc_stylize
 
        res = desc_stylize(sample)
 
        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)
 
        self.assertTrue('<div class="metatag" tag="obsolete">obsolete</div>' in res)
 
        self.assertTrue('<div class="metatag" tag="stale">stale</div>' in res)
 
        self.assertTrue('<div class="metatag" tag="lang">python</div>' in res)
 
        self.assertTrue('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res)
 
        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)
 

	
 
    def test_alternative_gravatar(self):
 
        from rhodecode.lib.helpers import gravatar_url
 
        _md5 = lambda s: hashlib.md5(s).hexdigest()
 

	
 
        def fake_conf(**kwargs):
 
            from pylons import config
 
            config['app_conf'] = {}
 
            config['app_conf']['use_gravatar'] = True
 
            config['app_conf'].update(kwargs)
 
            return config
 

	
 
        class fake_url():
 
            @classmethod
 
            def current(cls, *args, **kwargs):
 
                return 'https://server.com'
 

	
 
        with mock.patch('pylons.url', fake_url):
 
            fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')
 
            with mock.patch('pylons.config', fake):
 
                    from pylons import url
 
                    assert url.current() == 'https://server.com'
 
                    grav = gravatar_url(email_address='test@foo.com', size=24)
 
                    assert grav == 'http://test.com/test@foo.com'
 

	
 
            fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')
 
            with mock.patch('pylons.config', fake):
 
                grav = gravatar_url(email_address='test@foo.com', size=24)
 
                assert grav == 'http://test.com/test@foo.com'
 

	
 
            fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}')
 
            with mock.patch('pylons.config', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://test.com/%s' % (_md5(em))
 

	
 
            fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}/{size}')
 
            with mock.patch('pylons.config', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)
 

	
 
            fake = fake_conf(alternative_gravatar_url='{scheme}://{netloc}/{md5email}/{size}')
 
            with mock.patch('pylons.config', fake):
 
                em = 'test@foo.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)
 

	
 
    @parameterized.expand([
 
      ("",
 
       ""),
 
      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
 
       "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),
 
      ("from rev 000000000000",
 
       "from rev url[000000000000]"),
 
      ("from rev 000000000000123123 also rev 000000000000",
 
       "from rev url[000000000000123123] also rev url[000000000000]"),
 
      ("this should-000 00",
 
       "this should-000 00"),
 
      ("longtextffffffffff rev 123123123123",
 
       "longtextffffffffff rev url[123123123123]"),
 
      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
 
       "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),
 
      ("ffffffffffff some text traalaa",
 
       "url[ffffffffffff] some text traalaa"),
 
       ("""Multi line
 
       123123123123 
 
       some text 123123123123""",
 
       """Multi line
 
       url[123123123123] 
 
       some text url[123123123123]""")
 
    ])
 
    def test_urlify_changesets(self, sample, expected):
 
        import re
 

	
 
        def fake_url(self, *args, **kwargs):
 
            return '/some-url'
 

	
 
        #quickly change expected url[] into a link
 
        URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')
 

	
 
        def url_func(match_obj):
 
            _url = match_obj.groups()[0]
 
            tmpl = """<a class="revision-link" href="/some-url">%s</a>"""
 
            return tmpl % _url
 

	
 
        expected = URL_PAT.sub(url_func, expected)
 

	
 
        with mock.patch('pylons.url', fake_url):
 
            from rhodecode.lib.helpers import urlify_changesets
 
            self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)
0 comments (0 inline, 0 general)