Changeset - e5a7f8f41370
[Not reviewed]
default
0 3 0
Thomas De Schampheleire - 8 years ago 2018-02-16 22:30:51
thomas.de_schampheleire@nokia.com
issues: backout special whitespace handling

This is essentially a backout of commit 32e1e0745d3c.

That commit checked for whitespace at the beginning of the matched issue
reference, and explicitly retained it in the resulting link text.

The way this was handled is not only suboptimal, e.g. a set of 4 spaces
would still be reduced to 1, but is also not actually necessary: if
whitespace before the issue reference is not required, then it does not need
to be specified in the issue pattern, and if it _is_ required, then a
positive lookbehind assertion can be used instead.
3 files changed with 20 insertions and 7 deletions:
0 comments (0 inline, 0 general)
kallithea/lib/helpers.py
Show inline comments
 
@@ -391,884 +391,882 @@ class _Message(object):
 
    Converting the message to a string returns the message text. Instances
 
    also have the following attributes:
 

	
 
    * ``message``: the message text.
 
    * ``category``: the category specified when the message was created.
 
    """
 

	
 
    def __init__(self, category, message):
 
        self.category = category
 
        self.message = message
 

	
 
    def __str__(self):
 
        return self.message
 

	
 
    __unicode__ = __str__
 

	
 
    def __html__(self):
 
        return escape(safe_unicode(self.message))
 

	
 

	
 
class Flash(_Flash):
 

	
 
    def __call__(self, message, category=None, ignore_duplicate=False, logf=None):
 
        """
 
        Show a message to the user _and_ log it through the specified function
 

	
 
        category: notice (default), warning, error, success
 
        logf: a custom log function - such as log.debug
 

	
 
        logf defaults to log.info, unless category equals 'success', in which
 
        case logf defaults to log.debug.
 
        """
 
        if logf is None:
 
            logf = log.info
 
            if category == 'success':
 
                logf = log.debug
 

	
 
        logf('Flash %s: %s', category, message)
 

	
 
        super(Flash, self).__call__(message, category, ignore_duplicate)
 

	
 
    def pop_messages(self):
 
        """Return all accumulated messages and delete them from the session.
 

	
 
        The return value is a list of ``Message`` objects.
 
        """
 
        from tg import session
 
        messages = session.pop(self.session_key, [])
 
        session.save()
 
        return [_Message(*m) for m in messages]
 

	
 

	
 
flash = Flash()
 

	
 
#==============================================================================
 
# SCM FILTERS available via h.
 
#==============================================================================
 
from kallithea.lib.vcs.utils import author_name, author_email
 
from kallithea.lib.utils2 import credentials_filter, age as _age
 

	
 
age = lambda x, y=False: _age(x, y)
 
capitalize = lambda x: x.capitalize()
 
email = author_email
 
short_id = lambda x: x[:12]
 
hide_credentials = lambda x: ''.join(credentials_filter(x))
 

	
 

	
 
def show_id(cs):
 
    """
 
    Configurable function that shows ID
 
    by default it's r123:fffeeefffeee
 

	
 
    :param cs: changeset instance
 
    """
 
    from kallithea import CONFIG
 
    def_len = safe_int(CONFIG.get('show_sha_length', 12))
 
    show_rev = str2bool(CONFIG.get('show_revision_number', False))
 

	
 
    raw_id = cs.raw_id[:def_len]
 
    if show_rev:
 
        return 'r%s:%s' % (cs.revision, raw_id)
 
    else:
 
        return raw_id
 

	
 

	
 
def fmt_date(date):
 
    if date:
 
        return date.strftime("%Y-%m-%d %H:%M:%S").decode('utf8')
 

	
 
    return ""
 

	
 

	
 
def is_git(repository):
 
    if hasattr(repository, 'alias'):
 
        _type = repository.alias
 
    elif hasattr(repository, 'repo_type'):
 
        _type = repository.repo_type
 
    else:
 
        _type = repository
 
    return _type == 'git'
 

	
 

	
 
def is_hg(repository):
 
    if hasattr(repository, 'alias'):
 
        _type = repository.alias
 
    elif hasattr(repository, 'repo_type'):
 
        _type = repository.repo_type
 
    else:
 
        _type = repository
 
    return _type == 'hg'
 

	
 

	
 
@cache_region('long_term', 'user_or_none')
 
def user_or_none(author):
 
    """Try to match email part of VCS committer string with a local user - or return None"""
 
    from kallithea.model.db import User
 
    email = author_email(author)
 
    if email:
 
        return User.get_by_email(email, cache=True) # cache will only use sql_cache_short
 
    return None
 

	
 

	
 
def email_or_none(author):
 
    """Try to match email part of VCS committer string with a local user.
 
    Return primary email of user, email part of the specified author name, or None."""
 
    if not author:
 
        return None
 
    user = user_or_none(author)
 
    if user is not None:
 
        return user.email # always use main email address - not necessarily the one used to find user
 

	
 
    # extract email from the commit string
 
    email = author_email(author)
 
    if email:
 
        return email
 

	
 
    # No valid email, not a valid user in the system, none!
 
    return None
 

	
 

	
 
def person(author, show_attr="username"):
 
    """Find the user identified by 'author', return one of the users attributes,
 
    default to the username attribute, None if there is no user"""
 
    from kallithea.model.db import User
 
    # attr to return from fetched user
 
    person_getter = lambda usr: getattr(usr, show_attr)
 

	
 
    # if author is already an instance use it for extraction
 
    if isinstance(author, User):
 
        return person_getter(author)
 

	
 
    user = user_or_none(author)
 
    if user is not None:
 
        return person_getter(user)
 

	
 
    # Still nothing?  Just pass back the author name if any, else the email
 
    return author_name(author) or email(author)
 

	
 

	
 
def person_by_id(id_, show_attr="username"):
 
    from kallithea.model.db import User
 
    # attr to return from fetched user
 
    person_getter = lambda usr: getattr(usr, show_attr)
 

	
 
    # maybe it's an ID ?
 
    if str(id_).isdigit() or isinstance(id_, int):
 
        id_ = int(id_)
 
        user = User.get(id_)
 
        if user is not None:
 
            return person_getter(user)
 
    return id_
 

	
 

	
 
def boolicon(value):
 
    """Returns boolean value of a value, represented as small html image of true/false
 
    icons
 

	
 
    :param value: value
 
    """
 

	
 
    if value:
 
        return HTML.tag('i', class_="icon-ok")
 
    else:
 
        return HTML.tag('i', class_="icon-minus-circled")
 

	
 

	
 
def action_parser(user_log, feed=False, parse_cs=False):
 
    """
 
    This helper will action_map the specified string action into translated
 
    fancy names with icons and links
 

	
 
    :param user_log: user log instance
 
    :param feed: use output for feeds (no html and fancy icons)
 
    :param parse_cs: parse Changesets into VCS instances
 
    """
 

	
 
    action = user_log.action
 
    action_params = ' '
 

	
 
    x = action.split(':')
 

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

	
 
    def get_cs_links():
 
        revs_limit = 3  # display this amount always
 
        revs_top_limit = 50  # show upto this amount of changesets hidden
 
        revs_ids = action_params.split(',')
 
        deleted = user_log.repository is None
 
        if deleted:
 
            return ','.join(revs_ids)
 

	
 
        repo_name = user_log.repository.repo_name
 

	
 
        def lnk(rev, repo_name):
 
            lazy_cs = False
 
            title_ = None
 
            url_ = '#'
 
            if isinstance(rev, BaseChangeset) or isinstance(rev, AttributeDict):
 
                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 %s not found') % lbl
 
            if parse_cs:
 
                return link_to(lbl, url_, title=title_, **{'data-toggle': 'tooltip'})
 
            return link_to(lbl, url_, class_='lazy-cs' if lazy_cs else '',
 
                           **{'data-raw_id': rev.raw_id, 'data-repo_name': repo_name})
 

	
 
        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" data-toggle="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():
 
        from kallithea.model.db import PullRequest
 
        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('[', '<b>') \
 
            .replace(']', '</b>')
 

	
 
    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
 

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

	
 
    # 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, \
 
    HasRepoPermissionLevel, HasRepoGroupPermissionLevel
 

	
 

	
 
#==============================================================================
 
# GRAVATAR URL
 
#==============================================================================
 
def gravatar_div(email_address, cls='', size=30, **div_attributes):
 
    """Return an html literal with a span around a gravatar if they are enabled.
 
    Extra keyword parameters starting with 'div_' will get the prefix removed
 
    and '_' changed to '-' and be used as attributes on the div. The default
 
    class is 'gravatar'.
 
    """
 
    from tg 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:].replace('_', '-'), escape(v)))
 
    return literal("""<span%s>%s</span>""" %
 
                   (''.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 tg 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 = ('<i class="icon-gravatar {cls}"'
 
                ' style="font-size: {size}px;background-size: {size}px;background-image: url(\'{src}\')"'
 
                '></i>').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 kallithea.model.db import User
 
    from tg 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
 

	
 

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

	
 
    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 progress-bar" style="width:100%%">%s</div>' % (bin_op, lbl)
 
        b_a = '<div class="bin bin1" style="width:0%"></div>'
 
        return literal('<div style="width:%spx" class="progress">%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 progress-bar" style="width:%s%%">%s</div>' % (
 
        a_p, a_v
 
    )
 
    d_d = '<div class="deleted progress-bar" style="width:%s%%">%s</div>' % (
 
        d_p, d_v
 
    )
 
    return literal('<div class="progress" 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="changeset_hash" href="%(url)s">%(hash)s</a>' % {
 
                 'url': url('changeset_home', repo_name=repo_name, revision=hash_),
 
                 'hash': hash_,
 
                }
 
        bold = match_obj.group('bold')
 
        if bold is not None:
 
            return '<b>*%s*</b>' % _urlify(bold[1:-1])
 
        if stylize:
 
            seen = match_obj.group('seen')
 
            if seen:
 
                return '<div class="label label-meta" data-tag="see">see =&gt; %s</div>' % seen
 
            license = match_obj.group('license')
 
            if license:
 
                return '<div class="label label-meta" data-tag="license"><a href="http:\/\/www.opensource.org/licenses/%s">%s</a></div>' % (license, license)
 
            tagtype = match_obj.group('tagtype')
 
            if tagtype:
 
                tagvalue = match_obj.group('tagvalue')
 
                return '<div class="label label-meta" data-tag="%s">%s =&gt; <a href="/%s">%s</a></div>' % (tagtype, tagtype, tagvalue, tagvalue)
 
            lang = match_obj.group('lang')
 
            if lang:
 
                return '<div class="label label-meta" data-tag="lang">%s</div>' % lang
 
            tag = match_obj.group('tag')
 
            if tag:
 
                return '<div class="label label-meta" data-tag="%s">%s</div>' % (tag, tag)
 
        return match_obj.group(0)
 

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

	
 
    if truncate is None:
 
        s = s.rstrip()
 
    else:
 
        s = truncatef(s, truncate, whole_word=True)
 
    s = html_escape(s)
 
    s = _urlify(s)
 
    if repo_name is not None:
 
        s = urlify_issues(s, repo_name)
 
    if link_ is not None:
 
        # make href around everything that isn't a href already
 
        s = linkify_others(s, link_)
 
    s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')
 
    # Turn HTML5 into more valid HTML4 as required by some mail readers.
 
    # (This is not done in one step in html_escape, because character codes like
 
    # &#123; risk to be seen as an issue reference due to the presence of '#'.)
 
    s = s.replace("&apos;", "&#39;")
 
    return literal(s)
 

	
 

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

	
 
    return ''.join(links)
 

	
 

	
 
# Global variable that will hold the actual urlify_issues function body.
 
# Will be set on first use when the global configuration has been read.
 
_urlify_issues_f = None
 

	
 

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

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

	
 
        issue_pat_re = re.compile(r'issue_pat(.*)')
 
        for k in CONFIG.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 not issue_pat or not issue_server_link or issue_prefix is None: # issue_prefix can be empty but should be present
 
                log.error('skipping incomplete issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_prefix)
 
                continue
 

	
 
            # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound
 
            try:
 
                issue_re = re.compile(issue_pat)
 
            except re.error as e:
 
                log.error('skipping invalid issue pattern %r: %r -> %r %r. Error: %s', suffix, issue_pat, issue_server_link, issue_prefix, str(e))
 
                continue
 

	
 
            log.debug('issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_prefix)
 

	
 
            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">'
 
                    '<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)))
 

	
 
        # Set tmp function globally - atomically
 
        _urlify_issues_f = tmp_urlify_issues_f
 

	
 
    return _urlify_issues_f(newtext)
 

	
 

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

	
 

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

	
 

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

	
 

	
 
def changeset_status(repo, revision):
 
    from kallithea.model.changeset_status import ChangesetStatusModel
 
    return ChangesetStatusModel().get_status(repo, revision)
 

	
 

	
 
def changeset_status_lbl(changeset_status):
 
    from kallithea.model.db import ChangesetStatus
 
    return ChangesetStatus.get_status_lbl(changeset_status)
 

	
 

	
 
def get_permission_name(key):
 
    from kallithea.model.db import Permission
 
    return dict(Permission.PERMS).get(key)
 

	
 

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

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

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

	
 

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

	
 

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

	
 

	
 
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/lib/paster_commands/template.ini.mako
Show inline comments
 
## -*- coding: utf-8 -*-
 
<%text>################################################################################</%text>
 
<%text>################################################################################</%text>
 
# Kallithea - config file generated with kallithea-config                      #
 
#                                                                              #
 
# The %(here)s variable will be replaced with the parent directory of this file#
 
<%text>################################################################################</%text>
 
<%text>################################################################################</%text>
 

	
 
[DEFAULT]
 

	
 
<%text>################################################################################</%text>
 
<%text>## Email settings                                                             ##</%text>
 
<%text>##                                                                            ##</%text>
 
<%text>## Refer to the documentation ("Email settings") for more details.            ##</%text>
 
<%text>##                                                                            ##</%text>
 
<%text>## It is recommended to use a valid sender address that passes access         ##</%text>
 
<%text>## validation and spam filtering in mail servers.                             ##</%text>
 
<%text>################################################################################</%text>
 

	
 
<%text>## 'From' header for application emails. You can optionally add a name.</%text>
 
<%text>## Default:</%text>
 
#app_email_from = Kallithea
 
<%text>## Examples:</%text>
 
#app_email_from = Kallithea <kallithea-noreply@example.com>
 
#app_email_from = kallithea-noreply@example.com
 

	
 
<%text>## Subject prefix for application emails.</%text>
 
<%text>## A space between this prefix and the real subject is automatically added.</%text>
 
<%text>## Default:</%text>
 
#email_prefix =
 
<%text>## Example:</%text>
 
#email_prefix = [Kallithea]
 

	
 
<%text>## Recipients for error emails and fallback recipients of application mails.</%text>
 
<%text>## Multiple addresses can be specified, comma-separated.</%text>
 
<%text>## Only addresses are allowed, do not add any name part.</%text>
 
<%text>## Default:</%text>
 
#email_to =
 
<%text>## Examples:</%text>
 
#email_to = admin@example.com
 
#email_to = admin@example.com,another_admin@example.com
 
email_to =
 

	
 
<%text>## 'From' header for error emails. You can optionally add a name.</%text>
 
<%text>## Default: (none)</%text>
 
<%text>## Examples:</%text>
 
#error_email_from = Kallithea Errors <kallithea-noreply@example.com>
 
#error_email_from = kallithea_errors@example.com
 
error_email_from =
 

	
 
<%text>## SMTP server settings</%text>
 
<%text>## If specifying credentials, make sure to use secure connections.</%text>
 
<%text>## Default: Send unencrypted unauthenticated mails to the specified smtp_server.</%text>
 
<%text>## For "SSL", use smtp_use_ssl = true and smtp_port = 465.</%text>
 
<%text>## For "STARTTLS", use smtp_use_tls = true and smtp_port = 587.</%text>
 
smtp_server =
 
#smtp_username =
 
#smtp_password =
 
smtp_port =
 
#smtp_use_ssl = false
 
#smtp_use_tls = false
 

	
 
%if http_server != 'uwsgi':
 
<%text>## Entry point for 'gearbox serve'</%text>
 
[server:main]
 
host = ${host}
 
port = ${port}
 

	
 
%if http_server == 'gearbox':
 
<%text>## Gearbox default web server ##</%text>
 
use = egg:gearbox#wsgiref
 
<%text>## nr of worker threads to spawn</%text>
 
threadpool_workers = 1
 
<%text>## max request before thread respawn</%text>
 
threadpool_max_requests = 100
 
<%text>## option to use threads of process</%text>
 
use_threadpool = true
 

	
 
%elif http_server == 'gevent':
 
<%text>## Gearbox gevent web server ##</%text>
 
use = egg:gearbox#gevent
 

	
 
%elif http_server == 'waitress':
 
<%text>## WAITRESS ##</%text>
 
use = egg:waitress#main
 
<%text>## number of worker threads</%text>
 
threads = 1
 
<%text>## MAX BODY SIZE 100GB</%text>
 
max_request_body_size = 107374182400
 
<%text>## use poll instead of select, fixes fd limits, may not work on old</%text>
 
<%text>## windows systems.</%text>
 
#asyncore_use_poll = True
 

	
 
%elif http_server == 'gunicorn':
 
<%text>## GUNICORN ##</%text>
 
use = egg:gunicorn#main
 
<%text>## number of process workers. You must set `instance_id = *` when this option</%text>
 
<%text>## is set to more than one worker</%text>
 
workers = 4
 
<%text>## process name</%text>
 
proc_name = kallithea
 
<%text>## type of worker class, one of sync, eventlet, gevent, tornado</%text>
 
<%text>## recommended for bigger setup is using of of other than sync one</%text>
 
worker_class = sync
 
max_requests = 1000
 
<%text>## amount of time a worker can handle request before it gets killed and</%text>
 
<%text>## restarted</%text>
 
timeout = 3600
 

	
 
%endif
 
%else:
 
<%text>## UWSGI ##</%text>
 
<%text>## run with uwsgi --ini-paste-logged <inifile.ini></%text>
 
[uwsgi]
 
socket = /tmp/uwsgi.sock
 
master = true
 
http = ${host}:${port}
 

	
 
<%text>## set as daemon and redirect all output to file</%text>
 
#daemonize = ./uwsgi_kallithea.log
 

	
 
<%text>## master process PID</%text>
 
pidfile = ./uwsgi_kallithea.pid
 

	
 
<%text>## stats server with workers statistics, use uwsgitop</%text>
 
<%text>## for monitoring, `uwsgitop 127.0.0.1:1717`</%text>
 
stats = 127.0.0.1:1717
 
memory-report = true
 

	
 
<%text>## log 5XX errors</%text>
 
log-5xx = true
 

	
 
<%text>## Set the socket listen queue size.</%text>
 
listen = 128
 

	
 
<%text>## Gracefully Reload workers after the specified amount of managed requests</%text>
 
<%text>## (avoid memory leaks).</%text>
 
max-requests = 1000
 

	
 
<%text>## enable large buffers</%text>
 
buffer-size = 65535
 

	
 
<%text>## socket and http timeouts ##</%text>
 
http-timeout = 3600
 
socket-timeout = 3600
 

	
 
<%text>## Log requests slower than the specified number of milliseconds.</%text>
 
log-slow = 10
 

	
 
<%text>## Exit if no app can be loaded.</%text>
 
need-app = true
 

	
 
<%text>## Set lazy mode (load apps in workers instead of master).</%text>
 
lazy = true
 

	
 
<%text>## scaling ##</%text>
 
<%text>## set cheaper algorithm to use, if not set default will be used</%text>
 
cheaper-algo = spare
 

	
 
<%text>## minimum number of workers to keep at all times</%text>
 
cheaper = 1
 

	
 
<%text>## number of workers to spawn at startup</%text>
 
cheaper-initial = 1
 

	
 
<%text>## maximum number of workers that can be spawned</%text>
 
workers = 4
 

	
 
<%text>## how many workers should be spawned at a time</%text>
 
cheaper-step = 1
 

	
 
%endif
 
<%text>## middleware for hosting the WSGI application under a URL prefix</%text>
 
#[filter:proxy-prefix]
 
#use = egg:PasteDeploy#prefix
 
#prefix = /<your-prefix>
 

	
 
[app:main]
 
use = egg:kallithea
 
<%text>## enable proxy prefix middleware</%text>
 
#filter-with = proxy-prefix
 

	
 
full_stack = true
 
static_files = true
 

	
 
<%text>## Internationalization (see setup documentation for details)</%text>
 
<%text>## By default, the language requested by the browser is used if available.</%text>
 
#i18n.enable = false
 
<%text>## Fallback language, empty for English (valid values are the names of subdirectories in kallithea/i18n):</%text>
 
i18n.lang =
 

	
 
cache_dir = %(here)s/data
 
index_dir = %(here)s/data/index
 

	
 
<%text>## perform a full repository scan on each server start, this should be</%text>
 
<%text>## set to false after first startup, to allow faster server restarts.</%text>
 
initial_repo_scan = false
 

	
 
<%text>## uncomment and set this path to use archive download cache</%text>
 
archive_cache_dir = %(here)s/tarballcache
 

	
 
<%text>## change this to unique ID for security</%text>
 
app_instance_uuid = ${uuid()}
 

	
 
<%text>## cut off limit for large diffs (size in bytes)</%text>
 
cut_off_limit = 256000
 

	
 
<%text>## force https in Kallithea, fixes https redirects, assumes it's always https</%text>
 
force_https = false
 

	
 
<%text>## use Strict-Transport-Security headers</%text>
 
use_htsts = false
 

	
 
<%text>## number of commits stats will parse on each iteration</%text>
 
commit_parse_limit = 25
 

	
 
<%text>## path to git executable</%text>
 
git_path = git
 

	
 
<%text>## git rev filter option, --all is the default filter, if you need to</%text>
 
<%text>## hide all refs in changelog switch this to --branches --tags</%text>
 
#git_rev_filter = --branches --tags
 

	
 
<%text>## RSS feed options</%text>
 
rss_cut_off_limit = 256000
 
rss_items_per_page = 10
 
rss_include_diff = false
 

	
 
<%text>## options for showing and identifying changesets</%text>
 
show_sha_length = 12
 
show_revision_number = false
 

	
 
<%text>## Canonical URL to use when creating full URLs in UI and texts.</%text>
 
<%text>## Useful when the site is available under different names or protocols.</%text>
 
<%text>## Defaults to what is provided in the WSGI environment.</%text>
 
#canonical_url = https://kallithea.example.com/repos
 

	
 
<%text>## gist URL alias, used to create nicer urls for gist. This should be an</%text>
 
<%text>## url that does rewrites to _admin/gists/<gistid>.</%text>
 
<%text>## example: http://gist.example.com/{gistid}. Empty means use the internal</%text>
 
<%text>## Kallithea url, ie. http[s]://kallithea.example.com/_admin/gists/<gistid></%text>
 
gist_alias_url =
 

	
 
<%text>## white list of API enabled controllers. This allows to add list of</%text>
 
<%text>## controllers to which access will be enabled by api_key. eg: to enable</%text>
 
<%text>## api access to raw_files put `FilesController:raw`, to enable access to patches</%text>
 
<%text>## add `ChangesetController:changeset_patch`. This list should be "," separated</%text>
 
<%text>## Syntax is <ControllerClass>:<function>. Check debug logs for generated names</%text>
 
<%text>## Recommended settings below are commented out:</%text>
 
api_access_controllers_whitelist =
 
#    ChangesetController:changeset_patch,
 
#    ChangesetController:changeset_raw,
 
#    FilesController:raw,
 
#    FilesController:archivefile
 

	
 
<%text>## default encoding used to convert from and to unicode</%text>
 
<%text>## can be also a comma separated list of encoding in case of mixed encodings</%text>
 
default_encoding = utf8
 

	
 
<%text>## issue tracker for Kallithea (leave blank to disable, absent for default)</%text>
 
#bugtracker = https://bitbucket.org/conservancy/kallithea/issues
 

	
 
<%text>## issue tracking mapping for commits messages</%text>
 
<%text>## comment out issue_pat, issue_server, issue_prefix to enable</%text>
 

	
 
<%text>## pattern to get the issues from commit messages</%text>
 
<%text>## default one used here is #<numbers> with a regex passive group for `#`</%text>
 
<%text>## {id} will be all groups matched from this pattern</%text>
 
<%text>## To require mandatory whitespace before the issue pattern, use:</%text>
 
<%text>## (?:^|(?<=\s)) before the actual pattern, and for mandatory whitespace</%text>
 
<%text>## behind the issue pattern, use (?:$|(?=\s)) after the actual pattern</%text>
 

	
 
issue_pat = #(\d+)
 

	
 
<%text>## server url to the issue, each {id} will be replaced with match</%text>
 
<%text>## fetched from the regex and {repo} is replaced with full repository name</%text>
 
<%text>## including groups {repo_name} is replaced with just name of repo</%text>
 

	
 
issue_server_link = https://issues.example.com/{repo}/issue/{id}
 

	
 
<%text>## prefix to add to link to indicate it's an url</%text>
 
<%text>## #314 will be replaced by <issue_prefix><id></%text>
 

	
 
issue_prefix = #
 

	
 
<%text>## issue_pat, issue_server_link, issue_prefix can have suffixes to specify</%text>
 
<%text>## multiple patterns, to other issues server, wiki or others</%text>
 
<%text>## below an example how to create a wiki pattern</%text>
 
# wiki-some-id -> https://wiki.example.com/some-id
 

	
 
#issue_pat_wiki = (?:wiki-)(.+)
 
#issue_server_link_wiki = https://wiki.example.com/{id}
 
#issue_prefix_wiki = WIKI-
 

	
 
<%text>## alternative return HTTP header for failed authentication. Default HTTP</%text>
 
<%text>## response is 401 HTTPUnauthorized. Currently Mercurial clients have trouble with</%text>
 
<%text>## handling that. Set this variable to 403 to return HTTPForbidden</%text>
 
auth_ret_code =
 

	
 
<%text>## locking return code. When repository is locked return this HTTP code. 2XX</%text>
 
<%text>## codes don't break the transactions while 4XX codes do</%text>
 
lock_ret_code = 423
 

	
 
<%text>## allows to change the repository location in settings page</%text>
 
allow_repo_location_change = True
 

	
 
<%text>## allows to setup custom hooks in settings page</%text>
 
allow_custom_hooks_settings = True
 

	
 
<%text>## extra extensions for indexing, space separated and without the leading '.'.</%text>
 
# index.extensions =
 
#    gemfile
 
#    lock
 

	
 
<%text>## extra filenames for indexing, space separated</%text>
 
# index.filenames =
 
#    .dockerignore
 
#    .editorconfig
 
#    INSTALL
 
#    CHANGELOG
 

	
 
<%text>####################################</%text>
 
<%text>###        CELERY CONFIG        ####</%text>
 
<%text>####################################</%text>
 

	
 
use_celery = false
 

	
 
<%text>## Example: connect to the virtual host 'rabbitmqhost' on localhost as rabbitmq:</%text>
 
broker.url = amqp://rabbitmq:qewqew@localhost:5672/rabbitmqhost
 

	
 
celery.imports = kallithea.lib.celerylib.tasks
 
celery.accept.content = pickle
 
celery.result.backend = amqp
 
celery.result.dburi = amqp://
 
celery.result.serialier = json
 

	
 
#celery.send.task.error.emails = true
 
#celery.amqp.task.result.expires = 18000
 

	
 
celeryd.concurrency = 2
 
celeryd.max.tasks.per.child = 1
 

	
 
<%text>## If true, tasks will never be sent to the queue, but executed locally instead.</%text>
 
celery.always.eager = false
 

	
 
<%text>####################################</%text>
 
<%text>###         BEAKER CACHE        ####</%text>
 
<%text>####################################</%text>
 

	
 
beaker.cache.data_dir = %(here)s/data/cache/data
 
beaker.cache.lock_dir = %(here)s/data/cache/lock
 

	
 
beaker.cache.regions = short_term,long_term,sql_cache_short
 

	
 
beaker.cache.short_term.type = memory
 
beaker.cache.short_term.expire = 60
 
beaker.cache.short_term.key_length = 256
 

	
 
beaker.cache.long_term.type = memory
 
beaker.cache.long_term.expire = 36000
 
beaker.cache.long_term.key_length = 256
 

	
 
beaker.cache.sql_cache_short.type = memory
 
beaker.cache.sql_cache_short.expire = 10
 
beaker.cache.sql_cache_short.key_length = 256
 

	
 
<%text>####################################</%text>
 
<%text>###       BEAKER SESSION        ####</%text>
 
<%text>####################################</%text>
 

	
 
<%text>## Name of session cookie. Should be unique for a given host and path, even when running</%text>
 
<%text>## on different ports. Otherwise, cookie sessions will be shared and messed up.</%text>
 
beaker.session.key = kallithea
 
<%text>## Sessions should always only be accessible by the browser, not directly by JavaScript.</%text>
 
beaker.session.httponly = true
 
<%text>## Session lifetime. 2592000 seconds is 30 days.</%text>
 
beaker.session.timeout = 2592000
 

	
 
<%text>## Server secret used with HMAC to ensure integrity of cookies.</%text>
 
beaker.session.secret = ${uuid()}
 
<%text>## Further, encrypt the data with AES.</%text>
 
#beaker.session.encrypt_key = <key_for_encryption>
 
#beaker.session.validate_key = <validation_key>
 

	
 
<%text>## Type of storage used for the session, current types are</%text>
 
<%text>## dbm, file, memcached, database, and memory.</%text>
 

	
 
<%text>## File system storage of session data. (default)</%text>
 
#beaker.session.type = file
 

	
 
<%text>## Cookie only, store all session data inside the cookie. Requires secure secrets.</%text>
 
#beaker.session.type = cookie
 

	
 
<%text>## Database storage of session data.</%text>
 
#beaker.session.type = ext:database
 
#beaker.session.sa.url = postgresql://postgres:qwe@localhost/kallithea
 
#beaker.session.table_name = db_session
 

	
 
%if error_aggregation_service == 'appenlight':
 
<%text>############################</%text>
 
<%text>## ERROR HANDLING SYSTEMS ##</%text>
 
<%text>############################</%text>
 

	
 
# Propagate email settings to ErrorReporter of TurboGears2
 
# You do not normally need to change these lines
 
get trace_errors.error_email = email_to
 
get trace_errors.smtp_server = smtp_server
 
get trace_errors.smtp_port = smtp_port
 
get trace_errors.from_address = error_email_from
 

	
 
<%text>####################</%text>
 
<%text>### [appenlight] ###</%text>
 
<%text>####################</%text>
 

	
 
<%text>## AppEnlight is tailored to work with Kallithea, see</%text>
 
<%text>## http://appenlight.com for details how to obtain an account</%text>
 
<%text>## you must install python package `appenlight_client` to make it work</%text>
 

	
 
<%text>## appenlight enabled</%text>
 
appenlight = false
 

	
 
appenlight.server_url = https://api.appenlight.com
 
appenlight.api_key = YOUR_API_KEY
 

	
 
<%text>## TWEAK AMOUNT OF INFO SENT HERE</%text>
 

	
 
<%text>## enables 404 error logging (default False)</%text>
 
appenlight.report_404 = false
 

	
 
<%text>## time in seconds after request is considered being slow (default 1)</%text>
 
appenlight.slow_request_time = 1
 

	
 
<%text>## record slow requests in application</%text>
 
<%text>## (needs to be enabled for slow datastore recording and time tracking)</%text>
 
appenlight.slow_requests = true
 

	
 
<%text>## enable hooking to application loggers</%text>
 
#appenlight.logging = true
 

	
 
<%text>## minimum log level for log capture</%text>
 
#appenlight.logging.level = WARNING
 

	
 
<%text>## send logs only from erroneous/slow requests</%text>
 
<%text>## (saves API quota for intensive logging)</%text>
 
appenlight.logging_on_error = false
 

	
 
<%text>## list of additional keywords that should be grabbed from environ object</%text>
 
<%text>## can be string with comma separated list of words in lowercase</%text>
 
<%text>## (by default client will always send following info:</%text>
 
<%text>## 'REMOTE_USER', 'REMOTE_ADDR', 'SERVER_NAME', 'CONTENT_TYPE' + all keys that</%text>
 
<%text>## start with HTTP* this list be extended with additional keywords here</%text>
 
appenlight.environ_keys_whitelist =
 

	
 
<%text>## list of keywords that should be blanked from request object</%text>
 
<%text>## can be string with comma separated list of words in lowercase</%text>
 
<%text>## (by default client will always blank keys that contain following words</%text>
 
<%text>## 'password', 'passwd', 'pwd', 'auth_tkt', 'secret', 'csrf'</%text>
 
<%text>## this list be extended with additional keywords set here</%text>
 
appenlight.request_keys_blacklist =
 

	
 
<%text>## list of namespaces that should be ignores when gathering log entries</%text>
 
<%text>## can be string with comma separated list of namespaces</%text>
 
<%text>## (by default the client ignores own entries: appenlight_client.client)</%text>
 
appenlight.log_namespace_blacklist =
 

	
 
%elif error_aggregation_service == 'sentry':
 
<%text>################</%text>
 
<%text>### [sentry] ###</%text>
 
<%text>################</%text>
 

	
 
<%text>## sentry is a alternative open source error aggregator</%text>
 
<%text>## you must install python packages `sentry` and `raven` to enable</%text>
 

	
 
sentry.dsn = YOUR_DNS
 
sentry.servers =
 
sentry.name =
 
sentry.key =
 
sentry.public_key =
 
sentry.secret_key =
 
sentry.project =
 
sentry.site =
 
sentry.include_paths =
 
sentry.exclude_paths =
 

	
 
%endif
 
<%text>################################################################################</%text>
 
<%text>## WARNING: *DEBUG MODE MUST BE OFF IN A PRODUCTION ENVIRONMENT*              ##</%text>
 
<%text>## Debug mode will enable the interactive debugging tool, allowing ANYONE to  ##</%text>
 
<%text>## execute malicious code after an exception is raised.                       ##</%text>
 
<%text>################################################################################</%text>
 
debug = false
 

	
 
<%text>##################################</%text>
 
<%text>###       LOGVIEW CONFIG       ###</%text>
 
<%text>##################################</%text>
 

	
 
logview.sqlalchemy = #faa
 
logview.pylons.templating = #bfb
 
logview.pylons.util = #eee
 

	
 
<%text>#########################################################</%text>
 
<%text>### DB CONFIGS - EACH DB WILL HAVE IT'S OWN CONFIG    ###</%text>
 
<%text>#########################################################</%text>
 

	
 
%if database_engine == 'sqlite':
 
# SQLITE [default]
 
sqlalchemy.url = sqlite:///%(here)s/kallithea.db?timeout=60
 

	
 
%elif database_engine == 'postgres':
 
# POSTGRESQL
 
sqlalchemy.url = postgresql://user:pass@localhost/kallithea
 

	
 
%elif database_engine == 'mysql':
 
# MySQL
 
sqlalchemy.url = mysql://user:pass@localhost/kallithea?charset=utf8
 

	
 
%endif
 
# see sqlalchemy docs for others
 

	
 
sqlalchemy.pool_recycle = 3600
 

	
 
<%text>################################</%text>
 
<%text>### ALEMBIC CONFIGURATION   ####</%text>
 
<%text>################################</%text>
 

	
 
[alembic]
 
script_location = kallithea:alembic
 

	
 
<%text>################################</%text>
 
<%text>### LOGGING CONFIGURATION   ####</%text>
 
<%text>################################</%text>
 

	
 
[loggers]
 
keys = root, routes, kallithea, sqlalchemy, tg, gearbox, beaker, templates, whoosh_indexer, werkzeug, backlash
 

	
 
[handlers]
 
keys = console, console_sql
 

	
 
[formatters]
 
keys = generic, color_formatter, color_formatter_sql
 

	
 
<%text>#############</%text>
 
<%text>## LOGGERS ##</%text>
 
<%text>#############</%text>
 

	
 
[logger_root]
 
level = NOTSET
 
handlers = console
 

	
 
[logger_routes]
 
level = WARN
 
handlers =
 
qualname = routes.middleware
 
<%text>## "level = DEBUG" logs the route matched and routing variables.</%text>
 
propagate = 1
 

	
 
[logger_beaker]
 
level = WARN
 
handlers =
 
qualname = beaker.container
 
propagate = 1
 

	
 
[logger_templates]
 
level = WARN
 
handlers =
 
qualname = pylons.templating
 
propagate = 1
 

	
 
[logger_kallithea]
 
level = WARN
 
handlers =
 
qualname = kallithea
 
propagate = 1
 

	
 
[logger_tg]
 
level = WARN
 
handlers =
 
qualname = tg
 
propagate = 1
 

	
 
[logger_gearbox]
 
level = WARN
 
handlers =
 
qualname = gearbox
 
propagate = 1
 

	
 
[logger_sqlalchemy]
 
level = WARN
 
handlers = console_sql
 
qualname = sqlalchemy.engine
 
propagate = 0
 

	
 
[logger_whoosh_indexer]
 
level = WARN
 
handlers =
 
qualname = whoosh_indexer
 
propagate = 1
 

	
 
[logger_werkzeug]
 
level = WARN
 
handlers =
 
qualname = werkzeug
 
propagate = 1
 

	
 
[logger_backlash]
 
level = WARN
 
handlers =
 
qualname = backlash
 
propagate = 1
 

	
 
<%text>##############</%text>
 
<%text>## HANDLERS ##</%text>
 
<%text>##############</%text>
 

	
 
[handler_console]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
[handler_console_sql]
 
class = StreamHandler
 
args = (sys.stderr,)
 
formatter = generic
 

	
 
<%text>################</%text>
 
<%text>## FORMATTERS ##</%text>
 
<%text>################</%text>
 

	
 
[formatter_generic]
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter]
 
class = kallithea.lib.colored_formatter.ColorFormatter
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
 

	
 
[formatter_color_formatter_sql]
 
class = kallithea.lib.colored_formatter.ColorFormatterSql
 
format = %(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s
 
datefmt = %Y-%m-%d %H:%M:%S
kallithea/tests/other/test_libs.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
# This program is free software: you can redistribute it and/or modify
 
# it under the terms of the GNU General Public License as published by
 
# the Free Software Foundation, either version 3 of the License, or
 
# (at your option) any later version.
 
#
 
# This program is distributed in the hope that it will be useful,
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
# GNU General Public License for more details.
 
#
 
# You should have received a copy of the GNU General Public License
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
"""
 
