Changeset - 71a37439dcee
[Not reviewed]
default
0 7 0
Mads Kiilerich (mads) - 5 years ago 2020-11-09 17:13:33
mads@kiilerich.com
Grafted from: ac8430b9dfab
lib: move urlification to webutils

Less use of helpers in model.
7 files changed with 225 insertions and 226 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/feed.py
Show inline comments
 
@@ -80,25 +80,25 @@ class FeedController(BaseRepoController)
 
                           '(%(added)s lines added, %(removed)s lines removed)'
 
                            % st)
 
        if diff_processor.limited_diff:
 
            changes = changes + ['\n ' +
 
                                 _('Changeset was too big and was cut off...')]
 

	
 
        # rev link
 
        _url = webutils.canonical_url('changeset_home', repo_name=c.db_repo.repo_name,
 
                   revision=cs.raw_id)
 
        desc_msg.append('changeset: <a href="%s">%s</a>' % (_url, cs.raw_id[:8]))
 

	
 
        desc_msg.append('<pre>')
 
        desc_msg.append(h.urlify_text(cs.message))
 
        desc_msg.append(webutils.urlify_text(cs.message))
 
        desc_msg.append('\n')
 
        desc_msg.extend(changes)
 
        if asbool(kallithea.CONFIG.get('rss_include_diff', False)):
 
            desc_msg.append('\n\n')
 
            desc_msg.append(safe_str(raw_diff))
 
        desc_msg.append('</pre>')
 
        return desc_msg
 

	
 
    def _feed(self, repo_name, feeder):
 
        """Produce a simple feed"""
 

	
 
        @cache_region('long_term_file', '_get_feed_from_cache')
kallithea/lib/helpers.py
Show inline comments
 
@@ -36,55 +36,57 @@ from kallithea.lib.annotate import annot
 
#==============================================================================
 
from kallithea.lib.auth import HasPermissionAny, HasRepoGroupPermissionLevel, HasRepoPermissionLevel
 
from kallithea.lib.diffs import BIN_FILENODE, CHMOD_FILENODE, DEL_FILENODE, MOD_FILENODE, NEW_FILENODE, RENAMED_FILENODE
 
from kallithea.lib.pygmentsutils import get_custom_lexer
 
from kallithea.lib.utils2 import (AttributeDict, age, asbool, credentials_filter, fmt_date, link_to_ref, safe_bytes, safe_int, safe_str, shorter,
 
                                  time_to_datetime)
 
from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset
 
from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError
 
#==============================================================================
 
# SCM FILTERS available via h.
 
#==============================================================================
 
from kallithea.lib.vcs.utils import author_email, author_name
 
from kallithea.lib.webutils import (HTML, MENTIONS_REGEX, Option, canonical_url, checkbox, chop_at, end_form, escape, form, format_byte_size, hidden,
 
                                    html_escape, js, jshtml, link_to, literal, password, pop_flash_messages, radio, reset, safeid, select,
 
                                    session_csrf_secret_name, session_csrf_secret_token, submit, text, textarea, truncate, url, url_re, wrap_paragraphs)
 
from kallithea.lib.webutils import (HTML, Option, canonical_url, checkbox, chop_at, end_form, escape, form, format_byte_size, hidden, js, jshtml, link_to,
 
                                    literal, password, pop_flash_messages, radio, render_w_mentions, reset, safeid, select, session_csrf_secret_name,
 
                                    session_csrf_secret_token, submit, text, textarea, url, urlify_text, wrap_paragraphs)
 
from kallithea.model import db
 
from kallithea.model.changeset_status import ChangesetStatusModel
 

	
 

	
 
# mute pyflakes "imported but unused"
 
# from webutils
 
assert HTML
 
assert Option
 
assert canonical_url
 
assert checkbox
 
assert chop_at
 
assert end_form
 
assert form
 
assert format_byte_size
 
assert hidden
 
assert js
 
assert jshtml
 
assert password
 
assert pop_flash_messages
 
assert radio
 
assert render_w_mentions
 
