kallithea Changeset - 3cef2caf44f3

Changeset - 3cef2caf44f3

Parent rev.

Child rev.

[Not reviewed]

default

0 9 0

Mads Kiilerich (mads) - 5 years ago 2020-11-13 17:11:20
mads@kiilerich.com

Grafted from: 1e8e18e583df

lib: move some template filter functions from utils2 to webutils

While quite Kallithea specific, we prefer to have these functions in webutils
where they soon can be exposed when templates don't need the whole helpers
module.

9 files changed with 155 insertions and 145 deletions:

kallithea/controllers/admin/users.py

kallithea/controllers/feed.py

kallithea/lib/helpers.py

kallithea/lib/utils2.py

121

kallithea/lib/webutils.py

134

kallithea/model/comment.py

kallithea/model/notification.py

kallithea/model/pull_request.py

kallithea/tests/other/test_libs.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/users.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.controllers.admin.users
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Users crud controller
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 4, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 import traceback
 import formencode
 from formencode import htmlfill
 from sqlalchemy.sql.expression import func
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 import kallithea
 import kallithea.lib.helpers as h
 from kallithea.controllers import base
 from kallithea.lib import auth_modules, webutils
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.exceptions import DefaultUserException, UserCreationError, UserOwnsReposException
 from kallithea.lib.utils2 import datetime_to_time, fmt_date, generate_api_key, safe_int
 from kallithea.lib.webutils import url
 from kallithea.lib.utils2 import datetime_to_time, generate_api_key, safe_int
 from kallithea.lib.webutils import fmt_date, url
 from kallithea.model import db, meta, userlog
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.forms import CustomDefaultPermissionsForm, UserForm
 from kallithea.model.ssh_key import SshKeyModel, SshKeyModelException
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class UsersController(base.BaseController):
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(UsersController, self)._before(*args, **kwargs)
     def index(self, format='html'):
         c.users_list = db.User.query().order_by(db.User.username) \
                         .filter_by(is_default_user=False) \
                         .order_by(func.lower(db.User.username)) \
                         .all()
         users_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         grav_tmpl = '<div class="gravatar">%s</div>'
         def username(user_id, username):
             return template.get_def("user_name") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         def user_actions(user_id, username):
             return template.get_def("user_actions") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         for user in c.users_list:
             users_data.append({
                 "gravatar": grav_tmpl % h.gravatar(user.email, size=20),
                 "raw_name": user.username,
                 "username": username(user.user_id, user.username),
                 "firstname": webutils.escape(user.name),
                 "lastname": webutils.escape(user.lastname),
                 "last_login": fmt_date(user.last_login),
                 "last_login_raw": datetime_to_time(user.last_login),
                 "active": h.boolicon(user.active),
                 "admin": h.boolicon(user.admin),
                 "extern_type": user.extern_type,
                 "extern_name": user.extern_name,
                 "action": user_actions(user.user_id, user.username),
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": users_data
+        }
         return base.render('admin/users/users.html')
     def create(self):
         c.default_extern_type = db.User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         user_model = UserModel()
         user_form = UserForm()()
         try:
             form_result = user_form.to_python(dict(request.POST))
             user = user_model.create(form_result)
             userlog.action_logger(request.authuser, 'admin_created_user:%s' % user.username,
                           None, request.ip_addr)
             webutils.flash(_('Created user %s') % user.username,
                     category='success')
             meta.Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 base.render('admin/users/user_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except UserCreationError as e:
             webutils.flash(e, 'error')
         except Exception:
             log.error(traceback.format_exc())
             webutils.flash(_('Error occurred during creation of user %s')
                     % request.POST.get('username'), category='error')
         raise HTTPFound(location=url('edit_user', id=user.user_id))
     def new(self, format='html'):
         c.default_extern_type = db.User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         return base.render('admin/users/user_add.html')
     def update(self, id):

kallithea/controllers/feed.py

➞

Show inline comments

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

kallithea/lib/helpers.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/>.
 """
 Helper functions
 Consists of functions to typically be used within templates, but also
 available to Controllers. This module is available to both as 'h'.
 """
 import hashlib
 import logging
 import re
 import textwrap
 import urllib.parse
 from beaker.cache import cache_region
 from pygments import highlight as code_highlight
 from pygments.formatters.html import HtmlFormatter
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 import kallithea
 from kallithea.lib.annotate import annotate_highlight
 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.utils2 import AttributeDict, asbool, credentials_filter, link_to_ref, safe_bytes, safe_int, safe_str, time_to_datetime
 from kallithea.lib.vcs.backends.base import BaseChangeset, EmptyChangeset
 from kallithea.lib.vcs.exceptions import ChangesetDoesNotExistError
 from kallithea.lib.vcs.utils import author_email, author_name
 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.lib.webutils import (HTML, Option, age, canonical_url, checkbox, chop_at, end_form, escape, fmt_date, 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, shorter, 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 Option
 assert age
 assert canonical_url
 assert checkbox
 assert chop_at
 assert end_form
 assert fmt_date
 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 shorter
 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 credentials_filter
 assert fmt_date
 assert link_to_ref
 assert shorter
 assert time_to_datetime
 # from vcs
 assert EmptyChangeset
 log = logging.getLogger(__name__)
 def FID(raw_id, path):
     """
     Creates a unique ID for filenode based on it's hash of path and revision
     it's safe to use in urls
     """
     return 'C-%s-%s' % (short_id(raw_id), hashlib.md5(safe_bytes(path)).hexdigest()[:12])
 def get_ignore_whitespace_diff(GET):
     """Return true if URL requested whitespace to be ignored"""
     return bool(GET.get('ignorews'))
 def ignore_whitespace_link(GET, anchor=None):
     """Return snippet with link to current URL with whitespace ignoring toggled"""
     params = dict(GET)  # ignoring duplicates
     if get_ignore_whitespace_diff(GET):
         params.pop('ignorews')
         title = _("Show whitespace changes")
     else:
         params['ignorews'] = '1'
         title = _("Ignore whitespace changes")
     params['anchor'] = anchor
     return link_to(
         literal('<i class="icon-strike"></i>'),
         url.current(**params),
         title=title,
         **{'data-toggle': 'tooltip'})
 def get_diff_context_size(GET):
     """Return effective context size requested in URL"""
     return safe_int(GET.get('context'), default=3)
 def increase_context_link(GET, anchor=None):
     """Return snippet with link to current URL with double context size"""
     context = get_diff_context_size(GET) * 2
     params = dict(GET)  # ignoring duplicates
     params['context'] = str(context)
     params['anchor'] = anchor
     return link_to(
         literal('<i class="icon-sort"></i>'),
         url.current(**params),
         title=_('Increase diff context to %(num)s lines') % {'num': context},
         **{'data-toggle': 'tooltip'})
 def files_breadcrumbs(repo_name, rev, paths):
     url_l = [link_to(repo_name, url('files_home',
                                     repo_name=repo_name,
                                     revision=rev, f_path=''),
                      class_='ypjax-link')]
     paths_l = paths.split('/')
     for cnt, p in enumerate(paths_l):
         if p != '':
             url_l.append(link_to(p,
                                  url('files_home',
                                      repo_name=repo_name,
                                      revision=rev,
                                      f_path='/'.join(paths_l[:cnt + 1])
                                      ),
                                  class_='ypjax-link'
+                                 )
+                         )
     return literal('/'.join(url_l))
 class CodeHtmlFormatter(HtmlFormatter):
     """
     My code Html Formatter for source codes
     """
     def wrap(self, source, outfile):
         return self._wrap_div(self._wrap_pre(self._wrap_code(source)))
     def _wrap_code(self, source):
         for cnt, it in enumerate(source):
             i, t = it
             t = '<span id="L%s">%s</span>' % (cnt + 1, t)
             yield i, t
     def _wrap_tablelinenos(self, inner):
         inner_lines = []
         lncount = 0
         for t, line in inner:
             if t:
                 lncount += 1

kallithea/lib/utils2.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.lib.utils2
 ~~~~~~~~~~~~~~~~~~~~
 Some simple helper functions.
 Note: all these functions should be independent of Kallithea classes, i.e.
 models, controllers, etc.  to prevent import cycles.
 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: Jan 5, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import binascii
 import datetime
 import hashlib
 import json
 import logging
 import os
 import re
 import string
 import sys
 import time
 import urllib.parse
 from distutils.version import StrictVersion
 import bcrypt
 import urlobject
 from dateutil import relativedelta
 from sqlalchemy.engine import url as sa_url
 from sqlalchemy.exc import ArgumentError
 from tg import tmpl_context
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from tg.support.converters import asbool, aslist
 from webhelpers2.text import collapse, remove_formatting, strip_tags
 import kallithea
 from kallithea.lib import webutils
 from kallithea.lib.vcs.backends.base import BaseRepository, EmptyChangeset
 from kallithea.lib.vcs.backends.git.repository import GitRepository
 from kallithea.lib.vcs.conf import settings
 from kallithea.lib.vcs.exceptions import RepositoryError
 from kallithea.lib.vcs.utils import ascii_bytes, ascii_str, safe_bytes, safe_str  # re-export
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 try:
     import pwd
 except ImportError:
     pass
 log = logging.getLogger(__name__)
 # mute pyflakes "imported but unused"
 assert asbool
 assert aslist
 assert ascii_bytes
 assert ascii_str
 assert safe_bytes
 assert safe_str
 assert LazyProperty
 # get current umask value without changing it
 umask = os.umask(0)
 os.umask(umask)
 def convert_line_endings(line, mode):
     """
     Converts a given line  "line end" according to given mode
     Available modes are::
 - Unix
 - Mac
 - DOS
     :param line: given line to convert
     :param mode: mode to convert to
     :rtype: str
     :return: converted line according to mode
     """
     if mode == 0:
         line = line.replace('\r\n', '\n')
         line = line.replace('\r', '\n')
     elif mode == 1:
         line = line.replace('\r\n', '\r')
         line = line.replace('\n', '\r')
     elif mode == 2:
         line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
     return line
 def detect_mode(line, default):
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
     :param line: str line
     :param default: default
     :rtype: int
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 def generate_api_key():
     """
     Generates a random (presumably unique) API key.
     This value is used in URLs and "Bearer" HTTP Authorization headers,
     which in practice means it should only contain URL-safe characters
     (RFC 3986):
         unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
     """
     # Hexadecimal certainly qualifies as URL-safe.
     return ascii_str(binascii.hexlify(os.urandom(20)))
 def safe_int(val, default=None):
     """
     Returns int() of val if val is not convertable to int use default
     instead
     :param val:
     :param default:
     """
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
     return val
 def remove_suffix(s, suffix):
     if s.endswith(suffix):
         s = s[:-1 * len(suffix)]
     return s
 def remove_prefix(s, prefix):
     if s.startswith(prefix):
         s = s[len(prefix):]
     return s
 def shorter(s, size=20, firstline=False, postfix='...'):
     """Truncate s to size, including the postfix string if truncating.
     If firstline, truncate at newline.
     """
     if firstline:
         s = s.split('\n', 1)[0].rstrip()
     if len(s) > size:
         return s[:size - len(postfix)] + postfix
     return s
 def age(prevdate, show_short_version=False, now=None):
     """
     turns a datetime into an age string.
     If show_short_version is True, then it will generate a not so accurate but shorter string,
     example: 2days ago, instead of 2 days and 23 hours ago.
     :param prevdate: datetime object
     :param show_short_version: if it should approximate the date and return a shorter string
     :rtype: str
     :returns: str words describing age
     """
     now = now or datetime.datetime.now()
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     future = False
     if prevdate > now:
         now, prevdate = prevdate, now
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     for part in order:
         d = relativedelta.relativedelta(now, prevdate)
         deltas[part] = getattr(d, part + 's')
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
         part = order[num]
         carry_part = order[num - 1]
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if prevdate.month == 2 and (prevdate.year % 4 == 0 and
             (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)
         ):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[prevdate.month - 1]
         deltas['month'] -= 1
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
     # In short version, we want nicer handling of ages of more than a year
     if show_short_version:
         if deltas['year'] == 1:
             # ages between 1 and 2 years: show as months
             deltas['month'] += 12
             deltas['year'] = 0
         if deltas['year'] >= 2:
             # ages 2+ years: round
             if deltas['month'] > 6:
                 deltas['year'] += 1
                 deltas['month'] = 0
     # Format the result
     fmt_funcs = {
         'year': lambda d: ungettext('%d year', '%d years', d) % d,
         'month': lambda d: ungettext('%d month', '%d months', d) % d,
         'day': lambda d: ungettext('%d day', '%d days', d) % d,
         'hour': lambda d: ungettext('%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext('%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext('%d second', '%d seconds', d) % d,
+    }
     for i, part in enumerate(order):
         value = deltas[part]
         if value == 0:
             continue
         if i < 5:
             sub_part = order[i + 1]
             sub_value = deltas[sub_part]
         else:
             sub_value = 0
         if sub_value == 0 or show_short_version:
             if future:
                 return _('in %s') % fmt_funcs[part](value)
             else:
                 return _('%s ago') % fmt_funcs[part](value)
         if future:
             return _('in %s and %s') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
         else:
             return _('%s and %s ago') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
     return _('just now')
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S")
     return ""
 def uri_filter(uri):
     """
     Removes user:password from given url string
     :param uri:
     :rtype: str
     :returns: filtered list of strings
     """
     if not uri:
         return []
     proto = ''
     for pat in ('https://', 'http://', 'git://'):
         if uri.startswith(pat):
             uri = uri[len(pat):]
             proto = pat
             break
     # remove passwords and username
     uri = uri[uri.find('@') + 1:]
     # get the port
     cred_pos = uri.find(':')
     if cred_pos == -1:
         host, port = uri, None
     else:
         host, port = uri[:cred_pos], uri[cred_pos + 1:]
     return [_f for _f in [proto, host, port] if _f]
 def credentials_filter(uri):
     """
     Returns a url with removed credentials
     :param uri:
     """
     uri = uri_filter(uri)
     # check if we have port
     if len(uri) > 2 and uri[2]:
         uri[2] = ':' + uri[2]
     return ''.join(uri)
 def get_clone_url(clone_uri_tmpl, prefix_url, repo_name, repo_id, username=None):
     parsed_url = urlobject.URLObject(prefix_url)
     prefix = urllib.parse.unquote(parsed_url.path.rstrip('/'))
     try:
         system_user = pwd.getpwuid(os.getuid()).pw_name
     except NameError: # TODO: support all systems - especially Windows
         system_user = 'kallithea' # hardcoded default value ...
     args = {
         'scheme': parsed_url.scheme,
         'user': urllib.parse.quote(username or ''),
         'netloc': parsed_url.netloc + prefix,  # like "hostname:port/prefix" (with optional ":port" and "/prefix")
         'prefix': prefix, # undocumented, empty or starting with /
         'repo': repo_name,
         'repoid': str(repo_id),
         'system_user': system_user,
         'hostname': parsed_url.hostname,
+    }
     url = re.sub('{([^{}]+)}', lambda m: args.get(m.group(1), m.group(0)), clone_uri_tmpl)
     # remove leading @ sign if it's present. Case of empty user
     url_obj = urlobject.URLObject(url)
     if not url_obj.username:
         url_obj = url_obj.with_username(None)
     return str(url_obj)
 def short_ref_name(ref_type, ref_name):
     """Return short description of PR ref - revs will be truncated"""
     if ref_type == 'rev':
         return ref_name[:12]
     return ref_name
 def link_to_ref(repo_name, ref_type, ref_name, rev=None):
     """
     Return full markup for a PR ref 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_name(ref_type, ref_name)
     if ref_type == 'branch':
         u = webutils.url('changelog_home', repo_name=repo_name, branch=ref_name)
     else:
         u = webutils.url('changeset_home', repo_name=repo_name, revision=ref_name)
     l = webutils.link_to(repo_name + '#' + txt, u)
     if rev and ref_type != 'rev':
         l = webutils.literal('%s (%s)' % (l, webutils.link_to(rev[:12], webutils.url('changeset_home', repo_name=repo_name, revision=rev))))

kallithea/lib/webutils.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.lib.webutils
 ~~~~~~~~~~~~~~~~~~~~~~
 Helper functions that may rely on the current WSGI request, exposed in the TG2
 thread-local "global" variables. It should have few dependencies so it can be
 imported anywhere - just like the global variables can be used everywhere.
 """
 import datetime
 import json
 import logging
 import random
 import re
 from dateutil import relativedelta
 from tg import request, session
 from tg.i18n import ugettext as _
 from tg.i18n import ungettext
 from webhelpers2.html import HTML, escape, literal
 from webhelpers2.html.tags import NotGiven, Option, Options, _input
 from webhelpers2.html.tags import _make_safe_id_component as safeid
 from webhelpers2.html.tags import checkbox, end_form
 from webhelpers2.html.tags import form as insecure_form
 from webhelpers2.html.tags import hidden, link_to, password, radio
 from webhelpers2.html.tags import select as webhelpers2_select
 from webhelpers2.html.tags import submit, text, textarea
 from webhelpers2.number import format_byte_size
 from webhelpers2.text import chop_at, truncate, wrap_paragraphs
 import kallithea
 log = logging.getLogger(__name__)
 # mute pyflakes "imported but unused"
 assert Option
 assert checkbox
 assert chop_at
 assert end_form
 assert escape
 assert format_byte_size
 assert link_to
 assert literal
 assert password
 assert radio
 assert safeid
 assert submit
 assert text
 assert textarea
 assert truncate
 assert wrap_paragraphs
+#
 # General Kallithea URL handling
+#
 class UrlGenerator(object):
     """Emulate pylons.url in providing a wrapper around routes.url
     This code was added during migration from Pylons to Turbogears2. Pylons
     already provided a wrapper like this, but Turbogears2 does not.
     When the routing of Kallithea is changed to use less Routes and more
     Turbogears2-style routing, this class may disappear or change.
     url() (the __call__ method) returns the URL based on a route name and
     arguments.
     url.current() returns the URL of the current page with arguments applied.
     Refer to documentation of Routes for details:
     https://routes.readthedocs.io/en/latest/generating.html#generation
     """
     def __call__(self, *args, **kwargs):
         return request.environ['routes.url'](*args, **kwargs)
     def current(self, *args, **kwargs):
         return request.environ['routes.url'].current(*args, **kwargs)
 url = UrlGenerator()
 def canonical_url(*args, **kargs):
     '''Like url(x, qualified=True), but returns url that not only is qualified
     but also canonical, as configured in canonical_url'''
     try:
         parts = kallithea.CONFIG.get('canonical_url', '').split('://', 1)
         kargs['host'] = parts[1]
         kargs['protocol'] = parts[0]
     except IndexError:
         kargs['qualified'] = True
     return url(*args, **kargs)
 def canonical_hostname():
     '''Return canonical hostname of system'''
     try:
         parts = kallithea.CONFIG.get('canonical_url', '').split('://', 1)
         return parts[1].split('/', 1)[0]
     except IndexError:
         parts = url('home', qualified=True).split('://', 1)
         return parts[1].split('/', 1)[0]
+#
 # Custom Webhelpers2 stuff
+#
 def html_escape(s):
     """Return string with all html escaped.
     This is also safe for javascript in html but not necessarily correct.
     """
@@ @@ -439,96 +443,226 @@ def _linkify_others(t, l): @@
         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)
+#
 # Simple filters
+#
 def shorter(s, size=20, firstline=False, postfix='...'):
     """Truncate s to size, including the postfix string if truncating.
     If firstline, truncate at newline.
     """
     if firstline:
         s = s.split('\n', 1)[0].rstrip()
     if len(s) > size:
         return s[:size - len(postfix)] + postfix
     return s
 def age(prevdate, show_short_version=False, now=None):
     """
     turns a datetime into an age string.
     If show_short_version is True, then it will generate a not so accurate but shorter string,
     example: 2days ago, instead of 2 days and 23 hours ago.
     :param prevdate: datetime object
     :param show_short_version: if it should approximate the date and return a shorter string
     :rtype: str
     :returns: str words describing age
     """
     now = now or datetime.datetime.now()
     order = ['year', 'month', 'day', 'hour', 'minute', 'second']
     deltas = {}
     future = False
     if prevdate > now:
         now, prevdate = prevdate, now
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     for part in order:
         d = relativedelta.relativedelta(now, prevdate)
         deltas[part] = getattr(d, part + 's')
     # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,
     # not 1 hour, -59 minutes and -59 seconds)
     for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours
         part = order[num]
         carry_part = order[num - 1]
         if deltas[part] < 0:
             deltas[part] += length
             deltas[carry_part] -= 1
     # Same thing for days except that the increment depends on the (variable)
     # number of days in the month
     month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
     if deltas['day'] < 0:
         if prevdate.month == 2 and (prevdate.year % 4 == 0 and
             (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)
         ):
             deltas['day'] += 29
         else:
             deltas['day'] += month_lengths[prevdate.month - 1]
         deltas['month'] -= 1
     if deltas['month'] < 0:
         deltas['month'] += 12
         deltas['year'] -= 1
     # In short version, we want nicer handling of ages of more than a year
     if show_short_version:
         if deltas['year'] == 1:
             # ages between 1 and 2 years: show as months
             deltas['month'] += 12
             deltas['year'] = 0
         if deltas['year'] >= 2:
             # ages 2+ years: round
             if deltas['month'] > 6:
                 deltas['year'] += 1
                 deltas['month'] = 0
     # Format the result
     fmt_funcs = {
         'year': lambda d: ungettext('%d year', '%d years', d) % d,
         'month': lambda d: ungettext('%d month', '%d months', d) % d,
         'day': lambda d: ungettext('%d day', '%d days', d) % d,
         'hour': lambda d: ungettext('%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext('%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext('%d second', '%d seconds', d) % d,
+    }
     for i, part in enumerate(order):
         value = deltas[part]
         if value == 0:
             continue
         if i < 5:
             sub_part = order[i + 1]
             sub_value = deltas[sub_part]
         else:
             sub_value = 0
         if sub_value == 0 or show_short_version:
             if future:
                 return _('in %s') % fmt_funcs[part](value)
             else:
                 return _('%s ago') % fmt_funcs[part](value)
         if future:
             return _('in %s and %s') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
         else:
             return _('%s and %s ago') % (fmt_funcs[part](value),
                 fmt_funcs[sub_part](sub_value))
     return _('just now')
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S")
     return ""
 def capitalize(x):
     return x.capitalize()
 def short_id(x):
     return x[:12]

kallithea/model/comment.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.model.comment
 ~~~~~~~~~~~~~~~~~~~~~~~
 comments model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Nov 11, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import logging
 from collections import defaultdict
 from kallithea.lib import webutils
 from kallithea.lib.utils import extract_mentioned_users
 from kallithea.lib.utils2 import shorter
 from kallithea.model import db, meta, notification
 log = logging.getLogger(__name__)
 def _list_changeset_commenters(revision):
     return (meta.Session().query(db.User)
         .join(db.ChangesetComment.author)
         .filter(db.ChangesetComment.revision == revision)
         .all())
 def _list_pull_request_commenters(pull_request):
     return (meta.Session().query(db.User)
         .join(db.ChangesetComment.author)
         .filter(db.ChangesetComment.pull_request_id == pull_request.pull_request_id)
         .all())
 class ChangesetCommentsModel(object):
     def create_notification(self, repo, comment, author, comment_text,
                                 line_no=None, revision=None, pull_request=None,
                                 status_change=None, closing_pr=False):
         # changeset
         if revision:
             notification_type = notification.NotificationModel.TYPE_CHANGESET_COMMENT
             cs = repo.scm_instance.get_changeset(revision)
             threading = ['%s-rev-%s@%s' % (repo.repo_name, revision, webutils.canonical_hostname())]
             if line_no: # TODO: url to file _and_ line number
                 threading.append('%s-rev-%s-line-%s@%s' % (repo.repo_name, revision, line_no,
                                                            webutils.canonical_hostname()))
             comment_url = webutils.canonical_url('changeset_home',
                 repo_name=repo.repo_name,
                 revision=revision,
                 anchor='comment-%s' % comment.comment_id)
             # get the current participants of this changeset
             recipients = _list_changeset_commenters(revision)
             # add changeset author if it's known locally
             cs_author = db.User.get_from_cs_author(cs.author)
             if not cs_author:
                 # use repo owner if we cannot extract the author correctly
                 # FIXME: just use committer name even if not a user
                 cs_author = repo.owner
             recipients.append(cs_author)
             email_kwargs = {
                 'status_change': status_change,
                 'cs_comment_user': author.full_name_and_username,
                 'cs_target_repo': webutils.canonical_url('summary_home', repo_name=repo.repo_name),
                 'cs_comment_url': comment_url,
                 'cs_url': webutils.canonical_url('changeset_home', repo_name=repo.repo_name, revision=revision),
                 'message': cs.message,
                 'message_short': shorter(cs.message, 50, firstline=True),
+                'message_short': webutils.shorter(cs.message, 50, firstline=True),
                 'cs_author': cs_author,
                 'cs_author_username': cs_author.username,
                 'repo_name': repo.repo_name,
                 'short_id': revision[:12],
                 'branch': cs.branch,
                 'threading': threading,
+            }
         # pull request
         elif pull_request:
             notification_type = notification.NotificationModel.TYPE_PULL_REQUEST_COMMENT
             _org_ref_type, org_ref_name, _org_rev = comment.pull_request.org_ref.split(':')
             _other_ref_type, other_ref_name, _other_rev = comment.pull_request.other_ref.split(':')
             threading = ['%s-pr-%s@%s' % (pull_request.other_repo.repo_name,
                                           pull_request.pull_request_id,
                                           webutils.canonical_hostname())]
             if line_no: # TODO: url to file _and_ line number
                 threading.append('%s-pr-%s-line-%s@%s' % (pull_request.other_repo.repo_name,
                                                           pull_request.pull_request_id, line_no,
                                                           webutils.canonical_hostname()))
             comment_url = pull_request.url(canonical=True,
                 anchor='comment-%s' % comment.comment_id)
             # get the current participants of this pull request
             recipients = _list_pull_request_commenters(pull_request)
             recipients.append(pull_request.owner)
             recipients += pull_request.get_reviewer_users()
             # set some variables for email notification
             email_kwargs = {
                 'pr_title': pull_request.title,
                 'pr_title_short': shorter(pull_request.title, 50),
+                'pr_title_short': webutils.shorter(pull_request.title, 50),
                 'pr_nice_id': pull_request.nice_id(),
                 'status_change': status_change,
                 'closing_pr': closing_pr,
                 'pr_comment_url': comment_url,
                 'pr_url': pull_request.url(canonical=True),
                 'pr_comment_user': author.full_name_and_username,
                 'pr_target_repo': webutils.canonical_url('summary_home',
                                    repo_name=pull_request.other_repo.repo_name),
                 'pr_target_branch': other_ref_name,
                 'pr_source_repo': webutils.canonical_url('summary_home',
                                    repo_name=pull_request.org_repo.repo_name),
                 'pr_source_branch': org_ref_name,
                 'pr_owner': pull_request.owner,
                 'pr_owner_username': pull_request.owner.username,
                 'repo_name': pull_request.other_repo.repo_name,
                 'threading': threading,
+            }
         email_kwargs['is_mention'] = False
         # create notification objects, and emails
         notification.NotificationModel().create(
             created_by=author, body=comment_text,
             recipients=recipients, type_=notification_type,
             email_kwargs=email_kwargs,
+        )
         mention_recipients = extract_mentioned_users(comment_text).difference(recipients)
         if mention_recipients:
             email_kwargs['is_mention'] = True
             notification.NotificationModel().create(
                 created_by=author, body=comment_text,
                 recipients=mention_recipients,
                 type_=notification_type,
                 email_kwargs=email_kwargs
+            )
     def create(self, text, repo, author, revision=None, pull_request=None,
                f_path=None, line_no=None, status_change=None, closing_pr=False,
                send_email=True):
         """
         Creates a new comment for either a changeset or a pull request.
         status_change and closing_pr is only for the optional email.
         Returns the created comment.
         """
         if not status_change and not text:
             log.warning('Missing text for comment, skipping...')
             return None
         repo = db.Repository.guess_instance(repo)
         author = db.User.guess_instance(author)
         comment = db.ChangesetComment()
         comment.repo = repo
         comment.author = author
         comment.text = text
         comment.f_path = f_path
         comment.line_no = line_no
         if revision is not None:
             comment.revision = revision
         elif pull_request is not None:
             pull_request = db.PullRequest.guess_instance(pull_request)
             comment.pull_request = pull_request
         else:
             raise Exception('Please specify revision or pull_request_id')
         meta.Session().add(comment)
         meta.Session().flush()
         if send_email:
             self.create_notification(
                 repo, comment, author, text, line_no, revision, pull_request,
                 status_change, closing_pr
+            )
         return comment
     def delete(self, comment):
         comment = db.ChangesetComment.guess_instance(comment)
         meta.Session().delete(comment)
         return comment
     def get_comments(self, repo_id, revision=None, pull_request=None):
         """
         Gets general comments for either revision or pull_request.
         Returns a list, ordered by creation date.
         """
         return self._get_comments(repo_id, revision=revision, pull_request=pull_request,
                                   inline=False)
     def get_inline_comments(self, repo_id, revision=None, pull_request=None,
                 f_path=None, line_no=None):
         """

kallithea/model/notification.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.model.notification
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Model for notifications
 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: Nov 20, 2011
 :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
     TYPE_REGISTRATION = 'registration'
     TYPE_PULL_REQUEST = 'pull_request'
     TYPE_PULL_REQUEST_COMMENT = 'pull_request_comment'
     def create(self, created_by, body, recipients=None,
                type_=TYPE_MESSAGE, with_email=True,
                email_kwargs=None, repo_name=None):
         """
         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
         """
         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)
                 else:
                     # TODO: inform user that requested operation couldn't be completed
                     log.error('cannot email unknown user %r', u)
             log.debug('sending notifications %s to %s',
                 type_, recipients_objs
+            )
         elif recipients is None:
             # empty recipients means to all admins
             recipients_objs = db.User.query().filter(db.User.admin == True).all()
             log.debug('sending notifications %s to admins: %s',
                 type_, recipients_objs
+            )
         #else: silently skip notification mails?
         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())
+        created_on = webutils.fmt_date(datetime.datetime.now())
         html_kwargs = {
                   '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)
         email_subject = EmailNotificationModel() \
                             .get_email_description(type_, **txt_kwargs)
         email_txt_body = EmailNotificationModel() \
                             .get_email_tmpl(type_, 'txt', **txt_kwargs)
         email_html_body = EmailNotificationModel() \
                             .get_email_tmpl(type_, 'html', **html_kwargs)
         # don't send email to the person who caused the notification, except for
         # notifications about new pull requests where the author is explicitly
         # added.
         rec_mails = set(obj.email for obj in recipients_objs)
         if type_ == NotificationModel.TYPE_PULL_REQUEST:
             rec_mails.add(created_by_obj.email)
         else:
             rec_mails.discard(created_by_obj.email)
         # send email with notification to participants
         for rec_mail in sorted(rec_mails):
             async_tasks.send_email([rec_mail], email_subject, email_txt_body,
                      email_html_body, headers,
                      from_name=created_by_obj.full_name_or_username)
 class EmailNotificationModel(object):
     TYPE_CHANGESET_COMMENT = NotificationModel.TYPE_CHANGESET_COMMENT
     TYPE_MESSAGE = NotificationModel.TYPE_MESSAGE # only used for testing
     # NotificationModel.TYPE_MENTION is not used
     TYPE_PASSWORD_RESET = 'password_link'
     TYPE_REGISTRATION = NotificationModel.TYPE_REGISTRATION
     TYPE_PULL_REQUEST = NotificationModel.TYPE_PULL_REQUEST
     TYPE_PULL_REQUEST_COMMENT = NotificationModel.TYPE_PULL_REQUEST_COMMENT
     TYPE_DEFAULT = 'default'
     def __init__(self):
         super(EmailNotificationModel, self).__init__()
         self._tmpl_lookup = app_globals.mako_lookup
         self.email_types = {
             self.TYPE_CHANGESET_COMMENT: 'changeset_comment',
             self.TYPE_PASSWORD_RESET: 'password_reset',
             self.TYPE_REGISTRATION: 'registration',
             self.TYPE_DEFAULT: 'default',
             self.TYPE_PULL_REQUEST: 'pull_request',
             self.TYPE_PULL_REQUEST_COMMENT: 'pull_request_comment',
+        }
         self._subj_map = {
             self.TYPE_CHANGESET_COMMENT: _('[Comment] %(repo_name)s changeset %(short_id)s "%(message_short)s" on %(branch)s by %(cs_author_username)s'),
             self.TYPE_MESSAGE: 'Test Message',
             # self.TYPE_PASSWORD_RESET
             self.TYPE_REGISTRATION: _('New user %(new_username)s registered'),
             # self.TYPE_DEFAULT
             self.TYPE_PULL_REQUEST: _('[Review] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),
             self.TYPE_PULL_REQUEST_COMMENT: _('[Comment] %(repo_name)s PR %(pr_nice_id)s "%(pr_title_short)s" from %(pr_source_branch)s by %(pr_owner_username)s'),
+        }
     def get_email_description(self, type_, **kwargs):
         """
         return subject for email based on given type
         """
         tmpl = self._subj_map[type_]
         try:
             subj = tmpl % kwargs
         except KeyError as e:
             log.error('error generating email subject for %r from %s: %s', type_, ', '.join(self._subj_map), e)
             raise
         # gmail doesn't do proper threading but will ignore leading square
         # bracket content ... so that is where we put status info
         bracket_tags = []
         status_change = kwargs.get('status_change')
         if status_change:
             bracket_tags.append(str(status_change))  # apply str to evaluate LazyString before .join
         if kwargs.get('closing_pr'):
             bracket_tags.append(_('Closing'))
         if bracket_tags:
             if subj.startswith('['):
                 subj = '[' + ', '.join(bracket_tags) + ': ' + subj[1:]
             else:
                 subj = '[' + ', '.join(bracket_tags) + '] ' + subj
         return subj
     def get_email_tmpl(self, type_, content_type, **kwargs):
         """

kallithea/model/pull_request.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.model.pull_request
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 pull request model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 6, 2012
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import logging
 import re
 from tg import request
 from tg.i18n import ugettext as _
 from kallithea.lib import auth, hooks, webutils
 from kallithea.lib.utils import extract_mentioned_users
-from kallithea.lib.utils2 import ascii_bytes, short_ref_name, shorter
 from kallithea.lib.utils2 import ascii_bytes, short_ref_name
 from kallithea.model import changeset_status, comment, db, meta, notification
 log = logging.getLogger(__name__)
 def _assert_valid_reviewers(seq):
     """Sanity check: elements are actual User objects, and not the default user."""
     assert not any(user.is_default_user for user in seq)
 class PullRequestModel(object):
     def add_reviewers(self, user, pr, reviewers, mention_recipients=None):
         """Add reviewer and send notification to them.
         """
         reviewers = set(reviewers)
         _assert_valid_reviewers(reviewers)
         if mention_recipients is not None:
             mention_recipients = set(mention_recipients) - reviewers
             _assert_valid_reviewers(mention_recipients)
         redundant_reviewers = set(db.User.query() \
             .join(db.PullRequestReviewer) \
             .filter(db.PullRequestReviewer.pull_request == pr) \
             .filter(db.PullRequestReviewer.user_id.in_(r.user_id for r in reviewers))
             .all())
         if redundant_reviewers:
             log.debug('Following reviewers were already part of pull request %s: %s', pr.pull_request_id, redundant_reviewers)
             reviewers -= redundant_reviewers
         log.debug('Adding reviewers to pull request %s: %s', pr.pull_request_id, reviewers)
         for reviewer in reviewers:
             prr = db.PullRequestReviewer(reviewer, pr)
             meta.Session().add(prr)
         # notification to reviewers
         pr_url = pr.url(canonical=True)
         threading = ['%s-pr-%s@%s' % (pr.other_repo.repo_name,
                                       pr.pull_request_id,
                                       webutils.canonical_hostname())]
         body = pr.description
         _org_ref_type, org_ref_name, _org_rev = pr.org_ref.split(':')
         _other_ref_type, other_ref_name, _other_rev = pr.other_ref.split(':')
         revision_data = [(x.raw_id, x.message)
                          for x in map(pr.org_repo.get_changeset, pr.revisions)]
         email_kwargs = {
             'pr_title': pr.title,
             'pr_title_short': shorter(pr.title, 50),
+            'pr_title_short': webutils.shorter(pr.title, 50),
             'pr_user_created': user.full_name_and_username,
             'pr_repo_url': webutils.canonical_url('summary_home', repo_name=pr.other_repo.repo_name),
             'pr_url': pr_url,
             'pr_revisions': revision_data,
             'repo_name': pr.other_repo.repo_name,
             'org_repo_name': pr.org_repo.repo_name,
             'pr_nice_id': pr.nice_id(),
             'pr_target_repo': webutils.canonical_url('summary_home',
                                repo_name=pr.other_repo.repo_name),
             'pr_target_branch': other_ref_name,
             'pr_source_repo': webutils.canonical_url('summary_home',
                                repo_name=pr.org_repo.repo_name),
             'pr_source_branch': org_ref_name,
             'pr_owner': pr.owner,
             'pr_owner_username': pr.owner.username,
             'pr_username': user.username,
             'threading': threading,
             'is_mention': False,
+            }
         if reviewers:
             notification.NotificationModel().create(created_by=user, body=body,
                                        recipients=reviewers,
                                        type_=notification.NotificationModel.TYPE_PULL_REQUEST,
                                        email_kwargs=email_kwargs)
         if mention_recipients:
             email_kwargs['is_mention'] = True
             notification.NotificationModel().create(created_by=user, body=body,
                                        recipients=mention_recipients,
                                        type_=notification.NotificationModel.TYPE_PULL_REQUEST,
                                        email_kwargs=email_kwargs)
         return reviewers, redundant_reviewers
     def mention_from_description(self, user, pr, old_description=''):
         mention_recipients = (extract_mentioned_users(pr.description) -
                               extract_mentioned_users(old_description))
         log.debug("Mentioning %s", mention_recipients)
         self.add_reviewers(user, pr, set(), mention_recipients)
     def remove_reviewers(self, user, pull_request, reviewers):
         """Remove specified users from being reviewers of the PR."""
         if not reviewers:
             return # avoid SQLAlchemy warning about empty sequence for IN-predicate
         db.PullRequestReviewer.query() \
             .filter_by(pull_request=pull_request) \
             .filter(db.PullRequestReviewer.user_id.in_(r.user_id for r in reviewers)) \
             .delete(synchronize_session='fetch') # the default of 'evaluate' is not available
     def delete(self, pull_request):
         pull_request = db.PullRequest.guess_instance(pull_request)
         meta.Session().delete(pull_request)
         if pull_request.org_repo.scm_instance.alias == 'git':
             # remove a ref under refs/pull/ so that commits can be garbage-collected
             try:
                 del pull_request.org_repo.scm_instance._repo[b"refs/pull/%d/head" % pull_request.pull_request_id]
             except KeyError:
                 pass
     def close_pull_request(self, pull_request):
         pull_request = db.PullRequest.guess_instance(pull_request)
         pull_request.status = db.PullRequest.STATUS_CLOSED
         pull_request.updated_on = datetime.datetime.now()
 class CreatePullRequestAction(object):
     class ValidationError(Exception):
         pass
     class Empty(ValidationError):
         pass
     class AmbiguousAncestor(ValidationError):
         pass
     class Unauthorized(ValidationError):
         pass
     @staticmethod
     def is_user_authorized(org_repo, other_repo):
         """Performs authorization check with only the minimum amount of
         information needed for such a check, rather than a full command
         object.
         """
         if (auth.HasRepoPermissionLevel('read')(org_repo.repo_name) and
             auth.HasRepoPermissionLevel('read')(other_repo.repo_name)
         ):
             return True
         return False
     def __init__(self, org_repo, other_repo, org_ref, other_ref, title, description, owner, reviewers):
         reviewers = set(reviewers)
@@ @@ -252,140 +252,140 @@ class CreatePullRequestAction(object): @@
         pr.other_ref = self.other_ref
         pr.revisions = self.revisions
         pr.title = self.title
         pr.description = self.description
         pr.owner = self.owner
         meta.Session().add(pr)
         meta.Session().flush() # make database assign pull_request_id
         if self.org_repo.scm_instance.alias == 'git':
             # create a ref under refs/pull/ so that commits don't get garbage-collected
             self.org_repo.scm_instance._repo[b"refs/pull/%d/head" % pr.pull_request_id] = ascii_bytes(self.org_rev)
         # reset state to under-review
         new_comment = comment.ChangesetCommentsModel().create(
             text='',
             repo=self.org_repo,
             author=created_by,
             pull_request=pr,
             send_email=False,
             status_change=db.ChangesetStatus.STATUS_UNDER_REVIEW,
+        )
         changeset_status.ChangesetStatusModel().set_status(
             self.org_repo,
             db.ChangesetStatus.STATUS_UNDER_REVIEW,
             created_by,
             new_comment,
             pull_request=pr,
+        )
         mention_recipients = extract_mentioned_users(self.description)
         PullRequestModel().add_reviewers(created_by, pr, self.reviewers, mention_recipients)
         hooks.log_create_pullrequest(pr.get_dict(), created_by)
         return pr
 class CreatePullRequestIterationAction(object):
     @staticmethod
     def is_user_authorized(old_pull_request):
         """Performs authorization check with only the minimum amount of
         information needed for such a check, rather than a full command
         object.
         """
         if auth.HasPermissionAny('hg.admin')():
             return True
         # Authorized to edit the old PR?
         if request.authuser.user_id != old_pull_request.owner_id:
             return False
         # Authorized to create a new PR?
         if not CreatePullRequestAction.is_user_authorized(old_pull_request.org_repo, old_pull_request.other_repo):
             return False
         return True
     def __init__(self, old_pull_request, new_org_rev, new_other_rev, title, description, owner, reviewers):
         self.old_pull_request = old_pull_request
         org_repo = old_pull_request.org_repo
         org_ref_type, org_ref_name, org_rev = old_pull_request.org_ref.split(':')
         other_repo = old_pull_request.other_repo
         other_ref_type, other_ref_name, other_rev = old_pull_request.other_ref.split(':') # other_rev is ancestor
         #assert other_ref_type == 'branch', other_ref_type # TODO: what if not?
         new_org_ref = '%s:%s:%s' % (org_ref_type, org_ref_name, new_org_rev)
         new_other_ref = '%s:%s:%s' % (other_ref_type, other_ref_name, new_other_rev)
         self.create_action = CreatePullRequestAction(org_repo, other_repo, new_org_ref, new_other_ref, None, None, owner, reviewers)
         # Generate complete title/description
         old_revisions = set(old_pull_request.revisions)
         revisions = self.create_action.revisions
         new_revisions = [r for r in revisions if r not in old_revisions]
         lost = old_revisions.difference(revisions)
         infos = ['This is a new iteration of %s "%s".' %
                  (webutils.canonical_url('pullrequest_show', repo_name=old_pull_request.other_repo.repo_name,
                       pull_request_id=old_pull_request.pull_request_id),
                   old_pull_request.title)]
         if lost:
             infos.append(_('Missing changesets since the previous iteration:'))
             for r in old_pull_request.revisions:
                 if r in lost:
                     rev_desc = org_repo.get_changeset(r).message.split('\n')[0]
                     infos.append('  %s %s' % (r[:12], rev_desc))
         if new_revisions:
             infos.append(_('New changesets on %s %s since the previous iteration:') % (org_ref_type, org_ref_name))
             for r in reversed(revisions):
                 if r in new_revisions:
                     rev_desc = org_repo.get_changeset(r).message.split('\n')[0]
                     infos.append('  %s %s' % (r[:12], shorter(rev_desc, 80)))
+                    infos.append('  %s %s' % (r[:12], webutils.shorter(rev_desc, 80)))
             if self.create_action.other_ref == old_pull_request.other_ref:
                 infos.append(_("Ancestor didn't change - diff since previous iteration:"))
                 infos.append(webutils.canonical_url('compare_url',
                                  repo_name=org_repo.repo_name, # other_repo is always same as repo_name
                                  org_ref_type='rev', org_ref_name=org_rev[:12], # use old org_rev as base
                                  other_ref_type='rev', other_ref_name=new_org_rev[:12],
                                  )) # note: linear diff, merge or not doesn't matter
             else:
                 infos.append(_('This iteration is based on another %s revision and there is no simple diff.') % other_ref_name)
         else:
             infos.append(_('No changes found on %s %s since previous iteration.') % (org_ref_type, org_ref_name))
             # TODO: fail?
         v = 2
         m = re.match(r'(.*)\(v(\d+)\)\s*$', title)
         if m is not None:
             title = m.group(1)
             v = int(m.group(2)) + 1
         self.create_action.title = '%s (v%s)' % (title.strip(), v)
         # using a mail-like separator, insert new iteration info in description with latest first
         descriptions = description.replace('\r\n', '\n').split('\n-- \n', 1)
         description = descriptions[0].strip() + '\n\n-- \n' + '\n'.join(infos)
         if len(descriptions) > 1:
             description += '\n\n' + descriptions[1].strip()
         self.create_action.description = description
         if not CreatePullRequestIterationAction.is_user_authorized(self.old_pull_request):
             raise CreatePullRequestAction.Unauthorized(_('You are not authorized to create the pull request'))
     def execute(self):
         pull_request = self.create_action.execute()
         # Close old iteration
         comment.ChangesetCommentsModel().create(
             text=_('Closed, next iteration: %s .') % pull_request.url(canonical=True),
             repo=self.old_pull_request.other_repo_id,
             author=request.authuser.user_id,
             pull_request=self.old_pull_request.pull_request_id,
             closing_pr=True)
         PullRequestModel().close_pull_request(self.old_pull_request.pull_request_id)
         return pull_request

kallithea/tests/other/test_libs.py

➞

Show inline comments

@@ @@ -48,235 +48,235 @@ TEST_URLS = [ @@
      '%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 TestLibs(base.TestController):
     @base.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
     @base.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
     @base.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),
     ])
     def test_asbool(self, str_bool, expected):
         from kallithea.lib.utils2 import asbool
         assert asbool(str_bool) == expected
     def test_mention_extractor(self):
         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(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'),
         (dict(years= -1, months= -10), '1 year and 10 months ago'),
         (dict(years= -2, months= -4), '2 years and 4 months ago'),
         (dict(years= -2, months= -11), '2 years and 11 months ago'),
         (dict(years= -3, months= -2), '3 years and 2 months ago'),
     ])
     def test_age(self, age_args, expected):
-        from kallithea.lib.utils2 import age
+        from kallithea.lib.webutils import age
         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
     @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 ago'),
         (dict(months= -1, days= -20), '1 month ago'),
         (dict(years= -1, months= -1), '13 months ago'),
         (dict(years= -1, months= -10), '22 months ago'),
         (dict(years= -2, months= -4), '2 years ago'),
         (dict(years= -2, months= -11), '3 years ago'),
         (dict(years= -3, months= -2), '3 years ago'),
         (dict(years= -4, months= -8), '5 years ago'),
     ])
     def test_age_short(self, age_args, expected):