kallithea.tests.other.test_libs
 
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 

	
 
Package for testing various lib/helper functions in kallithea
 

	
 
This file was forked by the Kallithea project in July 2014.
 
Original author and date, and relevant copyright and licensing information is below:
 
:created_on: Jun 9, 2011
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import datetime
 
import hashlib
 
import mock
 
from kallithea.tests.base import *
 
from kallithea.lib.utils2 import AttributeDict
 
from kallithea.model.db import Repository
 
from tg.util.webtest import test_context
 

	
 
proto = 'http'
 
TEST_URLS = [
 
    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username: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://example.com' % proto, ['%s://' % proto, 'example.com'],
 
     '%s://example.com' % proto),
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
 
                                                '8080'],
 
     '%s://example.com: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://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
 
     '%s://127.0.0.1' % proto),
 
    ('%s://username: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://example.com' % proto, ['%s://' % proto, 'example.com'],
 
     '%s://example.com' % proto),
 
    ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
 
                                                '8080'],
 
     '%s://example.com:8080' % proto),
 
]
 

	
 

	
 
class FakeUrlGenerator(object):
 

	
 
    def __init__(self, current_url=None, default_route=None, **routes):
 
        """Initialize using specified 'current' URL template,
 
        default route template, and all other aguments describing known
 
        routes (format: route=template)"""
 
        self.current_url = current_url
 
        self.default_route = default_route
 
        self.routes = routes
 

	
 
    def __call__(self, route_name, *args, **kwargs):
 
        if route_name in self.routes:
 
            return self.routes[route_name] % kwargs
 

	
 
        return self.default_route % kwargs
 

	
 
    def current(self, *args, **kwargs):
 
        return self.current_url % kwargs
 

	
 

	
 