assert reset
 
assert safeid
 
assert select
 
assert session_csrf_secret_name
 
assert session_csrf_secret_token
 
assert submit
 
assert text
 
assert textarea
 
assert urlify_text
 
assert wrap_paragraphs
 
# from kallithea.lib.auth
 
assert HasPermissionAny
 
assert HasRepoGroupPermissionLevel
 
assert HasRepoPermissionLevel
 
# from utils2
 
assert age
 
assert fmt_date
 
assert link_to_ref
 
assert shorter
 
assert time_to_datetime
 
# from vcs
 
@@ -840,230 +842,24 @@ def fancy_file_stats(stats):
 
    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.
 
    >>> urlify_text("Urlify http://example.com/ and 'https://example.com' *and* <b>markup/b>")
 
    literal('Urlify <a href="http://example.com/">http://example.com/</a> and &#39;<a href="https://example.com&apos">https://example.com&apos</a>; <b>*and*</b> &lt;b&gt;markup/b&gt;')
 
    """
 

	
 
    def _replace(match_obj):
 
        match_url = match_obj.group('url')
 
        if match_url is not None:
 
            return '<a href="%(url)s">%(url)s</a>' % {'url': match_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:
 
            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:
 
        assert kallithea.CONFIG['sqlalchemy.url'] # make sure config has been loaded
 

	
 
        # Build chain of urlify functions, starting with not doing any transformation
 
        def tmp_urlify_issues_f(s):
 
            return s
 

	
 
        issue_pat_re = re.compile(r'issue_pat(.*)')
 
        for k in kallithea.CONFIG:
 
            # 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 = kallithea.CONFIG.get(k)
 
            issue_server_link = kallithea.CONFIG.get('issue_server_link%s' % suffix)
 
            issue_sub = kallithea.CONFIG.get('issue_sub%s' % suffix)
 
            issue_prefix = kallithea.CONFIG.get('issue_prefix%s' % suffix)
 
            if issue_prefix:
 
                log.error('found unsupported issue_prefix%s = %r - use issue_sub%s instead', suffix, issue_prefix, suffix)
 
            if not issue_pat:
 
                log.error('skipping incomplete issue pattern %r: it needs a regexp', k)
 
                continue
 
            if not issue_server_link:
 
                log.error('skipping incomplete issue pattern %r: it needs issue_server_link%s', k, suffix)
 
                continue
 
            if issue_sub is None: # issue_sub can be empty but should be present
 
                log.error('skipping incomplete issue pattern %r: it needs (a potentially empty) issue_sub%s', k, suffix)
 
                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', k, issue_pat, issue_server_link, issue_sub, str(e))
 
                continue
 

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

	
 
            def issues_replace(match_obj,
 
                               issue_server_link=issue_server_link, issue_sub=issue_sub):
 
                try:
 
                    issue_url = match_obj.expand(issue_server_link)
 
                except (IndexError, re.error) as e:
 
                    log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
 
                    issue_url = issue_server_link
 
                issue_url = issue_url.replace('{repo}', repo_name)
 
                issue_url = issue_url.replace('{repo_name}', repo_name.split(kallithea.URL_SEP)[-1])
 
                # if issue_sub is empty use the matched issue reference verbatim
 
                if not issue_sub:
 
                    issue_text = match_obj.group()
 
                else:
 
                    try:
 
                        issue_text = match_obj.expand(issue_sub)
 
                    except (IndexError, re.error) as e:
 
                        log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
 
                        issue_text = match_obj.group()
 

	
 
                return (
 
                    '<a class="issue-tracker-link" href="%(url)s">'
 
                    '%(text)s'
 
                    '</a>'
 
                    ) % {
 
                     'url': issue_url,
 
                     'text': issue_text,
 
                    }
 

	
 
            def tmp_urlify_issues_f(s, issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f):
 
                return 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 = urlify_text(source, repo_name=repo_name)
 
    return literal('<div class="formatted-fixed">%s</div>' % s)
 

	
 

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

	
 

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

	
 

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

	
 

	
kallithea/lib/webutils.py
Show inline comments
 
@@ -318,12 +318,217 @@ url_re = re.compile(r'''\bhttps?://(?:[\
 
# Matching is greedy so we don't have to look beyond the end.
 
MENTIONS_REGEX = re.compile(r'(?:^|(?<=[^a-zA-Z0-9]))@([a-zA-Z0-9][-_.a-zA-Z0-9]*[a-zA-Z0-9])')
 

	
 

	
 
def extract_mentioned_usernames(text):
 
    r"""
 
    Returns list of (possible) usernames @mentioned in given text.
 

	
 
    >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
 
    ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'gg', 'hh', 'zz']
 
    """
 
    return MENTIONS_REGEX.findall(text)
 

	
 

	
 
_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.
 
    >>> urlify_text("Urlify http://example.com/ and 'https://example.com' *and* <b>markup/b>")
 
    literal('Urlify <a href="http://example.com/">http://example.com/</a> and &#39;<a href="https://example.com&apos">https://example.com&apos</a>; <b>*and*</b> &lt;b&gt;markup/b&gt;')
 
    """
 

	
 
    def _replace(match_obj):
 
        match_url = match_obj.group('url')
 
        if match_url is not None:
 
            return '<a href="%(url)s">%(url)s</a>' % {'url': match_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:
 
            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:
 
        assert kallithea.CONFIG['sqlalchemy.url'] # make sure config has been loaded
 

	
 
        # Build chain of urlify functions, starting with not doing any transformation
 
        def tmp_urlify_issues_f(s):
 
            return s
 

	
 
        issue_pat_re = re.compile(r'issue_pat(.*)')
 
        for k in kallithea.CONFIG:
 
            # 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 = kallithea.CONFIG.get(k)
 
            issue_server_link = kallithea.CONFIG.get('issue_server_link%s' % suffix)
 
            issue_sub = kallithea.CONFIG.get('issue_sub%s' % suffix)
 
            issue_prefix = kallithea.CONFIG.get('issue_prefix%s' % suffix)
 
            if issue_prefix:
 
                log.error('found unsupported issue_prefix%s = %r - use issue_sub%s instead', suffix, issue_prefix, suffix)
 
            if not issue_pat:
 
                log.error('skipping incomplete issue pattern %r: it needs a regexp', k)
 
                continue
 
            if not issue_server_link:
 
                log.error('skipping incomplete issue pattern %r: it needs issue_server_link%s', k, suffix)
 
                continue
 
            if issue_sub is None: # issue_sub can be empty but should be present
 
                log.error('skipping incomplete issue pattern %r: it needs (a potentially empty) issue_sub%s', k, suffix)
 
                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', k, issue_pat, issue_server_link, issue_sub, str(e))
 
                continue
 

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

	
 
            def issues_replace(match_obj,
 
                               issue_server_link=issue_server_link, issue_sub=issue_sub):
 
                try:
 
                    issue_url = match_obj.expand(issue_server_link)
 
                except (IndexError, re.error) as e:
 
                    log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
 
                    issue_url = issue_server_link
 
                issue_url = issue_url.replace('{repo}', repo_name)
 
                issue_url = issue_url.replace('{repo_name}', repo_name.split(kallithea.URL_SEP)[-1])
 
                # if issue_sub is empty use the matched issue reference verbatim
 
                if not issue_sub:
 
                    issue_text = match_obj.group()
 
                else:
 
                    try:
 
                        issue_text = match_obj.expand(issue_sub)
 
                    except (IndexError, re.error) as e:
 
                        log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
 
                        issue_text = match_obj.group()
 

	
 
                return (
 
                    '<a class="issue-tracker-link" href="%(url)s">'
 
                    '%(text)s'
 
                    '</a>'
 
                    ) % {
 
                     'url': issue_url,
 
                     'text': issue_text,
 
                    }
 

	
 
            def tmp_urlify_issues_f(s, issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f):
 
                return 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 = urlify_text(source, repo_name=repo_name)
 
    return literal('<div class="formatted-fixed">%s</div>' % s)
kallithea/model/notification.py
Show inline comments
 
@@ -24,24 +24,25 @@ Original author and date, and relevant c
 
:author: marcink
 
:copyright: (c) 2013 RhodeCode GmbH, and others.
 
:license: GPLv3, see LICENSE.md for more details.
 
"""
 

	
 
import datetime
 
import logging
 

	
 
from tg import app_globals
 
from tg import tmpl_context as c
 
from tg.i18n import ugettext as _
 

	
 
from kallithea.lib import webutils
 
from kallithea.lib.utils2 import fmt_date
 
from kallithea.model import async_tasks, db
 

	
 

	
 
log = logging.getLogger(__name__)
 

	
 

	
 
class NotificationModel(object):
 

	
 
    TYPE_CHANGESET_COMMENT = 'cs_comment'
 
    TYPE_MESSAGE = 'message'
 
    TYPE_MENTION = 'mention' # not used
 
@@ -56,25 +57,24 @@ class NotificationModel(object):
 

	
 
        Creates notification of given type
 

	
 
        :param created_by: int, str or User instance. User who created this
 
            notification
 
        :param body:
 
        :param recipients: list of int, str or User objects, when None
 
            is given send to all admins
 
        :param type_: type of notification
 
        :param with_email: send email with this notification
 
        :param email_kwargs: additional dict to pass as args to email template
 
        """
 
        import kallithea.lib.helpers as h
 
        email_kwargs = email_kwargs or {}
 
        if recipients and not getattr(recipients, '__iter__', False):
 
            raise Exception('recipients must be a list or iterable')
 

	
 
        created_by_obj = db.User.guess_instance(created_by)
 

	
 
        recipients_objs = set()
 
        if recipients:
 
            for u in recipients:
 
                obj = db.User.guess_instance(u)
 
                if obj is not None:
 
                    recipients_objs.add(obj)
 
@@ -94,25 +94,25 @@ class NotificationModel(object):
 

	
 
        if not with_email:
 
            return
 

	
 
        headers = {}
 
        headers['X-Kallithea-Notification-Type'] = type_
 
        if 'threading' in email_kwargs:
 
            headers['References'] = ' '.join('<%s>' % x for x in email_kwargs['threading'])
 

	
 
        # this is passed into template
 
        created_on = fmt_date(datetime.datetime.now())
 
        html_kwargs = {
 
                  'body': None if body is None else h.render_w_mentions(body, repo_name),
 
                  'body': None if body is None else webutils.render_w_mentions(body, repo_name),
 
                  'when': created_on,
 
                  'user': created_by_obj.username,
 
                  }
 

	
 
        txt_kwargs = {
 
                  'body': body,
 
                  'when': created_on,
 
                  'user': created_by_obj.username,
 
                  }
 

	
 
        html_kwargs.update(email_kwargs)
 
        txt_kwargs.update(email_kwargs)
kallithea/model/repo.py
Show inline comments
 
@@ -131,26 +131,25 @@ class RepoModel(object):
 
        def rss_lnk(repo_name):
 
            return _render("rss", repo_name)
 

	
 
        def atom_lnk(repo_name):
 
            return _render("atom", repo_name)
 

	
 
        def last_rev(repo_name, cs_cache):
 
            return _render('revision', repo_name, cs_cache.get('revision'),
 
                           cs_cache.get('raw_id'), cs_cache.get('author'),
 
                           cs_cache.get('message'))
 

	
 
        def desc(desc):
 
            import kallithea.lib.helpers as h
 
            return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
 
            return webutils.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
 

	
 
        def state(repo_state):
 
            return _render("repo_state", repo_state)
 

	
 
        def repo_actions(repo_name):
 
            return _render('repo_actions', repo_name)
 

	
 
        def owner_actions(owner_id, username):
 
            return _render('user_name', owner_id, username)
 

	
 
        repos_data = []
 

	
kallithea/tests/functional/test_files.py
Show inline comments
 
# -*- coding: utf-8 -*-
 
import json
 
import mimetypes
 
import posixpath
 

	
 
import kallithea.lib.helpers
 
from kallithea.lib import webutils
 
from kallithea.model import db, meta
 
from kallithea.tests import base
 
from kallithea.tests.fixture import Fixture
 

	
 

	
 
fixture = Fixture()
 

	
 
ARCHIVE_SPECS = {
 
    '.tar.bz2': ('application/x-bzip2', 'tbz2', ''),
 
    '.tar.gz': ('application/x-gzip', 'tgz', ''),
 
    '.zip': ('application/zip', 'zip', ''),
 
}
 
@@ -87,25 +87,25 @@ class TestFilesController(base.TestContr
 
                  (0, 'b986218ba1c9b0d6a259fac9b050b1724ed8e545')]:
 

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

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

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

	
 
        response.mustcontain("""<div class="formatted-fixed">Partially implemented <a class="issue-tracker-link" href="https://issues.example.com/vcs_test_hg/issue/16">#16</a>. filecontent/commit message/author/node name are safe_unicode now.<br/>"""
 
"""In addition some other __str__ are unicode as well<br/>"""
 
"""Added test for unicode<br/>"""
 
"""Improved test to clone into uniq repository.<br/>"""
 
"""removed extra unicode conversion in diff.</div>
 
""")
kallithea/tests/other/test_libs.py
Show inline comments
 
@@ -102,37 +102,36 @@ class TestLibs(base.TestController):
 
                           ('TRUE', True),
 
                           ('T', True),
 
                           ('False', False),
 
                           ('F', False),
 
                           ('FALSE', False),
 
                           ('0', False),
 
    ])
 
    def test_asbool(self, str_bool, expected):
 
        from kallithea.lib.utils2 import asbool
 
        assert asbool(str_bool) == expected
 

	
 
    def test_mention_extractor(self):
 
        from kallithea.lib.webutils 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))
 
        assert expected == set(webutils.extract_mentioned_usernames(sample))
 

	
 
    @base.parametrize('age_args,expected', [
 
        (dict(), 'just now'),
 
        (dict(seconds= -1), '1 second ago'),
 
        (dict(seconds= -60 * 2), '2 minutes ago'),
 
        (dict(hours= -1), '1 hour ago'),
 
        (dict(hours= -24), '1 day ago'),
 
        (dict(hours= -24 * 5), '5 days ago'),
 
        (dict(months= -1), '1 month ago'),
 
        (dict(months= -1, days= -2), '1 month and 2 days ago'),
 
        (dict(months= -1, days= -20), '1 month and 19 days ago'),
 
        (dict(years= -1, months= -1), '1 year and 1 month ago'),
 
@@ -188,25 +187,25 @@ class TestLibs(base.TestController):
 
        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]]"
 
        )
 
        res = h.urlify_text(sample, stylize=True)
 
        res = webutils.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):
 
        _md5 = lambda s: hashlib.md5(safe_bytes(s)).hexdigest()
 

	
 
        # mock tg.tmpl_context
 
        def fake_tmpl_context(_url):
 
