kallithea Changeset - ba5fee3879c8

Changeset - ba5fee3879c8

Parent rev.

Child rev.

[Not reviewed]

default

0 4 0

Søren Løvborg - 10 years ago 2016-04-06 14:50:47
sorenl@unity3d.com

utils: improve extract_mentioned_users usefulness

Previously, extract_mentioned_users performed a regex search, converted
to a set to remove duplicates, converted that set back into a list, and
sorted the list on username, ignoring case. Every single caller of the
function then promptly took the carefully pruned and sorted result and
put it back into a set, rendering the entire exercise pointless.

In addition, every caller also resolved the usernames to database User
objects.

In this changeset, extract_mentioned_users is changed to return a set
of database User objects directly. A new extract_mentioned_usernames
function, which does only username extraction, is kept for the sole
purpose of providing a more testable interface (no database needed).

Bonus feature: The new extract_mentioned_users will prune non-existent
users, as well as the DEFAULT user. This means it is no longer possible
to @mention (and send notifications to) the DEFAULT user.

4 files changed with 23 insertions and 31 deletions:

kallithea/lib/utils2.py

kallithea/model/comment.py

kallithea/model/pull_request.py

kallithea/tests/other/test_libs.py

0 comments (0 inline, 0 general)

kallithea/lib/utils2.py

➞

Show inline comments