class TestLibs(TestController):
 

	
 
    @parametrize('test_url,expected,expected_creds', TEST_URLS)
 
    def test_uri_filter(self, test_url, expected, expected_creds):
 
        from kallithea.lib.utils2 import uri_filter
 
        assert uri_filter(test_url) == expected
 

	
 
    @parametrize('test_url,expected,expected_creds', TEST_URLS)
 
    def test_credentials_filter(self, test_url, expected, expected_creds):
 
        from kallithea.lib.utils2 import credentials_filter
 
        assert credentials_filter(test_url) == expected_creds
 

	
 
    @parametrize('str_bool,expected', [
 
                           ('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)
 
    ])
 
    def test_str2bool(self, str_bool, expected):
 
        from kallithea.lib.utils2 import str2bool
 
        assert str2bool(str_bool) == expected
 

	
 
    def test_mention_extractor(self):
 
        from kallithea.lib.utils2 import extract_mentioned_usernames
 
        sample = (
 
            "@first hi there @world here's my email username@example.com "
 
            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
 
            "@UPPER    @cAmEL @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@example.com"
 
        )
 

	
 
        expected = set([
 
            '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
 
            'marian.user', 'marco-polo', 'marco_polo', 'world'])
 
        assert expected == set(extract_mentioned_usernames(sample))
 

	
 
    @parametrize('age_args,expected', [
 
        (dict(), u'just now'),
 
        (dict(seconds= -1), u'1 second ago'),
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),
 
        (dict(hours= -1), u'1 hour ago'),
 
        (dict(hours= -24), u'1 day ago'),
 
        (dict(hours= -24 * 5), u'5 days ago'),
 
        (dict(months= -1), u'1 month ago'),
 
        (dict(months= -1, days= -2), u'1 month and 2 days ago'),
 
        (dict(months= -1, days= -20), u'1 month and 19 days ago'),
 
        (dict(years= -1, months= -1), u'1 year and 1 month ago'),
 
        (dict(years= -1, months= -10), u'1 year and 10 months ago'),
 
        (dict(years= -2, months= -4), u'2 years and 4 months ago'),
 
        (dict(years= -2, months= -11), u'2 years and 11 months ago'),
 
        (dict(years= -3, months= -2), u'3 years and 2 months ago'),
 
    ])
 
    def test_age(self, age_args, expected):
 
        from kallithea.lib.utils2 import age
 
        from dateutil import relativedelta
 
        with test_context(self.app):
 
            n = datetime.datetime(year=2012, month=5, day=17)
 
            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
            assert age(n + delt(**age_args), now=n) == expected
 

	
 
    @parametrize('age_args,expected', [
 
        (dict(), u'just now'),
 
        (dict(seconds= -1), u'1 second ago'),
 
        (dict(seconds= -60 * 2), u'2 minutes ago'),
 
        (dict(hours= -1), u'1 hour ago'),
 
        (dict(hours= -24), u'1 day ago'),
 
        (dict(hours= -24 * 5), u'5 days ago'),
 
        (dict(months= -1), u'1 month ago'),
 
        (dict(months= -1, days= -2), u'1 month ago'),
 
        (dict(months= -1, days= -20), u'1 month ago'),
 
        (dict(years= -1, months= -1), u'13 months ago'),
 
        (dict(years= -1, months= -10), u'22 months ago'),
 
        (dict(years= -2, months= -4), u'2 years ago'),
 
        (dict(years= -2, months= -11), u'3 years ago'),
 
        (dict(years= -3, months= -2), u'3 years ago'),
 
        (dict(years= -4, months= -8), u'5 years ago'),
 
    ])
 
    def test_age_short(self, age_args, expected):
 
        from kallithea.lib.utils2 import age
 
        from dateutil import relativedelta
 
        with test_context(self.app):
 
            n = datetime.datetime(year=2012, month=5, day=17)
 
            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
            assert age(n + delt(**age_args), show_short_version=True, now=n) == expected
 

	
 
    @parametrize('age_args,expected', [
 
        (dict(), u'just now'),
 
        (dict(seconds=1), u'in 1 second'),
 
        (dict(seconds=60 * 2), u'in 2 minutes'),
 
        (dict(hours=1), u'in 1 hour'),
 
        (dict(hours=24), u'in 1 day'),
 
        (dict(hours=24 * 5), u'in 5 days'),
 
        (dict(months=1), u'in 1 month'),
 
        (dict(months=1, days=1), u'in 1 month and 1 day'),
 
        (dict(years=1, months=1), u'in 1 year and 1 month')
 
    ])
 
    def test_age_in_future(self, age_args, expected):
 
        from kallithea.lib.utils2 import age
 
        from dateutil import relativedelta
 
        with test_context(self.app):
 
            n = datetime.datetime(year=2012, month=5, day=17)
 
            delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
 
            assert age(n + delt(**age_args), now=n) == expected
 

	
 
    def test_tag_extractor(self):
 
        sample = (
 
            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
 
            "[requires] [stale] [see<>=>] [see => http://example.com]"
 
            "[requires => url] [lang => python] [just a tag]"
 
            "[,d] [ => ULR ] [obsolete] [desc]]"
 
        )
 
        from kallithea.lib.helpers import urlify_text
 
        res = urlify_text(sample, stylize=True)
 
        assert '<div class="label label-meta" data-tag="tag">tag</div>' in res
 
        assert '<div class="label label-meta" data-tag="obsolete">obsolete</div>' in res
 
        assert '<div class="label label-meta" data-tag="stale">stale</div>' in res
 
        assert '<div class="label label-meta" data-tag="lang">python</div>' in res
 
        assert '<div class="label label-meta" data-tag="requires">requires =&gt; <a href="/url">url</a></div>' in res
 
        assert '<div class="label label-meta" data-tag="tag">tag</div>' in res
 

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

	
 
        # mock tg.tmpl_context
 
        def fake_tmpl_context(_url):
 
            _c = AttributeDict()
 
            _c.visual = AttributeDict()
 
            _c.visual.use_gravatar = True
 
            _c.visual.gravatar_url = _url
 

	
 
            return _c
 

	
 
        fake_url = FakeUrlGenerator(current_url='https://example.com')
 
        with mock.patch('kallithea.config.routing.url', fake_url):
 
            fake = fake_tmpl_context(_url='http://example.com/{email}')
 
            with mock.patch('tg.tmpl_context', fake):
 
                    from kallithea.config.routing import url
 
                    assert url.current() == 'https://example.com'
 
                    grav = gravatar_url(email_address='test@example.com', size=24)
 
                    assert grav == 'http://example.com/test@example.com'
 

	
 
            fake = fake_tmpl_context(_url='http://example.com/{email}')
 
            with mock.patch('tg.tmpl_context', fake):
 
                grav = gravatar_url(email_address='test@example.com', size=24)
 
                assert grav == 'http://example.com/test@example.com'
 

	
 
            fake = fake_tmpl_context(_url='http://example.com/{md5email}')
 
            with mock.patch('tg.tmpl_context', fake):
 
                em = 'test@example.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://example.com/%s' % (_md5(em))
 

	
 
            fake = fake_tmpl_context(_url='http://example.com/{md5email}/{size}')
 
            with mock.patch('tg.tmpl_context', fake):
 
                em = 'test@example.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'http://example.com/%s/%s' % (_md5(em), 24)
 

	
 
            fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')
 
            with mock.patch('tg.tmpl_context', fake):
 
                em = 'test@example.com'
 
                grav = gravatar_url(email_address=em, size=24)
 
                assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)
 

	
 
    @parametrize('tmpl,repo_name,overrides,prefix,expected', [
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
 
        (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),
 
        ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
 
        ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
 
        ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),
 
        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.example.com/group/repo1'),
 
        ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.example.com/group/repo1'),
 
        ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.example.com/username/group/repo1'),
 
    ])
 
    def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):
 
        from kallithea.lib.utils2 import get_clone_url
 
        clone_url = get_clone_url(uri_tmpl=tmpl, qualified_home_url='http://vps1:8000'+prefix,
 
                                  repo_name=repo_name, repo_id=23, **overrides)
 
        assert clone_url == expected
 

	
 
    def _quick_url(self, text, tmpl="""<a class="changeset_hash" href="%s">%s</a>""", url_=None):
 
        """
 
        Changes `some text url[foo]` => `some text <a href="/">foo</a>
 

	
 
        :param text:
 
        """
 
        import re
 
        # quickly change expected url[] into a link
 
        url_pattern = re.compile(r'(?:url\[)(.+?)(?:\])')
 

	
 
        def url_func(match_obj):
 
            _url = match_obj.groups()[0]
 
            return tmpl % (url_ or '/repo_name/changeset/%s' % _url, _url)
 
        return url_pattern.sub(url_func, text)
 

	
 
    @parametrize('sample,expected', [
 
      ("",
 
       ""),
 
      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
 
       """git-svn-id: <a href="https://svn.apache.org/repos/asf/libcloud/trunk@1441655">https://svn.apache.org/repos/asf/libcloud/trunk@1441655</a> 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
 
       sometimes !
 
       """,
 
       """Multi line<br/>"""
 
       """       url[123123123123]<br/>"""
 
       """       some text url[123123123123]<br/>"""
 
       """       sometimes !"""),
 
    ])
 
    def test_urlify_text(self, sample, expected):
 
        expected = self._quick_url(expected)
 
        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
 
        with mock.patch('kallithea.config.routing.url', fake_url):
 
            from kallithea.lib.helpers import urlify_text
 
            assert urlify_text(sample, 'repo_name') == expected
 

	
 
    @parametrize('sample,expected,url_', [
 
      ("",
 
       "",
 
       ""),
 
      ("https://svn.apache.org/repos",
 
       """url[https://svn.apache.org/repos]""",
 
       "https://svn.apache.org/repos"),
 
      ("http://svn.apache.org/repos",
 
       """url[http://svn.apache.org/repos]""",
 
       "http://svn.apache.org/repos"),
 
      ("from rev a also rev http://google.com",
 
       """from rev a also rev url[http://google.com]""",
 
       "http://google.com"),
 
      ("http://imgur.com/foo.gif inline http://imgur.com/foo.gif ending http://imgur.com/foo.gif",
 
       """url[http://imgur.com/foo.gif] inline url[http://imgur.com/foo.gif] ending url[http://imgur.com/foo.gif]""",
 
       "http://imgur.com/foo.gif"),
 
      ("""Multi line
 
       https://foo.bar.example.com
 
       some text lalala""",
 
       """Multi line<br/>"""
 
       """       url[https://foo.bar.example.com]<br/>"""
 
       """       some text lalala""",
 
       "https://foo.bar.example.com"),
 
      ("@mention @someone",
 
       """<b>@mention</b> <b>@someone</b>""",
 
       ""),
 
      ("deadbeefcafe 123412341234",
 
       """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a> <a class="changeset_hash" href="/repo_name/changeset/123412341234">123412341234</a>""",
 
       ""),
 
      ("We support * markup for *bold* markup of *single or multiple* words, "
 
       "*a bit @like http://slack.com*. "
 
       "The first * must come after whitespace and not be followed by whitespace, "
 
       "contain anything but * and newline until the next *, "
 
       "which must not come after whitespace "
 
       "and not be followed by * or alphanumerical *characters*.",
 
       """We support * markup for <b>*bold*</b> markup of <b>*single or multiple*</b> words, """
 
       """<b>*a bit <b>@like</b> <a href="http://slack.com">http://slack.com</a>*</b>. """
 
       """The first * must come after whitespace and not be followed by whitespace, """
 
       """contain anything but * and newline until the next *, """
 
       """which must not come after whitespace """
 
       """and not be followed by * or alphanumerical <b>*characters*</b>.""",
 
       "-"),
 
      ("HTML escaping: <abc> 'single' \"double\" &pointer",
 
       "HTML escaping: &lt;abc&gt; &#39;single&#39; &quot;double&quot; &amp;pointer",
 
       "-"),
 
      # tags are covered by test_tag_extractor
 
    ])
 
    def test_urlify_test(self, sample, expected, url_):
 
        expected = self._quick_url(expected,
 
                                   tmpl="""<a href="%s">%s</a>""", url_=url_)
 
        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
 
        with mock.patch('kallithea.config.routing.url', fake_url):
 
            from kallithea.lib.helpers import urlify_text
 
            assert urlify_text(sample, 'repo_name', stylize=True) == expected
 

	
 
    @parametrize('sample,expected', [
 
      ("deadbeefcafe @mention, and http://foo.bar/ yo",
 
       """<a class="changeset_hash" href="/repo_name/changeset/deadbeefcafe">deadbeefcafe</a>"""
 
       """<a class="message-link" href="#the-link"> <b>@mention</b>, and </a>"""
 
       """<a href="http://foo.bar/">http://foo.bar/</a>"""
 
       """<a class="message-link" href="#the-link"> yo</a>"""),
 
    ])
 
    def test_urlify_link(self, sample, expected):
 
        fake_url = FakeUrlGenerator(changeset_home='/%(repo_name)s/changeset/%(revision)s')
 
        with mock.patch('kallithea.config.routing.url', fake_url):
 
            from kallithea.lib.helpers import urlify_text
 
            assert urlify_text(sample, 'repo_name', link_='#the-link') == expected
 

	
 
    @parametrize('issue_pat,issue_server,issue_prefix,sample,expected', [
 
        (r'#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'issue #123 and issue#456',
 
            """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
 
        # following test case shows the result of a backward incompatible change that was made: the
 
        # space between 'issue' and '#123' is removed, because the space is part of the pattern.
 
        (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'issue #123 and issue#456',
 
            """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
 
        # to require whitespace before the issue reference, one may be tempted to use \b...
 
        (r'\bPR(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'issue PR123 and issuePR456',
 
            """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issuePR456"""),
 
        # following test case shows that \b does not work well in combination with '#': the expectations
 
        # ... but it turns out that \b does not work well in combination with '#': the expectations
 
        # are reversed from what is actually happening.
 
        (r'\b#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'issue #123 and issue#456',
 
            """issue #123 and """
 
            """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/456">#456</a>"""),
 
        (r'[ \t]#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'issue #123 and issue#456',
 
        # ... so maybe try to be explicit? Unfortunately the whitespace before the issue
 
        # reference is not retained, again, because it is part of the pattern.
 
        (r'(?:^|\s)#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            '#15 and issue #123 and issue#456',
 
            """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
 
            """issue<a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issue#456"""),
 
        # ... instead, use lookbehind assertions.
 
        (r'(?:^|(?<=\s))#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            '#15 and issue #123 and issue#456',
 
            """<a class="issue-tracker-link" href="http://foo/repo_name/issue/15">#15</a> and """
 
            """issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a> and """
 
            """issue#456"""),
 
        (r'(?:pullrequest|pull request|PR|pr) ?#?(\d+)', 'http://foo/{repo}/issue/{id}', 'PR#',
 
            'fixed with pullrequest #1, pull request#2, PR 3, pr4',
 
            """fixed with <a class="issue-tracker-link" href="http://foo/repo_name/issue/1">PR#1</a>, """
 
            """<a class="issue-tracker-link" href="http://foo/repo_name/issue/2">PR#2</a>, """
 
            """<a class="issue-tracker-link" href="http://foo/repo_name/issue/3">PR#3</a>, """
 
            """<a class="issue-tracker-link" href="http://foo/repo_name/issue/4">PR#4</a>"""),
 
        (r'#(\d+)', 'http://foo/{repo}/issue/{id}', 'PR',
 
            'interesting issue #123',
 
            """interesting issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">PR123</a>"""),
 
        (r'BUG\d{5}', 'https://bar/{repo}/{id}', 'BUG',
 
            'silly me, I did not parenthesize the {id}, BUG12345.',
 
            """silly me, I did not parenthesize the {id}, <a class="issue-tracker-link" href="https://bar/repo_name/">BUG</a>."""),
 
        (r'BUG(\d{5})', 'https://bar/{repo}/', 'BUG',
 
            'silly me, the URL does not contain {id}, BUG12345.',
 
            """silly me, the URL does not contain {id}, <a class="issue-tracker-link" href="https://bar/repo_name/">BUG12345</a>."""),
 
        (r'(PR-\d+)', 'http://foo/{repo}/issue/{id}', '',
 
            'interesting issue #123, err PR-56',
 
            """interesting issue #123, err <a class="issue-tracker-link" href="http://foo/repo_name/issue/PR-56">PR-56</a>"""),
 
        (r'#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            "some 'standard' text with apostrophes",
 
            """some &#39;standard&#39; text with apostrophes"""),
 
        (r'#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            "some 'standard' issue #123",
 
            """some &#39;standard&#39; issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>"""),
 
        (r'#(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'an issue   #123       with extra whitespace',
 
            """an issue   <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
 
        # Note: whitespace is squashed
 
        (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/{id}', '#',
 
            'an issue   #123       with extra whitespace',
 
            """an issue <a class="issue-tracker-link" href="http://foo/repo_name/issue/123">#123</a>       with extra whitespace"""),
 
        # invalid issue pattern
 
        (r'(PR\d+', 'http://foo/{repo}/issue/{id}', '',
 
            'PR135',
 
            """PR135"""),
 
    ])
 
    def test_urlify_issues(self, issue_pat, issue_server, issue_prefix, sample, expected):
 
        from kallithea.lib.helpers import urlify_text
 
        config_stub = {
 
            'sqlalchemy.url': 'foo',
 
            'issue_pat': issue_pat,
 
            'issue_server_link': issue_server,
 
            'issue_prefix': issue_prefix,
 
        }
 
        # force recreation of lazy function
 
        with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
 
            with mock.patch('kallithea.CONFIG', config_stub):
 
                assert urlify_text(sample, 'repo_name') == expected
 

	
 
    @parametrize('sample,expected', [
 
        ('abc X5', 'abc <a class="issue-tracker-link" href="http://main/repo_name/main/5/">#5</a>'),
 
        ('abc pullrequest #6 xyz', 'abc <a class="issue-tracker-link" href="http://pr/repo_name/pr/6">PR#6</a> xyz'),
 
        ('pull request7 #', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/7">PR#7</a> #'),
 
        ('look PR9 and pr #11', 'look <a class="issue-tracker-link" href="http://pr/repo_name/pr/9">PR#9</a> and <a class="issue-tracker-link" href="http://pr/repo_name/pr/11">PR#11</a>'),
 
        ('pullrequest#10 solves issue 9', '<a class="issue-tracker-link" href="http://pr/repo_name/pr/10">PR#10</a> solves <a class="issue-tracker-link" href="http://bug/repo_name/bug/9">bug#9</a>'),
 
        ('issue FAIL67', 'issue <a class="issue-tracker-link" href="http://fail/repo_name/67">67</a>'),
 
        ('issue FAILMORE89', 'issue FAILMORE89'), # no match because absent prefix
 
    ])
 
    def test_urlify_issues_multiple_issue_patterns(self, sample, expected):
 
        from kallithea.lib.helpers import urlify_text
 
        config_stub = {
 
            'sqlalchemy.url': 'foo',
 
            'issue_pat': 'X(\d+)',
 
            'issue_server_link': 'http://main/{repo}/main/{id}/',
 
            'issue_prefix': '#',
 
            'issue_pat_pr': '(?:pullrequest|pull request|PR|pr) ?#?(\d+)',
 
            'issue_server_link_pr': 'http://pr/{repo}/pr/{id}',
 
            'issue_prefix_pr': 'PR#',
 
            'issue_pat_bug': '(?:BUG|bug|issue) ?#?(\d+)',
 
            'issue_server_link_bug': 'http://bug/{repo}/bug/{id}',
 
            'issue_prefix_bug': 'bug#',
 
            'issue_pat_empty_prefix': 'FAIL(\d+)',
 
            'issue_server_link_empty_prefix': 'http://fail/{repo}/{id}',
 
            'issue_prefix_empty_prefix': '',
 
            'issue_pat_absent_prefix': 'FAILMORE(\d+)',
 
            'issue_server_link_absent_prefix': 'http://failmore/{repo}/{id}',
 
        }
 
        # force recreation of lazy function
 
        with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
 
            with mock.patch('kallithea.CONFIG', config_stub):
 
                assert urlify_text(sample, 'repo_name') == expected
 

	
 
    @parametrize('test,expected', [
 
      ("", None),
 
      ("/_2", '2'),
 
      ("_2", '2'),
 
      ("/_2/", '2'),
 
      ("_2/", '2'),
 

	
 
      ("/_21", '21'),
 
      ("_21", '21'),
 
      ("/_21/", '21'),
 
      ("_21/", '21'),
 

	
 
      ("/_21/foobar", '21'),
 
      ("_21/121", '21'),
 
      ("/_21/_12", '21'),
 
      ("_21/prefix/foo", '21'),
 
    ])
 
    def test_get_repo_by_id(self, test, expected):
 
        from kallithea.lib.utils import _extract_id_from_repo_name
 
        _test = _extract_id_from_repo_name(test)
 
        assert _test == expected, 'url:%s, got:`%s` expected: `%s`' % (test, _test, expected)
0 comments (0 inline, 0 general)