-        from kallithea.lib.utils2 import age
+        from kallithea.lib.webutils import age
         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
     @base.parametrize('age_args,expected', [
         (dict(), 'just now'),
         (dict(seconds=1), 'in 1 second'),
         (dict(seconds=60 * 2), 'in 2 minutes'),
         (dict(hours=1), 'in 1 hour'),
         (dict(hours=24), 'in 1 day'),
         (dict(hours=24 * 5), 'in 5 days'),
         (dict(months=1), 'in 1 month'),
         (dict(months=1, days=1), 'in 1 month and 1 day'),
         (dict(years=1, months=1), 'in 1 year and 1 month')
     ])
     def test_age_in_future(self, age_args, expected):
-        from kallithea.lib.utils2 import age
+        from kallithea.lib.webutils import age
         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 = 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):
             _c = AttributeDict()
             _c.visual = AttributeDict()
             _c.visual.use_gravatar = True
             _c.visual.gravatar_url = _url
             return _c
         with mock.patch('kallithea.lib.webutils.url.current', lambda *a, **b: 'https://example.com'):
             fake = fake_tmpl_context(_url='http://example.com/{email}')
             with mock.patch('kallithea.lib.helpers.c', fake):
                     assert webutils.url.current() == 'https://example.com'
                     grav = h.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('kallithea.lib.helpers.c', fake):
                 grav = h.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('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.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('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.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('kallithea.lib.helpers.c', fake):
                 em = 'test@example.com'
                 grav = h.gravatar_url(email_address=em, size=24)
                 assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)
     @base.parametrize('clone_uri_tmpl,repo_name,username,prefix,expected', [
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', None, '', 'http://vps1:8000/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '', 'http://username@vps1:8000/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', None, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'user', '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'user', '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
         (db.Repository.DEFAULT_CLONE_URI, 'group/repo1', 'username', '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', None, '', 'http://vps1:8000/_23'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://username@vps1:8000/_23'),
         ('http://{user}@{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://username@vps1:8000/_23'),
         ('http://{netloc}/_{repoid}', 'group/repo1', 'username', '', 'http://vps1:8000/_23'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', 'username', '', 'https://username@proxy1.example.com/group/repo1'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', None, '', 'https://proxy1.example.com/group/repo1'),
         ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', 'username', '', 'https://proxy1.example.com/username/group/repo1'),
     ])
     def test_clone_url_generator(self, clone_uri_tmpl, repo_name, username, prefix, expected):
         clone_url = get_clone_url(clone_uri_tmpl=clone_uri_tmpl, prefix_url='http://vps1:8000' + prefix,
                                   repo_name=repo_name, repo_id=23, username=username)
         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:
         """
         # 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)

0 comments (0 inline, 0 general)