@@ -306,25 +305,25 @@ class TestLibs(base.TestController):
 
       """,
 
       """Multi line<br/>"""
 
       """       url[123123123123]<br/>"""
 
       """       some text url[123123123123]<br/>"""
 
       """       sometimes !"""),
 
    ])
 
    def test_urlify_text(self, sample, expected):
 
        expected = self._quick_url(expected)
 

	
 
        with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
 
            lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
 
        ):
 
            assert h.urlify_text(sample, 'repo_name') == expected
 
            assert webutils.urlify_text(sample, 'repo_name') == expected
 

	
 
    @base.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",
 
@@ -361,38 +360,38 @@ class TestLibs(base.TestController):
 
       "-"),
 
      ("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_)
 
        with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
 
            lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
 
        ):
 
            assert h.urlify_text(sample, 'repo_name', stylize=True) == expected
 
            assert webutils.urlify_text(sample, 'repo_name', stylize=True) == expected
 

	
 
    @base.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):
 
        with mock.patch('kallithea.lib.webutils.UrlGenerator.__call__',
 
            lambda self, name, **kwargs: dict(changeset_home='/%(repo_name)s/changeset/%(revision)s')[name] % kwargs,
 
        ):
 
            assert h.urlify_text(sample, 'repo_name', link_='#the-link') == expected
 
            assert webutils.urlify_text(sample, 'repo_name', link_='#the-link') == expected
 

	
 
    @base.parametrize('issue_pat,issue_server,issue_sub,sample,expected', [
 
        (r'#(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
 
            '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>"""),
 
        (r'(?:\s*#)(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
 
            '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>"""),
 
        # to require whitespace before the issue reference, one may be tempted to use \b...
 
        (r'\bPR(\d+)', 'http://foo/{repo}/issue/\\1', '#\\1',
 
@@ -462,27 +461,27 @@ class TestLibs(base.TestController):
 
            'pullrequest CPH-789 is similar to PRBRU-747',
 
            """<a class="issue-tracker-link" href="http://foo/CPH/pullrequest/789/">PR-CPH-789</a> is similar to """
 
            """<a class="issue-tracker-link" href="http://foo/BRU/pullrequest/747/">PR-BRU-747</a>"""),
 
    ])
 
    def test_urlify_issues(self, issue_pat, issue_server, issue_sub, sample, expected):
 
        config_stub = {
 
            'sqlalchemy.url': 'foo',
 
            'issue_pat': issue_pat,
 
            'issue_server_link': issue_server,
 
            'issue_sub': issue_sub,
 
        }
 
        # force recreation of lazy function
 
        with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
 
        with mock.patch('kallithea.lib.webutils._urlify_issues_f', None):
 
            with mock.patch('kallithea.CONFIG', config_stub):
 
                assert h.urlify_text(sample, 'repo_name') == expected
 
                assert webutils.urlify_text(sample, 'repo_name') == expected
 

	
 
    @base.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">FAIL67</a>'),
 
        ('issue FAILMORE89', 'issue FAILMORE89'), # no match because absent prefix
 
    ])
 
    def test_urlify_issues_multiple_issue_patterns(self, sample, expected):
 
        config_stub = {
 
@@ -494,27 +493,27 @@ class TestLibs(base.TestController):
 
            'issue_server_link_pr': r'http://pr/{repo}/pr/\1',
 
            'issue_sub_pr': r'PR#\1',
 
            'issue_pat_bug': r'(?:BUG|bug|issue) ?#?(\d+)',
 
            'issue_server_link_bug': r'http://bug/{repo}/bug/\1',
 
            'issue_sub_bug': r'bug#\1',
 
            'issue_pat_empty_prefix': r'FAIL(\d+)',
 
            'issue_server_link_empty_prefix': r'http://fail/{repo}/\1',
 
            'issue_sub_empty_prefix': r'',
 
            'issue_pat_absent_prefix': r'FAILMORE(\d+)',
 
            'issue_server_link_absent_prefix': r'http://failmore/{repo}/\1',
 
        }
 
        # force recreation of lazy function
 
        with mock.patch('kallithea.lib.helpers._urlify_issues_f', None):
 
        with mock.patch('kallithea.lib.webutils._urlify_issues_f', None):
 
            with mock.patch('kallithea.CONFIG', config_stub):
 
                assert h.urlify_text(sample, 'repo_name') == expected
 
                assert webutils.urlify_text(sample, 'repo_name') == expected
 

	
 
    @base.parametrize('test,expected', [
 
      ("", None),
 
      ("/_2", None),
 
      ("_2", 2),
 
      ("_2/", None),
 
    ])
 
    def test_get_permanent_id(self, test, expected):
 
        from kallithea.lib.utils import _get_permanent_id
 
        extracted = _get_permanent_id(test)
 
        assert extracted == expected, 'url:%s, got:`%s` expected: `%s`' % (test, base._test, expected)
 

	
0 comments (0 inline, 0 general)