@@ @@ -383,397 +383,402 @@ def age(prevdate, show_short_version=Fal @@
     if prevdate > now:
         now, prevdate = prevdate, now
         future = True
     if future:
         prevdate = prevdate.replace(microsecond=0)
     # Get date parts deltas
     from dateutil import relativedelta
     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(u'%d year', '%d years', d) % d,
         'month': lambda d: ungettext(u'%d month', '%d months', d) % d,
         'day': lambda d: ungettext(u'%d day', '%d days', d) % d,
         'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,
         'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,
         'second': lambda d: ungettext(u'%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 uri_filter(uri):
     """
     Removes user:password from given url string
     :param uri:
     :rtype: unicode
     :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 filter(None, [proto, host, port])
 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(uri_tmpl, qualified_home_url, repo_name, repo_id, **override):
     parsed_url = urlobject.URLObject(qualified_home_url)
     decoded_path = safe_unicode(urllib.unquote(parsed_url.path.rstrip('/')))
     args = {
         'scheme': parsed_url.scheme,
         'user': '',
         'netloc': parsed_url.netloc+decoded_path,  # path if we use proxy-prefix
         'prefix': decoded_path,
         'repo': repo_name,
         'repoid': str(repo_id)
+    }
     args.update(override)
     args['user'] = urllib.quote(safe_str(args['user']))
     for k, v in args.items():
         uri_tmpl = uri_tmpl.replace('{%s}' % k, v)
     # remove leading @ sign if it's present. Case of empty user
     url_obj = urlobject.URLObject(uri_tmpl)
     url = url_obj.with_netloc(url_obj.netloc.lstrip('@'))
     return safe_unicode(url)
 def get_changeset_safe(repo, rev):
     """
     Safe version of get_changeset if this changeset doesn't exists for a
     repo it returns a Dummy one instead
     :param repo:
     :param rev:
     """
     from kallithea.lib.vcs.backends.base import BaseRepository
     from kallithea.lib.vcs.exceptions import RepositoryError
     from kallithea.lib.vcs.backends.base import EmptyChangeset
     if not isinstance(repo, BaseRepository):
         raise Exception('You must pass an Repository '
                         'object as first argument got %s', type(repo))
     try:
         cs = repo.get_changeset(rev)
     except (RepositoryError, LookupError):
         cs = EmptyChangeset(requested_revision=rev)
     return cs
 def datetime_to_time(dt):
     if dt:
         return time.mktime(dt.timetuple())
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, basestring):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
 # Must match regexp in kallithea/public/js/base.js MentionsAutoComplete()
 # Check char before @ - it must not look like we are in an email addresses.
 # 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_users(s):
+def extract_mentioned_usernames(text):
     r"""
-    Returns unique usernames from given string s that have @mention
+    Returns list of (possible) usernames @mentioned in given text.
     :param s: string to get mentions
     >>> extract_mentioned_users('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
     >>> 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', 'hh', 'zz']
     """
     usrs = set()
     for username in MENTIONS_REGEX.findall(s):
         usrs.add(username)
     return MENTIONS_REGEX.findall(text)
     return sorted(list(usrs), key=lambda k: k.lower())
 def extract_mentioned_users(text):
     """ Returns set of actual database Users @mentioned in given text. """
     from kallithea.model.db import User
     result = set()
     for name in extract_mentioned_usernames(text):
         user = User.get_by_username(name, case_insensitive=True)
         if user is not None and user.username != User.DEFAULT_USER:
             result.add(user)
     return result
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__
 def fix_PATH(os_=None):
     """
     Get current active python path, and append it to PATH variable to fix issues
     of subprocess calls and different python versions
     """
     if os_ is None:
         import os
     else:
         os = os_
     cur_path = os.path.split(sys.executable)[0]
     if not os.environ['PATH'].startswith(cur_path):
         os.environ['PATH'] = '%s:%s' % (cur_path, os.environ['PATH'])
 def obfuscate_url_pw(engine):
     from sqlalchemy.engine import url as sa_url
     from sqlalchemy.exc import ArgumentError
     try:
         _url = sa_url.make_url(engine or '')
     except ArgumentError:
         return engine
     if _url.password:
         _url.password = 'XXXXX'
     return str(_url)
 def get_server_url(environ):
     req = webob.Request(environ)
     return req.host_url + req.script_name
 def _extract_extras(env=None):
     """
     Extracts the Kallithea extras data from os.environ, and wraps it into named
     AttributeDict object
     """
     if not env:
         env = os.environ
     try:
         extras = json.loads(env['KALLITHEA_EXTRAS'])
     except KeyError:
         extras = {}
     try:
         for k in ['username', 'repository', 'locked_by', 'scm', 'make_lock',
                   'action', 'ip']:
             extras[k]
     except KeyError as e:
         raise Exception('Missing key %s in os.environ %s' % (e, extras))
     return AttributeDict(extras)
 def _set_extras(extras):
     # RC_SCM_DATA can probably be removed in the future, but for compatibility now...
     os.environ['KALLITHEA_EXTRAS'] = os.environ['RC_SCM_DATA'] = json.dumps(extras)
 def unique_id(hexlen=32):
     alphabet = "23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz"
     return suuid(truncate_to=hexlen, alphabet=alphabet)
 def suuid(url=None, truncate_to=22, alphabet=None):
     """
     Generate and return a short URL safe UUID.
     If the url parameter is provided, set the namespace to the provided
     URL and generate a UUID.
     :param url to get the uuid for
     :truncate_to: truncate the basic 22 UUID to shorter version
     The IDs won't be universally unique any longer, but the probability of
     a collision will still be very low.
     """
     # Define our alphabet.
     _ALPHABET = alphabet or "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
     # If no URL is given, generate a random UUID.
     if url is None:
         unique_id = uuid.uuid4().int
     else:
         unique_id = uuid.uuid3(uuid.NAMESPACE_URL, url).int
     alphabet_length = len(_ALPHABET)
     output = []
     while unique_id > 0:
         digit = unique_id % alphabet_length
         output.append(_ALPHABET[digit])
         unique_id = int(unique_id / alphabet_length)
     return "".join(output)[:truncate_to]
 def get_current_authuser():
     """
     Gets kallithea user from threadlocal tmpl_context variable if it's
     defined, else returns None.
     """
     from pylons import tmpl_context
     if hasattr(tmpl_context, 'authuser'):
         return tmpl_context.authuser
     return None
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute. Example::
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid)
             # calls
     """
     def __init__(self, attr_name):
         self.attr_name = attr_name
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
     def __call__(self):
         return self
 #alias
 OAttr = OptionalAttr
 class Optional(object):
     """
     Defines an optional parameter::
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
     is equivalent of::
         param = Optional.extract(param)
     """
     def __init__(self, type_):
         self.type_ = type_
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
     def __call__(self):
         return self.getval()
     def getval(self):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             # use params name
             return self.type_.attr_name
         return self.type_
     @classmethod
     def extract(cls, val):
         """
         Extracts value from Optional() instance
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             return val.getval()
         return val
 def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):
     return _cleanstringsub('_', safe_str(s)).rstrip('_')

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 pylons.i18n.translation import _
 from collections import defaultdict
 from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode
 from kallithea.lib import helpers as h
 from kallithea.model import BaseModel
 from kallithea.model.db import ChangesetComment, User, \
     Notification, PullRequest
 from kallithea.model.notification import NotificationModel
 from kallithea.model.meta import Session
 log = logging.getLogger(__name__)
 class ChangesetCommentsModel(BaseModel):
     cls = ChangesetComment
     def __get_changeset_comment(self, changeset_comment):
         return self._get_instance(ChangesetComment, changeset_comment)
     def __get_pull_request(self, pull_request):
         return self._get_instance(PullRequest, pull_request)
     def _extract_mentions(self, s):
         user_objects = []
         for username in extract_mentioned_users(s):
             user_obj = User.get_by_username(username, case_insensitive=True)
             if user_obj:
                 user_objects.append(user_obj)
         return user_objects
     def _get_notification_data(self, repo, comment, user, comment_text,
                                line_no=None, revision=None, pull_request=None,
                                status_change=None, closing_pr=False):
         """
         :returns: tuple (subj,body,recipients,notification_type,email_kwargs)
         """
         # make notification
         body = comment_text  # text of the comment
         line = ''
         if line_no:
             line = _('on line %s') % line_no
         #changeset
         if revision:
             notification_type = Notification.TYPE_CHANGESET_COMMENT
             cs = repo.scm_instance.get_changeset(revision)
             desc = cs.short_id
             threading = ['%s-rev-%s@%s' % (repo.repo_name, revision, h.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,
                                                            h.canonical_hostname()))
             comment_url = h.canonical_url('changeset_home',
                 repo_name=repo.repo_name,
                 revision=revision,
                 anchor='comment-%s' % comment.comment_id)
             subj = safe_unicode(
                 h.link_to('Re changeset: %(desc)s %(line)s' % \
                           {'desc': desc, 'line': line},
                           comment_url)
+            )
             # get the current participants of this changeset
             recipients = ChangesetComment.get_users(revision=revision)
             # add changeset author if it's known locally
             cs_author = User.get_from_cs_author(cs.author)
             if not cs_author:
                 #use repo owner if we cannot extract the author correctly
                 cs_author = repo.user
             recipients += [cs_author]
             email_kwargs = {
                 'status_change': status_change,
                 'cs_comment_user': user.full_name_and_username,
                 'cs_target_repo': h.canonical_url('summary_home', repo_name=repo.repo_name),
                 'cs_comment_url': comment_url,
                 'raw_id': revision,
                 'message': cs.message,
                 'repo_name': repo.repo_name,
                 'short_id': h.short_id(revision),
                 'branch': cs.branch,
                 'comment_username': user.username,
                 'threading': threading,
+            }
         #pull request
         elif pull_request:
             notification_type = Notification.TYPE_PULL_REQUEST_COMMENT
             desc = comment.pull_request.title
             _org_ref_type, org_ref_name, _org_rev = comment.pull_request.org_ref.split(':')
             threading = ['%s-pr-%s@%s' % (pull_request.other_repo.repo_name,
                                           pull_request.pull_request_id,
                                           h.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,
                                                           h.canonical_hostname()))
             comment_url = pull_request.url(canonical=True,
                 anchor='comment-%s' % comment.comment_id)
             subj = safe_unicode(
                 h.link_to('Re pull request %(pr_nice_id)s: %(desc)s %(line)s' % \
                           {'desc': desc,
                            'pr_nice_id': comment.pull_request.nice_id(),
                            'line': line},
                           comment_url)
+            )
             # get the current participants of this pull request
             recipients = ChangesetComment.get_users(pull_request_id=
                                                 pull_request.pull_request_id)
             # add pull request author
             recipients += [pull_request.owner]
             # add the reviewers to notification
             recipients += pull_request.get_reviewer_users()
             #set some variables for email notification
             email_kwargs = {
                 'pr_title': pull_request.title,
                 'pr_nice_id': pull_request.nice_id(),
                 'status_change': status_change,
                 'closing_pr': closing_pr,
                 'pr_comment_url': comment_url,
                 'pr_comment_user': user.full_name_and_username,
                 'pr_target_repo': h.canonical_url('summary_home',
                                    repo_name=pull_request.other_repo.repo_name),
                 'repo_name': pull_request.other_repo.repo_name,
                 'ref': org_ref_name,
                 'comment_username': user.username,
                 'threading': threading,
+            }
         return subj, body, recipients, notification_type, email_kwargs
     def create(self, text, repo, user, 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 = self._get_repo(repo)
         user = self._get_user(user)
         comment = ChangesetComment()
         comment.repo = repo
         comment.author = user
         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 = self.__get_pull_request(pull_request)
             comment.pull_request = pull_request
         else:
             raise Exception('Please specify revision or pull_request_id')
         Session().add(comment)
         Session().flush()
         if send_email:
             (subj, body, recipients, notification_type,
              email_kwargs) = self._get_notification_data(
                                 repo, comment, user,
                                 comment_text=text,
                                 line_no=line_no,
                                 revision=revision,
                                 pull_request=pull_request,
                                 status_change=status_change,
                                 closing_pr=closing_pr)
             email_kwargs['is_mention'] = False
             # create notification objects, and emails
             NotificationModel().create(
                 created_by=user, subject=subj, body=body,
                 recipients=recipients, type_=notification_type,
                 email_kwargs=email_kwargs,
+            )
             mention_recipients = set(self._extract_mentions(body)) \
                                     .difference(recipients)
             mention_recipients = extract_mentioned_users(body).difference(recipients)
             if mention_recipients:
                 email_kwargs['is_mention'] = True
                 subj = _('[Mention]') + ' ' + subj
                 NotificationModel().create(
                     created_by=user, subject=subj, body=body,
                     recipients=mention_recipients,
                     type_=notification_type,
                     email_kwargs=email_kwargs
+                )
         return comment
     def delete(self, comment):
         comment = self.__get_changeset_comment(comment)
         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):
         """
         Gets inline comments for either revision or pull_request.
         Returns a list of tuples with file path and list of comments per line number.
         """
         comments = self._get_comments(repo_id, revision=revision, pull_request=pull_request,
                                       inline=True)
         paths = defaultdict(lambda: defaultdict(list))
         for co in comments:
             paths[co.f_path][co.line_no].append(co)
         return paths.items()
     def _get_comments(self, repo_id, revision=None, pull_request=None, inline=False):
         """
         Gets comments for either revision or pull_request_id, either inline or general.
         """
         q = Session().query(ChangesetComment)
         if inline:
             q = q.filter(ChangesetComment.line_no != None) \
                 .filter(ChangesetComment.f_path != None)
         else:
             q = q.filter(ChangesetComment.line_no == None) \
                 .filter(ChangesetComment.f_path == None)
         if revision is not None:
             q = q.filter(ChangesetComment.revision == revision) \
                 .filter(ChangesetComment.repo_id == repo_id)
         elif pull_request is not None:
             pull_request = self.__get_pull_request(pull_request)
             q = q.filter(ChangesetComment.pull_request == pull_request)
         else:
             raise Exception('Please specify either revision or pull_request')
         return q.order_by(ChangesetComment.created_on).all()

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 logging
 import datetime
 from pylons.i18n.translation import _
 from kallithea.model.meta import Session
 from kallithea.lib import helpers as h
 from kallithea.lib.exceptions import UserInvalidException
 from kallithea.model import BaseModel
 from kallithea.model.db import PullRequest, PullRequestReviewers, Notification, \
-    ChangesetStatus, User
     ChangesetStatus
 from kallithea.model.notification import NotificationModel
 from kallithea.lib.utils2 import extract_mentioned_users, safe_unicode
 log = logging.getLogger(__name__)
 class PullRequestModel(BaseModel):
     cls = PullRequest
     def __get_pull_request(self, pull_request):
         return self._get_instance(PullRequest, pull_request)
     def get_pullrequest_cnt_for_user(self, user):
         return PullRequest.query() \
                                 .join(PullRequestReviewers) \
                                 .filter(PullRequestReviewers.user_id == user) \
                                 .filter(PullRequest.status != PullRequest.STATUS_CLOSED) \
                                 .count()
     def get_all(self, repo_name, from_=False, closed=False):
         """Get all PRs for repo.
         Default is all PRs to the repo, PRs from the repo if from_.
         Closed PRs are only included if closed is true."""
         repo = self._get_repo(repo_name)
         q = PullRequest.query()
         if from_:
             q = q.filter(PullRequest.org_repo == repo)
         else:
             q = q.filter(PullRequest.other_repo == repo)
         if not closed:
             q = q.filter(PullRequest.status != PullRequest.STATUS_CLOSED)
         return q.order_by(PullRequest.created_on.desc()).all()
     def create(self, created_by, org_repo, org_ref, other_repo, other_ref,
                revisions, reviewers, title, description=None):
         from kallithea.model.changeset_status import ChangesetStatusModel
         created_by_user = self._get_user(created_by)
         org_repo = self._get_repo(org_repo)
         other_repo = self._get_repo(other_repo)
         new = PullRequest()
         new.org_repo = org_repo
         new.org_ref = org_ref
         new.other_repo = other_repo
         new.other_ref = other_ref
         new.revisions = revisions
         new.title = title
         new.description = description
         new.owner = created_by_user
         Session().add(new)
         Session().flush()
         #reset state to under-review
         from kallithea.model.comment import ChangesetCommentsModel
         comment = ChangesetCommentsModel().create(
             text=u'',
             repo=org_repo,
             user=new.owner,
             pull_request=new,
             send_email=False,
             status_change=ChangesetStatus.STATUS_UNDER_REVIEW,
+        )
         ChangesetStatusModel().set_status(
             org_repo,
             ChangesetStatus.STATUS_UNDER_REVIEW,
             new.owner,
             comment,
             pull_request=new
+        )
         mention_recipients = set(User.get_by_username(username, case_insensitive=True)
                                  for username in extract_mentioned_users(new.description))
         mention_recipients = extract_mentioned_users(new.description)
         self.__add_reviewers(created_by_user, new, reviewers, mention_recipients)
         return new
     def __add_reviewers(self, user, pr, reviewers, mention_recipients=None):
         #members
         for member in set(reviewers):
             _usr = self._get_user(member)
             if _usr is None:
                 raise UserInvalidException(member)
             reviewer = PullRequestReviewers(_usr, pr)
             Session().add(reviewer)
         revision_data = [(x.raw_id, x.message)
                          for x in map(pr.org_repo.get_changeset, pr.revisions)]
         #notification to reviewers
         pr_url = pr.url(canonical=True)
         threading = ['%s-pr-%s@%s' % (pr.other_repo.repo_name,
                                       pr.pull_request_id,
                                       h.canonical_hostname())]
         subject = safe_unicode(
             h.link_to(
               _('%(user)s wants you to review pull request %(pr_nice_id)s: %(pr_title)s') % \
                 {'user': user.username,
                  'pr_title': pr.title,
                  'pr_nice_id': pr.nice_id()},
                 pr_url)
+            )
         body = pr.description
         _org_ref_type, org_ref_name, _org_rev = pr.org_ref.split(':')
         email_kwargs = {
             'pr_title': pr.title,
             'pr_user_created': user.full_name_and_username,
             'pr_repo_url': h.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(),
             'ref': org_ref_name,
             'pr_username': user.username,
             'threading': threading,
             'is_mention': False,
+            }
         if reviewers:
             NotificationModel().create(created_by=user, subject=subject, body=body,
                                        recipients=reviewers,
                                        type_=Notification.TYPE_PULL_REQUEST,
                                        email_kwargs=email_kwargs)
         if mention_recipients:
             mention_recipients.discard(None)
             mention_recipients.difference_update(reviewers)
         if mention_recipients:
             email_kwargs['is_mention'] = True
             subject = _('[Mention]') + ' ' + subject
             NotificationModel().create(created_by=user, subject=subject, body=body,
                                        recipients=mention_recipients,
                                        type_=Notification.TYPE_PULL_REQUEST,
                                        email_kwargs=email_kwargs)
     def mention_from_description(self, user, pr, old_description=''):
         mention_recipients = set(User.get_by_username(username, case_insensitive=True)
                                  for username in extract_mentioned_users(pr.description))
         mention_recipients.difference_update(User.get_by_username(username, case_insensitive=True)
                                              for username in extract_mentioned_users(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, [], mention_recipients)
     def update_reviewers(self, user, pull_request, reviewers_ids):
         reviewers_ids = set(reviewers_ids)
         pull_request = self.__get_pull_request(pull_request)
         current_reviewers = PullRequestReviewers.query() \
                             .filter(PullRequestReviewers.pull_request==
                                    pull_request) \
                             .all()
         current_reviewers_ids = set([x.user.user_id for x in current_reviewers])
         to_add = reviewers_ids.difference(current_reviewers_ids)
         to_remove = current_reviewers_ids.difference(reviewers_ids)
         log.debug("Adding %s reviewers", to_add)
         self.__add_reviewers(user, pull_request, to_add)
         log.debug("Removing %s reviewers", to_remove)
         for uid in to_remove:
             reviewer = PullRequestReviewers.query() \
                     .filter(PullRequestReviewers.user_id==uid,
                             PullRequestReviewers.pull_request==pull_request) \
                     .scalar()
             if reviewer:
                 Session().delete(reviewer)
     def delete(self, pull_request):
         pull_request = self.__get_pull_request(pull_request)
         Session().delete(pull_request)
     def close_pull_request(self, pull_request):
         pull_request = self.__get_pull_request(pull_request)
         pull_request.status = PullRequest.STATUS_CLOSED
         pull_request.updated_on = datetime.datetime.now()
         Session().add(pull_request)

kallithea/tests/other/test_libs.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.tests.other.test_libs
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 Package for testing various lib/helper functions in kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 9, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import datetime
 import hashlib
 import mock
 from kallithea.tests import *
 from kallithea.lib.utils2 import AttributeDict
 from kallithea.model.db import Repository
 proto = 'http'
 TEST_URLS = [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
      '%s://example.com' % proto),
     ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
                                                 '8080'],
      '%s://example.com:8080' % proto),
+]
 proto = 'https'
 TEST_URLS += [
     ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://username:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],
      '%s://127.0.0.1' % proto),
     ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],
      '%s://127.0.0.1:8080' % proto),
     ('%s://example.com' % proto, ['%s://' % proto, 'example.com'],
      '%s://example.com' % proto),
     ('%s://user:pass@example.com:8080' % proto, ['%s://' % proto, 'example.com',
                                                 '8080'],
      '%s://example.com:8080' % proto),
+]
 class TestLibs(BaseTestCase):
     @parameterized.expand(TEST_URLS)
     def test_uri_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import uri_filter
         self.assertEqual(uri_filter(test_url), expected)
     @parameterized.expand(TEST_URLS)
     def test_credentials_filter(self, test_url, expected, expected_creds):
         from kallithea.lib.utils2 import credentials_filter
         self.assertEqual(credentials_filter(test_url), expected_creds)
     @parameterized.expand([('t', True),
                            ('true', True),
                            ('y', True),
                            ('yes', True),
                            ('on', True),
                            ('1', True),
                            ('Y', True),
                            ('yeS', True),
                            ('Y', True),
                            ('TRUE', True),
                            ('T', True),
                            ('False', False),
                            ('F', False),
                            ('FALSE', False),
                            ('0', False),
                            ('-1', False),
                            ('', False)
     ])
     def test_str2bool(self, str_bool, expected):
         from kallithea.lib.utils2 import str2bool
         self.assertEqual(str2bool(str_bool), expected)
     def test_mention_extractor(self):
-        from kallithea.lib.utils2 import extract_mentioned_users
+        from kallithea.lib.utils2 import extract_mentioned_usernames
         sample = (
             "@first hi there @world here's my email username@example.com "
             "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "
             "@UPPER    @cAmEL @2one_more22 @john please see this http://org.pl "
             "@marian.user just do it @marco-polo and next extract @marco_polo "
             "user.dot  hej ! not-needed maril@example.com"
+        )
-        s = sorted([
+        expected = set([
             '2one_more22', 'first', 'lukaszb', 'one', 'one_more22', 'UPPER', 'cAmEL', 'john',
             'marian.user', 'marco-polo', 'marco_polo', 'world'], key=lambda k: k.lower())
         self.assertEqual(s, extract_mentioned_users(sample))
             'marian.user', 'marco-polo', 'marco_polo', 'world'])
         self.assertEqual(expected, set(extract_mentioned_usernames(sample)))
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds= -1), u'1 second ago'),
         (dict(seconds= -60 * 2), u'2 minutes ago'),
         (dict(hours= -1), u'1 hour ago'),
         (dict(hours= -24), u'1 day ago'),
         (dict(hours= -24 * 5), u'5 days ago'),
         (dict(months= -1), u'1 month ago'),
         (dict(months= -1, days= -2), u'1 month and 2 days ago'),
         (dict(months= -1, days= -20), u'1 month and 19 days ago'),
         (dict(years= -1, months= -1), u'1 year and 1 month ago'),
         (dict(years= -1, months= -10), u'1 year and 10 months ago'),
         (dict(years= -2, months= -4), u'2 years and 4 months ago'),
         (dict(years= -2, months= -11), u'2 years and 11 months ago'),
         (dict(years= -3, months= -2), u'3 years and 2 months ago'),
     ])
     def test_age(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), now=n), expected)
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds= -1), u'1 second ago'),
         (dict(seconds= -60 * 2), u'2 minutes ago'),
         (dict(hours= -1), u'1 hour ago'),
         (dict(hours= -24), u'1 day ago'),
         (dict(hours= -24 * 5), u'5 days ago'),
         (dict(months= -1), u'1 month ago'),
         (dict(months= -1, days= -2), u'1 month ago'),
         (dict(months= -1, days= -20), u'1 month ago'),
         (dict(years= -1, months= -1), u'13 months ago'),
         (dict(years= -1, months= -10), u'22 months ago'),
         (dict(years= -2, months= -4), u'2 years ago'),
         (dict(years= -2, months= -11), u'3 years ago'),
         (dict(years= -3, months= -2), u'3 years ago'),
         (dict(years= -4, months= -8), u'5 years ago'),
     ])
     def test_age_short(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), show_short_version=True, now=n), expected)
     @parameterized.expand([
         (dict(), u'just now'),
         (dict(seconds=1), u'in 1 second'),
         (dict(seconds=60 * 2), u'in 2 minutes'),
         (dict(hours=1), u'in 1 hour'),
         (dict(hours=24), u'in 1 day'),
         (dict(hours=24 * 5), u'in 5 days'),
         (dict(months=1), u'in 1 month'),
         (dict(months=1, days=1), u'in 1 month and 1 day'),
         (dict(years=1, months=1), u'in 1 year and 1 month')
     ])
     def test_age_in_future(self, age_args, expected):
         from kallithea.lib.utils2 import age
         from dateutil import relativedelta
         n = datetime.datetime(year=2012, month=5, day=17)
         delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)
         self.assertEqual(age(n + delt(**age_args), now=n), expected)
     def test_tag_extractor(self):
         sample = (
             "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"
             "[requires] [stale] [see<>=>] [see => http://example.com]"
             "[requires => url] [lang => python] [just a tag]"
             "[,d] [ => ULR ] [obsolete] [desc]]"
+        )
         from kallithea.lib.helpers import urlify_text
         res = urlify_text(sample, stylize=True)
         self.assertIn('<div class="metatag" tag="tag">tag</div>', res)
         self.assertIn('<div class="metatag" tag="obsolete">obsolete</div>', res)
         self.assertIn('<div class="metatag" tag="stale">stale</div>', res)
         self.assertIn('<div class="metatag" tag="lang">python</div>', res)
         self.assertIn('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>', res)
         self.assertIn('<div class="metatag" tag="tag">tag</div>', res)
     def test_alternative_gravatar(self):
         from kallithea.lib.helpers import gravatar_url
         _md5 = lambda s: hashlib.md5(s).hexdigest()
         #mock pylons.url
         class fake_url(object):
             @classmethod
             def current(cls, *args, **kwargs):
                 return 'https://example.com'
         #mock pylons.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('pylons.url', fake_url):
             fake = fake_tmpl_context(_url='http://example.com/{email}')
             with mock.patch('pylons.tmpl_context', fake):
                     from pylons import url
                     assert url.current() == 'https://example.com'
                     grav = gravatar_url(email_address='test@example.com', size=24)
                     assert grav == 'http://example.com/test@example.com'
             fake = fake_tmpl_context(_url='http://example.com/{email}')
             with mock.patch('pylons.tmpl_context', fake):
                 grav = gravatar_url(email_address='test@example.com', size=24)
                 assert grav == 'http://example.com/test@example.com'
             fake = fake_tmpl_context(_url='http://example.com/{md5email}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@example.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'http://example.com/%s' % (_md5(em))
             fake = fake_tmpl_context(_url='http://example.com/{md5email}/{size}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@example.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'http://example.com/%s/%s' % (_md5(em), 24)
             fake = fake_tmpl_context(_url='{scheme}://{netloc}/{md5email}/{size}')
             with mock.patch('pylons.tmpl_context', fake):
                 em = 'test@example.com'
                 grav = gravatar_url(email_address=em, size=24)
                 assert grav == 'https://example.com/%s/%s' % (_md5(em), 24)
     @parameterized.expand([
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '', 'http://vps1:8000/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {}, '/prefix', 'http://vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix', 'http://user@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix', 'http://username@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'user'}, '/prefix/', 'http://user@vps1:8000/prefix/group/repo1'),
         (Repository.DEFAULT_CLONE_URI, 'group/repo1', {'user': 'username'}, '/prefix/', 'http://username@vps1:8000/prefix/group/repo1'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {}, '', 'http://vps1:8000/_23'),
         ('{scheme}://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
         ('http://{user}@{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://username@vps1:8000/_23'),
         ('http://{netloc}/_{repoid}', 'group/repo1', {'user': 'username'}, '', 'http://vps1:8000/_23'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://username@proxy1.example.com/group/repo1'),
         ('https://{user}@proxy1.example.com/{repo}', 'group/repo1', {}, '', 'https://proxy1.example.com/group/repo1'),
         ('https://proxy1.example.com/{user}/{repo}', 'group/repo1', {'user': 'username'}, '', 'https://proxy1.example.com/username/group/repo1'),
     ])
     def test_clone_url_generator(self, tmpl, repo_name, overrides, prefix, expected):
         from kallithea.lib.utils2 import get_clone_url
         clone_url = get_clone_url(uri_tmpl=tmpl, qualified_home_url='http://vps1:8000'+prefix,
                                   repo_name=repo_name, repo_id=23, **overrides)
         self.assertEqual(clone_url, expected)
     def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):
         """
         Changes `some text url[foo]` => `some text <a href="/">foo</a>
         :param text:
         """
         import re
         # quickly change expected url[] into a link
         URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')
         def url_func(match_obj):
             _url = match_obj.groups()[0]
             return tmpl % (url_ or '/some-url', _url)
         return URL_PAT.sub(url_func, text)
     @parameterized.expand([
       ("",
        ""),
       ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",
        "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),
       ("from rev 000000000000",
        "from rev url[000000000000]"),
       ("from rev 000000000000123123 also rev 000000000000",
        "from rev url[000000000000123123] also rev url[000000000000]"),
       ("this should-000 00",
        "this should-000 00"),
       ("longtextffffffffff rev 123123123123",
        "longtextffffffffff rev url[123123123123]"),
       ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",
        "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),
       ("ffffffffffff some text traalaa",
        "url[ffffffffffff] some text traalaa"),
        ("""Multi line
        123123123123
        some text 123123123123
        sometimes !
        """,

0 comments (0 inline, 0 general)